fix(insights): follow ups from the profile analysis review

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

@@ -478,9 +478,11 @@ def _read_records(self) -> list[dict]:
 
     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), encoding="utf-8"
-        )
+        document = yaml.safe_load(self.path.read_text(encoding="utf-8")) if self.path.exists() else None
+        if not isinstance(document, dict):
+            document = {}
+        document["insights"] = records
+        self.path.write_text(yaml.safe_dump(document, 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()

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

@@ -27,6 +27,7 @@
 )
 from nemo_insights_plugin.preflight import (
     AnalysisProbes,
+    check_credentials,
     check_environment,
     check_profile,
     read_agent_spec,
@@ -62,6 +63,9 @@ def _load_profile_or_error(profile_path: Path | None) -> tuple[AnalysisProfile |
     except ProfileError as exc:
         if profile_path is not None:
             raise
+        loaded = load_env_file(found.parent / ".env")
+        if loaded:
+            typer.echo(f"Loaded .env from {found.parent / '.env'} ({len(loaded)} vars)", err=True)
         return None, str(exc)
     if profile_path is None:
         typer.echo(f"Using profile: {found} (agent: {profile.agent})", err=True)
@@ -97,16 +101,19 @@ def _resolve_analysis(
 ) -> _ResolvedAnalysis:
     profile, profile_error = _load_profile_or_error(profile_path)
     if profile_error is not None:
-        if agent is None or workspace is None:
+        if agent 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)
+    resolved_agent = agent if agent is not None else (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)
+    if workspace is not None:
+        resolved_workspace = workspace
+    else:
+        resolved_workspace = profile.workspace if profile is not None else DEFAULT_WORKSPACE
 
     spec_path = agent_spec
     spec_error: str | None = None
@@ -138,15 +145,7 @@ def _resolve_analysis(
 
 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,
-        )
-    )
+    checks.extend(check_credentials(analysis.profile_dir, probes=_PREFLIGHT_PROBES))
     _preflight_or_exit(checks)
 
     if analysis.profile_output is not None:
@@ -170,7 +169,7 @@ async def _run_analysis(analysis: _ResolvedAnalysis, *, verbose: bool) -> str:
             err=True,
         )
         raise typer.Exit(1) from None
-    except (ClientConstructionError, NeMoPlatformError, httpx.HTTPError, OSError) as exc:
+    except (ClientConstructionError, NeMoPlatformError, httpx.HTTPError) as exc:
         detail = _one_line_error(exc).rstrip(".")
         typer.echo(
             f"Error: analysis failed: {detail}. Check --base-url/NMP_BASE_URL, "
@@ -309,16 +308,15 @@ def doctor(
                 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,
-                            )
+                    results.extend(
+                        await check_environment(
+                            agent=profile.agent if profile is not None else None,
+                            workspace=profile.workspace if profile is not None else None,
+                            base_url=resolve_base_url(base_url),
+                            profile_dir=profile.profile_dir if profile is not None else None,
+                            probes=_PREFLIGHT_PROBES,
                         )
+                    )
                     return results
 
                 results = asyncio.run(_flow())

plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py

@@ -38,7 +38,10 @@ def discover_profile(start: Path | None = None) -> Path | None:
 
 def resolve_profile_path(value: str, profile_dir: Path) -> Path:
     """Resolve an absolute, home-relative, or profile-relative path."""
-    path = Path(value).expanduser()
+    try:
+        path = Path(value).expanduser()
+    except RuntimeError as exc:
+        raise ProfileError(f"Could not resolve path {value!r}: {exc}") from None
     return path.resolve() if path.is_absolute() else (profile_dir / path).resolve()
 
 
@@ -80,7 +83,9 @@ def load_env_file(path: Path, env: MutableMapping[str, str] = os.environ) -> lis
             continue
         key, _, value = line.removeprefix("export ").partition("=")
         key = key.strip()
-        value = value.strip().strip("'\"")
+        value = value.strip()
+        if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"":
+            value = value[1:-1]
         if key and key not in env:
             env[key] = value
             loaded.append(key)
@@ -103,4 +108,7 @@ def resolve_agent_spec_path(profile_dir: Path, configured: str | None) -> Path |
 
 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
+    if explicit is not None:
+        return explicit
+    env_url = env.get("NMP_BASE_URL")
+    return env_url if env_url is not None else DEFAULT_BASE_URL

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

@@ -28,7 +28,7 @@ def _default_http_ok(base_url: str) -> bool:
             ).status_code
             < 500
         )
