diff --git a/.github/workflows/insights-testbed.yml b/.github/workflows/insights-testbed.yml
index ee8a82c6ac..d9f4389de0 100644
--- a/.github/workflows/insights-testbed.yml
+++ b/.github/workflows/insights-testbed.yml
@@ -24,10 +24,6 @@ on:
         description: "Override subject num_trials (empty = testbeds.toml value)"
         type: string
         default: ""
-      publish_state:
-        description: "produce only: upload the new state version to the testbed-state release"
-        type: boolean
-        default: true
       reason:
         description: "produce only: why this fixture exists (one line for the release catalog)"
         type: string
@@ -43,6 +39,7 @@ on:
 env:
   TAU2_JUDGE_LLM: ${{ vars.TAU2_JUDGE_LLM || 'openai/nvidia/nvidia/evals-nemotron-ultra' }}
   TAU2_BENCH_REF: 8ebb7499622fc2be9b9d510d6f7a7653461f4f29
+  TESTBED_STATE_REPO: ${{ vars.TESTBED_STATE_REPO || 'NVIDIA-dev/NeMo-Optimizer' }}
 
 jobs:
   plugin-tests:
@@ -108,9 +105,11 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 180
     concurrency: { group: testbed-state-produce, cancel-in-progress: false }
-    permissions: { contents: write }   # release asset upload
+    permissions: { contents: read }
+    # Keep these credentials exclusively in this required-review environment,
+    # never as repository- or organization-level Actions secrets.
+    environment: insights-testbed
     env:
-      GH_TOKEN: ${{ github.token }}
       INFERENCE_API_KEY: ${{ secrets.NVIDIA_INFERENCE_KEY }}
       # Same gateway key; litellm/tau2 read it under the OpenAI-conventional name.
       OPENAI_API_KEY: ${{ secrets.NVIDIA_INFERENCE_KEY }}
@@ -150,7 +149,8 @@ jobs:
       - name: Round-trip fidelity guard
         working-directory: nemo-platform/plugins/nemo-insights
         # Re-ingest the candidate into scratch workspaces on the in-job stack, re-export,
-        # doc-diff; any mismatch fails the job before the candidate can be published.
+        # doc-diff; any mismatch fails the job and marks the uploaded candidate unverified.
+        # Maintainers publish manually only after confirming this step passed.
         # platform-root auto-resolves to the containing nemo-platform checkout.
         run: uv run --project ../.. python -m testbed roundtrip "$RUNNER_TEMP/bundles/candidate.tar.zst" --base http://localhost:8080
       - name: Upload state artifact
@@ -158,17 +158,10 @@ jobs:
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           # run_attempt suffix: upload-artifact@v4 409s on duplicate names within a
-          # run, which would fail this always() step on a re-run and block publish.
+          # run, which would otherwise hide the diagnostic candidate on a re-run.
           name: state-candidate-${{ github.run_id }}-${{ github.run_attempt }}
           path: ${{ runner.temp }}/bundles/candidate.tar.zst
           if-no-files-found: warn
-      - name: Publish to testbed-state release
-        if: success() && inputs.publish_state
-        working-directory: nemo-platform/plugins/nemo-insights
-        env:
-          REASON: ${{ inputs.reason }}
-        # --no-verify: the round-trip fidelity guard already ran as its own step above.
-        run: uv run --project ../.. python -m testbed publish "$RUNNER_TEMP/bundles/candidate.tar.zst" --reason "$REASON" --no-verify
       - name: Platform log on failure
         if: failure()
         run: tail -200 "$RUNNER_TEMP/platform.log" || true
@@ -184,6 +177,9 @@ jobs:
        (github.event.action == 'labeled' &&
         github.event.label.name == 'run-insights' &&
         github.event.pull_request.head.repo.full_name == github.repository))
+    # Keep these credentials exclusively in this required-review environment:
+    # analyze executes same-repository PR code with inference and fixture-read secrets.
+    environment: insights-testbed
     runs-on: ubuntu-latest
     timeout-minutes: 60
     permissions: { contents: read }
@@ -192,11 +188,14 @@ jobs:
       matrix:
         subject: ${{ fromJSON(needs.plan.outputs.subjects) }}
     env:
-      GH_TOKEN: ${{ github.token }}
       INFERENCE_API_KEY: ${{ secrets.NVIDIA_INFERENCE_KEY }}
     steps:
       - name: Require secrets
-        run: '[ -n "$INFERENCE_API_KEY" ] || { echo "secret NVIDIA_INFERENCE_KEY is not set"; exit 1; }'
+        env:
+          GH_TOKEN: ${{ secrets.TESTBED_STATE_GH_READ_TOKEN }}
+        run: |
+          [ -n "$GH_TOKEN" ] || { echo "secret TESTBED_STATE_GH_READ_TOKEN is not set"; exit 1; }
+          [ -n "$INFERENCE_API_KEY" ] || { echo "secret NVIDIA_INFERENCE_KEY is not set"; exit 1; }
       - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
         with: { path: nemo-platform, persist-credentials: false }
       - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
@@ -211,6 +210,7 @@ jobs:
       - name: Generate insights
         working-directory: nemo-platform/plugins/nemo-insights
         env:
+          GH_TOKEN: ${{ secrets.TESTBED_STATE_GH_READ_TOKEN }}
           SUBJECT: ${{ matrix.subject }}
           STATE: ${{ inputs.state }}
         # Empty STATE = bare analyze = the subject's state.lock pin (the
diff --git a/plugins/nemo-insights/README.md b/plugins/nemo-insights/README.md
index f783ec02eb..cc3a9d28a4 100644
--- a/plugins/nemo-insights/README.md
+++ b/plugins/nemo-insights/README.md
@@ -12,6 +12,41 @@ The plugin is intentionally not part of `enabled-plugins`.
 
 ## CLI
 
