diff --git a/.github/workflows/insights-testbed.yml b/.github/workflows/insights-testbed.yml
index d9f4389de0..223e1f4e30 100644
--- a/.github/workflows/insights-testbed.yml
+++ b/.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()
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py
index f8a9e7cde8..09b3bd9e33 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py
+++ b/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:
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py b/plugins/nemo-insights/src/nemo_insights_plugin/cli.py
index 12826e8d4c..918002c4f1 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py
+++ b/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,
         )
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/jobs/analyze.py b/plugins/nemo-insights/src/nemo_insights_plugin/jobs/analyze.py
index 8a39ba3d6f..bc93fc59e6 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/jobs/analyze.py
+++ b/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,
                 )
diff --git a/plugins/nemo-insights/testbed/README.md b/plugins/nemo-insights/testbed/README.md
index 3f72ee7618..a547fbc625 100644
--- a/plugins/nemo-insights/testbed/README.md
+++ b/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;
diff --git a/plugins/nemo-insights/testbed/adapters.py b/plugins/nemo-insights/testbed/adapters.py
index 02f4d3afe3..54a0d7c108 100644
--- a/plugins/nemo-insights/testbed/adapters.py
+++ b/plugins/nemo-insights/testbed/adapters.py
@@ -12,6 +12,8 @@
 
 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,
@@ -19,6 +21,7 @@
     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,
diff --git a/plugins/nemo-insights/testbed/artifact.py b/plugins/nemo-insights/testbed/artifact.py
index 2e205ae627..f1084d252a 100644
--- a/plugins/nemo-insights/testbed/artifact.py
+++ b/plugins/nemo-insights/testbed/artifact.py
@@ -23,7 +23,9 @@
 from pathlib import Path
 
 import httpx
+from nemo_platform import AsyncNeMoPlatform
 from testbed import export
+from testbed.intake_client import build_basic_auth_intake_client
 from testbed.registry import Subject
 
 LOCAL_URL = "http://localhost:8080"  # the local NeMo Platform (the default restore/analyze target)
@@ -127,6 +129,28 @@ def workspaces_for_subject(subject: Subject) -> list[str]:
     )
 
 
+def _basic_auth_intake_client_for(subject: Subject, source_url: str) -> AsyncNeMoPlatform | None:
+    """Build this subject's configured basic-auth Intake client, if needed."""
+    if subject.config.get("auth") != "basic":
+        return None
+    config = subject.config
+    credentials: dict[str, str] = {}
+    for role, key in (("username", "auth_user_env"), ("password", "auth_password_env")):
+        env_name = config.get(key)
+        value = os.environ.get(str(env_name)) if env_name else None
+        if not value:
+            env_label = str(env_name) if env_name else key
+            sys.exit(f"snapshot: subject '{subject.name}' is missing basic-auth {role} credential: {env_label}")
+        credentials[role] = value
+    real_prefix = str(config.get("intake_path_prefix", "/api/intake")).rstrip("/") + "/"
+    return build_basic_auth_intake_client(
+        base_url=source_url,
+        username=credentials["username"],
+        password=credentials["password"],
+        real_prefix=real_prefix,
+    )
+
+
 def backup_records(testbed_tmp: Path, names: list[str], *, backup_dir: Path | None = None) -> Path:
     """Move the named local files into *backup_dir* (default ``<testbed_tmp>/backup-<UTC stamp>/``);
     returns that dir.
@@ -189,11 +213,13 @@ def snapshot_export(
     Source URL is each subject's stanza ``base_url`` (they must agree; the
     CLI's ``--base`` override rewrites every stanza before this is called).
     """
-    workspaces: list[str] = []
+    subject_workspaces: list[tuple[Subject, list[str]]] = []
+    claimed_workspaces: set[str] = set()
     for subject in subjects:
-        for workspace in workspaces_for_subject(subject):
-            if workspace not in workspaces:
-                workspaces.append(workspace)
+        workspaces = [workspace for workspace in workspaces_for_subject(subject) if workspace not in claimed_workspaces]
+        if workspaces:
+            subject_workspaces.append((subject, workspaces))
+            claimed_workspaces.update(workspaces)
     # Every selected subject must carry a base_url: a partial miss would silently
     # let the agreement check pass on the configured subset and export the
     # unconfigured subject from the others' platform.