-    except (httpx.HTTPError, ValueError):
+    except (httpx.HTTPError, httpx.InvalidURL, ValueError):
         return False
 
 
@@ -153,15 +153,11 @@ def read_agent_spec(
     ]
 
 
-async def check_environment(
-    *,
-    agent: str,
-    workspace: str,
-    base_url: str,
+def check_credentials(
     profile_dir: Path | None,
     probes: AnalysisProbes | None = None,
 ) -> list[CheckResult]:
-    """Run credential and advisory platform checks without persisting state."""
+    """Check that the analyst's required inference credential is present."""
     active = probes or AnalysisProbes()
     env_path = profile_dir / ".env" if profile_dir is not None else None
     credential_hint = (
@@ -170,8 +166,6 @@ async def check_environment(
         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",
@@ -181,7 +175,27 @@ async def check_environment(
             "INFERENCE_API_KEY set",
             "INFERENCE_API_KEY not set",
             hint=credential_hint,
-        ),
+        )
+    ]
+
+
+async def check_environment(
+    *,
+    agent: str | None,
+    workspace: str | None,
+    base_url: str,
+    profile_dir: Path | None,
+    probes: AnalysisProbes | None = None,
+) -> list[CheckResult]:
+    """Run credential and advisory platform checks without persisting state.
+
+    The workspace probe is profile-dependent and skipped when *agent* or
+    *workspace* is unknown; the credential and reachability checks always run.
+    """
+    active = probes or AnalysisProbes()
+    results = check_credentials(profile_dir, active)
+    reachable = active.http_ok(base_url)
+    results.append(
         make_check_result(
             "platform-reachable",
             "platform",
@@ -190,14 +204,19 @@ async def check_environment(
             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",
-        ),
-    ]
+        )
+    )
+    if agent is not None and workspace is not None:
+        queryable = await active.workspace_ok(base_url, workspace, agent)
+        results.append(
+            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",
+            )
+        )
+    return results

plugins/nemo-insights/testbed/publish.py