+From an agent directory, Insights discovers `optimizer.yaml` in the current
+directory or its parents. Start by checking the profile and its environment,
+then run analysis:
+
+```bash
+cd <agent-directory>
+uv run nemo insights doctor
+uv run nemo insights analyze
+```
+
+The profile contract consumed by Insights is deliberately small:
+
+```yaml
+agent: research-agent
+agent_spec: AGENT-SPEC.md  # optional
+workspace: default         # optional; defaults to "default"
+```
+
+Only `agent`, `agent_spec`, and `workspace` are consumed by Insights.
+Unknown experiment-owned fields are ignored, while the reserved `profile_dir`
+field is rejected. `agent` is required. Relative `agent_spec` paths are
+resolved relative to the profile. When it is omitted, Insights looks for
+`AGENT-SPEC.md`, then `README.md`, beside the profile.
+
+An adjacent `.env` is loaded when a profile is found, without replacing
+variables already set in the shell. For this shared profile workflow,
+`NMP_BASE_URL` is the only base-URL environment variable. Resolution order is
+explicit command-line flags, then profile values (for `agent`, `agent_spec`,
+and `workspace`) or `NMP_BASE_URL` (for the base URL), then the built-in
+defaults. `--base-url` takes precedence over `NMP_BASE_URL`.
+
+With a discovered profile, analysis reads and writes the shared local output at
+`.nemo-optimizer/insights.yaml` beside `optimizer.yaml`. Pass
+`--insights-file-output` to use a different file explicitly.
+
 ```bash
 uv run nemo insights analyze \
   --agent research-agent \
diff --git a/plugins/nemo-insights/pyproject.toml b/plugins/nemo-insights/pyproject.toml
index af78dfe97a..c1538d7e8f 100644
--- a/plugins/nemo-insights/pyproject.toml
+++ b/plugins/nemo-insights/pyproject.toml
@@ -45,7 +45,3 @@ packages = ["src/nemo_insights_plugin"]
 asyncio_mode = "auto"
 pythonpath = ["src", "."]
 testpaths = ["tests"]
-
-[tool.uv.sources]
-nemo-platform = { workspace = true }
-nemo-platform-plugin = { workspace = true }
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/analyst_backend.py b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/analyst_backend.py
index a47833de98..12f43d213c 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/analyst_backend.py
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/analyst_backend.py
@@ -473,12 +473,14 @@ def __init__(self, *, client: AsyncNeMoPlatform, path: Path) -> None:
     def _read_records(self) -> list[dict]:
         if not self.path.exists():
             return []
-        raw = yaml.safe_load(self.path.read_text()) or {}
+        raw = yaml.safe_load(self.path.read_text(encoding="utf-8")) or {}
         return list(raw.get("insights", []))
 
     def _write_records(self, records: list[dict]) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
-        self.path.write_text(yaml.safe_dump({"insights": records}, sort_keys=False, allow_unicode=True))
+        self.path.write_text(
+            yaml.safe_dump({"insights": records}, sort_keys=False, allow_unicode=True), encoding="utf-8"
+        )
 
     async def persist_result(self, *, workspace: str, agent: str, result: AnalystResult) -> str:
         records = self._read_records()
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 249921baf4..f8a9e7cde8 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/analyst/run.py
@@ -30,6 +30,10 @@
 _VERBOSE_TRUNCATE = 2000
 
 
+class ClientConstructionError(Exception):
+    """The analyst's NeMo Platform client could not be constructed."""
+
+
 async def run_analyst(
     *,
     agent: str,
@@ -56,7 +60,10 @@ async def run_analyst(
         since: Optional incremental lower bound enforced on trace/span reads.
         evaluation_id: Optional run scope; AND-pinned onto every span read.
     """
-    client = make_client(base_url)
+    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(
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py b/plugins/nemo-insights/src/nemo_insights_plugin/cli.py
index a78c310bb3..36475eb69a 100644
--- a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/cli.py
@@ -6,17 +6,178 @@
 import asyncio
 import json
 import os
+from dataclasses import dataclass
 from importlib.metadata import entry_points
 from pathlib import Path
 from typing import ClassVar
 
+import httpx
 import typer
-from nemo_insights_plugin.analyst.run import run_analyst
+from nemo_insights_plugin.analyst.run import ClientConstructionError, run_analyst
 from nemo_insights_plugin.client import make_client
+from nemo_insights_plugin.contracts.checks import CheckResult, advisories, format_report, required_failures
+from nemo_insights_plugin.contracts.insights import InsightsFileError, validate_insights_file
+from nemo_insights_plugin.contracts.profile import (
+    DEFAULT_BASE_URL,
+    EnvFileError,
+    ProfileError,
+    discover_profile,
+    load_env_file,
+    resolve_base_url,
+)
+from nemo_insights_plugin.preflight import (
+    AnalysisProbes,
+    check_environment,
+    check_profile,
+    read_agent_spec,
+)
+from nemo_insights_plugin.profile import AnalysisProfile, load_profile, pick_agent_spec
+from nemo_platform import NeMoPlatformError
 from nemo_platform_plugin.cli import NemoCLI
+from pydantic_ai import AgentRunError
 
-DEFAULT_BASE_URL = "http://localhost:8080"
 DEFAULT_WORKSPACE = "default"
+_PREFLIGHT_PROBES: AnalysisProbes | None = None
+
+
+@dataclass(frozen=True)
+class _ResolvedAnalysis:
+    agent: str
+    agent_spec: str | None
+    workspace: str
+    base_url: str
+    insights_output: Path | None
+    profile_output: Path | None
+    profile_dir: Path | None
+    spec_checks: tuple[CheckResult, ...]
+
+
+def _load_profile_or_error(profile_path: Path | None) -> tuple[AnalysisProfile | None, str | None]:
+    """Load an explicit or discovered profile, preserving non-explicit failures."""
+    found = profile_path or discover_profile()
+    if found is None:
+        return None, None
+    try:
+        profile = load_profile(found)
+    except ProfileError as exc:
+        if profile_path is not None:
+            raise
+        return None, str(exc)
+    if profile_path is None:
+        typer.echo(f"Using profile: {found} (agent: {profile.agent})", err=True)
+    loaded = load_env_file(found.parent / ".env")
+    if loaded:
+        typer.echo(f"Loaded .env from {found.parent / '.env'} ({len(loaded)} vars)", err=True)
+    return profile, None
+
+
+def _preflight_or_exit(checks: list[CheckResult]) -> None:
+    """Print blockers and stop before an analyst run."""
+    if required_failures(checks):
+        typer.echo(format_report(checks), err=True)
+        raise typer.Exit(code=1)
+    warnings = advisories(checks)
+    if warnings:
+        typer.echo(format_report(warnings), err=True)
+
+
+def _one_line_error(exc: BaseException) -> str:
+    """Collapse expected CLI failures to one readable terminal line."""
+    return " ".join(str(exc).splitlines()).strip() or type(exc).__name__
+
+
+def _resolve_analysis(
+    *,
+    agent: str | None,
+    agent_spec: Path | None,
+    workspace: str | None,
+    base_url: str | None,
+    profile_path: Path | None,
+    insights_output: Path | None,
+) -> _ResolvedAnalysis:
+    profile, profile_error = _load_profile_or_error(profile_path)
+    if profile_error is not None:
+        if agent is None or workspace is None:
+            raise ProfileError(profile_error)
+        typer.echo(f"warning: ignoring discovered profile: {profile_error}", err=True)
+
+    resolved_agent = agent or (profile.agent if profile is not None else None)
+    if resolved_agent is None:
+        raise ProfileError(
+            "No --agent given and no optimizer.yaml profile found. Pass --agent or run from a directory with a profile."
+        )
+    resolved_workspace = workspace or (profile.workspace if profile is not None else DEFAULT_WORKSPACE)
+
+    spec_path = agent_spec
+    spec_error: str | None = None
+    if spec_path is None and profile is not None:
+        try:
+            spec_path = pick_agent_spec(profile)
+        except ProfileError as exc:
+            spec_error = str(exc)
+    spec_content, spec_checks = read_agent_spec(spec_path, spec_error)
+
+    resolved_base_url = resolve_base_url(base_url)
+    profile_output = None
+    if insights_output is None and profile is not None:
+        profile_output = profile.profile_dir / ".nemo-optimizer" / "insights.yaml"
+    resolved_output = insights_output if insights_output is not None else profile_output
+    validate_insights_file(resolved_output)
+
+    return _ResolvedAnalysis(
+        agent=resolved_agent,
+        agent_spec=spec_content,
+        workspace=resolved_workspace,
+        base_url=resolved_base_url,
+        insights_output=resolved_output,
+        profile_output=profile_output,
+        profile_dir=profile.profile_dir if profile is not None else None,
+        spec_checks=tuple(spec_checks),
+    )
+
+
+async def _run_analysis(analysis: _ResolvedAnalysis, *, verbose: bool) -> str:
+    checks = list(analysis.spec_checks)
+    checks.extend(
+        await check_environment(
+            agent=analysis.agent,
+            workspace=analysis.workspace,
+            base_url=analysis.base_url,
+            profile_dir=analysis.profile_dir,
+            probes=_PREFLIGHT_PROBES,
+        )
+    )
+    _preflight_or_exit(checks)
+
+    if analysis.profile_output is not None:
+        analysis.profile_output.parent.mkdir(parents=True, exist_ok=True)
+        typer.echo(f"Insights file: {analysis.profile_output}", err=True)
+    try:
+        return await run_analyst(
+            agent=analysis.agent,
+            agent_spec=analysis.agent_spec,
+            workspace=analysis.workspace,
+            base_url=analysis.base_url,
+            insights_output=analysis.insights_output,
+            verbose=verbose,
+        )
+    except AgentRunError as exc:
+        detail = _one_line_error(exc).rstrip(".")
+        typer.echo(
+            f"Error: analyst run failed: {detail}. "
+            "Check inference model access and credentials, "
+            "then retry or adjust usage limits.",
+            err=True,
+        )
+        raise typer.Exit(1) from None
+    except (ClientConstructionError, NeMoPlatformError, httpx.HTTPError, OSError) as exc:
+        detail = _one_line_error(exc).rstrip(".")
+        typer.echo(
+            f"Error: analysis failed: {detail}. Check --base-url/NMP_BASE_URL, "
+            "authentication, workspace, and Intake availability.",
+            err=True,
+        )
+        raise typer.Exit(1) from None
 
 
 class InsightsCLI(NemoCLI):
@@ -40,8 +201,8 @@ def _root() -> None:
 
         @app.command("analyze")
         def analyze(
-            agent: str = typer.Option(
-                ...,
+            agent: str | None = typer.Option(
+                None,
                 "--agent",
                 help="Name of the agent (agent under test) the analyst should focus on.",
             ),
@@ -52,16 +213,23 @@ def analyze(
                 exists=True,
                 readable=True,
             ),
-            workspace: str = typer.Option(
-                DEFAULT_WORKSPACE,
+            workspace: str | None = typer.Option(
+                None,
                 "--workspace",
                 help="Workspace the analyst should operate in.",
             ),
-            base_url: str = typer.Option(
-                os.environ.get("NMP_BASE_URL", DEFAULT_BASE_URL),
+            base_url: str | None = typer.Option(
+                None,
                 "--base-url",
                 help="Base URL of the running NMP instance the analyst's tools should call.",
-                envvar="NMP_BASE_URL",
+            ),
+            profile_path: Path | None = typer.Option(
+                None,
+                "--profile",
+                help="Path to optimizer.yaml. Default: discovered by walking up from cwd.",
+                exists=True,
+                dir_okay=False,
+                readable=True,
             ),
             insights_output: Path | None = typer.Option(
                 None,
@@ -92,18 +260,75 @@ def analyze(
             to ``--agent`` / ``--workspace`` / ``--base-url``, runs it, and
             prints whatever the agent returns.
             """
-            output = asyncio.run(
-                run_analyst(
+            try:
+                analysis = _resolve_analysis(
                     agent=agent,
-                    agent_spec=agent_spec.read_text() if agent_spec else None,
+                    agent_spec=agent_spec,
                     workspace=workspace,
                     base_url=base_url,
+                    profile_path=profile_path,
                     insights_output=insights_output,
-                    verbose=verbose,
                 )
-            )
+                output = asyncio.run(_run_analysis(analysis, verbose=verbose))
+            except (ProfileError, EnvFileError, InsightsFileError, OSError, UnicodeError) as exc:
+                typer.echo(f"Error: {_one_line_error(exc)}", err=True)
+                raise typer.Exit(1) from None
             typer.echo(output)
 
+        @app.command("doctor")
+        def doctor(
+            profile_path: Path | None = typer.Option(
+                None,
+                "--profile",
+                help="Path to optimizer.yaml. Default: discovered by walking up from cwd.",
+                exists=True,
+                dir_okay=False,
+                readable=True,
+            ),
+            base_url: str | None = typer.Option(
+                None,
+                "--base-url",
+                help="Base URL of the running NMP instance to check.",
+            ),
+        ) -> None:
+            """Check whether the current profile is ready for analysis."""
+            try:
+                try:
+                    profile, profile_error = _load_profile_or_error(profile_path)
+                except ProfileError as exc:
+                    profile, profile_error = None, str(exc)
+                spec_path: Path | None = None
+                spec_error: str | None = None
+                if profile is not None:
+                    try:
+                        spec_path = pick_agent_spec(profile)
+                    except ProfileError as exc:
+                        spec_error = str(exc)
+                _, spec_results = read_agent_spec(spec_path, spec_error)
+
+                async def _flow() -> list[CheckResult]:
+                    results = check_profile(profile, profile_error)
+                    results.extend(spec_results)
+                    if profile is not None:
+                        results.extend(
+                            await check_environment(
+                                agent=profile.agent,
+                                workspace=profile.workspace,
+                                base_url=resolve_base_url(base_url),
+                                profile_dir=profile.profile_dir,
+                                probes=_PREFLIGHT_PROBES,
+                            )
+                        )
+                    return results
+
+                results = asyncio.run(_flow())
+            except (EnvFileError, OSError, UnicodeError) as exc:
+                typer.echo(f"Error: {_one_line_error(exc)}", err=True)
+                raise typer.Exit(1) from None
+            typer.echo(format_report(results))
+            if required_failures(results):
+                raise typer.Exit(code=1)
+
         @analysis_app.command("enable")
         def enable_analysis(
             agent: str = typer.Option(
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/contracts/checks.py b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/checks.py
new file mode 100644
index 0000000000..f9b230ad6c
--- /dev/null
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/checks.py
@@ -0,0 +1,72 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Command-neutral readiness result construction and presentation."""
+
+from typing import Literal
+
+from pydantic import BaseModel
+
+CheckStatus = Literal["pass", "warn", "fail"]
+CheckSeverity = Literal["required", "advisory"]
+
+
+class CheckResult(BaseModel):
+    """One required or advisory readiness check."""
+
+    name: str
+    group: str
+    status: CheckStatus
+    severity: CheckSeverity
+    message: str
+    hint: str | None = None
+
+
+def make_check_result(
+    name: str,
+    group: str,
+    ok: bool,
+    severity: CheckSeverity,
+    pass_message: str,
+    fail_message: str,
+    *,
+    hint: str | None = None,
+) -> CheckResult:
+    """Build a passing, blocking, or advisory result from a boolean probe."""
+    if ok:
+        status: CheckStatus = "pass"
+    elif severity == "required":
+        status = "fail"
+    else:
+        status = "warn"
+    return CheckResult(
+        name=name,
+        group=group,
+        status=status,
+        severity=severity,
+        message=pass_message if ok else fail_message,
+        hint=None if ok else hint,
+    )
+
+
+def format_report(results: list[CheckResult]) -> str:
+    """Format checks into deterministic grouped terminal output."""
+    marks: dict[CheckStatus, str] = {"pass": "✓", "warn": "⚠", "fail": "✗"}
+    lines: list[str] = []
+    for group in sorted({result.group for result in results}):
+        lines.append(group.capitalize())
+        for result in (item for item in results if item.group == group):
+            lines.append(f"  {marks[result.status]} {result.message}")
+            if result.hint and result.status != "pass":
+                lines.append(f"      hint: {result.hint}")
+    return "\n".join(lines)
+
+
+def required_failures(results: list[CheckResult]) -> list[CheckResult]:
+    """Return required failures that block a command."""
+    return [result for result in results if result.status == "fail" and result.severity == "required"]
+
+
+def advisories(results: list[CheckResult]) -> list[CheckResult]:
+    """Return non-blocking warnings."""
+    return [result for result in results if result.status == "warn"]
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/contracts/insights.py b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/insights.py
new file mode 100644
index 0000000000..d1d593ed7b
--- /dev/null
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/insights.py
@@ -0,0 +1,64 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Structural contract for the local Insights YAML document."""
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+class InsightsFileError(ValueError):
+    """A shared Insights file is unreadable or structurally invalid."""
+
+
+def _read_and_validate(path: Path) -> dict[str, Any]:
+    """Read, parse, and shape-check one Insights YAML document.
+
+    The read is the single existence boundary: a missing file surfaces as a
+    bare ``FileNotFoundError`` (not wrapped) so callers can each decide
+    whether that counts as absent input or a hard error, without a separate
+    ``stat()`` racing the actual read.
+    """
+    try:
+        text = path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        raise
+    except UnicodeError as exc:
+        raise InsightsFileError(f"insights file {path} is not valid UTF-8: {exc}") from None
+    except OSError as exc:
+        raise InsightsFileError(f"insights file {path} could not be read: {exc}") from None
+    try:
+        payload = yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        detail = " ".join(str(exc).split())
+        raise InsightsFileError(f"insights file {path} must contain valid YAML: {detail}") from None
+    if not isinstance(payload, dict):
+        raise InsightsFileError(f"insights file {path} must contain a YAML mapping at its root")
+    if "insights" in payload:
+        records = payload["insights"]
+        if not isinstance(records, list):
+            raise InsightsFileError(f"insights file {path}: `insights` must be a list")
+        for index, record in enumerate(records, start=1):
+            if not isinstance(record, dict):
+                raise InsightsFileError(f"insights file {path}: `insights` item {index} must be a YAML mapping")
+    return dict(payload)
+
+
+def load_insights_document(path: Path) -> dict[str, Any]:
+    """Read and validate one existing UTF-8 Insights YAML document."""
+    try:
+        return _read_and_validate(path)
+    except FileNotFoundError as exc:
+        raise InsightsFileError(f"insights file {path} could not be read: {exc}") from None
+
+
+def validate_insights_file(path: Path | None) -> None:
+    """Validate an existing file; allow absent optional output files."""
+    if path is None:
+        return
+    try:
+        _read_and_validate(path)
+    except FileNotFoundError:
+        return
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py
new file mode 100644
index 0000000000..77bfbd25fb
--- /dev/null
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py
@@ -0,0 +1,106 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Shared mechanics for optimizer.yaml without a universal profile schema."""
+
+import os
+from collections.abc import Mapping, MutableMapping
+from pathlib import Path
+from typing import TypeVar
+
+import yaml
+from pydantic import BaseModel, ValidationError
+
+PROFILE_FILENAME = "optimizer.yaml"
+DEFAULT_BASE_URL = "http://localhost:8080"
+_AGENT_SPEC_FILENAMES = ("AGENT-SPEC.md", "README.md")
+
+ProfileModel = TypeVar("ProfileModel", bound=BaseModel)
+
+
+class ProfileError(ValueError):
+    """A profile file or profile-owned path is invalid."""
+
+
+class EnvFileError(ValueError):
+    """An adjacent environment file could not be read safely."""
+
+
+def discover_profile(start: Path | None = None) -> Path | None:
+    """Walk from *start* or cwd to the filesystem root for optimizer.yaml."""
+    current = (start or Path.cwd()).resolve()
+    for directory in (current, *current.parents):
+        candidate = directory / PROFILE_FILENAME
+        if candidate.is_file():
+            return candidate
+    return None
+
+
+def resolve_profile_path(value: str, profile_dir: Path) -> Path:
+    """Resolve an absolute, home-relative, or profile-relative path."""
+    path = Path(value).expanduser()
+    return path.resolve() if path.is_absolute() else (profile_dir / path).resolve()
+
+
+def load_profile_model(path: Path, model: type[ProfileModel]) -> ProfileModel:
+    """Load optimizer.yaml into a caller-owned strict or tolerant model."""
+    try:
+        payload = yaml.safe_load(path.read_text(encoding="utf-8"))
+    except UnicodeError as exc:
+        raise ProfileError(f"Could not parse profile {path}: expected readable UTF-8 YAML ({exc})") from None
+    except (OSError, yaml.YAMLError) as exc:
+        raise ProfileError(f"Could not parse profile {path}: {exc}") from None
+    if not isinstance(payload, dict):
+        raise ProfileError(f"Could not parse profile {path}: expected a YAML mapping")
+    if "profile_dir" in payload:
+        raise ProfileError(f"Invalid profile {path}: 'profile_dir' is reserved")
+    values = dict(payload)
+    values["profile_dir"] = path.parent.resolve()
+    try:
+        return model.model_validate(values)
+    except ValidationError as exc:
+        details = "; ".join(f"{'.'.join(str(item) for item in error['loc'])}: {error['msg']}" for error in exc.errors())
+        raise ProfileError(f"Invalid profile {path}: {details}") from None
+
+
+def load_env_file(path: Path, env: MutableMapping[str, str] = os.environ) -> list[str]:
+    """Load simple KEY=VALUE entries without replacing existing environment keys."""
+    try:
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError:
+        return []
+    except (OSError, UnicodeError) as exc:
+        raise EnvFileError(
+            f"Could not read environment file {path}: {exc}. Check that the file is readable UTF-8 text, then retry."
+        ) from None
+    loaded: list[str] = []
+    for raw in lines:
+        line = raw.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, _, value = line.removeprefix("export ").partition("=")
+        key = key.strip()
+        value = value.strip().strip("'\"")
+        if key and key not in env:
+            env[key] = value
+            loaded.append(key)
+    return loaded
+
+
+def resolve_agent_spec_path(profile_dir: Path, configured: str | None) -> Path | None:
+    """Resolve a configured agent spec or the conventional profile-local file."""
+    if configured is not None:
+        path = resolve_profile_path(configured, profile_dir)
+        if not path.is_file():
+            raise ProfileError(f"Profile agent_spec {configured!r} does not exist (resolved to {path})")
+        return path
+    for name in _AGENT_SPEC_FILENAMES:
+        candidate = profile_dir / name
+        if candidate.is_file():
+            return candidate
+    return None
+
+
+def resolve_base_url(explicit: str | None, env: Mapping[str, str] = os.environ) -> str:
+    """Apply explicit, NMP_BASE_URL, then localhost precedence."""
+    return explicit or env.get("NMP_BASE_URL") or DEFAULT_BASE_URL
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py b/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py
new file mode 100644
index 0000000000..1dbf601971
--- /dev/null
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py
@@ -0,0 +1,203 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Read-only readiness checks for Insights analysis."""
+
+import os
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import httpx
+from nemo_insights_plugin.analyst.analyst_backend import make_analyst_backend
+from nemo_insights_plugin.client import make_client
+from nemo_insights_plugin.contracts.checks import CheckResult, make_check_result
+from nemo_insights_plugin.profile import AnalysisProfile
+from nemo_platform import AsyncNeMoPlatform, NeMoPlatformError
+
+_EXPECTED_PLATFORM_ERRORS = (NeMoPlatformError, httpx.HTTPError, OSError, RuntimeError, ValueError)
+
+
+def _default_http_ok(base_url: str) -> bool:
+    try:
+        return (
+            httpx.get(
+                f"{base_url.rstrip('/')}/health/ready",
+                timeout=5,
+                follow_redirects=True,
+            ).status_code
+            < 500
+        )
+    except (httpx.HTTPError, ValueError):
+        return False
+
+
+async def _default_workspace_ok(base_url: str, workspace: str, agent: str) -> bool:
+    client: AsyncNeMoPlatform | None = None
+    try:
+        client = make_client(base_url)
+        backend = make_analyst_backend(client=client, insights_output=None)
+        await backend.count_agent_sessions(agent=agent, workspace=workspace)
+        return True
+    except _EXPECTED_PLATFORM_ERRORS:
+        return False
+    finally:
+        if client is not None:
+            try:
+                await client.close()
+            except _EXPECTED_PLATFORM_ERRORS:
+                pass
+
+
+@dataclass(frozen=True)
+class AnalysisProbes:
+    """Dependencies used by read-only environment checks."""
+
+    env: Mapping[str, str] = field(default_factory=lambda: os.environ)
+    http_ok: Callable[[str], bool] = _default_http_ok
+    workspace_ok: Callable[[str, str, str], Awaitable[bool]] = _default_workspace_ok
+
+
+def check_profile(
+    profile: AnalysisProfile | None,
+    profile_error: str | None,
+) -> list[CheckResult]:
+    """Check that a profile was found and parsed."""
+    if profile_error is not None:
+        return [
+            CheckResult(
+                name="profile-parse",
+                group="profile",
+                status="fail",
+                severity="required",
+                message=profile_error,
+                hint="fix optimizer.yaml or pass --profile with a valid file",
+            )
+        ]
+    if profile is None:
+        return [
+            CheckResult(
+                name="profile-found",
+                group="profile",
+                status="fail",
+                severity="required",
+                message="no optimizer.yaml found (searched cwd and parents)",
+                hint="create optimizer.yaml with at least `agent: <name>`",
+            )
+        ]
+    return [
+        CheckResult(
+            name="profile-found",
+            group="profile",
+            status="pass",
+            severity="required",
+            message=f"profile for agent {profile.agent!r} at {profile.profile_dir}",
+        )
+    ]
+
+
+def check_agent_spec(
+    spec_path: Path | None,
+    spec_error: str | None,
+) -> list[CheckResult]:
+    """Check the optional agent-spec artifact, including explicit UTF-8 readability."""
+    return read_agent_spec(spec_path, spec_error)[1]
+
+
+def read_agent_spec(
+    spec_path: Path | None,
+    spec_error: str | None,
+) -> tuple[str | None, list[CheckResult]]:
+    """Read the optional agent spec as UTF-8 and return its readiness check."""
+    if spec_error is not None:
+        return None, [
+            CheckResult(
+                name="agent-spec",
+                group="artifacts",
+                status="fail",
+                severity="required",
+                message=spec_error,
+            )
+        ]
+    if spec_path is None:
+        return None, [
+            CheckResult(
+                name="agent-spec",
+                group="artifacts",
+                status="pass",
+                severity="advisory",
+                message="agent spec omitted (optional)",
+            )
+        ]
+    try:
+        content = spec_path.read_text(encoding="utf-8")
+    except (OSError, UnicodeError) as exc:
+        return None, [
+            CheckResult(
+                name="agent-spec",
+                group="artifacts",
+                status="fail",
+                severity="required",
+                message=f"Could not read agent spec {spec_path} as UTF-8: {exc}",
+                hint="ensure the file is readable and encoded as UTF-8",
+            )
+        ]
+    return content, [
+        CheckResult(
+            name="agent-spec",
+            group="artifacts",
+            status="pass",
+            severity="required",
+            message=f"agent spec readable at {spec_path}",
+        )
+    ]
+
+
+async def check_environment(
+    *,
+    agent: str,
+    workspace: str,
+    base_url: str,
+    profile_dir: Path | None,
+    probes: AnalysisProbes | None = None,
+) -> list[CheckResult]:
+    """Run credential and advisory platform checks without persisting state."""
+    active = probes or AnalysisProbes()
+    env_path = profile_dir / ".env" if profile_dir is not None else None
+    credential_hint = (
+        f"save it in {env_path} or export INFERENCE_API_KEY=<key>"
+        if env_path is not None
+        else "export INFERENCE_API_KEY=<key>"
+    )
+    credential = bool(active.env.get("INFERENCE_API_KEY", "").strip())
+    reachable = active.http_ok(base_url)
+    queryable = await active.workspace_ok(base_url, workspace, agent)
+    return [
+        make_check_result(
+            "INFERENCE_API_KEY",
+            "credentials",
+            credential,
+            "required",
+            "INFERENCE_API_KEY set",
+            "INFERENCE_API_KEY not set",
+            hint=credential_hint,
+        ),
+        make_check_result(
+            "platform-reachable",
+            "platform",
+            reachable,
+            "advisory",
+            f"{base_url} reachable",
+            f"{base_url} unreachable",
+            hint="check --base-url/NMP_BASE_URL and platform health",
+        ),
+        make_check_result(
+            "workspace-query",
+            "platform",
+            queryable,
+            "advisory",
+            f"workspace {workspace!r} can be queried for agent {agent!r}",
+            f"workspace {workspace!r} could not be queried for agent {agent!r}",
+            hint="check the workspace, authentication context, and Intake availability",
+        ),
+    ]
diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/profile.py b/plugins/nemo-insights/src/nemo_insights_plugin/profile.py
new file mode 100644
index 0000000000..f5dd08b282
--- /dev/null
+++ b/plugins/nemo-insights/src/nemo_insights_plugin/profile.py
@@ -0,0 +1,30 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Analysis-owned view of an optimizer.yaml agent profile."""
+
+from pathlib import Path
+
+from nemo_insights_plugin.contracts.profile import load_profile_model, resolve_agent_spec_path
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class AnalysisProfile(BaseModel):
+    """Only fields consumed by ``nemo insights``; all other keys are tolerated."""
+
+    model_config = ConfigDict(extra="ignore")
+
+    agent: str = Field(min_length=1)
+    agent_spec: str | None = None
+    workspace: str = "default"
+    profile_dir: Path
+
+
+def load_profile(path: Path) -> AnalysisProfile:
+    """Load the analysis-owned subset of optimizer.yaml."""
+    return load_profile_model(path, AnalysisProfile)
+
+
+def pick_agent_spec(profile: AnalysisProfile) -> Path | None:
+    """Resolve the profile's configured or conventional agent spec."""
+    return resolve_agent_spec_path(profile.profile_dir, profile.agent_spec)
diff --git a/plugins/nemo-insights/testbed/README.md b/plugins/nemo-insights/testbed/README.md
index ce9a70492f..3f72ee7618 100644
--- a/plugins/nemo-insights/testbed/README.md
+++ b/plugins/nemo-insights/testbed/README.md
@@ -13,6 +13,7 @@ uv run python -m testbed analyze tau2-airline --live  # analyze the recorded run
 uv run python -m testbed analyze nvq --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-vN --into WORKSPACE
 ```
 
 Bare `analyze <subject>` is a fully reproducible run: pinned data (the subject's
@@ -65,6 +66,15 @@ base_url recorded at `run` time, so set `--base` on `run`.)
 Immutable per-subject fixtures on the `testbed-state` GitHub release, pinned in
 `testbed/state.lock` — analyst changes get measured against fixed data.
 