@@ -215,7 +241,33 @@ def snapshot_export(
     with tempfile.TemporaryDirectory(dir=tmp_dir) as tmp:
         state = Path(tmp) / "state"
         (state / "tmp").mkdir(parents=True)
-        stats = export.export_workspaces(source_url, workspaces, state, since=since)
+        exported = [
+            export.export_workspaces(
+                source_url,
+                workspaces,
+                state,
+                since=since,
+                client=_basic_auth_intake_client_for(subject, source_url),
+            )
+            for subject, workspaces in subject_workspaces
+        ]
+        min_bounds = [
+            datetime.datetime.fromisoformat(stats["min_start_time"])
+            for stats in exported
+            if stats["min_start_time"] is not None
+        ]
+        max_bounds = [
+            datetime.datetime.fromisoformat(stats["max_start_time"])
+            for stats in exported
+            if stats["max_start_time"] is not None
+        ]
+        stats = {
+            "workspaces": {
+                workspace: counts for result in exported for workspace, counts in result["workspaces"].items()
+            },
+            "min_start_time": min(min_bounds).isoformat() if min_bounds else None,
+            "max_start_time": max(max_bounds).isoformat() if max_bounds else None,
+        }
         for rec in records:
             shutil.copy2(rec, state / "tmp" / rec.name)
         manifest = build_export_manifest(
diff --git a/plugins/nemo-insights/testbed/cli.py b/plugins/nemo-insights/testbed/cli.py
index 7d1b03ed51..d781155dfc 100644
--- a/plugins/nemo-insights/testbed/cli.py
+++ b/plugins/nemo-insights/testbed/cli.py
@@ -6,6 +6,7 @@
     uv run python -m testbed doctor [<name>]
     uv run python -m testbed run <name> [--base URL] [--set KEY=VALUE ...]                     # benchmark: produce traces + record the run
     uv run python -m testbed analyze <name> [--update-insights] [--base URL] [--platform-root PATH]  # reproducible default: restore the subject's pinned state, then analyze
+    uv run python -m testbed analyze all                                                       # transactionally refresh every pinned benchmark/intake baseline
     uv run python -m testbed analyze <name> --state (state-vN | FILE)                          # same, against another published state or a local bundle file
     uv run python -m testbed analyze <name> --live [--since S] [-v] [--set KEY=VALUE ...]      # analyze the platform's live traces (no restore)
     uv run python -m testbed snapshot <subject> [...] [-o FILE] [--base URL] [--since S]       # export subject workspaces (read API) into a state bundle
@@ -35,6 +36,7 @@
 import argparse
 import asyncio
 import datetime
+import hashlib
 import json
 import os
 import re
@@ -42,6 +44,8 @@
 import subprocess
 import sys
 import tempfile
+import tomllib
+from collections.abc import Mapping
 from pathlib import Path
 
 import yaml
@@ -56,8 +60,17 @@
 HERE = Path(__file__).parent
 REGISTRY_PATH = HERE / "testbeds.toml"
 TMP = HERE / "tmp"
+INSIGHTS_DIR = HERE / "insights"
+ANALYST_SOURCE_ROOT = HERE.parent / "src" / "nemo_insights_plugin"
+ANALYST_LOCKFILE = HERE.parents[2] / "uv.lock"
+ANALYST_LOCK_ROOT = "nemo-insights-plugin"
 ENV_PATH = HERE / ".env"
 LOCAL_URL = artifact.LOCAL_URL  # the local NeMo Platform (the default restore/analyze target)
+_REMOTE_AUTH_KEYS = ("auth", "intake_path_prefix", "auth_user_env", "auth_password_env")
+INSIGHTS_SPDX_HEADER = (
+    "# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.\n"
+    "# SPDX-License-Identifier: Apache-2.0\n"
+)
 
 
 def _load_dotenv(path: Path = ENV_PATH) -> None:
@@ -104,11 +117,252 @@ def _doctor(subjects: dict[str, Subject], name: str | None) -> None:
             print(f"✓ {subject_name} ({subject.type}) — ready: uv run python -m testbed analyze {subject_name} --live")
 
 
-def _with_base(subject: Subject, base: str | None) -> Subject:
+def _atomic_write_text(path: Path, contents: str) -> None:
+    """Replace a text file atomically after writing it beside the destination."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    descriptor, temporary_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    temporary = Path(temporary_name)
+    try:
+        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+            stream.write(contents)
+        os.replace(temporary, path)
+    except BaseException:
+        temporary.unlink(missing_ok=True)
+        raise
+
+
+def _check_in_insights(
+    subject_name: str,
+    source: Path,
+    *,
+    directory: Path | None = None,
+) -> Path:
+    """Copy one successful runtime result into the tracked Insights directory."""
+    if not source.is_file():
+        sys.exit(f"analyze: expected Insights output at {source}, but the Analyst did not write it")
+    destination = (directory or INSIGHTS_DIR) / f"{subject_name}.yaml"
+    _atomic_write_text(destination, INSIGHTS_SPDX_HEADER + source.read_text(encoding="utf-8"))
+    return destination
+
+
+def _resolved_lock_entries(lockfile: dict, lockfile_path: Path) -> list[str]:
+    """Return deterministic lock entries in the Analyst plugin's dependency closure."""
+    packages_by_name: dict[str, list[dict]] = {}
+    for package in lockfile.get("package", []):
+        packages_by_name.setdefault(str(package["name"]), []).append(package)
+
+    pending = [{"name": ANALYST_LOCK_ROOT}]
+    visited: set[tuple[str, tuple[str, ...]]] = set()
+    selected: set[str] = set()
+    while pending:
+        dependency = pending.pop()
+        name = str(dependency["name"])
+        extras = tuple(sorted(str(extra) for extra in dependency.get("extra", [])))
+        request = (name, extras)
+        if request in visited:
+            continue
+        visited.add(request)
+        packages = packages_by_name.get(name)
+        if not packages:
+            raise ValueError(f"{lockfile_path} is missing Analyst dependency: {name}")
+        for package in packages:
+            optional = package.get("optional-dependencies", {})
+            entry = {key: value for key, value in package.items() if key not in {"metadata", "optional-dependencies"}}
+            if extras:
+                entry["selected-optional-dependencies"] = {extra: optional.get(extra, []) for extra in extras}
+            selected.add(json.dumps(entry, sort_keys=True, separators=(",", ":"), ensure_ascii=False))
+            pending.extend(package.get("dependencies", []))
+            for extra in extras:
+                pending.extend(optional.get(extra, []))
+    return sorted(selected)
+
+
+def _analyst_sha256(
+    root: Path = ANALYST_SOURCE_ROOT,
+    lockfile_path: Path = ANALYST_LOCKFILE,
+) -> str:
+    """Fingerprint Insights plugin source and the resolved Analyst dependency closure."""
+    digest = hashlib.sha256()
+    for path in sorted(root.rglob("*.py")):
+        digest.update(path.relative_to(root).as_posix().encode())
+        digest.update(b"\0")
+        digest.update(path.read_bytes())
+        digest.update(b"\0")
+
+    lockfile = tomllib.loads(lockfile_path.read_text(encoding="utf-8"))
+    for canonical in _resolved_lock_entries(lockfile, lockfile_path):
+        digest.update(b"dependency\0")
+        digest.update(canonical.encode())
+        digest.update(b"\0")
+    return digest.hexdigest()
+
+
+def _write_insights_manifest(
+    states: Mapping[str, str],
+    *,
+    merge: bool = False,
+    directory: Path | None = None,
+) -> Path:
+    """Write snapshot hashes and provenance, optionally preserving other subjects."""
+    insights_dir = directory or INSIGHTS_DIR
+    path = insights_dir / "manifest.yaml"
+    snapshots: dict[str, dict[str, str]] = {}
+    if merge and path.is_file():
+        existing = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
+        if not isinstance(existing, dict) or not isinstance(existing.get("snapshots", {}), dict):
+            raise ValueError(f"{path} does not contain a snapshots mapping")
+        snapshots.update(existing.get("snapshots", {}))
+
+    analyst_sha256 = _analyst_sha256()
+    for name, state in states.items():
+        snapshots[name] = {
+            "analyst_sha256": analyst_sha256,
+            "insights_sha256": hashlib.sha256((insights_dir / f"{name}.yaml").read_bytes()).hexdigest(),
+            "state": state,
+        }
+    contents = yaml.safe_dump({"snapshots": snapshots}, sort_keys=True)
+    _atomic_write_text(path, INSIGHTS_SPDX_HEADER + contents)
+    return path
+
+
+def _remove_stale_insights(names: list[str], *, directory: Path | None = None) -> None:
+    """Remove YAML files outside the complete expected snapshot set."""
+    insights_dir = directory or INSIGHTS_DIR
+    expected = {"manifest.yaml", *(f"{name}.yaml" for name in names)}
+    for path in insights_dir.glob("*.yaml"):
+        if path.name not in expected:
+            path.unlink()
+            print(f"✓ Removed stale checked-in Insights {path}")
+
+
+def _swap_insights_directory(staged: Path) -> None:
+    """Promote a complete staged directory, restoring the old one on swap failure."""
+    INSIGHTS_DIR.parent.mkdir(parents=True, exist_ok=True)
+    backup: Path | None = None
+    if INSIGHTS_DIR.exists():
+        backup = Path(
+            tempfile.mkdtemp(
+                prefix=f".{INSIGHTS_DIR.name}.backup-",
+                dir=INSIGHTS_DIR.parent,
+            )
+        )
+        backup.rmdir()
+        os.replace(INSIGHTS_DIR, backup)
+    try:
+        os.replace(staged, INSIGHTS_DIR)
+    except BaseException:
+        if backup is not None:
+            try:
+                os.replace(backup, INSIGHTS_DIR)
+            except BaseException as rollback_error:
+                raise RuntimeError(
+                    f"failed to promote {staged} and could not restore backup {backup}"
+                ) from rollback_error
+        raise
+    if backup is not None:
+        try:
+            shutil.rmtree(backup)
+        except OSError as exc:
+            print(f"warning: promoted checked-in Insights but could not remove backup {backup}: {exc}")
+
+
+def _check_in_single_insights(subject_name: str, source: Path, state: str) -> tuple[Path, Path]:
+    """Transactionally update one Insight and its manifest entry."""
+    INSIGHTS_DIR.parent.mkdir(parents=True, exist_ok=True)
+    staged = Path(
+        tempfile.mkdtemp(
+            prefix=f".{INSIGHTS_DIR.name}.staging-",
+            dir=INSIGHTS_DIR.parent,
+        )
+    )
+    try:
+        if INSIGHTS_DIR.is_dir():
+            shutil.copytree(INSIGHTS_DIR, staged, dirs_exist_ok=True)
+        _check_in_insights(subject_name, source, directory=staged)
+        _write_insights_manifest({subject_name: state}, merge=True, directory=staged)
+        _swap_insights_directory(staged)
+    finally:
+        if staged.exists():
+            shutil.rmtree(staged)
+    return INSIGHTS_DIR / f"{subject_name}.yaml", INSIGHTS_DIR / "manifest.yaml"
+
+
+def _analyze_all(args: argparse.Namespace, subjects: dict[str, Subject]) -> None:
+    """Run every reproducible subject, then transactionally promote all outputs."""
+    if args.live or args.state is not None or args.since is not None or args.sets or args.update_insights:
+        sys.exit(
+            "analyze all uses each subject's pinned state and cannot be combined with "
+            "--live, --state, --since, --set, or --update-insights"
+        )
+
+    names = sorted(name for name, subject in subjects.items() if subject.type in ("benchmark", "intake"))
+    resolved_pins = {name: release.lock_ref(HERE / "state.lock", name) for name in names}
+    missing_pins = [name for name, pin in resolved_pins.items() if pin is None]
+    if missing_pins:
+        sys.exit(
+            "analyze all requires a state.lock pin for every analyzable subject; missing: " + ", ".join(missing_pins)
+        )
+    pins = {name: pin for name, pin in resolved_pins.items() if pin is not None}
+
+    stale_outputs = [f"insights_{name}.yaml" for name in names if (TMP / f"insights_{name}.yaml").is_file()]
+    if stale_outputs:
+        backup_dir = artifact.backup_records(TMP, stale_outputs)
+        print(f"analyze all: moved prior runtime Insights to {backup_dir}: {', '.join(stale_outputs)}")
+
+    for name in names:
+        command = [sys.executable, "-m", "testbed", "analyze", name, "--no-baseline-update"]
+        if args.base is not None:
+            command += ["--base", args.base]
+        if args.platform_root is not None:
+            command += ["--platform-root", args.platform_root]
+        if args.summary_md is not None:
+            command += ["--summary-md", args.summary_md]
+        if args.verbose:
+            command.append("--verbose")
+        try:
+            subprocess.run(command, check=True)
+        except subprocess.CalledProcessError as exc:
+            sys.exit(f"analyze all: {name} failed with exit code {exc.returncode}; no checked-in Insights were updated")
+
+    outputs = {name: TMP / f"insights_{name}.yaml" for name in names}
+    missing_outputs = [name for name, path in outputs.items() if not path.is_file()]
+    if missing_outputs:
+        sys.exit(
+            "analyze all completed without Insights output for: "
+            + ", ".join(missing_outputs)
+            + "; no checked-in Insights were updated"
+        )
+    if args.no_baseline_update:
+        return
+
+    INSIGHTS_DIR.parent.mkdir(parents=True, exist_ok=True)
+    staged = Path(
+        tempfile.mkdtemp(
+            prefix=f".{INSIGHTS_DIR.name}.staging-",
+            dir=INSIGHTS_DIR.parent,
+        )
+    )
+    try:
+        for name, source in outputs.items():
+            _check_in_insights(name, source, directory=staged)
+        _remove_stale_insights(names, directory=staged)
+        _write_insights_manifest(pins, directory=staged)
+        _swap_insights_directory(staged)
+    finally:
+        if staged.exists():
+            shutil.rmtree(staged)
+    print(f"✓ Checked in complete Insights set to {INSIGHTS_DIR}")
+
+
+def _with_base(subject: Subject, base: str | None, *, drop_auth: bool = True) -> Subject:
     """Rebuild *subject* pointed at *base* (the uniform ``--base`` override); unchanged when None."""
     if base is None:
         return subject
-    return Subject(subject.name, subject.type, {**subject.config, "base_url": base})
+    config = {**subject.config, "base_url": base}
+    if drop_auth:
+        for key in _REMOTE_AUTH_KEYS:
+            config.pop(key, None)
+    return Subject(subject.name, subject.type, config)
 
 
 def _local_source(args: argparse.Namespace) -> str | None:
@@ -445,7 +699,7 @@ def main() -> None:
         help="Generate Insights for a subject (default: restore its pinned state onto the "
         "local platform, then analyze — fully reproducible).",
     )
-    p_ins.add_argument("name", help="Subject name from testbeds.toml.")
+    p_ins.add_argument("name", help="Subject name from testbeds.toml, or 'all'.")
     p_ins_mode = p_ins.add_mutually_exclusive_group()
     p_ins_mode.add_argument(
         "--live",
@@ -491,6 +745,11 @@ def main() -> None:
         help="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.",
     )
+    p_ins.add_argument(
+        "--no-baseline-update",
+        action="store_true",
+        help="leave the generated Insights only in testbed/tmp instead of updating testbed/insights.",
+    )
     p_ins.add_argument(
         "--platform-root",
         default=None,
@@ -680,7 +939,7 @@ def main() -> None:
         # all-stanzas-must-agree check trivially holds; without it the stanza
         # agreement (or disagreement) stands as-is.
         result = artifact.snapshot_export(
-            [_with_base(subjects[n], args.base) for n in names],
+            [_with_base(subjects[n], args.base, drop_auth=False) for n in names],
             out,
             TMP,
             since=since,
@@ -740,6 +999,9 @@ def main() -> None:
         return
 
     # args.cmd == "analyze" — pinned/--state mode restores a bundle first; --live doesn't.
+    if args.name == "all":
+        _analyze_all(args, subjects)
+        return
     subject = subjects.get(args.name)
     if subject is None:
         sys.exit(f"Unknown testbed '{args.name}'. Available: {', '.join(sorted(subjects)) or '(none)'}")
@@ -751,7 +1013,11 @@ def main() -> None:
     # Target platform: --base wins everywhere; without it, pinned/state mode
     # restores onto the local platform (reproducible default) and --live reads
     # the stanza's own base_url.
-    subject = _with_base(subject, args.base if args.live else (args.base or LOCAL_URL))
+    subject = _with_base(
+        subject,
+        args.base if args.live else (args.base or LOCAL_URL),
+        drop_auth=not args.live,
+    )
     subject = _apply_overrides(subject, args.sets)
     if not os.environ.get("INFERENCE_API_KEY"):
         sys.exit("Set INFERENCE_API_KEY (NVIDIA Inference Gateway sk-... key) and re-run.")
@@ -883,6 +1149,10 @@ def main() -> None:
     report = asyncio.run(adapter.analyze(record=record, since=since, verbose=args.verbose, out_path=out))
     print(report)
     print(f"\n✓ Insights written to {out}")
+    if not args.no_baseline_update:
+        checked_in, manifest_path = _check_in_single_insights(args.name, out, label)
+        print(f"✓ Checked in to {checked_in}")
+        print(f"✓ Recorded provenance in {manifest_path}")
     if args.summary_md:
         with Path(args.summary_md).open("a", encoding="utf-8") as fh:
             fh.write(render_summary_md(out, args.name))
diff --git a/plugins/nemo-insights/testbed/export.py b/plugins/nemo-insights/testbed/export.py
index 8fbe94fc54..46138364b6 100644
--- a/plugins/nemo-insights/testbed/export.py
+++ b/plugins/nemo-insights/testbed/export.py
@@ -77,7 +77,14 @@ def note(self, doc: dict) -> None:
             self.max = ts
 
 
-def export_workspaces(base_url: str, workspaces: list[str], out_dir: Path, *, since: datetime | None) -> dict:
+def export_workspaces(
+    base_url: str,
+    workspaces: list[str],
+    out_dir: Path,
+    *,
+    since: datetime | None,
+    client: AsyncNeMoPlatform | None = None,
+) -> dict:
     """Drain spans/annotations/evaluator-results per workspace into JSONL files.
 
     Writes ``out_dir/export/<workspace>/{spans,annotations,evaluator_results}.jsonl``
@@ -85,14 +92,21 @@ def export_workspaces(base_url: str, workspaces: list[str], out_dir: Path, *, si
     "max_start_time": ...}`` (time bounds from span ``started_at``; ISO strings or
     None when no spans matched).
     """
-    return asyncio.run(_export_workspaces(base_url, workspaces, out_dir, since=since))
+    return asyncio.run(_export_workspaces(base_url, workspaces, out_dir, since=since, client=client))
 
 
-async def _export_workspaces(base_url: str, workspaces: list[str], out_dir: Path, *, since: datetime | None) -> dict:
+async def _export_workspaces(
+    base_url: str,
+    workspaces: list[str],
+    out_dir: Path,
+    *,
+    since: datetime | None,
+    client: AsyncNeMoPlatform | None = None,
+) -> dict:
     lower = (since or EPOCH).isoformat()
     bounds = _StartBounds()
     counts: dict[str, dict[str, int]] = {}
-    client = make_client(base_url)
+    client = client if client is not None else make_client(base_url)
     try:
         for workspace in workspaces:
             ws_dir = out_dir / "export" / workspace
diff --git a/plugins/nemo-insights/testbed/intake_client.py b/plugins/nemo-insights/testbed/intake_client.py
new file mode 100644
index 0000000000..a919f4f55d
--- /dev/null
+++ b/plugins/nemo-insights/testbed/intake_client.py
@@ -0,0 +1,65 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Build an SDK client for a basic-auth Intake deployment."""
+
+from collections.abc import Awaitable, Callable
+
+import httpx
+from nemo_platform import AsyncNeMoPlatform
+
+_SDK_INTAKE_PREFIX = "/apis/intake/"
+_DEFAULT_REAL_PREFIX = "/api/intake/"
+
+
+def _make_request_rewriter(
+    sdk_prefix: str,
+    real_prefix: str,
+) -> Callable[[httpx.Request], Awaitable[None]]:
+    """Return a request hook that rewrites an Intake path prefix."""
+    sdk = sdk_prefix.encode()
+    real = real_prefix.encode()
+
+    async def _rewrite(request: httpx.Request) -> None:
+        raw_path = request.url.raw_path
+        if raw_path.startswith(sdk):
+            request.url = request.url.copy_with(raw_path=real + raw_path[len(sdk) :])
+
+    return _rewrite
+
+
+def build_rewriting_http_client(
+    *,
+    username: str,
+    password: str,
+    real_prefix: str = _DEFAULT_REAL_PREFIX,
+    sdk_prefix: str = _SDK_INTAKE_PREFIX,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> httpx.AsyncClient:
+    """Build an HTTP client with basic auth and an Intake path-prefix rewrite."""
+    return httpx.AsyncClient(
+        auth=httpx.BasicAuth(username, password),
+        event_hooks={"request": [_make_request_rewriter(sdk_prefix, real_prefix)]},
+        transport=transport,
+        timeout=60.0,
+    )
+
+
+def build_basic_auth_intake_client(
+    *,
+    base_url: str,
+    username: str,
+    password: str,
+    real_prefix: str = _DEFAULT_REAL_PREFIX,
+    sdk_prefix: str = _SDK_INTAKE_PREFIX,
+    transport: httpx.AsyncBaseTransport | None = None,
+) -> AsyncNeMoPlatform:
+    """Build an SDK client for a basic-auth Intake mounted at ``real_prefix``."""
+    http_client = build_rewriting_http_client(
+        username=username,
+        password=password,
+        real_prefix=real_prefix,
+        sdk_prefix=sdk_prefix,
+        transport=transport,
+    )
+    return AsyncNeMoPlatform(base_url=base_url, http_client=http_client)
diff --git a/plugins/nemo-insights/testbed/reingest.py b/plugins/nemo-insights/testbed/reingest.py
index 3afd953c08..832824feaa 100644
--- a/plugins/nemo-insights/testbed/reingest.py
+++ b/plugins/nemo-insights/testbed/reingest.py
@@ -74,7 +74,8 @@
 _EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
 _EPOCH_ISO = "1970-01-01T00:00:00Z"
 _STATUS_CODE_ERROR = 2  # opentelemetry.proto.trace.v1.Status.STATUS_CODE_ERROR
-SPAN_BATCH = 100  # spans per OTLP request (ingest caps bodies at 5 MiB; tau2 spans are a few KB)
+OTLP_REQUEST_MAX_BYTES = 4 * 1024 * 1024
+OTLP_REQUEST_MAX_SPANS = 100
 
 # Where the attribute catalog lives inside a nemo-platform checkout.
 CATALOG_RELPATH = Path("services/intake/src/nmp/intake/spans/span_attribute_catalog.py")
@@ -335,6 +336,52 @@ def build_trace_request(otlp_spans: list[dict]) -> ExportTraceServiceRequest:
     return request
 
 
+def build_trace_requests(
+    docs: list[dict],
+    catalog,
+    *,
+    max_bytes: int | None = None,
+    max_spans: int | None = None,
+) -> list[ExportTraceServiceRequest]:
+    """Convert span docs into OTLP export requests bounded by span count and protobuf size.
+
+    Each doc is converted once via :func:`doc_to_otlp`. A candidate batch is flushed
+    when adding the next span would exceed ``max_spans`` or ``max_bytes`` (measured
+    with protobuf ``ByteSize()``). A single span whose request exceeds ``max_bytes``
+    raises before any request is returned.
+    """
+    if max_bytes is None:
+        max_bytes = OTLP_REQUEST_MAX_BYTES
+    if max_spans is None:
+        max_spans = OTLP_REQUEST_MAX_SPANS
+    requests: list[ExportTraceServiceRequest] = []
+    batch: list[dict] = []
+
+    def _reject_oversized(otlp: dict, doc: dict) -> None:
+        size = build_trace_request([otlp]).ByteSize()
+        if size > max_bytes:
+            raise RuntimeError(f"span {doc.get('span_id')}: OTLP body exceeds {max_bytes} bytes ({size})")
+
+    for doc in docs:
+        otlp = doc_to_otlp(doc, catalog)
+        if not batch:
+            _reject_oversized(otlp, doc)
+            batch = [otlp]
+            continue
+        candidate = batch + [otlp]
+        candidate_request = build_trace_request(candidate)
+        if len(candidate) > max_spans or candidate_request.ByteSize() > max_bytes:
+            requests.append(build_trace_request(batch))
+            _reject_oversized(otlp, doc)
+            batch = [otlp]
+        else:
+            batch = candidate
+
+    if batch:
+        requests.append(build_trace_request(batch))
+    return requests
+
+
 def _collection_count(
     base_url: str, workspace: str, collection: str, time_field: str, *, client: httpx.Client | None = None
 ) -> int:
@@ -700,9 +747,10 @@ def ingest_bundle(
             healing = expected_spans > 0 and not ingest_spans
             if ingest_spans:
                 print(f"ingesting {len(spans)} spans into {target}")
-                for start in range(0, len(spans), SPAN_BATCH):
-                    batch = [doc_to_otlp(doc, catalog) for doc in spans[start : start + SPAN_BATCH]]
-                    request = build_trace_request(batch)
+                # Materialize every request before posting: higher memory use buys up-front
+                # validation (including oversized spans) and prevents a partial restore.
+                trace_requests = build_trace_requests(spans, catalog)
+                for request in trace_requests:
                     export_trace_request(base_url, target, request, client=client)
                 if spans:
                     _wait_for_spans(base_url, target, expected_spans or len(spans), client=client, sleep=sleep)
diff --git a/plugins/nemo-insights/testbed/state.lock b/plugins/nemo-insights/testbed/state.lock
index 9921c3d4d8..797600f11a 100644
--- a/plugins/nemo-insights/testbed/state.lock
+++ b/plugins/nemo-insights/testbed/state.lock
@@ -1,6 +1,13 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 # Analyze-mode CI reads each subject's pinned state from here (one PR = one
 # state per subject). A missing entry fails loudly — add the subject after
 # minting its fixture, or pass --state <state-vN> explicitly.
 [subjects]
-tau2-airline = "state-v6"
 nvq = "state-v7"
+tau2-airline = "state-v6"
+tau2-retail = "state-v10"
+tau2-telecom = "state-v10"
+glamr = "state-v8"
+nemo-oo-airline = "state-v9"
diff --git a/plugins/nemo-insights/testbed/tau2run.py b/plugins/nemo-insights/testbed/tau2run.py
index 1814a6a6d0..4930c74d5c 100644
--- a/plugins/nemo-insights/testbed/tau2run.py
+++ b/plugins/nemo-insights/testbed/tau2run.py
@@ -136,13 +136,16 @@ def read_policy(data_dir: Path, domain: str) -> str | None:
     """Return the domain policy markdown (the analyst's agent_spec), or None.
 
     A tau2 checkout nests domains under ``tau2/domains/<domain>/``; some data
-    dirs are flat (``domains/<domain>/``). Tries both and returns the first
-    ``policy.md`` found, else ``None`` (the analyst then runs without the spec).
+    dirs are flat (``domains/<domain>/``). For each layout, tries ``policy.md``
+    then ``main_policy.md`` (Telecom uses the latter). Returns the first file
+    found, else ``None`` (the analyst then runs without the spec).
     """
     for base in (data_dir / "tau2" / "domains", data_dir / "domains"):
-        path = base / domain / "policy.md"
-        if path.exists():
-            return path.read_text(encoding="utf-8")
+        domain_dir = base / domain
+        for filename in ("policy.md", "main_policy.md"):
+            path = domain_dir / filename
+            if path.exists():
+                return path.read_text(encoding="utf-8")
     return None
 
 
diff --git a/plugins/nemo-insights/testbed/testbeds.toml b/plugins/nemo-insights/testbed/testbeds.toml
index b87a55428f..ed684fb102 100644
--- a/plugins/nemo-insights/testbed/testbeds.toml
+++ b/plugins/nemo-insights/testbed/testbeds.toml
@@ -1,3 +1,6 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
 # Testbed subjects for the analyst harness — one table per subject.
 # Top-level keys are shared defaults merged into every subject.
 base_url = "https://nemo-platform-freeplay.dev.aire.nvidia.com"
@@ -6,7 +9,26 @@ base_url = "https://nemo-platform-freeplay.dev.aire.nvidia.com"
 type = "intake"
 agent = "content-dedup"
 workspace = "nvq"
-# since = "7d"   # optional default lookback (Nd/Nh/Nm or an ISO date)
+
+[glamr]
+type = "intake"
+agent = "glamr"                                 # the main GLAMR agent (ux-agent); the only agent_name-tagged one
+workspace = "default"
+base_url = "https://agenthub.aire.nvidia.com"   # host root; the intake prefix is applied by the rewrite
+auth = "basic"                                  # this Intake is gated by HTTP basic auth (VPN-only, aire)
+intake_path_prefix = "/api/intake"              # this deployment mounts intake here, not at the SDK's /apis/intake
+auth_user_env = "GLAMR_INTAKE_USER"             # names the env var holding the basic-auth username (value in testbed/.env)
+auth_password_env = "GLAMR_INTAKE_PASSWORD"     # names the env var holding the basic-auth password (value in testbed/.env)
+since = "7d"                                     # default lookback for --live; override with --since (e.g. --since 1d, or '' for all-time)
+
+[nemo-oo-airline]
+# The reproducible-analysis face of the NeMo OO Agents tau2-airline corpus.
+# Pinned analyze supports benchmark/intake subjects, so this corpus is
+# represented by the intake workspace captured in state-v9.
+type = "intake"
+agent = "nemo-oo-airline"
+workspace = "tau2-airline-20260710-152942-4754"
+base_url = "http://localhost:8080"
 
 [tau2-airline]
 type = "benchmark"
@@ -26,6 +48,24 @@ include_rewards = true            # also ingest a "-oracle" workspace (answer ke
 # tau2_data_dir = "/custom/data"  # optional: override the derived <tau2_repo>/data
 # timeout = 600                   # optional per-task timeout (seconds)
 
+[tau2-telecom]
+type = "benchmark"
+domain = "telecom"
+base_url = "http://localhost:8080"
+workspace = "tau2-telecom"
+tau2_repo = "../tau2-bench"
+agent_llm = "openai/nvidia/nvidia/nemotron-3-super-v3"
+user_llm = "openai/nvidia/nvidia/nemotron-3-super-v3"
+task_split_name = "small"  # curated 20-task telecom split
+num_tasks = 20
+num_trials = 1
+max_concurrency = 8
+seed = 300
+include_rewards = true
+# tau2_bin = "/custom/tau2"
+# tau2_data_dir = "/custom/data"
+# timeout = 600
+
 [tau2-retail]
 type = "benchmark"
 domain = "retail"
diff --git a/plugins/nemo-insights/tests/test_analyst_run.py b/plugins/nemo-insights/tests/test_analyst_run.py
new file mode 100644
index 0000000000..dd34643e13
--- /dev/null
+++ b/plugins/nemo-insights/tests/test_analyst_run.py
@@ -0,0 +1,99 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""The ``run_analyst`` client injection contract."""
+
+import pytest
+from nemo_insights_plugin.analyst import run as run_module
+
+
+class FakeClient:
+    def __init__(self) -> None:
+        self.closed = False
+
+    async def close(self) -> None:
+        self.closed = True
+
+
+class FakeBackend:
+    async def persist_result(self, *, workspace: str, agent: str, result: object) -> str:
+        return "REPORT"
+
+
+def _stub_pipeline(monkeypatch: pytest.MonkeyPatch, seen: dict[str, object]) -> None:
+    def fake_make_backend(*, client: FakeClient, insights_output: str | None) -> FakeBackend:
+        seen["backend_client"] = client
+        return FakeBackend()
+
+    async def fake_run_agent(analyst: object, deps: object, *, verbose: bool) -> object:
+        return object()
+
+    monkeypatch.setattr(run_module, "make_analyst_backend", fake_make_backend)
+    monkeypatch.setattr(run_module, "build_analyst_agent", lambda **kwargs: object())
+    monkeypatch.setattr(run_module, "_run_agent", fake_run_agent)
+
+
+async def test_injected_client_is_used_and_closed(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = FakeClient()
+    seen: dict[str, object] = {}
+    _stub_pipeline(monkeypatch, seen)
+
+    report = await run_module.run_analyst(
+        agent="agent",
+        agent_spec=None,
+        workspace="workspace",
+        base_url="https://platform",
+        client=client,  # type: ignore[arg-type]
+    )
+
+    assert report == "REPORT"
+    assert seen["backend_client"] is client
+    assert client.closed
+
+
+async def test_client_closed_when_backend_construction_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = FakeClient()
+
+    def raising_backend(*, client: FakeClient, insights_output: str | None) -> FakeBackend:
+        raise RuntimeError("backend failed")
+
+    monkeypatch.setattr(run_module, "make_analyst_backend", raising_backend)
+
+    with pytest.raises(RuntimeError, match="backend failed"):
+        await run_module.run_analyst(
+            agent="agent",
+            agent_spec=None,
+            workspace="workspace",
+            base_url="https://platform",
+            client=client,  # type: ignore[arg-type]
+        )
+
+    assert client.closed
+
+
+async def test_client_closed_when_observability_shutdown_raises(monkeypatch: pytest.MonkeyPatch) -> None:
+    client = FakeClient()
+    seen: dict[str, object] = {}
+    _stub_pipeline(monkeypatch, seen)
+
+    class FailingObservability:
+        def shutdown(self) -> None:
+            raise RuntimeError("shutdown failed")
+
+    monkeypatch.setattr(run_module, "_analyst_observability_enabled", lambda: True)
+    monkeypatch.setattr(
+        run_module,
+        "setup_analyst_observability",
+        lambda **kwargs: FailingObservability(),
+    )
+
+    with pytest.raises(RuntimeError, match="shutdown failed"):
+        await run_module.run_analyst(
+            agent="agent",
+            agent_spec=None,
+            workspace="workspace",
+            base_url="https://platform",
+            client=client,  # type: ignore[arg-type]
+        )
+
+    assert client.closed
diff --git a/plugins/nemo-insights/tests/test_cli_profile.py b/plugins/nemo-insights/tests/test_cli_profile.py
index f5abcf29c4..fea8fe460e 100644
--- a/plugins/nemo-insights/tests/test_cli_profile.py
+++ b/plugins/nemo-insights/tests/test_cli_profile.py
@@ -5,7 +5,6 @@
 from pathlib import Path
 
 import httpx
-import nemo_insights_plugin.analyst.run as analyst_run
 import pytest
 import typer
 from nemo_insights_plugin import cli
@@ -46,6 +45,7 @@ async def queryable(base_url: str, workspace: str, agent: str) -> bool:
             workspace_ok=queryable,
         ),
     )
+    monkeypatch.setattr(cli, "make_client", lambda base_url: object())
 
 
 @pytest.fixture
@@ -504,7 +504,7 @@ def fail_to_construct(base_url: str | None) -> object:
         attempts += 1
         raise error_type("invalid\nremote client context")
 
-    monkeypatch.setattr(analyst_run, "make_client", fail_to_construct)
+    monkeypatch.setattr(cli, "make_client", fail_to_construct)
     monkeypatch.chdir(profile_tree)
 
     result = runner.invoke(app, ["analyze"])
diff --git a/plugins/nemo-insights/tests/test_periodic_analysis.py b/plugins/nemo-insights/tests/test_periodic_analysis.py
index 64e5cee976..d37af678b5 100644
--- a/plugins/nemo-insights/tests/test_periodic_analysis.py
+++ b/plugins/nemo-insights/tests/test_periodic_analysis.py
@@ -486,6 +486,7 @@ async def fake_run_analyst(**_: object) -> str:
         return "analysis report"
 
     monkeypatch.setattr("nemo_insights_plugin.jobs.analyze.run_analyst", fake_run_analyst)
+    monkeypatch.setattr("nemo_insights_plugin.jobs.analyze.make_client", lambda base_url: object())
     sdk = _SyncSdk()
 
     result = AnalyzeJob().run(
diff --git a/plugins/nemo-insights/tests/testbed/test_adapters.py b/plugins/nemo-insights/tests/testbed/test_adapters.py
index c19bdeb527..04645bdf17 100644
--- a/plugins/nemo-insights/tests/testbed/test_adapters.py
+++ b/plugins/nemo-insights/tests/testbed/test_adapters.py
@@ -43,6 +43,82 @@ def _intake_subject(**overrides) -> Subject:
     return Subject(name="nvq", type="intake", config=config)
 
 
+def _basic_intake_subject(**overrides) -> Subject:
+    config = {
+        "agent": "glamr",
+        "workspace": "default",
+        "base_url": "https://agenthub.aire.nvidia.com",
+        "auth": "basic",
+        "auth_user_env": "GLAMR_INTAKE_USER",
+        "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+        **overrides,
+    }
+    return Subject(name="glamr", type="intake", config=config)
+
+
+def test_intake_check_basic_auth_reports_missing_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("GLAMR_INTAKE_USER", raising=False)
+    monkeypatch.delenv("GLAMR_INTAKE_PASSWORD", raising=False)
+
+    missing = IntakeAdapter(_basic_intake_subject()).check()
+
+    assert any("GLAMR_INTAKE_USER" in item for item in missing)
+    assert any("GLAMR_INTAKE_PASSWORD" in item for item in missing)
+
+
+def test_intake_check_basic_auth_reports_missing_username_env_name(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "secret")
+
+    missing = IntakeAdapter(_basic_intake_subject(auth_user_env=None)).check()
+
+    assert any("auth_user_env" in item for item in missing)
+    assert not any("auth_password_env" in item for item in missing)
+
+
+def test_intake_check_basic_auth_reports_missing_password_env_name(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "intake")
+
+    missing = IntakeAdapter(_basic_intake_subject(auth_password_env=None)).check()
+
+    assert any("auth_password_env" in item for item in missing)
+    assert not any("auth_user_env" in item for item in missing)
+
+
+async def test_intake_analyze_basic_auth_injects_built_client(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "intake")
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "secret")
+    sentinel = object()
+    built: dict[str, object] = {}
+    calls: dict[str, object] = {}
+
+    def fake_builder(**kwargs: object) -> object:
+        built.update(kwargs)
+        return sentinel
+
+    async def fake_run_analyst(**kwargs: object) -> str:
+        calls.update(kwargs)
+        return "REPORT"
+
+    monkeypatch.setattr("testbed.adapters.build_basic_auth_intake_client", fake_builder)
+    monkeypatch.setattr("testbed.adapters.run_analyst", fake_run_analyst)
+
+    report = await IntakeAdapter(_basic_intake_subject()).analyze(
+        record=None,
+        since=None,
+        verbose=False,
+        out_path=tmp_path / "insights.json",
+    )
+
+    assert report == "REPORT"
+    assert calls["client"] is sentinel
+    assert built == {
+        "base_url": "https://agenthub.aire.nvidia.com",
+        "username": "intake",
+        "password": "secret",
+        "real_prefix": "/api/intake/",
+    }
+
+
 def test_build_adapter_returns_intake():
     assert isinstance(build_adapter(_intake_subject()), IntakeAdapter)
 
@@ -54,12 +130,14 @@ def test_build_adapter_unknown_type_exits():
 
 async def test_intake_analyze_calls_run_analyst(monkeypatch, tmp_path: Path):
     calls: dict[str, object] = {}
+    built_client = object()
 
     async def fake_run_analyst(**kwargs):
         calls.update(kwargs)
         return "REPORT"
 
     monkeypatch.setattr("testbed.adapters.run_analyst", fake_run_analyst)
+    monkeypatch.setattr("testbed.adapters.make_client", lambda base_url: built_client)
     out = tmp_path / "insights.json"
     report = await build_adapter(_intake_subject()).analyze(record=None, since=None, verbose=True, out_path=out)
     assert report == "REPORT"
@@ -67,6 +145,7 @@ async def fake_run_analyst(**kwargs):
     assert calls["workspace"] == "w"
     assert calls["base_url"] == "u"
     assert calls["agent_spec"] is None
+    assert calls["client"] is built_client
 
 
 async def test_intake_analyze_missing_keys_exits(tmp_path: Path):
@@ -235,6 +314,7 @@ async def fake_run_analyst(**kwargs):
         return "SHOULD-NOT-RUN"
 
     monkeypatch.setattr("testbed.adapters.run_analyst", fake_run_analyst)
+    monkeypatch.setattr("testbed.adapters.make_client", lambda base_url: object())
 
     cfg = {**_CFG, "tau2_data_dir": str(tmp_path), "tau2_bin": "tau2"}
     record = await BenchmarkAdapter(Subject("tau2-airline", "benchmark", cfg)).produce()
@@ -272,7 +352,7 @@ async def test_benchmark_analyze_uses_record(monkeypatch, tmp_path):
     seen: dict[str, object] = {}
 
     async def fake_run_analyst(
-        *, agent, agent_spec, workspace, base_url, insights_output, verbose, since, evaluation_id
+        *, agent, agent_spec, workspace, base_url, client, insights_output, verbose, since, evaluation_id
     ):
         seen.update(
             agent=agent,
@@ -284,6 +364,7 @@ async def fake_run_analyst(
         return "REPORT-OK"
 
     monkeypatch.setattr("testbed.adapters.run_analyst", fake_run_analyst)
+    monkeypatch.setattr("testbed.adapters.make_client", lambda base_url: object())
 
     cfg = {**_CFG, "tau2_data_dir": str(tmp_path), "tau2_bin": "tau2"}
     record = {
diff --git a/plugins/nemo-insights/tests/testbed/test_artifact_snapshot_auth.py b/plugins/nemo-insights/tests/testbed/test_artifact_snapshot_auth.py
new file mode 100644
index 0000000000..a497e4c30e
--- /dev/null
+++ b/plugins/nemo-insights/tests/testbed/test_artifact_snapshot_auth.py
@@ -0,0 +1,208 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Tests for authenticated Intake snapshot exports."""
+
+from pathlib import Path
+
+import pytest
+from testbed import artifact
+from testbed.registry import Subject
+
+
+def _glamr_subject(**overrides: object) -> Subject:
+    return Subject(
+        "glamr",
+        "intake",
+        {
+            "agent": "glamr",
+            "workspace": "default",
+            "base_url": "https://agenthub.aire.nvidia.com",
+            "auth": "basic",
+            "intake_path_prefix": "/glamr/intake",
+            "auth_user_env": "GLAMR_INTAKE_USER",
+            "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+            **overrides,
+        },
+    )
+
+
+def test_snapshot_basic_auth_client_uses_named_credentials_and_normalized_prefix(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "intake-user")
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "secret")
+    built: dict[str, object] = {}
+    sentinel = object()
+
+    def fake_builder(**kwargs: object) -> object:
+        built.update(kwargs)
+        return sentinel
+
+    monkeypatch.setattr(artifact, "build_basic_auth_intake_client", fake_builder)
+
+    client = artifact._basic_auth_intake_client_for(_glamr_subject(), "https://snapshot.example")
+
+    assert client is sentinel
+    assert built == {
+        "base_url": "https://snapshot.example",
+        "username": "intake-user",
+        "password": "secret",
+        "real_prefix": "/glamr/intake/",
+    }
+
+
+@pytest.mark.parametrize(
+    ("missing_env", "credential"),
+    [
+        ("GLAMR_INTAKE_USER", "username"),
+        ("GLAMR_INTAKE_PASSWORD", "password"),
+    ],
+)
+def test_snapshot_basic_auth_client_exits_for_missing_credential(
+    monkeypatch: pytest.MonkeyPatch,
+    missing_env: str,
+    credential: str,
+) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "intake-user")
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "secret")
+    monkeypatch.delenv(missing_env)
+
+    with pytest.raises(SystemExit, match=f"glamr.*{credential}.*{missing_env}"):
+        artifact._basic_auth_intake_client_for(_glamr_subject(), "https://snapshot.example")
+
+
+def test_snapshot_non_basic_subject_does_not_build_authenticated_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        artifact,
+        "build_basic_auth_intake_client",
+        lambda **kwargs: pytest.fail(f"unexpected basic-auth builder call: {kwargs}"),
+    )
+
+    assert (
+        artifact._basic_auth_intake_client_for(
+            Subject("plain", "intake", {"workspace": "default", "base_url": "https://snapshot.example"}),
+            "https://snapshot.example",
+        )
+        is None
+    )
+
+
+def test_snapshot_setup_failure_does_not_construct_authenticated_client(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    constructed = False
+
+    def fake_client(subject: Subject, source_url: str) -> object:
+        nonlocal constructed
+        constructed = True
+        return object()
+
+    def fail_pick_records(tmp_dir: Path, names: list[str]) -> list[Path]:
+        raise RuntimeError("setup failed")
+
+    monkeypatch.setattr(artifact, "_basic_auth_intake_client_for", fake_client)
+    monkeypatch.setattr(artifact, "pick_records", fail_pick_records)
+
+    with pytest.raises(RuntimeError, match="setup failed"):
+        artifact.snapshot_export([_glamr_subject()], tmp_path / "snapshot.tar.zst", tmp_path / "tmp", since=None)
+
+    assert not constructed
+
+
+def test_snapshot_export_scopes_authentication_to_each_subject(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "glamr-user")
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "glamr-secret")
+    monkeypatch.setenv("OTHER_INTAKE_USER", "other-user")
+    monkeypatch.setenv("OTHER_INTAKE_PASSWORD", "other-secret")
+    source_url = "https://snapshot.example"
+    other = Subject(
+        "other",
+        "intake",
+        {
+            "workspace": "other-workspace",
+            "base_url": source_url,
+            "auth": "basic",
+            "intake_path_prefix": "/other/intake",
+            "auth_user_env": "OTHER_INTAKE_USER",
+            "auth_password_env": "OTHER_INTAKE_PASSWORD",
+        },
+    )
+    plain = Subject("plain", "intake", {"workspace": "plain-workspace", "base_url": source_url})
+    duplicate_workspace = Subject("duplicate", "intake", {"workspace": "plain-workspace", "base_url": source_url})
+    built: list[tuple[dict[str, object], object]] = []
+    exports: list[tuple[list[str], object | None]] = []
+    manifest_stats: dict[str, object] = {}
+
+    def fake_builder(**kwargs: object) -> object:
+        client = object()
+        built.append((kwargs, client))
+        return client
+
+    def fake_export(
+        base_url: str,
+        workspaces: list[str],
+        out_dir: Path,
+        *,
+        since: object,
+        client: object | None,
+    ) -> dict:
+        exports.append((workspaces, client))
+        return {
+            "workspaces": {
+                workspace: {"spans": 0, "annotations": 0, "evaluator_results": 0} for workspace in workspaces
+            },
+            "min_start_time": f"2026-07-0{len(exports)}T00:00:00+00:00",
+            "max_start_time": f"2026-07-0{len(exports)}T01:00:00+00:00",
+        }
+
+    def fake_manifest(
+        subjects: list[str],
+        records: list[Path],
+        stats: dict,
+        *,
+        source_url: str,
+        platform_info: dict | None,
+        env: object,
+    ) -> dict:
+        manifest_stats.update(stats)
+        return {}
+
+    monkeypatch.setattr(artifact, "build_basic_auth_intake_client", fake_builder)
+    monkeypatch.setattr(artifact.export, "export_workspaces", fake_export)
+    monkeypatch.setattr(artifact, "build_export_manifest", fake_manifest)
+    monkeypatch.setattr(artifact, "fetch_platform_info", lambda base_url: None)
+    monkeypatch.setattr(artifact.subprocess, "run", lambda *args, **kwargs: None)
+
+    artifact.snapshot_export(
+        [_glamr_subject(base_url=source_url), other, plain, duplicate_workspace],
+        tmp_path / "snapshot.tar.zst",
+        tmp_path / "tmp",
+        since=None,
+    )
+
+    assert [kwargs for kwargs, _ in built] == [
+        {
+            "base_url": source_url,
+            "username": "glamr-user",
+            "password": "glamr-secret",
+            "real_prefix": "/glamr/intake/",
+        },
+        {
+            "base_url": source_url,
+            "username": "other-user",
+            "password": "other-secret",
+            "real_prefix": "/other/intake/",
+        },
+    ]
+    assert exports == [
+        (["default"], built[0][1]),
+        (["other-workspace"], built[1][1]),
+        (["plain-workspace"], None),
+    ]
+    assert set(manifest_stats["workspaces"]) == {"default", "other-workspace", "plain-workspace"}
+    assert manifest_stats["min_start_time"] == "2026-07-01T00:00:00+00:00"
+    assert manifest_stats["max_start_time"] == "2026-07-03T01:00:00+00:00"
diff --git a/plugins/nemo-insights/tests/testbed/test_checked_in_insights.py b/plugins/nemo-insights/tests/testbed/test_checked_in_insights.py
new file mode 100644
index 0000000000..39a49ead56
--- /dev/null
+++ b/plugins/nemo-insights/tests/testbed/test_checked_in_insights.py
@@ -0,0 +1,38 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import hashlib
+
+import pytest
+import yaml
+from testbed import cli, release
+from testbed.registry import load_registry
+
+
+def test_checked_in_insights_match_current_analyst_and_state_pins() -> None:
+    insights_dir = getattr(cli, "INSIGHTS_DIR", cli.HERE / "insights")
+    if not insights_dir.exists():
+        pytest.skip("Task 5 has not generated the checked-in Insights directory")
+
+    manifest_path = insights_dir / "manifest.yaml"
+    assert manifest_path.is_file(), "checked-in Insights directory exists without manifest.yaml"
+
+    subjects = load_registry(cli.REGISTRY_PATH)
+    names = sorted(subjects)
+    manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8"))
+    assert isinstance(manifest, dict)
+    snapshots = manifest.get("snapshots")
+    assert isinstance(snapshots, dict)
+
+    assert {path.name for path in insights_dir.glob("*.yaml")} == {
+        "manifest.yaml",
+        *(f"{name}.yaml" for name in names),
+    }
+    assert set(snapshots) == set(names)
+    for name in names:
+        insights_path = insights_dir / f"{name}.yaml"
+        assert snapshots[name] == {
+            "analyst_sha256": cli._analyst_sha256(),
+            "insights_sha256": hashlib.sha256(insights_path.read_bytes()).hexdigest(),
+            "state": release.lock_ref(cli.HERE / "state.lock", name),
+        }
diff --git a/plugins/nemo-insights/tests/testbed/test_cli.py b/plugins/nemo-insights/tests/testbed/test_cli.py
index d60e119f39..7a982c2d59 100644
--- a/plugins/nemo-insights/tests/testbed/test_cli.py
+++ b/plugins/nemo-insights/tests/testbed/test_cli.py
@@ -15,6 +15,52 @@
 import yaml
 from testbed import cli
 
+_REAL_LOAD_REGISTRY = cli.load_registry
+_REAL_LOCK_REF = cli.release.lock_ref
+
+
+@pytest.fixture(autouse=True)
+def isolate_checked_in_insights(monkeypatch, tmp_path):
+    from testbed.registry import Subject
+
+    def load_registry_with_test_intake(path):
+        subjects = dict(_REAL_LOAD_REGISTRY(path))
+        subjects["nvq"] = Subject(
+            "nvq",
+            "intake",
+            {
+                "agent": "content-dedup",
+                "workspace": "nvq",
+                "base_url": "https://nemo-platform-freeplay.dev.aire.nvidia.com",
+            },
+        )
+        return subjects
+
+    def lock_ref_with_test_intake(path, subject):
+        return "state-v7" if subject == "nvq" else _REAL_LOCK_REF(path, subject)
+
+    helpers = {
+        "check_in": getattr(cli, "_check_in_insights", None),
+        "write_manifest": getattr(cli, "_write_insights_manifest", None),
+    }
+    monkeypatch.setattr(cli, "load_registry", load_registry_with_test_intake)
+    monkeypatch.setattr(cli.release, "lock_ref", lock_ref_with_test_intake)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", tmp_path / "checked-in-insights", raising=False)
+    if helpers["check_in"] is not None:
+        monkeypatch.setattr(
+            cli,
+            "_check_in_insights",
+            lambda subject_name, _source, *, directory=None: (directory or cli.INSIGHTS_DIR) / f"{subject_name}.yaml",
+        )
+    if helpers["write_manifest"] is not None:
+        monkeypatch.setattr(
+            cli,
+            "_write_insights_manifest",
+            lambda *_args, **_kwargs: cli.INSIGHTS_DIR / "manifest.yaml",
+        )
+    return helpers
+
+
 # --------------------------------------------------------------------------- #
 # bundle fixtures: real tar.zst files, kind-switched exactly like production
 # --------------------------------------------------------------------------- #
@@ -172,6 +218,603 @@ async def fake_analyze(self, *, record, since, verbose, out_path):
     assert "Insights written" in out
 
 
+def test_analyze_checks_in_insights_with_spdx_and_provenance_by_default(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    monkeypatch.setenv("INFERENCE_API_KEY", "sk-test")
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    if isolate_checked_in_insights["check_in"] is not None:
+        monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    if isolate_checked_in_insights["write_manifest"] is not None:
+        monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+    monkeypatch.setattr(cli, "_analyst_sha256", lambda: "analyst-digest", raising=False)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "nvq", "--live"])
+
+    async def fake_analyze(self, *, record, since, verbose, out_path):
+        out_path.write_text("insights: []\n", encoding="utf-8")
+        return "REPORT-OK"
+
+    monkeypatch.setattr("testbed.adapters.IntakeAdapter.analyze", fake_analyze)
+    cli.main()
+
+    contents = (
+        "# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. "
+        "All rights reserved.\n"
+        "# SPDX-License-Identifier: Apache-2.0\n"
+        "insights: []\n"
+    )
+    assert (checked_in / "nvq.yaml").read_text(encoding="utf-8") == contents
+    manifest = yaml.safe_load((checked_in / "manifest.yaml").read_text(encoding="utf-8"))
+    assert manifest == {
+        "snapshots": {
+            "nvq": {
+                "analyst_sha256": "analyst-digest",
+                "insights_sha256": hashlib.sha256(contents.encode()).hexdigest(),
+                "state": "live",
+            }
+        }
+    }
+
+
+def test_analyze_checks_in_insights_and_merges_existing_manifest(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    prior = {
+        "analyst_sha256": "old-analyst",
+        "insights_sha256": "old-insights",
+        "state": "state-v1",
+    }
+    (checked_in / "manifest.yaml").write_text(
+        yaml.safe_dump({"snapshots": {"other": prior}}),
+        encoding="utf-8",
+    )
+    (checked_in / "other.yaml").write_text("prior insights\n", encoding="utf-8")
+    monkeypatch.setenv("INFERENCE_API_KEY", "sk-test")
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    if isolate_checked_in_insights["check_in"] is not None:
+        monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    if isolate_checked_in_insights["write_manifest"] is not None:
+        monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+    monkeypatch.setattr(cli, "_analyst_sha256", lambda: "new-analyst", raising=False)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "nvq", "--live"])
+
+    async def fake_analyze(self, *, record, since, verbose, out_path):
+        out_path.write_text("insights: []\n", encoding="utf-8")
+        return "REPORT-OK"
+
+    monkeypatch.setattr("testbed.adapters.IntakeAdapter.analyze", fake_analyze)
+    cli.main()
+
+    snapshots = yaml.safe_load((checked_in / "manifest.yaml").read_text(encoding="utf-8"))["snapshots"]
+    assert snapshots["other"] == prior
+    assert snapshots["nvq"]["state"] == "live"
+    assert (checked_in / "other.yaml").read_text(encoding="utf-8") == "prior insights\n"
+
+
+def test_check_in_single_insights_manifest_failure_leaves_directory_unchanged(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "nvq.yaml").write_text("old insights\n", encoding="utf-8")
+    (checked_in / "other.yaml").write_text("other insights\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    source = tmp_path / "new-insights.yaml"
+    source.write_text("insights: []\n", encoding="utf-8")
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+
+    def fail_manifest(*_args, **_kwargs):
+        raise RuntimeError("manifest failed")
+
+    monkeypatch.setattr(cli, "_write_insights_manifest", fail_manifest)
+
+    with pytest.raises(RuntimeError, match="manifest failed"):
+        cli._check_in_single_insights("nvq", source, "live")
+
+    assert _checked_in_bytes(checked_in) == before
+
+
+def test_analyze_checks_in_insights_requires_source_output(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    monkeypatch.setenv("INFERENCE_API_KEY", "sk-test")
+    monkeypatch.setattr(cli, "TMP", tmp_path / "tmp")
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", tmp_path / "insights")
+    if isolate_checked_in_insights["check_in"] is not None:
+        monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    if isolate_checked_in_insights["write_manifest"] is not None:
+        monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "nvq", "--live"])
+
+    async def fake_analyze(self, *, record, since, verbose, out_path):
+        return "REPORT-OK"
+
+    monkeypatch.setattr("testbed.adapters.IntakeAdapter.analyze", fake_analyze)
+
+    with pytest.raises(SystemExit, match="Analyst did not write"):
+        cli.main()
+
+
+def test_analyze_no_baseline_update_leaves_checked_in_files_untouched(monkeypatch, tmp_path):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    checked_path = checked_in / "nvq.yaml"
+    manifest_path = checked_in / "manifest.yaml"
+    checked_path.write_text("old insights\n", encoding="utf-8")
+    manifest_path.write_text("old manifest\n", encoding="utf-8")
+    monkeypatch.setenv("INFERENCE_API_KEY", "sk-test")
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "nvq", "--live", "--no-baseline-update"])
+
+    async def fake_analyze(self, *, record, since, verbose, out_path):
+        out_path.write_text("insights: []\n", encoding="utf-8")
+        return "REPORT-OK"
+
+    monkeypatch.setattr("testbed.adapters.IntakeAdapter.analyze", fake_analyze)
+    cli.main()
+
+    assert (runtime / "insights_nvq.yaml").read_text(encoding="utf-8") == "insights: []\n"
+    assert checked_path.read_text(encoding="utf-8") == "old insights\n"
+    assert manifest_path.read_text(encoding="utf-8") == "old manifest\n"
+
+
+def _write_analyst_lock(
+    path: Path,
+    *,
+    unrelated_version: str = "1",
+    transitive_version: str = "1",
+    anthropic_version: str = "1",
+) -> None:
+    path.write_text(
+        f"""
+version = 1
+
+[[package]]
+name = "unrelated"
+version = "{unrelated_version}"
+
+[[package]]
+name = "nemo-insights-plugin"
+version = "1"
+dependencies = [
+    {{ name = "pydantic-ai-harness" }},
+    {{ name = "pydantic-ai-slim", extra = ["anthropic"] }},
+]
+
+[[package]]
+name = "pydantic-ai-harness"
+version = "2"
+dependencies = [{{ name = "transitive" }}]
+
+[[package]]
+name = "pydantic-ai-slim"
+version = "3"
+
+[package.optional-dependencies]
+anthropic = [{{ name = "anthropic" }}]
+
+[[package]]
+name = "transitive"
+version = "{transitive_version}"
+
+[[package]]
+name = "anthropic"
+version = "{anthropic_version}"
+""",
+        encoding="utf-8",
+    )
+
+
+def test_analyst_hash_tracks_complete_plugin_source(tmp_path):
+    plugin_root = tmp_path / "nemo_insights_plugin"
+    analyst_root = plugin_root / "analyst"
+    analyst_root.mkdir(parents=True)
+    (analyst_root / "agent.py").write_text("AGENT = 1\n", encoding="utf-8")
+    entities = plugin_root / "entities.py"
+    schema = plugin_root / "schema.py"
+    entities.write_text("ENTITY = 1\n", encoding="utf-8")
+    schema.write_text("SCHEMA = 1\n", encoding="utf-8")
+    lockfile = tmp_path / "uv.lock"
+    _write_analyst_lock(lockfile)
+
+    initial = cli._analyst_sha256(plugin_root, lockfile)
+    entities.write_text("ENTITY = 2\n", encoding="utf-8")
+    assert cli._analyst_sha256(plugin_root, lockfile) != initial
+
+    entities.write_text("ENTITY = 1\n", encoding="utf-8")
+    schema.write_text("SCHEMA = 2\n", encoding="utf-8")
+    assert cli._analyst_sha256(plugin_root, lockfile) != initial
+
+
+def test_analyst_hash_tracks_resolved_dependency_closure(tmp_path):
+    plugin_root = tmp_path / "nemo_insights_plugin"
+    plugin_root.mkdir()
+    (plugin_root / "agent.py").write_text("VALUE = 1\n", encoding="utf-8")
+    lockfile = tmp_path / "uv.lock"
+    _write_analyst_lock(lockfile)
+    initial = cli._analyst_sha256(plugin_root, lockfile)
+
+    _write_analyst_lock(lockfile, unrelated_version="changed")
+    assert cli._analyst_sha256(plugin_root, lockfile) == initial
+
+    _write_analyst_lock(lockfile, unrelated_version="changed", transitive_version="changed")
+    transitive_changed = cli._analyst_sha256(plugin_root, lockfile)
+    assert transitive_changed != initial
+
+    _write_analyst_lock(
+        lockfile,
+        unrelated_version="changed",
+        transitive_version="changed",
+        anthropic_version="changed",
+    )
+    assert cli._analyst_sha256(plugin_root, lockfile) != transitive_changed
+
+
+def test_analyst_hash_defaults_to_platform_paths():
+    assert cli.ANALYST_SOURCE_ROOT == cli.HERE.parent / "src" / "nemo_insights_plugin"
+    assert cli.ANALYST_LOCKFILE == cli.HERE.parents[2] / "uv.lock"
+
+
+def test_analyst_hash_resolves_platform_lock():
+    assert re.fullmatch(r"[0-9a-f]{64}", cli._analyst_sha256())
+
+
+def _analyze_all_subjects():
+    from testbed.registry import Subject
+
+    return {
+        "zeta": Subject("zeta", "intake", {"workspace": "z"}),
+        "alpha": Subject("alpha", "benchmark", {"workspace": "a"}),
+        "producer": Subject("producer", "harbor", {"workspace": "p"}),
+    }
+
+
+def _checked_in_bytes(directory: Path) -> dict[str, bytes]:
+    return {
+        path.relative_to(directory).as_posix(): path.read_bytes() for path in directory.rglob("*") if path.is_file()
+    }
+
+
+def test_analyze_all_runs_every_pinned_analyzable_subject_in_sorted_order(
+    monkeypatch,
+    tmp_path,
+):
+    runtime = tmp_path / "tmp"
+    calls: list[list[str]] = []
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+
+    def fake_run(command, *, check):
+        calls.append(command)
+        name = command[command.index("analyze") + 1]
+        runtime.mkdir(parents=True, exist_ok=True)
+        (runtime / f"insights_{name}.yaml").write_text(f"insights: [{name}]\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "testbed",
+            "analyze",
+            "all",
+            "--no-baseline-update",
+            "--base",
+            "http://platform",
+            "--platform-root",
+            "/platform/root",
+            "--summary-md",
+            "/tmp/summary.md",
+            "--verbose",
+        ],
+    )
+
+    cli.main()
+
+    assert calls == [
+        [
+            sys.executable,
+            "-m",
+            "testbed",
+            "analyze",
+            "alpha",
+            "--no-baseline-update",
+            "--base",
+            "http://platform",
+            "--platform-root",
+            "/platform/root",
+            "--summary-md",
+            "/tmp/summary.md",
+            "--verbose",
+        ],
+        [
+            sys.executable,
+            "-m",
+            "testbed",
+            "analyze",
+            "zeta",
+            "--no-baseline-update",
+            "--base",
+            "http://platform",
+            "--platform-root",
+            "/platform/root",
+            "--summary-md",
+            "/tmp/summary.md",
+            "--verbose",
+        ],
+    ]
+
+
+def test_analyze_all_validates_every_pin_before_subprocess_execution(monkeypatch):
+    calls: list[list[str]] = []
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: None if name == "zeta" else "state-v1")
+    monkeypatch.setattr(cli.subprocess, "run", lambda command, *, check: calls.append(command))
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    with pytest.raises(SystemExit, match="missing: zeta"):
+        cli.main()
+
+    assert calls == []
+
+
+def test_analyze_all_child_failure_leaves_checked_in_directory_unchanged(
+    monkeypatch,
+    tmp_path,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "stale.yaml").write_text("old stale\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        if name == "zeta":
+            raise subprocess.CalledProcessError(7, command)
+        runtime.mkdir(parents=True, exist_ok=True)
+        (runtime / f"insights_{name}.yaml").write_text("new alpha\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    with pytest.raises(SystemExit, match="zeta failed with exit code 7"):
+        cli.main()
+
+    assert _checked_in_bytes(checked_in) == before
+
+
+def test_analyze_all_missing_child_output_leaves_checked_in_directory_unchanged(
+    monkeypatch,
+    tmp_path,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        if name == "alpha":
+            runtime.mkdir(parents=True, exist_ok=True)
+            (runtime / "insights_alpha.yaml").write_text("new alpha\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    with pytest.raises(SystemExit, match="without Insights output for: zeta"):
+        cli.main()
+
+    assert _checked_in_bytes(checked_in) == before
+
+
+def test_analyze_all_stale_output_cannot_mask_missing_child_output(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    runtime = tmp_path / "tmp"
+    runtime.mkdir()
+    stale = runtime / "insights_zeta.yaml"
+    stale.write_text("stale zeta\n", encoding="utf-8")
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+    monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        if name == "alpha":
+            (runtime / "insights_alpha.yaml").write_text("new alpha\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    with pytest.raises(SystemExit, match="without Insights output for: zeta"):
+        cli.main()
+
+    assert _checked_in_bytes(checked_in) == before
+    (backup,) = runtime.glob("backup-*")
+    assert (backup / stale.name).read_text(encoding="utf-8") == "stale zeta\n"
+
+
+def test_analyze_all_top_level_no_baseline_update_skips_promotion(monkeypatch, tmp_path):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        runtime.mkdir(parents=True, exist_ok=True)
+        (runtime / f"insights_{name}.yaml").write_text(f"new {name}\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all", "--no-baseline-update"])
+
+    cli.main()
+
+    assert _checked_in_bytes(checked_in) == before
+
+
+def test_analyze_all_successfully_swaps_complete_set_and_removes_stale_yaml(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "stale.yaml").write_text("old stale\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+    monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+    monkeypatch.setattr(cli, "_analyst_sha256", lambda: "analyst-digest")
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        runtime.mkdir(parents=True, exist_ok=True)
+        (runtime / f"insights_{name}.yaml").write_text(f"insights: [{name}]\n", encoding="utf-8")
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    cli.main()
+
+    assert {path.name for path in checked_in.glob("*.yaml")} == {
+        "alpha.yaml",
+        "zeta.yaml",
+        "manifest.yaml",
+    }
+    snapshots = yaml.safe_load((checked_in / "manifest.yaml").read_text(encoding="utf-8"))["snapshots"]
+    assert snapshots == {
+        "alpha": {
+            "analyst_sha256": "analyst-digest",
+            "insights_sha256": hashlib.sha256((checked_in / "alpha.yaml").read_bytes()).hexdigest(),
+            "state": "state-alpha",
+        },
+        "zeta": {
+            "analyst_sha256": "analyst-digest",
+            "insights_sha256": hashlib.sha256((checked_in / "zeta.yaml").read_bytes()).hexdigest(),
+            "state": "state-zeta",
+        },
+    }
+
+
+def test_analyze_all_swap_failure_rolls_back_old_directory(
+    monkeypatch,
+    tmp_path,
+    isolate_checked_in_insights,
+):
+    runtime = tmp_path / "tmp"
+    checked_in = tmp_path / "insights"
+    checked_in.mkdir()
+    (checked_in / "alpha.yaml").write_text("old alpha\n", encoding="utf-8")
+    (checked_in / "manifest.yaml").write_text("old manifest\n", encoding="utf-8")
+    before = _checked_in_bytes(checked_in)
+    monkeypatch.setattr(cli, "TMP", runtime)
+    monkeypatch.setattr(cli, "INSIGHTS_DIR", checked_in)
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.release, "lock_ref", lambda _path, name: f"state-{name}")
+    monkeypatch.setattr(cli, "_check_in_insights", isolate_checked_in_insights["check_in"])
+    monkeypatch.setattr(cli, "_write_insights_manifest", isolate_checked_in_insights["write_manifest"])
+    monkeypatch.setattr(cli, "_analyst_sha256", lambda: "analyst-digest")
+
+    def fake_run(command, *, check):
+        name = command[command.index("analyze") + 1]
+        runtime.mkdir(parents=True, exist_ok=True)
+        (runtime / f"insights_{name}.yaml").write_text(f"insights: [{name}]\n", encoding="utf-8")
+
+    real_replace = cli.os.replace
+
+    def fail_staging_swap(source, destination):
+        source_path = Path(source)
+        if source_path.is_dir() and source_path.name.startswith(".insights.staging-"):
+            raise OSError("simulated swap failure")
+        real_replace(source, destination)
+
+    monkeypatch.setattr(cli.subprocess, "run", fake_run)
+    monkeypatch.setattr(cli.os, "replace", fail_staging_swap)
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all"])
+
+    with pytest.raises(OSError, match="simulated swap failure"):
+        cli.main()
+
+    assert _checked_in_bytes(checked_in) == before
+
+
+@pytest.mark.parametrize(
+    "flags",
+    [
+        ["--live"],
+        ["--state", "state-v1"],
+        ["--since", "1d"],
+        ["--set", "seed=1"],
+        ["--update-insights"],
+    ],
+)
+def test_analyze_all_rejects_incompatible_flags_before_execution(monkeypatch, flags):
+    calls: list[list[str]] = []
+    monkeypatch.setattr(cli, "load_registry", lambda _path: _analyze_all_subjects())
+    monkeypatch.setattr(cli.subprocess, "run", lambda command, *, check: calls.append(command))
+    monkeypatch.setattr(sys, "argv", ["testbed", "analyze", "all", *flags])
+
+    with pytest.raises(SystemExit, match="cannot be combined"):
+        cli.main()
+
+    assert calls == []
+
+
 def test_missing_registry_exits(monkeypatch, tmp_path):
     monkeypatch.setattr(cli, "REGISTRY_PATH", tmp_path / "nope.toml")
     monkeypatch.setattr(sys, "argv", ["testbed", "list"])
@@ -346,6 +989,50 @@ async def fake_produce(self):
     assert seen["base_url"] == "http://localhost:8080"
 
 
+def test_with_base_drops_remote_auth_by_default() -> None:
+    from testbed.registry import Subject
+
+    glamr = Subject(
+        "glamr",
+        "intake",
+        {
+            "agent": "glamr",
+            "workspace": "default",
+            "base_url": "https://remote",
+            "auth": "basic",
+            "intake_path_prefix": "/api/intake",
+            "auth_user_env": "GLAMR_INTAKE_USER",
+            "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+        },
+    )
+
+    assert cli._with_base(glamr, "http://localhost:8080").config == {
+        "agent": "glamr",
+        "workspace": "default",
+        "base_url": "http://localhost:8080",
+    }
+
+
+def test_with_base_preserves_remote_auth_when_requested() -> None:
+    from testbed.registry import Subject
+
+    glamr = Subject(
+        "glamr",
+        "intake",
+        {
+            "agent": "glamr",
+            "workspace": "default",
+            "base_url": "https://remote",
+            "auth": "basic",
+            "intake_path_prefix": "/api/intake",
+            "auth_user_env": "GLAMR_INTAKE_USER",
+            "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+        },
+    )
+
+    assert cli._with_base(glamr, "https://remote", drop_auth=False).config["auth"] == "basic"
+
+
 def test_analyze_live_base_overrides_stanza(monkeypatch, tmp_path):
     monkeypatch.setattr(cli, "TMP", tmp_path)
     monkeypatch.setenv("INFERENCE_API_KEY", "sk")
@@ -361,6 +1048,65 @@ async def fake_analyze(self, *, record, since, verbose, out_path):
     assert seen["base_url"] == "http://localhost:8080"
 
 
+def test_analyze_glamr_live_base_preserves_remote_auth(monkeypatch, tmp_path):
+    from testbed.adapters import IntakeAdapter, TestbedAdapter
+    from testbed.registry import Subject
+
+    glamr = Subject(
+        "glamr",
+        "intake",
+        {
+            "agent": "glamr",
+            "workspace": "default",
+            "base_url": "https://original.example",
+            "auth": "basic",
+            "intake_path_prefix": "/glamr/intake",
+            "auth_user_env": "GLAMR_INTAKE_USER",
+            "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+        },
+    )
+    monkeypatch.setattr(cli, "load_registry", lambda _path: {"glamr": glamr})
+    monkeypatch.setattr(cli, "_load_dotenv", lambda *args, **kwargs: None)
+    monkeypatch.setattr(cli, "TMP", tmp_path)
+    monkeypatch.setenv("INFERENCE_API_KEY", "sk-test")
+    monkeypatch.setenv("GLAMR_INTAKE_USER", "intake-user")
+    monkeypatch.setenv("GLAMR_INTAKE_PASSWORD", "secret")
+    built: dict[str, object] = {}
+    real_build_adapter = cli.build_adapter
+
+    def capture_build_adapter(subject: Subject) -> TestbedAdapter:
+        built.update(subject.config)
+        return real_build_adapter(subject)
+
+    async def fake_analyze(
+        self: IntakeAdapter,
+        *,
+        record: dict[str, object] | None,
+        since: datetime | None,
+        verbose: bool,
+        out_path: Path,
+    ) -> str:
+        return "REPORT-OK"
+
+    monkeypatch.setattr(cli, "build_adapter", capture_build_adapter)
+    monkeypatch.setattr(IntakeAdapter, "analyze", fake_analyze)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["testbed", "analyze", "glamr", "--live", "--base", "https://override.example"],
+    )
+
+    cli.main()
+
+    assert built["base_url"] == "https://override.example"
+    assert {key: built[key] for key in cli._REMOTE_AUTH_KEYS} == {
+        "auth": "basic",
+        "intake_path_prefix": "/glamr/intake",
+        "auth_user_env": "GLAMR_INTAKE_USER",
+        "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+    }
+
+
 def test_analyze_live_base_retargets_record(monkeypatch, tmp_path):
     """--live --base must also retarget the benchmark run record."""
     from testbed.runstore import save_run
@@ -1181,6 +1927,63 @@ def test_snapshot_base_overrides_source_url(monkeypatch, tmp_path):
     assert seen["base_urls"] == ["http://ci-host:8080", "http://ci-host:8080"]
 
 
+def test_snapshot_glamr_base_preserves_remote_auth(monkeypatch, tmp_path):
+    from testbed.registry import Subject
+
+    glamr = Subject(
+        "glamr",
+        "intake",
+        {
+            "agent": "glamr",
+            "workspace": "default",
+            "base_url": "https://original.example",
+            "auth": "basic",
+            "intake_path_prefix": "/glamr/intake",
+            "auth_user_env": "GLAMR_INTAKE_USER",
+            "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+        },
+    )
+    monkeypatch.setattr(cli, "load_registry", lambda _path: {"glamr": glamr})
+    monkeypatch.setattr(cli, "TMP", tmp_path)
+    captured: dict[str, object] = {}
+
+    def fake_snapshot_export(
+        subjects: list[Subject],
+        out: Path,
+        tmp_dir: Path,
+        *,
+        since: datetime | None,
+    ) -> Path:
+        (subject,) = subjects
+        captured.update(subject.config)
+        return out
+
+    monkeypatch.setattr(cli.artifact, "snapshot_export", fake_snapshot_export)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        [
+            "testbed",
+            "snapshot",
+            "glamr",
+            "--base",
+            "https://override.example",
+            "-o",
+            str(tmp_path / "glamr.tar.zst"),
+        ],
+    )
+
+    cli.main()
+
+    assert captured["base_url"] == "https://override.example"
+    assert {key: captured[key] for key in cli._REMOTE_AUTH_KEYS} == {
+        "auth": "basic",
+        "intake_path_prefix": "/glamr/intake",
+        "auth_user_env": "GLAMR_INTAKE_USER",
+        "auth_password_env": "GLAMR_INTAKE_PASSWORD",
+    }
+
+
 def test_snapshot_subjects_json_for_ci(monkeypatch, tmp_path):
     seen: dict = {}
     monkeypatch.setattr(cli, "TMP", tmp_path)
diff --git a/plugins/nemo-insights/tests/testbed/test_export.py b/plugins/nemo-insights/tests/testbed/test_export.py
index 43bff81542..0635877aba 100644
--- a/plugins/nemo-insights/tests/testbed/test_export.py
+++ b/plugins/nemo-insights/tests/testbed/test_export.py
@@ -175,6 +175,14 @@ def test_export_closes_client(tmp_path, monkeypatch):
     assert client.closed
 
 
+def test_export_closes_injected_client(tmp_path):
+    client = FakeClient({})
+
+    export.export_workspaces("http://localhost:8080", ["ws-a"], tmp_path, since=None, client=client)
+
+    assert client.closed
+
+
 # --------------------------------------------------------------------------- #
 # subject scoping
 # --------------------------------------------------------------------------- #
@@ -340,10 +348,11 @@ def boom(url, timeout):
 
 
 def _fake_export(seen):
-    def fake(base_url, workspaces, out_dir, *, since):
+    def fake(base_url, workspaces, out_dir, *, since, client=None):
         seen["base_url"] = base_url
         seen["workspaces"] = list(workspaces)
         seen["since"] = since
+        seen["client"] = client
         for ws in workspaces:
             ws_dir = out_dir / "export" / ws
             ws_dir.mkdir(parents=True, exist_ok=True)
diff --git a/plugins/nemo-insights/tests/testbed/test_intake_client.py b/plugins/nemo-insights/tests/testbed/test_intake_client.py
new file mode 100644
index 0000000000..aa7e0f320a
--- /dev/null
+++ b/plugins/nemo-insights/tests/testbed/test_intake_client.py
@@ -0,0 +1,97 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""The basic-auth intake client builder: path rewrite + basic auth."""
+
+import base64
+
+import httpx
+import pytest
+from nemo_platform import AsyncNeMoPlatform
+from testbed import intake_client as intake_client_module
+from testbed.intake_client import build_basic_auth_intake_client, build_rewriting_http_client
+
+
+async def test_rewrites_sdk_prefix_and_attaches_basic_auth() -> None:
+    seen: dict[str, str | None] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        seen["auth"] = request.headers.get("authorization")
+        return httpx.Response(200, json={"data": []})
+
+    client = build_rewriting_http_client(
+        username="intake",
+        password="secret",
+        transport=httpx.MockTransport(handler),
+    )
+    try:
+        response = await client.get("https://agenthub.aire.nvidia.com/apis/intake/v2/workspaces/default/spans?page=1")
+    finally:
+        await client.aclose()
+
+    assert response.status_code == 200
+    assert seen["url"] == "https://agenthub.aire.nvidia.com/api/intake/v2/workspaces/default/spans?page=1"
+    assert seen["auth"] == f"Basic {base64.b64encode(b'intake:secret').decode()}"
+
+
+async def test_rewriting_http_client_uses_60_second_timeout() -> None:
+    client = build_rewriting_http_client(username="u", password="p")
+    try:
+        assert client.timeout == httpx.Timeout(60.0)
+    finally:
+        await client.aclose()
+
+
+async def test_leaves_non_intake_paths_untouched() -> None:
+    seen: dict[str, str | None] = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen["url"] = str(request.url)
+        return httpx.Response(200, json={})
+
+    client = build_rewriting_http_client(
+        username="u",
+        password="p",
+        transport=httpx.MockTransport(handler),
+    )
+    try:
+        await client.get("https://agenthub.aire.nvidia.com/other/path")
+    finally:
+        await client.aclose()
+
+    assert seen["url"] == "https://agenthub.aire.nvidia.com/other/path"
+
+
+async def test_build_basic_auth_intake_client_returns_sdk_client() -> None:
+    client = build_basic_auth_intake_client(
+        base_url="https://agenthub.aire.nvidia.com",
+        username="u",
+        password="p",
+    )
+    try:
+        assert isinstance(client, AsyncNeMoPlatform)
+    finally:
+        await client.close()
+
+
+async def test_sdk_client_closes_its_http_client(monkeypatch: pytest.MonkeyPatch) -> None:
+    http_client = httpx.AsyncClient(transport=httpx.MockTransport(lambda request: httpx.Response(200)))
+    monkeypatch.setattr(
+        intake_client_module,
+        "build_rewriting_http_client",
+        lambda **kwargs: http_client,
+    )
+    client = build_basic_auth_intake_client(
+        base_url="https://agenthub.aire.nvidia.com",
+        username="u",
+        password="p",
+    )
+
+    try:
+        assert not http_client.is_closed
+        await client.close()
+        assert http_client.is_closed
+    finally:
+        if not http_client.is_closed:
+            await http_client.aclose()
diff --git a/plugins/nemo-insights/tests/testbed/test_publish.py b/plugins/nemo-insights/tests/testbed/test_publish.py
index d4f3e50081..5715192aba 100644
--- a/plugins/nemo-insights/tests/testbed/test_publish.py
+++ b/plugins/nemo-insights/tests/testbed/test_publish.py
@@ -366,3 +366,4 @@ def test_workflow_protects_all_secrets_and_exports_state_repository():
     assert "    environment: insights-testbed\n" in produce_job
     assert "    environment: insights-testbed\n" in analyze_job
     assert "TESTBED_STATE_REPO: ${{ vars.TESTBED_STATE_REPO || 'NVIDIA-dev/NeMo-Optimizer' }}" in workflow
+    assert 'testbed analyze "$SUBJECT" ${STATE:+--state "$STATE"} --no-baseline-update ' in analyze_job
diff --git a/plugins/nemo-insights/tests/testbed/test_registry.py b/plugins/nemo-insights/tests/testbed/test_registry.py
index be4f68bda6..dd7b286aa3 100644
--- a/plugins/nemo-insights/tests/testbed/test_registry.py
+++ b/plugins/nemo-insights/tests/testbed/test_registry.py
@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 from pathlib import Path
 
+from testbed import cli, release
 from testbed.registry import load_registry
 
 
@@ -22,3 +23,57 @@ def test_per_subject_override_wins(tmp_path: Path):
     toml = tmp_path / "t.toml"
     toml.write_text('base_url = "https://shared"\n\n[local]\ntype = "intake"\nbase_url = "http://localhost:8000"\n')
     assert load_registry(toml)["local"].config["base_url"] == "http://localhost:8000"
+
+
+def test_registry_contains_only_expected_analyzable_subjects() -> None:
+    subjects = load_registry(cli.REGISTRY_PATH)
+
+    assert set(subjects) == {
+        "glamr",
+        "nemo-oo-airline",
+        "nvq",
+        "tau2-airline",
+        "tau2-retail",
+        "tau2-telecom",
+    }
+    assert all(subject.type in ("benchmark", "intake") for subject in subjects.values())
+    assert subjects["nvq"].config["agent"] == "content-dedup"
+
+
+def test_glamr_stores_credential_environment_names_only() -> None:
+    glamr = load_registry(cli.REGISTRY_PATH)["glamr"]
+
+    assert glamr.config["auth_user_env"] == "GLAMR_INTAKE_USER"
+    assert glamr.config["auth_password_env"] == "GLAMR_INTAKE_PASSWORD"
+    assert "auth_user" not in glamr.config
+    assert "auth_password" not in glamr.config
+
+
+def test_nemo_oo_airline_is_an_intake_subject() -> None:
+    subject = load_registry(cli.REGISTRY_PATH)["nemo-oo-airline"]
+
+    assert subject.type == "intake"
+    assert subject.config["agent"] == "nemo-oo-airline"
+
+
+def test_tau2_telecom_uses_small_split() -> None:
+    telecom = load_registry(cli.REGISTRY_PATH)["tau2-telecom"]
+
+    assert telecom.type == "benchmark"
+    assert telecom.config["domain"] == "telecom"
+    assert telecom.config["task_split_name"] == "small"
+
+
+def test_every_analyzable_subject_has_expected_state_pin() -> None:
+    expected = {
+        "glamr": "state-v8",
+        "nemo-oo-airline": "state-v9",
+        "nvq": "state-v7",
+        "tau2-airline": "state-v6",
+        "tau2-retail": "state-v10",
+        "tau2-telecom": "state-v10",
+    }
+
+    assert {
+        name: release.lock_ref(cli.HERE / "state.lock", name) for name in sorted(load_registry(cli.REGISTRY_PATH))
+    } == expected
diff --git a/plugins/nemo-insights/tests/testbed/test_reingest.py b/plugins/nemo-insights/tests/testbed/test_reingest.py
index 41885a1509..d56a220da8 100644
--- a/plugins/nemo-insights/tests/testbed/test_reingest.py
+++ b/plugins/nemo-insights/tests/testbed/test_reingest.py
@@ -202,6 +202,79 @@ def test_iso_to_ns_treats_naive_as_utc():
 # --- build_trace_request ---
 
 
+def test_build_trace_requests_respects_serialized_size(monkeypatch):
+    """ByteSize() drives batching: a small limit splits spans across requests."""
+    monkeypatch.setattr(reingest, "OTLP_REQUEST_MAX_BYTES", 512)
+    docs = [{**AGENT_DOC, "span_id": f"{i:016x}"} for i in range(5)]
+    requests = reingest.build_trace_requests(docs, CATALOG)
+    assert len(requests) > 1
+    for request in requests:
+        assert request.ByteSize() <= 512
+        assert len(request.resource_spans[0].scope_spans[0].spans) >= 1
+
+
+def test_build_trace_requests_accepts_exact_serialized_size_limit():
+    one_span_size = reingest.build_trace_request([reingest.doc_to_otlp(AGENT_DOC, CATALOG)]).ByteSize()
+
+    requests = reingest.build_trace_requests([AGENT_DOC], CATALOG, max_bytes=one_span_size)
+
+    assert len(requests) == 1
+    assert requests[0].ByteSize() == one_span_size
+    with pytest.raises(RuntimeError, match=rf"exceeds {one_span_size - 1} bytes"):
+        reingest.build_trace_requests([AGENT_DOC], CATALOG, max_bytes=one_span_size - 1)
+
+
+def test_build_trace_requests_respects_span_count_limit(monkeypatch):
+    monkeypatch.setattr(reingest, "OTLP_REQUEST_MAX_SPANS", 2)
+    docs = [{**AGENT_DOC, "span_id": f"{i:016x}"} for i in range(5)]
+    requests = reingest.build_trace_requests(docs, CATALOG)
+    sizes = [len(req.resource_spans[0].scope_spans[0].spans) for req in requests]
+    assert sizes == [2, 2, 1]
+
+
+def test_build_trace_requests_rejects_oversized_single_span():
+    huge = {**AGENT_DOC, "raw_attributes": json.dumps({"payload": "x" * (5 * 1024 * 1024)})}
+    with pytest.raises(RuntimeError, match="exceeds"):
+        reingest.build_trace_requests([huge], CATALOG)
+
+
+def test_ingest_bundle_splits_on_serialized_size(tmp_path, quiet_platform, monkeypatch):
+    """Small byte limit forces multiple OTLP posts even when span count is low."""
+    monkeypatch.setattr(reingest, "OTLP_REQUEST_MAX_BYTES", 512)
+    docs = [{**AGENT_DOC, "span_id": f"{i:016x}"} for i in range(5)]
+    export_dir = _write_export(tmp_path, "ws-a", docs)
+    quiet_platform["span_counts"] = [0, 5]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0]
+    reingest.ingest_bundle(
+        "http://x",
+        export_dir,
+        _manifest("ws-a", 5),
+        workspace_map={"ws-a": "ws-b"},
+        catalog=CATALOG,
+        sleep=lambda s: None,
+    )
+    assert len(quiet_platform["requests"]) > 1
+
+
+def test_ingest_bundle_oversized_span_raises_before_post(tmp_path, quiet_platform):
+    """One span above the byte limit must fail before the first export_trace_request call."""
+    huge = {**AGENT_DOC, "raw_attributes": json.dumps({"payload": "x" * (5 * 1024 * 1024)})}
+    export_dir = _write_export(tmp_path, "ws-a", [huge])
+    quiet_platform["span_counts"] = [0]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0]
+    with pytest.raises(RuntimeError, match="exceeds"):
+        reingest.ingest_bundle(
+            "http://x",
+            export_dir,
+            _manifest("ws-a", 1),
+            workspace_map={"ws-a": "ws-b"},
+            catalog=CATALOG,
+        )
+    assert quiet_platform["requests"] == []
+
+
 def test_build_request_groups_by_scope_and_encodes_protocol_fields():
     spans = [reingest.doc_to_otlp(AGENT_DOC, CATALOG), reingest.doc_to_otlp(LLM_DOC, CATALOG)]
     spans.append({**spans[1], "span_id": "aabbccdd11223344", "scope": None, "status_error": True})
@@ -625,7 +698,7 @@ def test_ingest_bundle_zero_ingests_everything(tmp_path, quiet_platform):
 
 
 def test_ingest_bundle_batches_spans(tmp_path, quiet_platform, monkeypatch):
-    monkeypatch.setattr(reingest, "SPAN_BATCH", 2)
+    monkeypatch.setattr(reingest, "OTLP_REQUEST_MAX_SPANS", 2)
     export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC, LLM_DOC, {**LLM_DOC, "span_id": "aabbccdd11223344"}])
     quiet_platform["span_counts"] = [0, 3]
     quiet_platform["annotation_counts"] = [0]
diff --git a/plugins/nemo-insights/tests/testbed/test_tau2run.py b/plugins/nemo-insights/tests/testbed/test_tau2run.py
index 15acae7ff7..259461ce2f 100644
--- a/plugins/nemo-insights/tests/testbed/test_tau2run.py
+++ b/plugins/nemo-insights/tests/testbed/test_tau2run.py
@@ -131,6 +131,34 @@ def test_read_policy_flat_and_nested_and_absent(tmp_path):
     assert read_policy(tmp_path, "telecom") is None
 
 
+def test_read_policy_main_policy_fallback_flat_and_nested(tmp_path):
+    """Telecom uses main_policy.md; policy.md wins when both exist."""
+    flat = tmp_path / "domains" / "telecom"
+    flat.mkdir(parents=True)
+    (flat / "main_policy.md").write_text("TELECOM FLAT MAIN")
+    nested = tmp_path / "tau2" / "domains" / "telecom-nested"
+    nested.mkdir(parents=True)
+    (nested / "main_policy.md").write_text("TELECOM NESTED MAIN")
+    both = tmp_path / "domains" / "telecom-both"
+    both.mkdir(parents=True)
+    (both / "policy.md").write_text("PRIMARY")
+    (both / "main_policy.md").write_text("FALLBACK")
+    assert read_policy(tmp_path, "telecom") == "TELECOM FLAT MAIN"
+    assert read_policy(tmp_path, "telecom-nested") == "TELECOM NESTED MAIN"
+    assert read_policy(tmp_path, "telecom-both") == "PRIMARY"
+
+
+def test_read_policy_nested_layout_precedes_flat_filename_priority(tmp_path):
+    nested = tmp_path / "tau2" / "domains" / "telecom"
+    nested.mkdir(parents=True)
+    (nested / "main_policy.md").write_text("NESTED MAIN")
+    flat = tmp_path / "domains" / "telecom"
+    flat.mkdir(parents=True)
+    (flat / "policy.md").write_text("FLAT POLICY")
+
+    assert read_policy(tmp_path, "telecom") == "NESTED MAIN"
+
+
 def test_resolve_paths_from_repo_absolute():
     tau2_bin, data_dir = resolve_paths({"tau2_repo": "/r"}, repo_root=Path("/root"))
     assert tau2_bin == str(Path("/r") / ".venv" / "bin" / "tau2")