@@ -137,9 +137,10 @@ def publish(candidate: Path, *, reason: str | None, env: Mapping[str, str] | Non
     tarball = candidate.parent / f"{ref}.tar.zst"
     shutil.copy2(candidate, tarball)
     _ensure_release()
-    # 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))
+    # --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._release_gh("upload", RELEASE_TAG, str(tarball), "--clobber")
     body = json.loads(release._release_gh("view", RELEASE_TAG, "--json", "body"))["body"]
     row = catalog_row(ref, manifest, reason=reason, env=env)
     release._release_gh("edit", RELEASE_TAG, "--notes", insert_catalog_row(body, row))

plugins/nemo-insights/testbed/reingest.py

@@ -491,6 +491,12 @@ def _first_span_id(base_url: str, workspace: str, *, client: httpx.Client | None
             client.close()
 
 
+def _doc_started_at(doc: dict) -> datetime:
+    """A span doc's chronological ``started_at`` (naive values are UTC; missing sorts first)."""
+    parsed = datetime.fromisoformat(str(doc.get("started_at") or _EPOCH_ISO))
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)
+
+
 def _assert_same_first_span(base_url: str, workspace: str, span_docs: list[dict], *, client: httpx.Client) -> None:
     """Harden the count-only skip guard: matching counts can still be a different corpus.
 
@@ -504,8 +510,8 @@ def _assert_same_first_span(base_url: str, workspace: str, span_docs: list[dict]
     live_first = _first_span_id(base_url, workspace, client=client)
     if live_first is None:
         return
-    earliest = min(str(doc.get("started_at") or "") for doc in span_docs)
-    expected = {doc.get("span_id") for doc in span_docs if str(doc.get("started_at") or "") == earliest}
+    earliest = min(_doc_started_at(doc) for doc in span_docs)
+    expected = {doc.get("span_id") for doc in span_docs if _doc_started_at(doc) == earliest}
     if live_first not in expected:
         raise RuntimeError(
             f"{workspace}: span count matches the bundle but its first span is {live_first!r} "
@@ -574,9 +580,8 @@ def ingest_bundle(
       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.
+    up front. This direct-restore mode is not idempotent into a populated
+    workspace.
 
     "already restored — skipping" is printed only when EVERY collection is
     satisfied. Returns
@@ -698,29 +703,21 @@ def ingest_bundle(
                 for start in range(0, len(spans), SPAN_BATCH):
                     batch = [doc_to_otlp(doc, catalog) for doc in spans[start : start + SPAN_BATCH]]
                     request = build_trace_request(batch)
-                    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 index, doc in enumerate(annotations):
+                for doc in 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 index, doc in enumerate(results):
+                for doc in 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,

plugins/nemo-insights/testbed/release.py

@@ -86,7 +86,7 @@ def _release_repo_accessible() -> None:
 
 
 def _release_missing(error: subprocess.CalledProcessError) -> bool:
-    return (error.stderr or "").strip().lower() == "release not found"
+    return "not found" in (error.stderr or "").lower()
 
 
 def _release_exists() -> bool:
@@ -150,17 +150,20 @@ def resolve_state(state: str | None, *, subject: str | None, lock_path: Path) ->
 def download_ref(ref: str, dest_dir: Path) -> Path:
     """Download a state version tarball from the testbed-state release.
 
-    Returns the path to the tarball. Published refs are immutable, so an
-    already-downloaded ``<ref>.tar.zst`` in *dest_dir* is reused without
+    Returns the path to the tarball, cached under a per-repo subdirectory of
+    *dest_dir* (the repo is configurable via ``TESTBED_STATE_REPO``, so one
+    repo's bundle must never satisfy another repo's ref). Published refs are
+    immutable, so an already-downloaded ``<ref>.tar.zst`` is reused without
     invoking gh (one printed line says so). On a fresh download, ``--clobber``
     overwrites any leftover file from a prior run (gh refuses to overwrite by
     default, which would make re-downloading the same ref crash). Failures
     surface gh stderr and propagate the exception.
     """
-    dest = dest_dir / f"{ref}.tar.zst"
+    repo_dir = dest_dir / state_repo().replace("/", "__")
+    dest = repo_dir / f"{ref}.tar.zst"
     if dest.is_file():
         print(f"using cached {ref}.tar.zst")
         return dest
-    dest_dir.mkdir(parents=True, exist_ok=True)
-    _release_gh("download", RELEASE_TAG, "--pattern", f"{ref}.tar.zst", "--dir", str(dest_dir), "--clobber")
+    repo_dir.mkdir(parents=True, exist_ok=True)
+    _release_gh("download", RELEASE_TAG, "--pattern", f"{ref}.tar.zst", "--dir", str(repo_dir), "--clobber")
     return dest

plugins/nemo-insights/tests/contracts/test_profile_contract.py

@@ -78,6 +78,11 @@ def test_resolve_profile_path_handles_relative_absolute_and_home(tmp_path: Path)
     assert absolute == (tmp_path / "agent").resolve()
 
 
+def test_resolve_profile_path_wraps_unresolvable_home_as_profile_error(tmp_path: Path) -> None:
+    with pytest.raises(ProfileError, match="no-such-user-anywhere"):
+        resolve_profile_path("~no-such-user-anywhere/agent", tmp_path)
+
+
 def test_load_env_file_parses_without_overriding(tmp_path: Path) -> None:
     path = tmp_path / ".env"
     path.write_text(
@@ -91,6 +96,26 @@ def test_load_env_file_parses_without_overriding(tmp_path: Path) -> None:
     assert load_env_file(tmp_path / "missing.env", env) == []
 
 
+def test_load_env_file_strips_exactly_one_matched_quote_pair(tmp_path: Path) -> None:
+    path = tmp_path / ".env"
+    path.write_text(
+        "SINGLE='quoted'\nAPOSTROPHE=\"it's\"\nMISMATCHED=\"value'\nTRAILING=value'\nNESTED=\"\"twice\"\"\nEMPTY=''\n",
+        encoding="utf-8",
+    )
+    env: dict[str, str] = {}
+
+    load_env_file(path, env)
+
+    assert env == {
+        "SINGLE": "quoted",
+        "APOSTROPHE": "it's",
+        "MISMATCHED": "\"value'",
+        "TRAILING": "value'",
+        "NESTED": '"twice"',
+        "EMPTY": "",
+    }
+
+
 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")
@@ -142,3 +167,4 @@ def test_resolve_base_url_uses_only_explicit_nmp_and_default() -> None:
     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
+    assert resolve_base_url("", env) == ""  # explicit empty string is not silently replaced