+The assets currently live in `NVIDIA-dev/NeMo-Optimizer`. Release operations
+target that repository explicitly; set `TESTBED_STATE_REPO=owner/repository` to
+use another fixture home. The `gh` token must be able to read that repository
+for restore/analyze and write releases there for local publish. A token scoped
+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.
+
 Which file do I touch?
 
 | Surface              | Owns                                          |
@@ -93,7 +103,7 @@ API's 30-day default lookback. (`--live` skips the restore entirely and
 analyzes the platform's live traces, with `since` from `--since`, the stanza,
 or a 30d default — in that order; the effective bound is always printed.)
 
-**Mint a new fixture (laptop or CI):**
+**Publish a verified candidate from a maintainer machine:**
 
 ```bash
 uv run python -m testbed snapshot nvq -o testbed/tmp/nvq.tar.zst
@@ -104,16 +114,27 @@ uv run python -m testbed publish testbed/tmp/nvq.tar.zst --base http://localhost
 `-oracle` twin) into JSONL + manifest — no ClickHouse, no Docker. `publish`
 refuses to mint unverified: `--base` runs the round-trip fidelity guard there
 first (re-ingest into scratch workspaces → re-export → doc diff), or pass
-`--no-verify` to skip out loud (CI does; its guard runs as a separate step).
+`--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`.
 
 **Restore without analyzing:** `uv run python -m testbed restore (FILE | --state state-v7) [--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
+workspace. The default restore remains fixture-scoped and idempotent.
 
 What restore touches:
 
-- **Platform: additive only.** Ingests into `<ws>-<ref>` (`<ws>-<sha256[:8]>`
-  for local files); never writes into existing workspaces. Per-collection
-  guard: counts match → skip, empty → ingest, anything else → hard error.
+- **Platform, default fixture restore:** additive, idempotent, and healing.
+  Ingests into `<ws>-<ref>` (`<ws>-<sha256[:8]>` for local files).
+  Per-collection guard: counts match → skip, empty → ingest, supported
+  interrupted states → heal, anything else → hard error.
+- **Platform, direct `--into` restore:** writes to the exact named workspace
+  only after proving all three collections are empty. It is fresh-target-only,
+  and rerunning into the now-populated workspace fails rather than acting
+  idempotently.
 - **Local `testbed/tmp`: run records seeded.** The bundle's run records
   replace yours — clobbered files are moved to `testbed/tmp/backup-<timestamp>/`
   first, and only the bundle's own subjects' files are ever touched. Bundles
@@ -189,17 +210,17 @@ uv run python -m testbed run tau2-retail
 uv run python -m testbed analyze tau2-retail --live
 ```
 
-## CI (`.github/workflows/testbed-insights.yml`)
+## CI (`.github/workflows/insights-testbed.yml`)
 
 CI runs the testbed against a **self-contained platform inside the job**
 (ClickHouse + `auth,entities,intake` from a `nemo-platform` checkout) — the
 freeplay remote is not reachable from GitHub-hosted runners. The workflow's
 steps are thin wrappers over the same CLI you run locally (`testbed restore`
-/ `analyze` / `snapshot` / `roundtrip` / `publish`); the shared
+/ `analyze` / `snapshot` / `roundtrip`); the shared
 helpers live in `testbed/eval/` (stdlib-only `plan.py`/`prep.py`/
 `run_subjects.py` run on the bare runner before `uv sync`). Dispatch inputs
 mirror the CLI flags 1:1 (`mode`, `subjects`, `state`, `num_tasks`,
-`num_trials`, `publish_state`, `reason`). Three modes, all validated **green
+`num_trials`, `reason`). Three modes, all validated **green
 on real GitHub Actions** (branch `testbed-ci-insights`; API-export pipeline:
 stack-check
 [run 28880210091](https://github.com/NVIDIA-dev/NeMo-Optimizer/actions/runs/28880210091),
@@ -207,30 +228,33 @@ analyze vs `state-v6`
 [run 28880474684](https://github.com/NVIDIA-dev/NeMo-Optimizer/actions/runs/28880474684)
 (966 spans re-ingested into `<ws>-state-v6` fixtures), produce smoke ×2 tasks
 [run 28881101719](https://github.com/NVIDIA-dev/NeMo-Optimizer/actions/runs/28881101719)
-— round-trip guard green in CI, candidate uploaded, publish skipped on the
-validation trigger):
+— round-trip guard green in CI and candidate uploaded for inspection):
+
+Configure the `insights-testbed` GitHub environment with required reviewers
+and self-review prevention. Store `NVIDIA_INFERENCE_KEY`,
+`NVIDIA_INFERENCE_URL`, and `TESTBED_STATE_GH_READ_TOKEN` exclusively as
+secrets in that environment; do not retain repository- or organization-level
+copies that PR workflow edits could access without approval. Set the
+`TESTBED_STATE_REPO` repository variable when fixtures live outside the
+default `NVIDIA-dev/NeMo-Optimizer` repository.
 
 - `stack-check` — bring the stack up, verify `/ping` + `/health/ready`, exit.
   Cheap CI doctor.
 - `produce` — **explicit dispatch only** (hard-gated to `workflow_dispatch`): a
-  human mints a new fixture when there's a reason — tau2-bench update, agent/sim
-  config change, staleness refresh. Never runs from PRs or any automatic
-  trigger. Fixtures mint from the fresh in-job stack (no base restore):
+  human generates a candidate when there's a reason — tau2-bench update,
+  agent/sim config change, staleness refresh. Never runs from PRs or any
+  automatic trigger. Candidates come from the fresh in-job stack (no base restore):
   `run_subjects.py` runs the subjects (`subjects=tau2-airline,...`,
   override size with `num_tasks=`/`num_trials=`; analyze retries absorb
   post-sim rate-limit heat) with `--base http://localhost:8080`, then
   `testbed snapshot` exports an **unminted candidate bundle** that is always
   uploaded as workflow artifact `state-candidate-<run_id>-<attempt>` (even on
-  failure — failed runs never mint a version, they just leave their candidate
-  for inspection; there are no `-failed` refs), and `testbed roundtrip` proves
-  the candidate re-ingests with full read-API fidelity. Only on success does
-  `testbed publish` (`--no-verify`: the guard ran as its own step) mint the
-  next `state-v<N>` — it resolves the next free version at publish time,
-  uploads the asset to the `testbed-state` release, and prepends a row to the
-  **fixture catalog** in the release notes (version, contents, the `reason`
-  input, minted-by). The same `testbed publish` runs from a laptop (that is how
-  `state-v6` was minted). Serialized via a `testbed-state-produce` concurrency
-  group so bundles never race.
+  failure), and `testbed roundtrip` proves the candidate re-ingests with full
+  read-API fidelity. While the fixture release remains in NeMo Optimizer, this
+  workflow stops at the candidate artifact; the canonical repository remains
+  the only automated publisher. A maintainer can download the candidate and
+  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
   analyze — each subject's own pin under `[subjects]` in `testbed/state.lock`
@@ -253,23 +277,23 @@ States are per-produce-dispatch compositions, so subjects pin different
 versions — each under `[subjects]` in `testbed/state.lock`; a subject with no
 entry errors (add its line after minting a fixture). Lock-bump PRs edit the
 subject's line; bump on `main` to advance that subject's shared baseline after
-a `produce` run. The dispatch `state` input still overrides the lock for
-**all** subjects in that run.
+the produced candidate is verified and manually published. The dispatch
+`state` input still overrides the lock for **all** subjects in that run.
 
 ### Dispatching a run
 
 `workflow_dispatch` (and `gh workflow run`) only becomes available once the
 workflow file has landed on the repository's **default branch** — a GitHub
 registration requirement (dispatching from a feature branch 404s). After the
-merge, verify with `gh workflow run testbed-insights.yml -f mode=stack-check`,
+merge, verify with `gh workflow run insights-testbed.yml -f mode=stack-check`,
 then:
 
 ```bash
-gh workflow run testbed-insights.yml -f mode=stack-check
-gh workflow run testbed-insights.yml -f mode=produce -f subjects=tau2-airline -f num_tasks=2 \
-  -f reason="2-task smoke after tau2-bench bump"   # reason lands in the release's fixture catalog
-gh workflow run testbed-insights.yml -f mode=analyze                    # each subject's state.lock pin
-gh workflow run testbed-insights.yml -f mode=analyze -f state=state-v6  # explicit override, all subjects
+gh workflow run insights-testbed.yml -f mode=stack-check
+gh workflow run insights-testbed.yml -f mode=produce -f subjects=tau2-airline -f num_tasks=2 \
+  -f reason="2-task smoke after tau2-bench bump"   # retained when the candidate is published
+gh workflow run insights-testbed.yml -f mode=analyze                    # each subject's state.lock pin
+gh workflow run insights-testbed.yml -f mode=analyze -f state=state-v6  # explicit override, all subjects
 ```
 
 ### Pre-merge checklist: validating workflow changes from a branch
@@ -282,8 +306,8 @@ from the feature branch with a **temporary push trigger** before merge:
    `inputs.x`, falling back on push events to a repo variable
    (`gh variable set TESTBED_X --body ...`) — every arm marked `# TEMPORARY`.
    produce's dispatch-only event gate needs a temporary
-   `|| github.event_name == 'push'` arm; leave the publish step
-   dispatch-gated so a validation run can never mint.
+   `|| github.event_name == 'push'` arm. The Platform workflow has no publish
+   step, so validation runs cannot mint fixture refs.
 2. Drive rounds by setting the variables (e.g. `gh variable set TESTBED_MODE
    --body produce`) plus a trivial trigger commit; watch `gh run list` /
    `gh run view --log-failed` (runs take 5–20 min).
@@ -321,7 +345,9 @@ 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;
 locally: `analyze <subject> --state state-vN`).
 
-Secrets: the workflow uses Actions secrets `NVIDIA_INFERENCE_KEY` and
-`NVIDIA_INFERENCE_URL`, exposing them under the `INFERENCE_API_KEY` and
-OpenAI-compatible environment names expected by the analyst and litellm/tau2.
-The analyst and tau2 sim LLMs need no VPN and work on public runners.
+Secrets: the workflow uses `TESTBED_STATE_GH_READ_TOKEN`, a least-privilege
+GitHub App/PAT credential with release-read access to `TESTBED_STATE_REPO`,
+plus `NVIDIA_INFERENCE_KEY` and `NVIDIA_INFERENCE_URL`. The latter are exposed
+under the `INFERENCE_API_KEY` and OpenAI-compatible environment names expected
+by the analyst and litellm/tau2. The analyst and tau2 sim LLMs need no VPN and
+work on public runners.
diff --git a/plugins/nemo-insights/testbed/cli.py b/plugins/nemo-insights/testbed/cli.py
index 01922573d6..7d1b03ed51 100644
--- a/plugins/nemo-insights/testbed/cli.py
+++ b/plugins/nemo-insights/testbed/cli.py
@@ -22,9 +22,11 @@
 
 Export bundles (`kind: testbed-export`) restore by re-ingesting through the real
 APIs into fixture-scoped workspaces (`<ws>-<ref>` for published refs,
-`<ws>-<sha256[:8]>` for local files) — additive and idempotent, never touching
-existing data. Legacy tar bundles (state-v1..v5) are restorable only from a
-pre-migration checkout; see testbed/README.md.
+`<ws>-<sha256[:8]>` for local files) — additive, idempotent, and healing.
+`restore --into WORKSPACE` is the direct alternative: it requires a fresh,
+empty target and is not idempotent into a populated workspace. Legacy tar
+bundles (state-v1..v5) are restorable only from a pre-migration checkout; see
+testbed/README.md.
 
 This drives the analyst (`nemo insights analyze`) against registered subjects; it is not
 the product CLI and is not shipped in the wheel.
@@ -212,6 +214,7 @@ def _restore_export_bundle(
     platform_root: str | None,
     tmp_dir: Path,
     backup_dir: Path | None = None,
+    into: str | None = None,
 ) -> tuple[dict, dict[str, str], list[str]]:
     """Re-ingest an export bundle into fixture workspaces.
 
@@ -219,13 +222,15 @@ def _restore_export_bundle(
     run-record files the bundle left in ``tmp_dir`` (callers use it to refuse
     analyzing a subject the bundle carries no run record for).
 
-    Additive + idempotent (``reingest.ingest_bundle`` guards on per-collection counts and
-    warns on stale bundles) — no destructive ceremony. The catalog inversion is
-    loaded up front so a missing nemo-platform checkout fails before any
-    extraction or network I/O. The bundle's ``tmp/`` run records are copied
-    beside the local ones (clobbered locals are backed up into *backup_dir*,
-    the caller's per-invocation destination); insights never travel in
-    bundles, so nothing else lands in ``tmp_dir``.
+    The default fixture-scoped path is additive, idempotent, and healing
+    (``reingest.ingest_bundle`` guards on per-collection counts). Direct
+    ``--into`` restores require a fresh, empty target and are not idempotent
+    into populated workspaces. Both warn on stale bundles. The catalog
+    inversion is loaded up front so a missing nemo-platform checkout fails
+    before any extraction or network I/O. The bundle's ``tmp/`` run records
+    are copied beside the local ones (clobbered locals are backed up into
+    *backup_dir*, the caller's per-invocation destination); insights never
+    travel in bundles, so nothing else lands in ``tmp_dir``.
     """
     catalog = reingest.load_catalog(reingest.resolve_platform_root(platform_root))
     TMP.mkdir(parents=True, exist_ok=True)
@@ -233,13 +238,18 @@ def _restore_export_bundle(
         subprocess.run(["tar", "--zstd", "-xf", str(bundle), "-C", tmp], check=True)
         state = Path(tmp) / "state"
         manifest = json.loads((state / "manifest.json").read_text(encoding="utf-8"))
-        workspace_map = reingest.fixture_workspace_map(manifest["workspaces"], suffix)
+        workspace_map = (
+            reingest.explicit_workspace_map(manifest["workspaces"], into)
+            if into is not None
+            else reingest.fixture_workspace_map(manifest["workspaces"], suffix)
+        )
         outcome = reingest.ingest_bundle(
             base_url,
             state / "export",
             manifest,
             workspace_map=workspace_map,
             catalog=catalog,
+            require_empty=into is not None,
         )
         # seed_records must run inside this `with` block: it copies the bundle's
         # records out of state/tmp (under the tempdir) before teardown.
@@ -528,6 +538,14 @@ def main() -> None:
         help="Restore a published state ref (state-vN); local files go through the positional FILE. "
         "Pins are per-subject: for the pinned state, use `analyze <subject>` instead.",
     )
+    p_res.add_argument(
+        "--into",
+        default=None,
+        metavar="WORKSPACE",
+        help="Restore a single-workspace bundle directly into this workspace instead of the "
+        "fixture-scoped <ws>-<ref> default. Requires a fresh, empty target and is "
+        "not idempotent into a populated workspace.",
+    )
     p_res.add_argument(
         "--base",
         default=None,
@@ -687,12 +705,16 @@ def main() -> None:
                 f"restore: {bundle_path.name} is a legacy tar bundle (state-v1..v5) — "
                 "restorable only from a pre-migration checkout; see testbed/README.md"
             )
+        if args.into is not None:
+            # Validate the direct target before catalog loading, extraction, or network I/O.
+            reingest.explicit_workspace_map(manifest["workspaces"], args.into)
         _restore_export_bundle(
             bundle_path,
             base_url=args.base or LOCAL_URL,
             suffix=suffix,
             platform_root=args.platform_root,
             tmp_dir=TMP,
+            into=args.into,
         )
         return
 
diff --git a/plugins/nemo-insights/testbed/publish.py b/plugins/nemo-insights/testbed/publish.py
index 1e46b18746..d37e4e9b29 100644
--- a/plugins/nemo-insights/testbed/publish.py
+++ b/plugins/nemo-insights/testbed/publish.py
@@ -2,13 +2,13 @@
 # SPDX-License-Identifier: Apache-2.0
 """Mint the next state ref, upload a candidate bundle, and update the fixture catalog.
 
-Laptop-first: ``uv run python -m testbed publish FILE --reason "why"`` works
-from any machine with ``gh`` auth'd against the repo; the CI produce job runs
-the same command. The catalog's "Minted by" column records the GitHub Actions
-run when ``GITHUB_RUN_ID``/``GITHUB_REPOSITORY`` are set, else
-``laptop (<user>)``; the "Contents" column comes from the bundle's own
-manifest (subjects + per-collection doc counts), so the catalog can never
-disagree with what the asset actually holds.
+``uv run python -m testbed publish FILE --reason "why"`` works from a
+maintainer machine with ``gh`` access to the fixture repository. The catalog's
+"Minted by" column records the GitHub Actions run when
+``GITHUB_RUN_ID``/``GITHUB_REPOSITORY`` are set, else ``laptop (<user>)``; the
+"Contents" column comes from the bundle's own manifest (subjects +
+per-collection doc counts), so the catalog can never disagree with what the
+asset actually holds.
 """
 
 import getpass
@@ -105,46 +105,44 @@ def insert_catalog_row(body: str, row: str) -> str:
 
 
 def _ensure_release() -> None:
-    try:
-        release._gh("release", "view", RELEASE_TAG)
-    except subprocess.CalledProcessError:
-        release._gh(
-            "release",
-            "create",
-            RELEASE_TAG,
-            "--title",
-            "Testbed state fixtures",
-            "--notes",
-            "Immutable testbed state fixtures. Do not delete assets.",
-            "--latest=false",
-        )
+    if release._release_exists():
+        return
+    release._release_gh(
+        "create",
+        RELEASE_TAG,
+        "--title",
+        "Testbed state fixtures",
+        "--notes",
+        "Immutable testbed state fixtures. Do not delete assets.",
+        "--latest=false",
+    )
 
 
 def publish(candidate: Path, *, reason: str | None, env: Mapping[str, str] | None = None) -> str:
     """Mint the next ref, upload the bundle, then edit the release notes with its catalog row.
 
-    ``reason=None`` falls back to the ``REASON`` env var. Returns the minted
-    ref; when ``GITHUB_OUTPUT`` is set (CI), also writes ``state_ref=<ref>``.
+    ``reason=None`` falls back to the ``REASON`` env var, then the candidate
+    manifest. Returns the minted ref; when ``GITHUB_OUTPUT`` is set, also
+    writes ``state_ref=<ref>``.
 
     A failure between the upload and the notes edit leaves an orphaned asset
     with no catalog row; recover by deleting the asset or hand-adding the
     row — a retry mints the next ref, not the orphaned one.
     """
     env = os.environ if env is None else env
-    if reason is None:
-        reason = env.get("REASON", "")
     manifest = read_manifest(candidate)
+    if reason is None:
+        reason = env.get("REASON") or str(manifest.get("reason") or "")
     ref = release.next_ref(release.latest_ref(release._release_asset_names()))
     tarball = candidate.parent / f"{ref}.tar.zst"
     shutil.copy2(candidate, tarball)
     _ensure_release()
-    # --clobber: a retry after a partial upload overwrites the broken asset
-    # instead of erroring (next_ref never collides with a *completed* publish —
-    # its ref would already be in the asset list).
-    release._gh("release", "upload", RELEASE_TAG, str(tarball), "--clobber")
-    body = json.loads(release._gh("release", "view", RELEASE_TAG, "--json", "body"))["body"]
+    # State refs are immutable: a concurrent publisher must fail on collision,
+    # never replace the asset that won the race.
+    release._release_gh("upload", RELEASE_TAG, str(tarball))
+    body = json.loads(release._release_gh("view", RELEASE_TAG, "--json", "body"))["body"]
     row = catalog_row(ref, manifest, reason=reason, env=env)
-    release._gh("release", "edit", RELEASE_TAG, "--notes", insert_catalog_row(body, row))
+    release._release_gh("edit", RELEASE_TAG, "--notes", insert_catalog_row(body, row))
     if env.get("GITHUB_OUTPUT"):
         with open(env["GITHUB_OUTPUT"], "a", encoding="utf-8") as fh:
             fh.write(f"state_ref={ref}\n")
diff --git a/plugins/nemo-insights/testbed/reingest.py b/plugins/nemo-insights/testbed/reingest.py
index 3c77457a10..24603fa705 100644
--- a/plugins/nemo-insights/testbed/reingest.py
+++ b/plugins/nemo-insights/testbed/reingest.py
@@ -4,8 +4,9 @@
 
 The consumption side of testbed export bundles: convert exported span docs back
 to OTLP protobuf and POST them to Intake's ingest route, then re-post
-annotations and evaluator results — additive and idempotent, into
-caller-specified workspaces, never touching existing data.
+annotations and evaluator results. Fixture-scoped restores are additive,
+idempotent, and healing. Direct restores use ``require_empty=True`` and are
+fresh-target-only: they fail once the target contains data.
 
 Doc -> OTLP inversion (validated against the live platform, 2026-07-06 spike):
 
@@ -183,6 +184,17 @@ def fixture_workspace_map(workspaces: list[str], suffix: str) -> dict[str, str]:
     return mapping
 
 
+def explicit_workspace_map(workspaces: list[str], into: str) -> dict[str, str]:
+    if len(workspaces) != 1:
+        sys.exit(
+            "restore --into requires a single-workspace bundle; "
+            f"found {len(workspaces)}: {', '.join(sorted(workspaces))}"
+        )
+    if not _WS_OK.fullmatch(into):
+        sys.exit(f"workspace {into!r} violates the platform naming rule ({_WS_OK.pattern})")
+    return {workspaces[0]: into}
+
+
 def manifest_since(manifest: dict) -> datetime:
     """The analyst's explicit lower bound for a restored bundle: ``min_start_time`` floored.
 
@@ -362,6 +374,22 @@ def evaluator_result_count(base_url: str, workspace: str, *, client: httpx.Clien
     return _collection_count(base_url, workspace, "evaluator-results", "created_at", client=client)
 
 
+def _require_zero(workspace: str, collection: str, count: int) -> None:
+    if count:
+        raise RuntimeError(
+            f"{workspace}: direct restore requires an empty target, but it has "
+            f"{count} {collection}. Choose a fresh workspace or explicitly delete "
+            "and recreate this one."
+        )
+
+
+def _collection_outcome(documents: list[dict], ingested: bool) -> dict[str, int]:
+    count = len(documents)
+    if ingested:
+        return {"ingested": count, "skipped": 0}
+    return {"ingested": 0, "skipped": count}
+
+
 def _post_created(client: httpx.Client, url: str, body: dict) -> None:
     resp = client.post(url, json=body)
     if not (200 <= resp.status_code < 300):
@@ -523,6 +551,7 @@ def ingest_bundle(
     *,
     workspace_map: dict[str, str],
     catalog,
+    require_empty: bool = False,
     sleep: Callable[[float], None] = time.sleep,
 ) -> dict:
     """Re-ingest a bundle's export into the mapped workspaces through the real APIs.
@@ -544,6 +573,11 @@ def ingest_bundle(
       error — annotation POSTs mint fresh server-side uuids, so re-posting a
       partial set would duplicate; there is no safe heal (delete + re-restore).
 
+    With ``require_empty=True``, all three target collections must be empty
+    before the first data write and each non-empty collection is rechecked
+    immediately before its first write. This direct-restore mode is not
+    idempotent into a populated workspace.
+
     "already restored — skipping" is printed only when EVERY collection is
     satisfied. Returns
     ``{source_ws: {"workspace": target, <collection>: {"ingested": n, "skipped": n}}}``.
@@ -598,50 +632,63 @@ def ingest_bundle(
             ensure_workspace(base_url, target, client=client)
 
             have_spans = span_count(base_url, target, client=client)
-            if have_spans == expected_spans:
-                ingest_spans = False
-                if expected_spans and spans:
-                    # Counts alone can't tell a restored corpus from a re-minted one — fingerprint it.
-                    _assert_same_first_span(base_url, target, spans, client=client)
-            elif have_spans == 0:
-                ingest_spans = True
-            else:
-                raise RuntimeError(
-                    f"{target}: has {have_spans} spans but the bundle expects {expected_spans} — the workspace "
-                    "is partially restored (or holds foreign data). Delete the workspace (or map to a fresh "
-                    "one) and restore again."
-                )
-            have_ann = annotation_count(base_url, target, client=client)
-            if have_ann == expected_ann:
-                post_annotations = False
-            elif have_ann == 0:
-                post_annotations = True
-            else:
-                raise RuntimeError(
-                    f"{target}: has {have_ann} annotations but the bundle expects {expected_ann} — annotation "
-                    "POSTs mint fresh server-side ids, so re-posting would duplicate what is already there "
-                    "(there is no safe partial re-post). Delete the fixture workspace (or map to a fresh one) "
-                    "and restore again."
-                )
-            have_res = evaluator_result_count(base_url, target, client=client)
-            if have_res == expected_res:
-                post_results = False
-            elif have_res < expected_res:
-                post_results = True  # upsert-safe: re-post the FULL set
+            if require_empty:
+                have_ann = annotation_count(base_url, target, client=client)
+                have_res = evaluator_result_count(base_url, target, client=client)
+                for collection, count in (
+                    ("spans", have_spans),
+                    ("annotations", have_ann),
+                    ("evaluator results", have_res),
+                ):
+                    _require_zero(target, collection, count)
+                ingest_spans = bool(spans)
+                post_annotations = bool(annotations)
+                post_results = bool(results)
             else:
-                raise RuntimeError(
-                    f"{target}: has {have_res} evaluator results but the bundle expects {expected_res} — the "
-                    "workspace holds foreign evaluator results. Delete the fixture workspace (or map to a "
-                    "fresh one) and restore again."
-                )
+                if have_spans == expected_spans:
+                    ingest_spans = False
+                    if expected_spans and spans:
+                        # Counts alone can't tell a restored corpus from a re-minted one — fingerprint it.
+                        _assert_same_first_span(base_url, target, spans, client=client)
+                elif have_spans == 0:
+                    ingest_spans = True
+                else:
+                    raise RuntimeError(
+                        f"{target}: has {have_spans} spans but the bundle expects {expected_spans} — the workspace "
+                        "is partially restored (or holds foreign data). Delete the workspace (or map to a fresh "
+                        "one) and restore again."
+                    )
+                have_ann = annotation_count(base_url, target, client=client)
+                if have_ann == expected_ann:
+                    post_annotations = False
+                elif have_ann == 0:
+                    post_annotations = True
+                else:
+                    raise RuntimeError(
+                        f"{target}: has {have_ann} annotations but the bundle expects {expected_ann} — annotation "
+                        "POSTs mint fresh server-side ids, so re-posting would duplicate what is already there "
+                        "(there is no safe partial re-post). Delete the fixture workspace (or map to a fresh one) "
+                        "and restore again."
+                    )
+                have_res = evaluator_result_count(base_url, target, client=client)
+                if have_res == expected_res:
+                    post_results = False
+                elif have_res < expected_res:
+                    post_results = True  # upsert-safe: re-post the FULL set
+                else:
+                    raise RuntimeError(
+                        f"{target}: has {have_res} evaluator results but the bundle expects {expected_res} — the "
+                        "workspace holds foreign evaluator results. Delete the fixture workspace (or map to a "
+                        "fresh one) and restore again."
+                    )
 
             if not (ingest_spans or post_annotations or post_results):
                 print(f"{target}: already restored ({have_spans} spans) — skipping")
                 outcome[source_ws] = {
                     "workspace": target,
-                    "spans": {"ingested": 0, "skipped": len(spans)},
-                    "annotations": {"ingested": 0, "skipped": len(annotations)},
-                    "evaluator_results": {"ingested": 0, "skipped": len(results)},
+                    "spans": _collection_outcome(spans, False),
+                    "annotations": _collection_outcome(annotations, False),
+                    "evaluator_results": _collection_outcome(results, False),
                 }
                 continue
             # Healing = posting into a workspace whose spans already landed (interrupted restore).
@@ -650,37 +697,36 @@ def ingest_bundle(
                 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]]
-                    export_trace_request(base_url, target, build_trace_request(batch), client=client)
+                    request = build_trace_request(batch)
+                    if require_empty and start == 0:
+                        _require_zero(target, "spans", span_count(base_url, target, client=client))
+                    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)
             root = f"{base_url.rstrip('/')}/apis/intake/v2/workspaces/{target}"
             if post_annotations:
                 if healing:
                     print(f"{target}: healing annotations: posting {len(annotations)}")
-                for doc in annotations:
+                for index, doc in enumerate(annotations):
                     body = {k: v for k, v in doc.items() if k not in _POST_DROP["annotations"]}
+                    if require_empty and index == 0:
+                        _require_zero(target, "annotations", annotation_count(base_url, target, client=client))
                     _post_created(client, f"{root}/annotations", body)
             if post_results:
                 if healing:
                     print(f"{target}: healing evaluator results: posting {len(results)}")
-                for doc in results:
+                for index, doc in enumerate(results):
                     body = {k: v for k, v in doc.items() if k not in _POST_DROP["evaluator_results"]}
+                    if require_empty and index == 0:
+                        _require_zero(
+                            target, "evaluator results", evaluator_result_count(base_url, target, client=client)
+                        )
                     _post_created(client, f"{root}/evaluator-results", body)
             outcome[source_ws] = {
                 "workspace": target,
-                "spans": (
-                    {"ingested": len(spans), "skipped": 0} if ingest_spans else {"ingested": 0, "skipped": len(spans)}
-                ),
-                "annotations": (
-                    {"ingested": len(annotations), "skipped": 0}
-                    if post_annotations
-                    else {"ingested": 0, "skipped": len(annotations)}
-                ),
-                "evaluator_results": (
-                    {"ingested": len(results), "skipped": 0}
-                    if post_results
-                    else {"ingested": 0, "skipped": len(results)}
-                ),
+                "spans": _collection_outcome(spans, ingest_spans),
+                "annotations": _collection_outcome(annotations, post_annotations),
+                "evaluator_results": _collection_outcome(results, post_results),
             }
     return outcome
 
diff --git a/plugins/nemo-insights/testbed/release.py b/plugins/nemo-insights/testbed/release.py
index 3df5b9d454..cbcbc979b8 100644
--- a/plugins/nemo-insights/testbed/release.py
+++ b/plugins/nemo-insights/testbed/release.py
@@ -7,13 +7,16 @@
 """
 
 import json
+import os
 import re
 import subprocess
 import sys
 import tomllib
+from collections.abc import Mapping
 from pathlib import Path
 
 RELEASE_TAG = "testbed-state"
+DEFAULT_STATE_REPO = "NVIDIA-dev/NeMo-Optimizer"
 _ASSET = re.compile(r"^state-v(\d+)\.tar\.zst$")
 
 
@@ -66,16 +69,48 @@ def _gh(*args: str) -> str:
         raise
 
 
+def state_repo(env: Mapping[str, str] | None = None) -> str:
+    """Return the explicit GitHub repository that owns testbed state assets."""
+    values = os.environ if env is None else env
+    return values.get("TESTBED_STATE_REPO", DEFAULT_STATE_REPO)
+
+
+def _release_gh(*args: str) -> str:
+    """Run a release command against the configured fixture repository."""
+    return _gh("release", *args, "--repo", state_repo())
+
+
+def _release_repo_accessible() -> None:
+    """Verify release-read access before interpreting a missing release."""
+    _gh("api", f"repos/{state_repo()}/releases?per_page=1")
+
+
+def _release_missing(error: subprocess.CalledProcessError) -> bool:
+    return (error.stderr or "").strip().lower() == "release not found"
+
+
+def _release_exists() -> bool:
+    _release_repo_accessible()
+    try:
+        _release_gh("view", RELEASE_TAG)
+    except subprocess.CalledProcessError as error:
+        if _release_missing(error):
+            return False
+        raise
+    return True
+
+
 def _release_asset_names() -> list[str]:
     """Fetch the list of asset names from the testbed-state release.
 
     Returns an empty list if the release does not exist (404).
     Raises on other failures (outages, auth errors, etc.).
     """
+    _release_repo_accessible()
     try:
-        out = _gh("release", "view", RELEASE_TAG, "--json", "assets")
-    except subprocess.CalledProcessError as e:
-        if "not found" in (e.stderr or "").lower():
+        out = _release_gh("view", RELEASE_TAG, "--json", "assets")
+    except subprocess.CalledProcessError as error:
+        if _release_missing(error):
             return []
         raise
     return [a["name"] for a in json.loads(out).get("assets", [])]
@@ -127,5 +162,5 @@ def download_ref(ref: str, dest_dir: Path) -> Path:
         print(f"using cached {ref}.tar.zst")
         return dest
     dest_dir.mkdir(parents=True, exist_ok=True)
-    _gh("release", "download", RELEASE_TAG, "--pattern", f"{ref}.tar.zst", "--dir", str(dest_dir), "--clobber")
+    _release_gh("download", RELEASE_TAG, "--pattern", f"{ref}.tar.zst", "--dir", str(dest_dir), "--clobber")
     return dest
diff --git a/plugins/nemo-insights/tests/contracts/test_checks.py b/plugins/nemo-insights/tests/contracts/test_checks.py
new file mode 100644
index 0000000000..49e0d54235
--- /dev/null
+++ b/plugins/nemo-insights/tests/contracts/test_checks.py
@@ -0,0 +1,58 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from nemo_insights_plugin.contracts.checks import (
+    CheckResult,
+    advisories,
+    format_report,
+    make_check_result,
+    required_failures,
+)
+
+
+def test_make_check_result_maps_outcome_and_severity_to_status() -> None:
+    passed = make_check_result("ready", "runtime", True, "required", "ready", "not ready")
+    failed = make_check_result("ready", "runtime", False, "required", "ready", "not ready", hint="fix it")
+    warned = make_check_result("remote", "runtime", False, "advisory", "reachable", "unreachable")
+
+    assert passed.status == "pass"
+    assert passed.hint is None
+    assert failed.status == "fail"
+    assert failed.hint == "fix it"
+    assert warned.status == "warn"
+
+
+def test_filters_return_only_blockers_or_warnings() -> None:
+    results = [
+        CheckResult(name="pass", group="profile", status="pass", severity="required", message="ok"),
+        CheckResult(name="fail", group="profile", status="fail", severity="required", message="broken"),
+        CheckResult(name="warn", group="platform", status="warn", severity="advisory", message="offline"),
+    ]
+
+    assert [result.name for result in required_failures(results)] == ["fail"]
+    assert [result.name for result in advisories(results)] == ["warn"]
+
+
+def test_format_report_sorts_groups_and_prints_nonpassing_hints() -> None:
+    report = format_report(
+        [
+            CheckResult(
+                name="remote",
+                group="platform",
+                status="warn",
+                severity="advisory",
+                message="platform unreachable",
+                hint="start the platform",
+            ),
+            CheckResult(
+                name="profile",
+                group="profile",
+                status="pass",
+                severity="required",
+                message="profile found",
+                hint="not printed",
+            ),
+        ]
+    )
+
+    assert report == ("Platform\n  ⚠ platform unreachable\n      hint: start the platform\nProfile\n  ✓ profile found")
diff --git a/plugins/nemo-insights/tests/contracts/test_insights.py b/plugins/nemo-insights/tests/contracts/test_insights.py
new file mode 100644
index 0000000000..e9268366e4
--- /dev/null
+++ b/plugins/nemo-insights/tests/contracts/test_insights.py
@@ -0,0 +1,91 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+
+import pytest
+from nemo_insights_plugin.contracts.insights import (
+    InsightsFileError,
+    load_insights_document,
+    validate_insights_file,
+)
+
+
+def test_validate_insights_file_allows_none_missing_and_mapping_without_records(tmp_path: Path) -> None:
+    validate_insights_file(None)
+    validate_insights_file(tmp_path / "missing.yaml")
+    path = tmp_path / "insights.yaml"
+    path.write_text("metadata: retained\n", encoding="utf-8")
+    validate_insights_file(path)
+
+
+def test_load_insights_document_returns_validated_mapping(tmp_path: Path) -> None:
+    path = tmp_path / "insights.yaml"
+    path.write_text("insights:\n  - id: one\n", encoding="utf-8")
+
+    assert load_insights_document(path) == {"insights": [{"id": "one"}]}
+
+
+@pytest.mark.parametrize(
+    ("content", "message"),
+    [
+        ("- id: one\n", "YAML mapping"),
+        ("insights: null\n", "`insights` must be a list"),
+        ("insights: 42\n", "`insights` must be a list"),
+        ("insights:\n  - id: one\n  - broken\n", "item 2 must be a YAML mapping"),
+    ],
+)
+def test_invalid_insights_shapes_are_actionable(tmp_path: Path, content: str, message: str) -> None:
+    path = tmp_path / "insights.yaml"
+    path.write_text(content, encoding="utf-8")
+
+    with pytest.raises(InsightsFileError, match=message):
+        load_insights_document(path)
+
+
+def test_invalid_yaml_error_includes_path_on_one_line(tmp_path: Path) -> None:
+    path = tmp_path / "insights.yaml"
+    path.write_text("insights: [\n", encoding="utf-8")
+
+    with pytest.raises(InsightsFileError, match="valid YAML") as exc_info:
+        load_insights_document(path)
+
+    message = str(exc_info.value)
+    assert str(path) in message
+    assert "\n" not in message
+
+
+def test_invalid_utf8_is_actionable_without_raw_chain(tmp_path: Path) -> None:
+    path = tmp_path / "insights.yaml"
+    path.write_bytes(b"\xff\xfe")
+
+    with pytest.raises(InsightsFileError, match="UTF-8") as exc_info:
+        load_insights_document(path)
+
+    assert exc_info.value.__cause__ is None
+
+
+def test_generic_os_error_uses_neutral_could_not_be_read_wording(tmp_path: Path) -> None:
+    path = tmp_path / "insights.yaml"
+    path.mkdir()
+
+    with pytest.raises(InsightsFileError, match="could not be read") as exc_info:
+        load_insights_document(path)
+
+    message = str(exc_info.value)
+    assert "UTF-8" not in message
+    assert exc_info.value.__cause__ is None
+
+
+def test_validate_treats_disappearance_at_read_boundary_as_missing(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    path = tmp_path / "insights.yaml"
+    path.write_text("insights: []\n", encoding="utf-8")
+
+    def _vanished(self: Path, *args: object, **kwargs: object) -> str:
+        raise FileNotFoundError(2, "No such file or directory", str(self))
+
+    monkeypatch.setattr(Path, "read_text", _vanished)
+
+    validate_insights_file(path)
diff --git a/plugins/nemo-insights/tests/contracts/test_profile_contract.py b/plugins/nemo-insights/tests/contracts/test_profile_contract.py
new file mode 100644
index 0000000000..ce2c4903dc
--- /dev/null
+++ b/plugins/nemo-insights/tests/contracts/test_profile_contract.py
@@ -0,0 +1,144 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+
+import pytest
+from nemo_insights_plugin.contracts.profile import (
+    DEFAULT_BASE_URL,
+    EnvFileError,
+    ProfileError,
+    discover_profile,
+    load_env_file,
+    load_profile_model,
+    resolve_agent_spec_path,
+    resolve_base_url,
+    resolve_profile_path,
+)
+from pydantic import BaseModel, ConfigDict
+
+
+class TolerantProfile(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    agent: str
+    profile_dir: Path
+
+
+class StrictProfile(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    agent: str
+    profile_dir: Path
+
+
+def test_load_profile_model_respects_caller_strictness_and_injects_directory(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_text("agent: a\nexperiment_only: true\n", encoding="utf-8")
+
+    assert load_profile_model(path, TolerantProfile).profile_dir == tmp_path.resolve()
+    with pytest.raises(ProfileError, match="experiment_only"):
+        load_profile_model(path, StrictProfile)
+
+
+def test_load_profile_model_rejects_reserved_directory(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_text("agent: a\nprofile_dir: /tmp/forged\n", encoding="utf-8")
+
+    with pytest.raises(ProfileError, match="reserved"):
+        load_profile_model(path, TolerantProfile)
+
+
+def test_load_profile_model_wraps_yaml_utf8_and_shape_errors(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_bytes(b"agent: \xff")
+    with pytest.raises(ProfileError, match="UTF-8"):
+        load_profile_model(path, TolerantProfile)
+
+    path.write_text("- not-a-mapping\n", encoding="utf-8")
+    with pytest.raises(ProfileError, match="YAML mapping"):
+        load_profile_model(path, TolerantProfile)
+
+
+def test_discover_profile_walks_up_and_returns_none_when_absent(tmp_path: Path) -> None:
+    child = tmp_path / "one" / "two"
+    child.mkdir(parents=True)
+    assert discover_profile(child) is None
+
+    profile = tmp_path / "optimizer.yaml"
+    profile.write_text("agent: a\n", encoding="utf-8")
+    assert discover_profile(child) == profile
+
+
+def test_resolve_profile_path_handles_relative_absolute_and_home(tmp_path: Path) -> None:
+    relative = resolve_profile_path("./agent", tmp_path)
+    absolute = resolve_profile_path(str(tmp_path / "agent"), Path("/elsewhere"))
+
+    assert relative == (tmp_path / "agent").resolve()
+    assert absolute == (tmp_path / "agent").resolve()
+
+
+def test_load_env_file_parses_without_overriding(tmp_path: Path) -> None:
+    path = tmp_path / ".env"
+    path.write_text(
+        '# comment\nPLAIN=value\nexport EXPORTED=ok\nQUOTED="with spaces"\nSET=file\n',
+        encoding="utf-8",
+    )
+    env = {"SET": "process"}
+
+    assert load_env_file(path, env) == ["PLAIN", "EXPORTED", "QUOTED"]
+    assert env == {"SET": "process", "PLAIN": "value", "EXPORTED": "ok", "QUOTED": "with spaces"}
+    assert load_env_file(tmp_path / "missing.env", env) == []
+
+
+def test_load_env_file_wraps_read_failures_without_raw_chain(tmp_path: Path) -> None:
+    path = tmp_path / ".env"
+    path.write_bytes(b"KEY=\xff")
+
+    with pytest.raises(EnvFileError, match="readable UTF-8") as exc_info:
+        load_env_file(path, {})
+
+    assert exc_info.value.__cause__ is None
+
+
+def test_load_env_file_wraps_permission_errors(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = tmp_path / ".env"
+    path.write_text("KEY=value\n", encoding="utf-8")
+    original_read_text = Path.read_text
+
+    def deny(candidate: Path, *args: object, **kwargs: object) -> str:
+        if candidate == path:
+            raise PermissionError("permission denied")
+        return original_read_text(candidate, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", deny)
+
+    with pytest.raises(EnvFileError, match="permission denied") as exc_info:
+        load_env_file(path, {})
+
+    assert exc_info.value.__cause__ is None
+
+
+def test_resolve_agent_spec_uses_configured_then_conventional_precedence(tmp_path: Path) -> None:
+    readme = tmp_path / "README.md"
+    readme.write_text("# Readme", encoding="utf-8")
+    assert resolve_agent_spec_path(tmp_path, None) == readme
+
+    spec = tmp_path / "AGENT-SPEC.md"
+    spec.write_text("# Spec", encoding="utf-8")
+    assert resolve_agent_spec_path(tmp_path, None) == spec
+    assert resolve_agent_spec_path(tmp_path, "./README.md") == readme.resolve()
+
+    with pytest.raises(ProfileError, match="does not exist"):
+        resolve_agent_spec_path(tmp_path, "./missing.md")
+
+
+def test_resolve_base_url_uses_only_explicit_nmp_and_default() -> None:
+    env = {"NMP_BASE_URL": "http://nmp", "NEMO_BASE_URL": "http://ignored"}
+
+    assert resolve_base_url("http://flag", env) == "http://flag"
+    assert resolve_base_url(None, env) == "http://nmp"
+    assert resolve_base_url(None, {"NEMO_BASE_URL": "http://ignored"}) == DEFAULT_BASE_URL
diff --git a/plugins/nemo-insights/tests/test_cli_profile.py b/plugins/nemo-insights/tests/test_cli_profile.py
new file mode 100644
index 0000000000..f68131f3b3
--- /dev/null
+++ b/plugins/nemo-insights/tests/test_cli_profile.py
@@ -0,0 +1,684 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+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, preflight
+from nemo_insights_plugin.contracts.profile import DEFAULT_BASE_URL
+from nemo_insights_plugin.preflight import AnalysisProbes
+from nemo_platform import NeMoPlatformError
+from pydantic_ai import AgentRunError
+from typer.testing import CliRunner
+
+runner = CliRunner()
+
+
+class AnalystRecorder:
+    def __init__(self) -> None:
+        self.kwargs: dict[str, object] | None = None
+
+    async def __call__(self, **kwargs: object) -> str:
+        self.kwargs = kwargs
+        return "analysis-summary"
+
+
+@pytest.fixture
+def app() -> typer.Typer:
+    return cli.InsightsCLI().get_cli()
+
+
+@pytest.fixture(autouse=True)
+def quiet_preflight(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def queryable(base_url: str, workspace: str, agent: str) -> bool:
+        return True
+
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: True,
+            workspace_ok=queryable,
+        ),
+    )
+
+
+@pytest.fixture
+def profile_tree(tmp_path: Path) -> Path:
+    (tmp_path / "optimizer.yaml").write_text(
+        "agent: flight-planner\n"
+        "task_template: ./evals/task_template\n"
+        "datasets:\n  train: ./evals/train\n  validation: ./evals/validation\n"
+        "workspace: flight-workspace\n",
+        encoding="utf-8",
+    )
+    (tmp_path / "AGENT-SPEC.md").write_text("# Flight planner", encoding="utf-8")
+    return tmp_path
+
+
+def test_analyze_runs_flag_free_from_profile(app: typer.Typer, profile_tree: Path, monkeypatch) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["agent"] == "flight-planner"
+    assert recorder.kwargs["workspace"] == "flight-workspace"
+    assert recorder.kwargs["agent_spec"] == "# Flight planner"
+    assert recorder.kwargs["insights_output"] == profile_tree / ".nemo-optimizer" / "insights.yaml"
+
+
+def test_analyze_flags_override_profile(app: typer.Typer, profile_tree: Path, monkeypatch) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze", "--agent", "other", "--workspace", "other-ws"])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["agent"] == "other"
+    assert recorder.kwargs["workspace"] == "other-ws"
+
+
+def test_profile_env_is_loaded_before_base_url_resolution(app: typer.Typer, profile_tree: Path, monkeypatch) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.delenv("NMP_BASE_URL", raising=False)
+    (profile_tree / ".env").write_text("NMP_BASE_URL=https://platform.example\n", encoding="utf-8")
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["base_url"] == "https://platform.example"
+
+
+def test_analyze_renders_invalid_profile_env_as_command_error(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    recorder = AnalystRecorder()
+    probe_calls: list[str] = []
+
+    async def record_workspace_probe(base_url: str, workspace: str, agent: str) -> bool:
+        probe_calls.append("workspace")
+        return True
+
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: probe_calls.append("http") or True,
+            workspace_ok=record_workspace_probe,
+        ),
+    )
+    env_file = profile_tree / ".env"
+    env_file.write_bytes(b"KEY=\xff")
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 1
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert len(error_lines) == 1
+    assert error_lines[0].startswith(f"Error: Could not read environment file {env_file}:")
+    assert "Check that the file is readable UTF-8 text, then retry." in error_lines[0]
+    assert "Traceback" not in result.output
+    assert recorder.kwargs is None
+    assert probe_calls == []
+
+
+def test_doctor_renders_invalid_profile_env_as_command_error(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    recorder = AnalystRecorder()
+    probe_calls: list[str] = []
+
+    async def record_workspace_probe(base_url: str, workspace: str, agent: str) -> bool:
+        probe_calls.append("workspace")
+        return True
+
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: probe_calls.append("http") or True,
+            workspace_ok=record_workspace_probe,
+        ),
+    )
+    env_file = profile_tree / ".env"
+    env_file.write_bytes(b"KEY=\xff")
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["doctor"])
+
+    assert result.exit_code == 1
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert len(error_lines) == 1
+    assert error_lines[0].startswith(f"Error: Could not read environment file {env_file}:")
+    assert "Check that the file is readable UTF-8 text, then retry." in error_lines[0]
+    assert "Traceback" not in result.output
+    assert recorder.kwargs is None
+    assert probe_calls == []
+
+
+@pytest.mark.parametrize(
+    ("arguments", "environment", "profile_env", "expected"),
+    [
+        (
+            ["--base-url", "https://flag.example"],
+            {"NMP_BASE_URL": "https://process.example"},
+            "NMP_BASE_URL=https://profile.example\n",
+            "https://flag.example",
+        ),
+        ([], {}, "NMP_BASE_URL=https://profile.example\n", "https://profile.example"),
+        (
+            [],
+            {"NMP_BASE_URL": "https://process.example"},
+            "NMP_BASE_URL=https://profile.example\n",
+            "https://process.example",
+        ),
+        ([], {"NEMO_BASE_URL": "https://legacy.example"}, None, DEFAULT_BASE_URL),
+    ],
+    ids=["explicit", "profile-env", "process-env", "legacy-ignored"],
+)
+def test_doctor_resolves_base_url_after_profile_env_loading(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    arguments: list[str],
+    environment: dict[str, str],
+    profile_env: str | None,
+    expected: str,
+) -> None:
+    http_urls: list[str] = []
+    workspace_urls: list[str] = []
+
+    async def record_workspace_probe(base_url: str, workspace: str, agent: str) -> bool:
+        workspace_urls.append(base_url)
+        return True
+
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: http_urls.append(base_url) or True,
+            workspace_ok=record_workspace_probe,
+        ),
+    )
+    monkeypatch.delenv("NMP_BASE_URL", raising=False)
+    monkeypatch.delenv("NEMO_BASE_URL", raising=False)
+    for name, value in environment.items():
+        monkeypatch.setenv(name, value)
+    if profile_env is not None:
+        (profile_tree / ".env").write_text(profile_env, encoding="utf-8")
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["doctor", *arguments])
+
+    assert result.exit_code == 0, result.output
+    assert http_urls == [expected]
+    assert workspace_urls == [expected]
+
+
+@pytest.mark.parametrize(
+    ("arguments", "environment", "expected"),
+    [
+        (
+            ["--base-url", "https://flag.example"],
+            {"NMP_BASE_URL": "https://nmp.example", "NEMO_BASE_URL": "https://legacy.example"},
+            "https://flag.example",
+        ),
+        (
+            [],
+            {"NMP_BASE_URL": "https://nmp.example", "NEMO_BASE_URL": "https://legacy.example"},
+            "https://nmp.example",
+        ),
+        ([], {"NEMO_BASE_URL": "https://legacy.example"}, DEFAULT_BASE_URL),
+    ],
+)
+def test_base_url_precedence_uses_only_nmp_base_url(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    arguments: list[str],
+    environment: dict[str, str],
+    expected: str,
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.delenv("NMP_BASE_URL", raising=False)
+    monkeypatch.delenv("NEMO_BASE_URL", raising=False)
+    for name, value in environment.items():
+        monkeypatch.setenv(name, value)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze", *arguments])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["base_url"] == expected
+
+
+def test_explicit_profile_is_used_outside_profile_directory(
+    app: typer.Typer, profile_tree: Path, tmp_path: Path, monkeypatch
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["analyze", "--profile", str(profile_tree / "optimizer.yaml")])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["agent"] == "flight-planner"
+
+
+def test_malformed_explicit_profile_errors(app: typer.Typer, tmp_path: Path, monkeypatch) -> None:
+    profile = tmp_path / "optimizer.yaml"
+    profile.write_text("agent: ''\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["analyze", "--profile", str(profile)])
+
+    assert result.exit_code != 0
+    assert "Invalid profile" in result.output
+
+
+def test_malformed_discovered_profile_warns_when_flags_are_complete(
+    app: typer.Typer, tmp_path: Path, monkeypatch
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    (tmp_path / "optimizer.yaml").write_text("agent: ''\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["analyze", "--agent", "other", "--workspace", "other-ws"])
+
+    assert result.exit_code == 0, result.output
+    assert "warning:" in result.output
+    assert "Invalid profile" in result.output
+
+
+def test_malformed_discovered_profile_errors_without_explicit_workspace(
+    app: typer.Typer, tmp_path: Path, monkeypatch
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    (tmp_path / "optimizer.yaml").write_text("agent: ''\n", encoding="utf-8")
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["analyze", "--agent", "other"])
+
+    assert result.exit_code != 0
+    assert "Invalid profile" in result.output
+    assert recorder.kwargs is None
+
+
+def test_explicit_output_overrides_profile_default(
+    app: typer.Typer, profile_tree: Path, tmp_path: Path, monkeypatch
+) -> None:
+    recorder = AnalystRecorder()
+    output = tmp_path / "custom.yaml"
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze", "--insights-file-output", str(output)])
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+    assert recorder.kwargs["insights_output"] == output
+
+
+def test_missing_profile_and_agent_errors(app: typer.Typer, tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code != 0
+    assert "No --agent given and no optimizer.yaml profile found" in result.output
+
+
+def test_analyze_blocks_before_runner_when_preflight_fails(
+    app: typer.Typer, profile_tree: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={},
+            http_ok=lambda base_url: True,
+            workspace_ok=lambda base_url, workspace, agent: _queryable(),
+        ),
+    )
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 1
+    assert "INFERENCE_API_KEY not set" in result.output
+    assert recorder.kwargs is None
+
+
+def test_analyze_prints_advisory_and_runs_analyst(
+    app: typer.Typer, profile_tree: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    recorder = AnalystRecorder()
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: True,
+            workspace_ok=lambda base_url, workspace, agent: _not_queryable(),
+        ),
+    )
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 0, result.output
+    assert "workspace 'flight-workspace' could not be queried" in result.stderr
+    assert recorder.kwargs is not None
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        NeMoPlatformError("Intake SDK failed"),
+        httpx.ConnectError("Intake unavailable", request=httpx.Request("GET", "https://platform.example")),
+        OSError("could not read SDK configuration"),
+    ],
+)
+def test_analyze_renders_expected_platform_failures_without_traceback(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    error: Exception,
+) -> None:
+    async def fail_analysis(**kwargs: object) -> str:
+        raise error
+
+    monkeypatch.setattr(cli, "run_analyst", fail_analysis)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: True,
+            workspace_ok=lambda base_url, workspace, agent: _not_queryable(),
+        ),
+    )
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 1
+    assert "workspace 'flight-workspace' could not be queried" in result.stderr
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert len(error_lines) == 1
+    assert "analysis failed" in error_lines[0]
+    assert "--base-url/NMP_BASE_URL" in error_lines[0]
+    assert "Traceback" not in result.output
+
+
+def test_analyze_renders_agent_run_error_with_model_and_usage_guidance(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def fail_analysis(**kwargs: object) -> str:
+        raise AgentRunError("request limit exceeded")
+
+    monkeypatch.setattr(cli, "run_analyst", fail_analysis)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert result.exit_code == 1
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert error_lines == [
+        "Error: analyst run failed: request limit exceeded. "
+        "Check inference model access and credentials, then retry or adjust usage limits."
+    ]
+    assert "--base-url/NMP_BASE_URL" not in result.stderr
+    assert "Intake availability" not in result.stderr
+    assert "Traceback" not in result.output
+
+
+@pytest.mark.parametrize("error_type", [RuntimeError, ValueError])
+def test_analyze_constructor_failure_warns_then_exits_cleanly(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    error_type: type[Exception],
+) -> None:
+    attempts = 0
+
+    def fail_to_construct(base_url: str | None) -> object:
+        nonlocal attempts
+        attempts += 1
+        raise error_type("invalid\nremote client context")
+
+    monkeypatch.setattr(preflight, "make_client", fail_to_construct)
+    monkeypatch.setattr(analyst_run, "make_client", fail_to_construct)
+    monkeypatch.setattr(
+        cli,
+        "_PREFLIGHT_PROBES",
+        AnalysisProbes(
+            env={"INFERENCE_API_KEY": "k"},
+            http_ok=lambda base_url: True,
+            workspace_ok=preflight._default_workspace_ok,
+        ),
+    )
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["analyze"])
+
+    assert attempts == 2
+    assert result.exit_code == 1
+    assert "workspace 'flight-workspace' could not be queried" in result.stderr
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert error_lines == [
+        "Error: analysis failed: invalid remote client context. "
+        "Check --base-url/NMP_BASE_URL, authentication, workspace, and Intake availability."
+    ]
+    assert "analyst run failed" not in result.stderr
+    assert "Traceback" not in result.output
+    assert "During handling of the above exception" not in result.output
+
+
+@pytest.mark.parametrize("explicit", [False, True], ids=["profile-default", "explicit-path"])
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        (b"insights: [\n", "valid YAML"),
+        (b"- id: first\n", "YAML mapping"),
+        (b"scalar\n", "YAML mapping"),
+        (b"\xff\xfe", "UTF-8"),
+    ],
+    ids=["malformed-yaml", "list-root", "scalar-root", "invalid-utf8"],
+)
+def test_analyze_rejects_invalid_existing_insights_file_before_runner(
+    app: typer.Typer,
+    profile_tree: Path,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    explicit: bool,
+    payload: bytes,
+    expected: str,
+) -> None:
+    recorder = AnalystRecorder()
+    output = tmp_path / "explicit-insights.yaml" if explicit else profile_tree / ".nemo-optimizer" / "insights.yaml"
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_bytes(payload)
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+    arguments = ["analyze", "--insights-file-output", str(output)] if explicit else ["analyze"]
+
+    result = runner.invoke(app, arguments)
+
+    assert result.exit_code == 1
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert len(error_lines) == 1
+    assert f"insights file {output}" in error_lines[0]
+    assert expected in error_lines[0]
+    assert "Traceback" not in result.output
+    assert recorder.kwargs is None
+
+
+@pytest.mark.parametrize("explicit", [False, True], ids=["profile-default", "explicit-path"])
+@pytest.mark.parametrize(
+    ("payload", "expected"),
+    [
+        (b"insights: null\n", "`insights` must be a list"),
+        (b"insights: 42\n", "`insights` must be a list"),
+        (b"insights: records\n", "`insights` must be a list"),
+        (b"insights: {id: one}\n", "`insights` must be a list"),
+        (
+            b"insights:\n  - {id: one}\n  - broken\n",
+            "`insights` item 2 must be a YAML mapping",
+        ),
+    ],
+    ids=["null", "numeric", "scalar", "mapping", "scalar-list-item"],
+)
+def test_analyze_rejects_invalid_insights_records_before_runner(
+    app: typer.Typer,
+    profile_tree: Path,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    explicit: bool,
+    payload: bytes,
+    expected: str,
+) -> None:
+    recorder = AnalystRecorder()
+    output = tmp_path / "explicit-insights.yaml" if explicit else profile_tree / ".nemo-optimizer" / "insights.yaml"
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_bytes(payload)
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+    arguments = ["analyze", "--insights-file-output", str(output)] if explicit else ["analyze"]
+
+    result = runner.invoke(app, arguments)
+
+    assert result.exit_code == 1
+    error_lines = [line for line in result.stderr.splitlines() if line.startswith("Error:")]
+    assert len(error_lines) == 1
+    assert f"insights file {output}" in error_lines[0]
+    assert expected in error_lines[0]
+    assert "Traceback" not in result.output
+    assert recorder.kwargs is None
+
+
+@pytest.mark.parametrize("explicit", [False, True], ids=["profile-default", "explicit-path"])
+def test_analyze_accepts_existing_insights_file_without_insights_key(
+    app: typer.Typer,
+    profile_tree: Path,
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    explicit: bool,
+) -> None:
+    recorder = AnalystRecorder()
+    output = tmp_path / "explicit-insights.yaml" if explicit else profile_tree / ".nemo-optimizer" / "insights.yaml"
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text("metadata: retained\n", encoding="utf-8")
+    monkeypatch.setattr(cli, "run_analyst", recorder)
+    monkeypatch.chdir(profile_tree)
+    arguments = ["analyze", "--insights-file-output", str(output)] if explicit else ["analyze"]
+
+    result = runner.invoke(app, arguments)
+
+    assert result.exit_code == 0, result.output
+    assert recorder.kwargs is not None
+
+
+@pytest.mark.parametrize("command", ["doctor", "analyze"])
+def test_commands_reject_invalid_utf8_agent_spec(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    command: str,
+) -> None:
+    (profile_tree / "AGENT-SPEC.md").write_bytes(b"\xff\xfe")
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, [command])
+
+    assert result.exit_code == 1
+    assert "agent spec" in result.output.lower()
+    assert "UTF-8" in result.output
+    assert "Traceback" not in result.output
+
+
+@pytest.mark.parametrize("command", ["doctor", "analyze"])
+def test_commands_reject_unreadable_agent_spec(
+    app: typer.Typer,
+    profile_tree: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    command: str,
+) -> None:
+    spec = profile_tree / "AGENT-SPEC.md"
+    original_read_text = Path.read_text
+
+    def deny_spec_read(path: Path, encoding: str | None = None, errors: str | None = None) -> str:
+        if path == spec:
+            raise PermissionError("permission denied")
+        return original_read_text(path, encoding=encoding, errors=errors)
+
+    monkeypatch.setattr(Path, "read_text", deny_spec_read)
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, [command])
+
+    assert result.exit_code == 1
+    assert "agent spec" in result.output.lower()
+    assert "permission denied" in result.output
+    assert "ensure the file is readable and encoded as UTF-8" in result.output
+    assert "Traceback" not in result.output
+
+
+async def _queryable() -> bool:
+    return True
+
+
+async def _not_queryable() -> bool:
+    return False
+
+
+def test_doctor_exits_nonzero_for_missing_profile(app: typer.Typer, tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.chdir(tmp_path)
+
+    result = runner.invoke(app, ["doctor"])
+
+    assert result.exit_code == 1
+    assert "no optimizer.yaml found" in result.output
+
+
+def test_doctor_reports_healthy_profile(app: typer.Typer, profile_tree: Path, monkeypatch) -> None:
+    monkeypatch.chdir(profile_tree)
+
+    result = runner.invoke(app, ["doctor"])
+
+    assert result.exit_code == 0, result.output
+    assert "Profile\n  ✓ profile for agent 'flight-planner'" in result.output
+    assert "Credentials\n  ✓ INFERENCE_API_KEY set" in result.output
diff --git a/plugins/nemo-insights/tests/test_periodic_analysis.py b/plugins/nemo-insights/tests/test_periodic_analysis.py
index fba5618914..637c71bc13 100644
--- a/plugins/nemo-insights/tests/test_periodic_analysis.py
+++ b/plugins/nemo-insights/tests/test_periodic_analysis.py
@@ -11,6 +11,7 @@
 import httpx
 import pytest
 from nemo_insights_plugin.analyst.analyst_backend import (
+    LocalAnalystBackend,
     RemoteAnalystBackend,
     _merge_eval_filter,
     _merge_since_filter,
@@ -94,6 +95,33 @@ async def test_remote_persist_validates_updates_without_trace_refs() -> None:
     assert "- updated: missing-insight" not in report
 
 
+def test_local_backend_reads_and_writes_insights_file_with_explicit_utf8(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    read_calls: list[dict[str, object]] = []
+    write_calls: list[dict[str, object]] = []
+    original_read_text = Path.read_text
+    original_write_text = Path.write_text
+
+    def spy_read_text(self: Path, *args: object, **kwargs: object) -> str:
+        read_calls.append(kwargs)
+        return original_read_text(self, *args, **kwargs)
+
+    def spy_write_text(self: Path, *args: object, **kwargs: object) -> int:
+        write_calls.append(kwargs)
+        return original_write_text(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "read_text", spy_read_text)
+    monkeypatch.setattr(Path, "write_text", spy_write_text)
+
+    backend = LocalAnalystBackend(client=SimpleNamespace(), path=tmp_path / "insights.yaml")  # type: ignore[arg-type]
+    backend._write_records([])
+    backend._read_records()
+
+    assert write_calls[-1].get("encoding") == "utf-8"
+    assert read_calls[-1].get("encoding") == "utf-8"
+
+
 def test_merge_eval_filter_pins_evaluation_id() -> None:
     assert _merge_eval_filter({"agent_name": "a"}, evaluation_id="run-1") == {
         "agent_name": "a",
diff --git a/plugins/nemo-insights/tests/test_preflight.py b/plugins/nemo-insights/tests/test_preflight.py
new file mode 100644
index 0000000000..8514b865e8
--- /dev/null
+++ b/plugins/nemo-insights/tests/test_preflight.py
@@ -0,0 +1,153 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import asyncio
+from pathlib import Path
+
+import httpx
+import pytest
+from nemo_insights_plugin import preflight
+from nemo_insights_plugin.contracts.checks import format_report, required_failures
+from nemo_insights_plugin.preflight import (
+    AnalysisProbes,
+    check_agent_spec,
+    check_environment,
+    check_profile,
+)
+from nemo_insights_plugin.profile import AnalysisProfile
+from nemo_platform import NeMoPlatformError
+
+
+async def always_queryable(base_url: str, workspace: str, agent: str) -> bool:
+    return True
+
+
+async def never_queryable(base_url: str, workspace: str, agent: str) -> bool:
+    return False
+
+
+def test_missing_inference_key_is_required_failure(tmp_path: Path) -> None:
+    results = asyncio.run(
+        check_environment(
+            agent="a",
+            workspace="default",
+            base_url="http://localhost:8080",
+            profile_dir=tmp_path,
+            probes=AnalysisProbes(
+                env={},
+                http_ok=lambda base_url: True,
+                workspace_ok=always_queryable,
+            ),
+        )
+    )
+
+    assert any(result.name == "INFERENCE_API_KEY" and result.status == "fail" for result in results)
+    assert required_failures(results)
+
+
+def test_workspace_query_failure_is_advisory(tmp_path: Path) -> None:
+    results = asyncio.run(
+        check_environment(
+            agent="a",
+            workspace="missing",
+            base_url="http://localhost:8080",
+            profile_dir=tmp_path,
+            probes=AnalysisProbes(
+                env={"INFERENCE_API_KEY": "k"},
+                http_ok=lambda base_url: True,
+                workspace_ok=never_queryable,
+            ),
+        )
+    )
+
+    warning = next(result for result in results if result.name == "workspace-query")
+    assert warning.status == "warn"
+    assert warning.severity == "advisory"
+
+
+@pytest.mark.parametrize(
+    "error",
+    [
+        httpx.ConnectError("OIDC discovery failed", request=httpx.Request("GET", "https://platform.example")),
+        NeMoPlatformError("SDK initialization failed"),
+        RuntimeError("NeMoPlatform client initialization failed: invalid context"),
+        ValueError("invalid remote configuration"),
+        OSError("could not read SDK configuration"),
+    ],
+)
+def test_remote_workspace_probe_treats_client_construction_failures_as_advisory(
+    monkeypatch: pytest.MonkeyPatch,
+    error: Exception,
+) -> None:
+    def fail_to_construct(base_url: str) -> object:
+        raise error
+
+    monkeypatch.setattr(preflight, "make_client", fail_to_construct)
+
+    assert asyncio.run(preflight._default_workspace_ok("https://platform.example", "default", "agent")) is False
+
+
+def test_profile_and_agent_spec_failures_are_required(tmp_path: Path) -> None:
+    profile_results = check_profile(None, None)
+    spec_results = check_agent_spec(None, "configured agent spec does not exist")
+
+    assert required_failures(profile_results) == profile_results
+    assert required_failures(spec_results) == spec_results
+
+
+def test_agent_spec_invalid_utf8_is_required_failure(tmp_path: Path) -> None:
+    spec = tmp_path / "AGENT-SPEC.md"
+    spec.write_bytes(b"\xff\xfe")
+
+    results = check_agent_spec(spec, None)
+
+    assert results[0].status == "fail"
+    assert results[0].severity == "required"
+    assert "UTF-8" in results[0].message
+    assert results[0].hint == "ensure the file is readable and encoded as UTF-8"
+
+
+def test_agent_spec_unreadable_is_required_failure(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    spec = tmp_path / "AGENT-SPEC.md"
+    spec.write_text("# Agent", encoding="utf-8")
+    original_read_text = Path.read_text
+
+    def deny_spec_read(path: Path, encoding: str | None = None, errors: str | None = None) -> str:
+        if path == spec:
+            raise PermissionError("permission denied")
+        return original_read_text(path, encoding=encoding, errors=errors)
+
+    monkeypatch.setattr(Path, "read_text", deny_spec_read)
+
+    results = check_agent_spec(spec, None)
+
+    assert results[0].status == "fail"
+    assert results[0].severity == "required"
+    assert "permission denied" in results[0].message
+    assert results[0].hint == "ensure the file is readable and encoded as UTF-8"
+
+
+def test_healthy_setup_formats_grouped_report(tmp_path: Path) -> None:
+    profile = AnalysisProfile(agent="a", profile_dir=tmp_path)
+    (tmp_path / "AGENT-SPEC.md").write_text("# Agent", encoding="utf-8")
+    results = check_profile(profile, None) + check_agent_spec(tmp_path / "AGENT-SPEC.md", None)
+    results += asyncio.run(
+        check_environment(
+            agent="a",
+            workspace="default",
+            base_url="http://localhost:8080",
+            profile_dir=tmp_path,
+            probes=AnalysisProbes(
+                env={"INFERENCE_API_KEY": "k"},
+                http_ok=lambda base_url: True,
+                workspace_ok=always_queryable,
+            ),
+        )
+    )
+
+    report = format_report(results)
+
+    assert "Profile\n  ✓ profile for agent 'a'" in report
+    assert "Credentials\n  ✓ INFERENCE_API_KEY set" in report
+    assert "Platform\n  ✓ http://localhost:8080 reachable" in report
+    assert not required_failures(results)
diff --git a/plugins/nemo-insights/tests/test_profile.py b/plugins/nemo-insights/tests/test_profile.py
new file mode 100644
index 0000000000..1d798d54fb
--- /dev/null
+++ b/plugins/nemo-insights/tests/test_profile.py
@@ -0,0 +1,51 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+
+import pytest
+from nemo_insights_plugin.contracts.profile import ProfileError
+from nemo_insights_plugin.profile import load_profile, pick_agent_spec
+
+FULL_PROFILE = """\
+agent: flight-planner
+task_template: ./evals/task_template
+datasets:
+  train: ./evals/train
+  validation: ./evals/validation
+experiment_config:
+  optimization:
+    rounds: 2
+framework_skills: [./skills]
+workspace: flight-workspace
+agent_spec: ./AGENT-SPEC.md
+"""
+
+
+def test_load_profile_reads_analysis_fields_and_ignores_experiment_fields(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_text(FULL_PROFILE, encoding="utf-8")
+
+    profile = load_profile(path)
+
+    assert profile.agent == "flight-planner"
+    assert profile.workspace == "flight-workspace"
+    assert profile.agent_spec == "./AGENT-SPEC.md"
+    assert profile.profile_dir == tmp_path.resolve()
+
+
+def test_profile_requires_nonempty_agent(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_text("agent: ''\n", encoding="utf-8")
+
+    with pytest.raises(ProfileError, match="agent"):
+        load_profile(path)
+
+
+def test_pick_agent_spec_is_profile_relative(tmp_path: Path) -> None:
+    path = tmp_path / "optimizer.yaml"
+    path.write_text("agent: a\nagent_spec: ./AGENT-SPEC.md\n", encoding="utf-8")
+    expected = tmp_path / "AGENT-SPEC.md"
+    expected.write_text("# Agent", encoding="utf-8")
+
+    assert pick_agent_spec(load_profile(path)) == expected.resolve()
diff --git a/plugins/nemo-insights/tests/testbed/test_cli.py b/plugins/nemo-insights/tests/testbed/test_cli.py
index e495221a95..d60e119f39 100644
--- a/plugins/nemo-insights/tests/testbed/test_cli.py
+++ b/plugins/nemo-insights/tests/testbed/test_cli.py
@@ -111,6 +111,7 @@ def fake_ingest(base_url, export_dir, manifest, *, workspace_map, catalog, **kw)
                 "manifest": manifest,
                 "workspace_map": dict(workspace_map),
                 "catalog": catalog,
+                "require_empty": kw.get("require_empty", False),
             }
         )
         return {
@@ -921,6 +922,45 @@ async def fake_analyze(self, *, record, since, verbose, out_path):
 # --------------------------------------------------------------------------- #
 
 
+def test_restore_into_uses_exact_target(monkeypatch, tmp_path, stub_reingest) -> None:
+    bundle = _make_export_bundle(tmp_path / "state.tar.zst", workspaces=("source-workspace",))
+    monkeypatch.setattr(sys, "argv", ["testbed", "restore", str(bundle), "--into", "stable-workspace"])
+
+    cli.main()
+
+    call = stub_reingest["ingest"][0]
+    assert call["workspace_map"] == {"source-workspace": "stable-workspace"}
+    assert call["require_empty"] is True
+
+
+def test_restore_into_rejects_invalid_target_before_catalog_or_ingest(
+    monkeypatch,
+    tmp_path,
+    stub_reingest,
+) -> None:
+    bundle = _make_export_bundle(tmp_path / "state.tar.zst", workspaces=("source-workspace",))
+    monkeypatch.setattr(sys, "argv", ["testbed", "restore", str(bundle), "--into", "INVALID!"])
+
+    with pytest.raises(SystemExit, match="platform naming rule"):
+        cli.main()
+
+    assert stub_reingest["platform_root"] == []
+    assert stub_reingest["catalog_root"] == []
+    assert stub_reingest["ingest"] == []
+
+
+def test_restore_into_help_states_fresh_target_contract(monkeypatch, capsys) -> None:
+    monkeypatch.setattr(sys, "argv", ["testbed", "restore", "--help"])
+
+    with pytest.raises(SystemExit) as exc:
+        cli.main()
+
+    assert exc.value.code == 0
+    output = " ".join(capsys.readouterr().out.split())
+    assert "fresh, empty target" in output
+    assert "not idempotent" in output
+
+
 def test_restore_export_bundle_reingests_with_digest_suffix(monkeypatch, tmp_path, capsys, stub_reingest):
     monkeypatch.setattr(cli, "TMP", tmp_path)
     bundle = _make_export_bundle(tmp_path / "b.tar.zst")
diff --git a/plugins/nemo-insights/tests/testbed/test_publish.py b/plugins/nemo-insights/tests/testbed/test_publish.py
index 4c4007f5eb..70c91b0f5c 100644
--- a/plugins/nemo-insights/tests/testbed/test_publish.py
+++ b/plugins/nemo-insights/tests/testbed/test_publish.py
@@ -14,6 +14,7 @@
 from testbed import cli, publish, release
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
+PLATFORM_ROOT = REPO_ROOT.parents[1]
 
 MANIFEST = {
     "kind": "testbed-export",
@@ -159,7 +160,19 @@ def test_publish_mints_next_ref_uploads_and_prepends_row(fake_gh, tmp_path, monk
     assert ref == "state-v7"
     assert (tmp_path / "state-v7.tar.zst").is_file()  # candidate copied to the ref name
     uploads = [c for c in fake_gh["calls"] if c[:2] == ("release", "upload")]
-    assert uploads == [("release", "upload", release.RELEASE_TAG, str(tmp_path / "state-v7.tar.zst"), "--clobber")]
+    assert uploads == [
+        (
+            "release",
+            "upload",
+            release.RELEASE_TAG,
+            str(tmp_path / "state-v7.tar.zst"),
+            "--repo",
+            release.DEFAULT_STATE_REPO,
+        )
+    ]
+    assert all(
+        call[-2:] == ("--repo", release.DEFAULT_STATE_REPO) for call in fake_gh["calls"] if call[:1] == ("release",)
+    )
     # new row lands right under the header separator, above the old row
     assert fake_gh["body"].index("state-v7") < fake_gh["body"].index("| state-v6 |")
     assert "laptop (ada)" in fake_gh["body"]
@@ -170,7 +183,7 @@ def test_publish_mints_next_ref_uploads_and_prepends_row(fake_gh, tmp_path, monk
 def test_publish_creates_release_when_missing(fake_gh, tmp_path, monkeypatch):
     def view_fails(*args):
         if args[:2] == ("release", "view") and "--json" not in args:
-            raise subprocess.CalledProcessError(1, ["gh", *args], stderr="not found")
+            raise subprocess.CalledProcessError(1, ["gh", *args], stderr="release not found")
         return original_gh(*args)
 
     original_gh = release._gh
@@ -180,6 +193,21 @@ def view_fails(*args):
     assert len(creates) == 1 and release.RELEASE_TAG in creates[0]
 
 
+def test_publish_does_not_create_release_after_auth_failure(fake_gh, tmp_path, monkeypatch):
+    original_gh = release._gh
+
+    def view_fails(*args):
+        if args[:2] == ("release", "view") and "--json" not in args:
+            raise subprocess.CalledProcessError(1, ["gh", *args], stderr="HTTP 403: Resource not accessible\n")
+        return original_gh(*args)
+
+    monkeypatch.setattr(release, "_gh", view_fails)
+    with pytest.raises(subprocess.CalledProcessError):
+        publish.publish(_make_bundle(tmp_path / "c.tar.zst"), reason="", env={})
+    assert not any(call[:2] == ("release", "create") for call in fake_gh["calls"])
+    assert not any(call[:2] == ("release", "upload") for call in fake_gh["calls"])
+
+
 def test_publish_writes_github_output_only_when_env_set(fake_gh, tmp_path):
     out_file = tmp_path / "gh_output"
     publish.publish(_make_bundle(tmp_path / "a.tar.zst"), reason="", env={"GITHUB_OUTPUT": str(out_file)})
@@ -187,6 +215,13 @@ def test_publish_writes_github_output_only_when_env_set(fake_gh, tmp_path):
     publish.publish(_make_bundle(tmp_path / "b.tar.zst"), reason="", env={})  # no env -> nothing written
 
 
+def test_publish_reason_falls_back_to_candidate_manifest(fake_gh, tmp_path):
+    manifest = {**MANIFEST, "reason": "captured during produce"}
+    publish.publish(_make_bundle(tmp_path / "candidate.tar.zst", manifest=manifest), reason=None, env={})
+
+    assert "captured during produce" in fake_gh["body"]
+
+
 def test_publish_rejects_non_export_bundles(fake_gh, tmp_path):
     legacy = _make_bundle(tmp_path / "legacy.tar.zst", manifest={"created_at": "2026-01-01"})
     with pytest.raises(SystemExit) as exc:
@@ -320,3 +355,12 @@ def test_cli_publish_base_and_no_verify_conflict(fake_gh, tmp_path, monkeypatch,
         cli.main()
     assert exc.value.code == 2  # argparse mutual-exclusion error
     assert all(c[:2] != ("release", "upload") for c in fake_gh["calls"])
+
+
+def test_workflow_protects_all_secrets_and_exports_state_repository():
+    workflow = (PLATFORM_ROOT / ".github" / "workflows" / "insights-testbed.yml").read_text(encoding="utf-8")
+    produce_job, analyze_job = workflow.split("\n  produce:\n", 1)[1].split("\n  analyze:\n", 1)
+
+    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
diff --git a/plugins/nemo-insights/tests/testbed/test_reingest.py b/plugins/nemo-insights/tests/testbed/test_reingest.py
index 96a120cb14..aa3f05abef 100644
--- a/plugins/nemo-insights/tests/testbed/test_reingest.py
+++ b/plugins/nemo-insights/tests/testbed/test_reingest.py
@@ -342,6 +342,7 @@ def quiet_platform(monkeypatch):
         "posts": [],
         "ensured": [],
         "counts": [],
+        "events": [],
         "span_counts": [],
         "annotation_counts": [],
         "result_counts": [],
@@ -351,24 +352,53 @@ def quiet_platform(monkeypatch):
     def _counter(name):
         def fake(base_url, workspace, *, client=None):
             calls["counts"].append((name, workspace))
+            calls["events"].append(("count", name, workspace))
             return calls[f"{name}_counts"].pop(0)
 
         return fake
 
     def fake_first(base_url, workspace, *, client=None):
         calls["counts"].append(("first_span", workspace))
+        calls["events"].append(("count", "first_span", workspace))
         # default (no preloaded id): probe unavailable -> count-only fallback
         return calls["first_ids"].pop(0) if calls["first_ids"] else None
 
+    def fake_ensure(url, workspace, *, client=None):
+        calls["ensured"].append(workspace)
+        calls["events"].append(("ensure", workspace))
+
+    def fake_export(url, workspace, request, *, client=None):
+        calls["requests"].append((workspace, request))
+        calls["events"].append(("write", "span", workspace))
+
+    def fake_post(client, url, body):
+        calls["posts"].append((url, body))
+        endpoint = "annotation" if url.endswith("/annotations") else "result"
+        workspace = url.split("/workspaces/", 1)[1].split("/", 1)[0]
+        calls["events"].append(("write", endpoint, workspace))
+
+    real_client = httpx.Client
+
+    def reject_unexpected_request(request: httpx.Request) -> httpx.Response:
+        if request.method in {"POST", "PUT", "PATCH", "DELETE"}:
+            calls["events"].append(("write", request.method, request.url.path))
+            raise AssertionError(f"unexpected {request.method} mutation: {request.url}")
+        raise AssertionError(f"unexpected HTTP request: {request.method} {request.url}")
+
+    transport = httpx.MockTransport(reject_unexpected_request)
+
+    def client_factory(*args, **kwargs):
+        kwargs["transport"] = transport
+        return real_client(*args, **kwargs)
+
     monkeypatch.setattr(reingest, "span_count", _counter("span"))
     monkeypatch.setattr(reingest, "annotation_count", _counter("annotation"))
     monkeypatch.setattr(reingest, "evaluator_result_count", _counter("result"))
     monkeypatch.setattr(reingest, "_first_span_id", fake_first, raising=False)
-    monkeypatch.setattr(reingest, "ensure_workspace", lambda url, ws, client=None: calls["ensured"].append(ws))
-    monkeypatch.setattr(
-        reingest, "export_trace_request", lambda url, ws, req, client=None: calls["requests"].append((ws, req))
-    )
-    monkeypatch.setattr(reingest, "_post_created", lambda client, url, body: calls["posts"].append((url, body)))
+    monkeypatch.setattr(reingest, "ensure_workspace", fake_ensure)
+    monkeypatch.setattr(reingest, "export_trace_request", fake_export)
+    monkeypatch.setattr(reingest, "_post_created", fake_post)
+    monkeypatch.setattr(reingest.httpx, "Client", client_factory)
     return calls
 
 
@@ -385,6 +415,264 @@ def fake_run(cmd, **kwargs):
     return runs
 
 
+@pytest.mark.parametrize(
+    ("counts_key", "message"),
+    [
+        ("span_counts", "spans"),
+        ("annotation_counts", "annotations"),
+        ("result_counts", "evaluator results"),
+    ],
+)
+def test_require_empty_rejects_existing_target_data(tmp_path, quiet_platform, counts_key: str, message: str) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC], [ANNOTATION_DOC], [RESULT_DOC])
+    quiet_platform["span_counts"] = [0]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0]
+    quiet_platform[counts_key] = [1]
+
+    with pytest.raises(RuntimeError, match=message):
+        reingest.ingest_bundle(
+            "http://x",
+            export_dir,
+            _manifest("ws-a", 1, 1, 1),
+            workspace_map={"ws-a": "target"},
+            catalog=CATALOG,
+            require_empty=True,
+        )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+    ]
+    assert quiet_platform["requests"] == []
+    assert quiet_platform["posts"] == []
+
+
+@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
+def test_require_empty_transport_rejects_all_http_mutations(quiet_platform, method: str) -> None:
+    with reingest.httpx.Client() as client:
+        with pytest.raises(AssertionError, match=f"unexpected {method} mutation"):
+            client.request(method, "http://x/hidden-mutation")
+
+    assert quiet_platform["events"] == [("write", method, "/hidden-mutation")]
+
+
+def test_require_empty_direct_restore_uses_real_http_transport(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC], [ANNOTATION_DOC], [RESULT_DOC])
+    counts = {"spans": 0, "annotations": 0, "evaluator-results": 0}
+    requests: list[tuple[str, str]] = []
+
+    def handle(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        requests.append((request.method, path))
+        if request.method == "POST" and path == "/apis/entities/v2/workspaces":
+            return httpx.Response(409)
+        if request.method == "GET":
+            collection = path.rsplit("/", 1)[-1]
+            return httpx.Response(200, json={"pagination": {"total_results": counts[collection]}, "data": []})
+        if request.method == "POST" and path.endswith("/ingest/otlp/v1/traces"):
+            counts["spans"] = 1
+            return httpx.Response(200, json={"errors": []})
+        if request.method == "POST" and path.endswith("/annotations"):
+            counts["annotations"] = 1
+            return httpx.Response(201, json={})
+        if request.method == "POST" and path.endswith("/evaluator-results"):
+            counts["evaluator-results"] = 1
+            return httpx.Response(201, json={})
+        raise AssertionError(f"unexpected request: {request.method} {path}")
+
+    real_client = httpx.Client
+    transport = httpx.MockTransport(handle)
+
+    def client_factory(*args, **kwargs):
+        kwargs["transport"] = transport
+        return real_client(*args, **kwargs)
+
+    monkeypatch.setattr(reingest.httpx, "Client", client_factory)
+
+    outcome = reingest.ingest_bundle(
+        "http://platform.example",
+        export_dir,
+        _manifest("ws-a", 1, 1, 1),
+        workspace_map={"ws-a": "target"},
+        catalog=CATALOG,
+        require_empty=True,
+        sleep=lambda seconds: None,
+    )
+
+    assert [request for request in requests if request[0] != "GET"] == [
+        ("POST", "/apis/entities/v2/workspaces"),
+        ("POST", "/apis/intake/v2/workspaces/target/ingest/otlp/v1/traces"),
+        ("POST", "/apis/intake/v2/workspaces/target/annotations"),
+        ("POST", "/apis/intake/v2/workspaces/target/evaluator-results"),
+    ]
+    assert outcome["ws-a"] == {
+        "workspace": "target",
+        "spans": {"ingested": 1, "skipped": 0},
+        "annotations": {"ingested": 1, "skipped": 0},
+        "evaluator_results": {"ingested": 1, "skipped": 0},
+    }
+
+
+def test_require_empty_rechecks_spans_before_ingest(tmp_path, quiet_platform) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC])
+    quiet_platform["span_counts"] = [0, 1]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0]
+
+    with pytest.raises(RuntimeError, match="spans"):
+        reingest.ingest_bundle(
+            "http://x",
+            export_dir,
+            _manifest("ws-a", 1),
+            workspace_map={"ws-a": "target"},
+            catalog=CATALOG,
+            require_empty=True,
+        )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+        ("count", "span", "target"),
+    ]
+    assert quiet_platform["requests"] == []
+    assert quiet_platform["posts"] == []
+
+
+def test_require_empty_rechecks_annotations_before_post(tmp_path, quiet_platform) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [], [ANNOTATION_DOC])
+    quiet_platform["span_counts"] = [0]
+    quiet_platform["annotation_counts"] = [0, 1]
+    quiet_platform["result_counts"] = [0]
+
+    with pytest.raises(RuntimeError, match="annotations"):
+        reingest.ingest_bundle(
+            "http://x",
+            export_dir,
+            _manifest("ws-a", 0, 1),
+            workspace_map={"ws-a": "target"},
+            catalog=CATALOG,
+            require_empty=True,
+        )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+        ("count", "annotation", "target"),
+    ]
+    assert quiet_platform["requests"] == []
+    assert quiet_platform["posts"] == []
+
+
+def test_require_empty_rechecks_results_before_post(tmp_path, quiet_platform) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [], [], [RESULT_DOC])
+    quiet_platform["span_counts"] = [0]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0, 1]
+
+    with pytest.raises(RuntimeError, match="evaluator results"):
+        reingest.ingest_bundle(
+            "http://x",
+            export_dir,
+            _manifest("ws-a", 0, 0, 1),
+            workspace_map={"ws-a": "target"},
+            catalog=CATALOG,
+            require_empty=True,
+        )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+        ("count", "result", "target"),
+    ]
+    assert quiet_platform["requests"] == []
+    assert quiet_platform["posts"] == []
+
+
+def test_require_empty_ingests_all_nonempty_collections_in_order(tmp_path, quiet_platform) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC], [ANNOTATION_DOC], [RESULT_DOC])
+    quiet_platform["span_counts"] = [0, 0, 1]
+    quiet_platform["annotation_counts"] = [0, 0]
+    quiet_platform["result_counts"] = [0, 0]
+
+    outcome = reingest.ingest_bundle(
+        "http://x",
+        export_dir,
+        _manifest("ws-a", 1, 1, 1),
+        workspace_map={"ws-a": "target"},
+        catalog=CATALOG,
+        require_empty=True,
+        sleep=lambda seconds: None,
+    )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+        ("count", "span", "target"),
+        ("write", "span", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("write", "annotation", "target"),
+        ("count", "result", "target"),
+        ("write", "result", "target"),
+    ]
+    assert len(quiet_platform["requests"]) == 1
+    assert [url.rsplit("/", 1)[-1] for url, _ in quiet_platform["posts"]] == [
+        "annotations",
+        "evaluator-results",
+    ]
+    assert outcome["ws-a"] == {
+        "workspace": "target",
+        "spans": {"ingested": 1, "skipped": 0},
+        "annotations": {"ingested": 1, "skipped": 0},
+        "evaluator_results": {"ingested": 1, "skipped": 0},
+    }
+
+
+def test_require_empty_all_empty_bundle_has_no_recounts_or_writes(tmp_path, quiet_platform) -> None:
+    export_dir = _write_export(tmp_path, "ws-a", [])
+    quiet_platform["span_counts"] = [0]
+    quiet_platform["annotation_counts"] = [0]
+    quiet_platform["result_counts"] = [0]
+
+    outcome = reingest.ingest_bundle(
+        "http://x",
+        export_dir,
+        _manifest("ws-a", 0),
+        workspace_map={"ws-a": "target"},
+        catalog=CATALOG,
+        require_empty=True,
+    )
+
+    assert quiet_platform["events"] == [
+        ("ensure", "target"),
+        ("count", "span", "target"),
+        ("count", "annotation", "target"),
+        ("count", "result", "target"),
+    ]
+    assert quiet_platform["requests"] == []
+    assert quiet_platform["posts"] == []
+    assert outcome["ws-a"] == {
+        "workspace": "target",
+        "spans": {"ingested": 0, "skipped": 0},
+        "annotations": {"ingested": 0, "skipped": 0},
+        "evaluator_results": {"ingested": 0, "skipped": 0},
+    }
+
+
 def test_ingest_bundle_zero_ingests_everything(tmp_path, quiet_platform):
     export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC, LLM_DOC], [ANNOTATION_DOC], [RESULT_DOC])
     quiet_platform["span_counts"] = [0, 2]  # guard sees empty; wait sees all spans
@@ -1248,3 +1536,14 @@ def test_manifest_since_naive_timestamp_is_utc():
 def test_manifest_since_missing_bound_is_epoch():
     assert reingest.manifest_since({}) == datetime(1970, 1, 1, tzinfo=timezone.utc)
     assert reingest.manifest_since({"min_start_time": None}) == datetime(1970, 1, 1, tzinfo=timezone.utc)
+
+
+def test_explicit_workspace_map_single_workspace() -> None:
+    assert reingest.explicit_workspace_map(["run-scoped-workspace"], "stable-workspace") == {
+        "run-scoped-workspace": "stable-workspace"
+    }
+
+
+def test_explicit_workspace_map_rejects_multi_workspace_bundle() -> None:
+    with pytest.raises(SystemExit, match="single-workspace"):
+        reingest.explicit_workspace_map(["realistic", "oracle"], "stable-workspace")
diff --git a/plugins/nemo-insights/tests/testbed/test_release.py b/plugins/nemo-insights/tests/testbed/test_release.py
index 1f784458c6..d52b0a17ba 100644
--- a/plugins/nemo-insights/tests/testbed/test_release.py
+++ b/plugins/nemo-insights/tests/testbed/test_release.py
@@ -30,6 +30,14 @@ def test_next_ref():
     assert release.next_ref(None) == "state-v1"
 
 
+def test_state_repo_defaults_to_existing_fixture_home():
+    assert release.state_repo({}) == "NVIDIA-dev/NeMo-Optimizer"
+
+
+def test_state_repo_allows_explicit_override():
+    assert release.state_repo({"TESTBED_STATE_REPO": "owner/repository"}) == "owner/repository"
+
+
 def test_lock_ref_returns_subject_entry(tmp_path):
     lock = _write_lock(tmp_path)
     assert release.lock_ref(lock, "tau2-airline") == "state-v6"
@@ -62,6 +70,8 @@ def test_repo_lock_file_is_per_subject_and_pins_tau2_airline():
 
 def test_release_asset_names_missing_release_returns_empty(monkeypatch):
     def fake_gh(*args):
+        if args[:1] == ("api",):
+            return "[]"
         raise subprocess.CalledProcessError(1, ["gh", *args], stderr="release not found\n")
 
     monkeypatch.setattr(release, "_gh", fake_gh)
@@ -70,6 +80,8 @@ def fake_gh(*args):
 
 def test_release_asset_names_other_failure_raises(monkeypatch):
     def fake_gh(*args):
+        if args[:1] == ("api",):
+            return "[]"
         raise subprocess.CalledProcessError(1, ["gh", *args], stderr="HTTP 500 (Internal Server Error)\n")
 
     monkeypatch.setattr(release, "_gh", fake_gh)
@@ -77,6 +89,30 @@ def fake_gh(*args):
         release._release_asset_names()
 
 
+def test_release_asset_names_inaccessible_repo_raises(monkeypatch):
+    def fake_gh(*args):
+        raise subprocess.CalledProcessError(1, ["gh", *args], stderr="HTTP 404: Not Found\n")
+
+    monkeypatch.setattr(release, "_gh", fake_gh)
+    with pytest.raises(subprocess.CalledProcessError):
+        release._release_asset_names()
+
+
+def test_release_asset_names_does_not_substring_match_auth_error(monkeypatch):
+    def fake_gh(*args):
+        if args[:1] == ("api",):
+            return "[]"
+        raise subprocess.CalledProcessError(
+            1,
+            ["gh", *args],
+            stderr="GraphQL: release not found because the token is unauthorized\n",
+        )
+
+    monkeypatch.setattr(release, "_gh", fake_gh)
+    with pytest.raises(subprocess.CalledProcessError):
+        release._release_asset_names()
+
+
 def test_gh_prints_stderr_on_failure(monkeypatch, capsys):
     def fake_run(*args, **kwargs):
         raise subprocess.CalledProcessError(1, args[0], stderr="gh: some auth error\n")
@@ -165,7 +201,18 @@ def fake_gh(*args):
     dest = tmp_path / "dl"
     result = release.download_ref("state-v4", dest)
     assert calls == [
-        ("release", "download", "testbed-state", "--pattern", "state-v4.tar.zst", "--dir", str(dest), "--clobber")
+        (
+            "release",
+            "download",
+            "testbed-state",
+            "--pattern",
+            "state-v4.tar.zst",
+            "--dir",
+            str(dest),
+            "--clobber",
+            "--repo",
+            "NVIDIA-dev/NeMo-Optimizer",
+        )
     ]
     assert dest.is_dir()
     assert result == tmp_path / "dl" / "state-v4.tar.zst"