From 0cd00ed0053bf61b2a8af00aaf77e79fe8b01d22 Mon Sep 17 00:00:00 2001 From: Aditya Pandey Date: Thu, 16 Jul 2026 23:25:52 -0600 Subject: [PATCH] fix(insights): follow-up fixes for profile-driven analysis Post-merge review follow-up to #718. Analyze keeps other top level keys when writing the shared insights file, .env values keep quotes that are part of the secret, --agent alone bypasses a broken profile, local file errors are no longer reported as platform errors, doctor runs its checks even without a profile, analyze drops the duplicate network probes, restore drops the mid-run recheck, the download cache is split per repo, and the first-span guard compares timestamps chronologically. Restores --clobber and substring release matching. Co-Authored-By: Claude Fable 5 Signed-off-by: Aditya Pandey --- .../analyst/analyst_backend.py | 8 +- .../src/nemo_insights_plugin/cli.py | 42 ++++--- .../nemo_insights_plugin/contracts/profile.py | 14 ++- .../src/nemo_insights_plugin/preflight.py | 61 ++++++---- plugins/nemo-insights/testbed/publish.py | 7 +- plugins/nemo-insights/testbed/reingest.py | 27 ++--- plugins/nemo-insights/testbed/release.py | 15 ++- .../tests/contracts/test_profile_contract.py | 26 +++++ .../nemo-insights/tests/test_cli_profile.py | 109 ++++++++++++------ .../tests/test_periodic_analysis.py | 14 +++ plugins/nemo-insights/tests/test_preflight.py | 22 ++++ .../tests/testbed/test_publish.py | 2 + .../tests/testbed/test_reingest.py | 108 ++++------------- .../tests/testbed/test_release.py | 47 ++++---- 14 files changed, 285 insertions(+), 217 deletions(-) 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 12f43d213c..b39affc94e 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 @@ -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() diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py b/plugins/nemo-insights/src/nemo_insights_plugin/cli.py index 36475eb69a..12826e8d4c 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/cli.py +++ b/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()) diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py b/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py index 77bfbd25fb..ad1682f17f 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/contracts/profile.py +++ b/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 diff --git a/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py b/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py index 1dbf601971..edbcb06234 100644 --- a/plugins/nemo-insights/src/nemo_insights_plugin/preflight.py +++ b/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=" ) 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 diff --git a/plugins/nemo-insights/testbed/publish.py b/plugins/nemo-insights/testbed/publish.py index d37e4e9b29..3e640526fb 100644 --- a/plugins/nemo-insights/testbed/publish.py +++ b/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)) diff --git a/plugins/nemo-insights/testbed/reingest.py b/plugins/nemo-insights/testbed/reingest.py index 24603fa705..3afd953c08 100644 --- a/plugins/nemo-insights/testbed/reingest.py +++ b/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,8 +703,6 @@ 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) @@ -707,20 +710,14 @@ def ingest_bundle( 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, diff --git a/plugins/nemo-insights/testbed/release.py b/plugins/nemo-insights/testbed/release.py index cbcbc979b8..ed16f60b8f 100644 --- a/plugins/nemo-insights/testbed/release.py +++ b/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 ``.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 ``.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 diff --git a/plugins/nemo-insights/tests/contracts/test_profile_contract.py b/plugins/nemo-insights/tests/contracts/test_profile_contract.py index ce2c4903dc..b1d19ca5ff 100644 --- a/plugins/nemo-insights/tests/contracts/test_profile_contract.py +++ b/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 diff --git a/plugins/nemo-insights/tests/test_cli_profile.py b/plugins/nemo-insights/tests/test_cli_profile.py index f68131f3b3..f5abcf29c4 100644 --- a/plugins/nemo-insights/tests/test_cli_profile.py +++ b/plugins/nemo-insights/tests/test_cli_profile.py @@ -1,13 +1,14 @@ # SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import os 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 import cli from nemo_insights_plugin.contracts.profile import DEFAULT_BASE_URL from nemo_insights_plugin.preflight import AnalysisProbes from nemo_platform import NeMoPlatformError @@ -317,9 +318,7 @@ def test_malformed_discovered_profile_warns_when_flags_are_complete( assert "Invalid profile" in result.output -def test_malformed_discovered_profile_errors_without_explicit_workspace( - app: typer.Typer, tmp_path: Path, monkeypatch -) -> None: +def test_malformed_discovered_profile_warns_with_agent_only(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") @@ -327,6 +326,36 @@ def test_malformed_discovered_profile_errors_without_explicit_workspace( result = runner.invoke(app, ["analyze", "--agent", "other"]) + assert result.exit_code == 0, result.output + assert "warning:" in result.output + assert "Invalid profile" in result.output + assert recorder.kwargs is not None + assert recorder.kwargs["workspace"] == "default" + + +def test_malformed_discovered_profile_still_loads_env_file(app: typer.Typer, tmp_path: Path, monkeypatch) -> None: + recorder = AnalystRecorder() + monkeypatch.setattr(cli, "run_analyst", recorder) + monkeypatch.delenv("INFERENCE_API_KEY", raising=False) + (tmp_path / "optimizer.yaml").write_text("agent: ''\n", encoding="utf-8") + (tmp_path / ".env").write_text("INFERENCE_API_KEY=from-env-file\n", encoding="utf-8") + monkeypatch.chdir(tmp_path) + + result = runner.invoke(app, ["analyze", "--agent", "other"]) + + assert result.exit_code == 0, result.output + assert os.environ["INFERENCE_API_KEY"] == "from-env-file" + assert recorder.kwargs is not None + + +def test_malformed_discovered_profile_errors_without_agent(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", "--workspace", "other-ws"]) + assert result.exit_code != 0 assert "Invalid profile" in result.output assert recorder.kwargs is None @@ -379,18 +408,24 @@ def test_analyze_blocks_before_runner_when_preflight_fails( assert recorder.kwargs is None -def test_analyze_prints_advisory_and_runs_analyst( +def test_analyze_runs_only_the_credential_check( 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 False + 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(), + http_ok=lambda base_url: probe_calls.append("http") or False, + workspace_ok=record_workspace_probe, ), ) monkeypatch.chdir(profile_tree) @@ -398,8 +433,8 @@ def test_analyze_prints_advisory_and_runs_analyst( 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 + assert probe_calls == [] @pytest.mark.parametrize( @@ -407,7 +442,6 @@ def test_analyze_prints_advisory_and_runs_analyst( [ 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( @@ -420,21 +454,11 @@ 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] @@ -480,24 +504,13 @@ def fail_to_construct(base_url: str | None) -> object: 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 attempts == 1 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. " @@ -661,10 +674,6 @@ 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) @@ -674,6 +683,36 @@ def test_doctor_exits_nonzero_for_missing_profile(app: typer.Typer, tmp_path: Pa assert "no optimizer.yaml found" in result.output +def test_doctor_missing_profile_still_runs_environment_checks( + app: typer.Typer, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> 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={}, + http_ok=lambda base_url: http_urls.append(base_url) or True, + workspace_ok=record_workspace_probe, + ), + ) + monkeypatch.chdir(tmp_path) + + result = runner.invoke(app, ["doctor", "--base-url", "https://flag.example"]) + + assert result.exit_code == 1 + assert "no optimizer.yaml found" in result.output + assert "INFERENCE_API_KEY not set" in result.output + assert http_urls == ["https://flag.example"] + assert workspace_urls == [] + + def test_doctor_reports_healthy_profile(app: typer.Typer, profile_tree: Path, monkeypatch) -> None: monkeypatch.chdir(profile_tree) diff --git a/plugins/nemo-insights/tests/test_periodic_analysis.py b/plugins/nemo-insights/tests/test_periodic_analysis.py index 637c71bc13..64e5cee976 100644 --- a/plugins/nemo-insights/tests/test_periodic_analysis.py +++ b/plugins/nemo-insights/tests/test_periodic_analysis.py @@ -10,6 +10,7 @@ import httpx import pytest +import yaml from nemo_insights_plugin.analyst.analyst_backend import ( LocalAnalystBackend, RemoteAnalystBackend, @@ -122,6 +123,19 @@ def spy_write_text(self: Path, *args: object, **kwargs: object) -> int: assert read_calls[-1].get("encoding") == "utf-8" +def test_local_backend_write_preserves_other_top_level_keys(tmp_path: Path) -> None: + path = tmp_path / "insights.yaml" + path.write_text("metadata: retained\ninsights:\n- id: stale\n", encoding="utf-8") + backend = LocalAnalystBackend(client=SimpleNamespace(), path=path) # type: ignore[arg-type] + + backend._write_records([{"id": "insight-1"}]) + + assert yaml.safe_load(path.read_text(encoding="utf-8")) == { + "metadata": "retained", + "insights": [{"id": "insight-1"}], + } + + 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 index 8514b865e8..3eb9d2d045 100644 --- a/plugins/nemo-insights/tests/test_preflight.py +++ b/plugins/nemo-insights/tests/test_preflight.py @@ -45,6 +45,28 @@ def test_missing_inference_key_is_required_failure(tmp_path: Path) -> None: assert required_failures(results) +def test_check_environment_without_profile_skips_workspace_probe(tmp_path: Path) -> None: + results = asyncio.run( + check_environment( + agent=None, + workspace=None, + base_url="http://localhost:8080", + profile_dir=None, + probes=AnalysisProbes( + env={"INFERENCE_API_KEY": "k"}, + http_ok=lambda base_url: True, + workspace_ok=never_queryable, + ), + ) + ) + + assert [result.name for result in results] == ["INFERENCE_API_KEY", "platform-reachable"] + + +def test_default_http_probe_treats_invalid_base_url_as_unreachable() -> None: + assert preflight._default_http_ok("http://localhost:not-a-port") is False + + def test_workspace_query_failure_is_advisory(tmp_path: Path) -> None: results = asyncio.run( check_environment( diff --git a/plugins/nemo-insights/tests/testbed/test_publish.py b/plugins/nemo-insights/tests/testbed/test_publish.py index 70c91b0f5c..d4f3e50081 100644 --- a/plugins/nemo-insights/tests/testbed/test_publish.py +++ b/plugins/nemo-insights/tests/testbed/test_publish.py @@ -155,6 +155,7 @@ def gh(*args): def test_publish_mints_next_ref_uploads_and_prepends_row(fake_gh, tmp_path, monkeypatch, capsys): monkeypatch.setattr("getpass.getuser", lambda: "ada") monkeypatch.delenv("GITHUB_OUTPUT", raising=False) + monkeypatch.delenv("TESTBED_STATE_REPO", raising=False) bundle = _make_bundle(tmp_path / "candidate.tar.zst") ref = publish.publish(bundle, reason="fresh corpus", env={}) assert ref == "state-v7" @@ -166,6 +167,7 @@ def test_publish_mints_next_ref_uploads_and_prepends_row(fake_gh, tmp_path, monk "upload", release.RELEASE_TAG, str(tmp_path / "state-v7.tar.zst"), + "--clobber", "--repo", release.DEFAULT_STATE_REPO, ) diff --git a/plugins/nemo-insights/tests/testbed/test_reingest.py b/plugins/nemo-insights/tests/testbed/test_reingest.py index aa3f05abef..41885a1509 100644 --- a/plugins/nemo-insights/tests/testbed/test_reingest.py +++ b/plugins/nemo-insights/tests/testbed/test_reingest.py @@ -519,93 +519,12 @@ def client_factory(*args, **kwargs): } -def test_require_empty_rechecks_spans_before_ingest(tmp_path, quiet_platform) -> None: - export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC]) +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, 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, @@ -621,12 +540,9 @@ def test_require_empty_ingests_all_nonempty_collections_in_order(tmp_path, quiet ("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 @@ -1115,6 +1031,26 @@ def test_skip_path_probe_tolerates_started_at_ties(tmp_path, quiet_platform, cap assert "already restored" in capsys.readouterr().out +def test_skip_path_probe_compares_started_at_chronologically(tmp_path, quiet_platform, capsys): + """Mixed timestamp serializations: the bundle's first span is the chronological min, not the lexicographic one.""" + # 14:14-05:00 is 19:14 UTC — lexicographically first but chronologically LATER than 18:14 UTC. + later_by_offset = {**LLM_DOC, "started_at": "2026-06-26T14:14:41.406179-05:00"} + earlier_utc = {**AGENT_DOC, "started_at": "2026-06-26T18:14:41.406179Z"} + export_dir = _write_export(tmp_path, "ws-a", [earlier_utc, later_by_offset]) + quiet_platform["span_counts"] = [2] + quiet_platform["first_ids"] = [AGENT_DOC["span_id"]] # the live first span IS the chronological min + quiet_platform["annotation_counts"] = [0] + quiet_platform["result_counts"] = [0] + reingest.ingest_bundle( + "http://x", + export_dir, + _manifest("ws-a", 2), + workspace_map={"ws-a": "ws-b"}, + catalog=CATALOG, + ) + assert "already restored" in capsys.readouterr().out + + def test_skip_path_probe_failure_falls_back_to_count_guard(tmp_path, quiet_platform, capsys): """Transient probe failure (fixture default: None) must NOT block the restore.""" export_dir = _write_export(tmp_path, "ws-a", [AGENT_DOC, LLM_DOC]) diff --git a/plugins/nemo-insights/tests/testbed/test_release.py b/plugins/nemo-insights/tests/testbed/test_release.py index d52b0a17ba..5d88743693 100644 --- a/plugins/nemo-insights/tests/testbed/test_release.py +++ b/plugins/nemo-insights/tests/testbed/test_release.py @@ -98,21 +98,6 @@ def fake_gh(*args): 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") @@ -197,8 +182,10 @@ def fake_gh(*args): calls.append(args) return "" + monkeypatch.delenv("TESTBED_STATE_REPO", raising=False) monkeypatch.setattr(release, "_gh", fake_gh) dest = tmp_path / "dl" + repo_dir = dest / "NVIDIA-dev__NeMo-Optimizer" result = release.download_ref("state-v4", dest) assert calls == [ ( @@ -208,23 +195,37 @@ def fake_gh(*args): "--pattern", "state-v4.tar.zst", "--dir", - str(dest), + str(repo_dir), "--clobber", "--repo", "NVIDIA-dev/NeMo-Optimizer", ) ] - assert dest.is_dir() - assert result == tmp_path / "dl" / "state-v4.tar.zst" + assert repo_dir.is_dir() + assert result == repo_dir / "state-v4.tar.zst" def test_download_ref_reuses_cached_file_without_gh(tmp_path, monkeypatch, capsys): """Refs are immutable: an already-downloaded tarball is reused, gh never invoked.""" - dest = tmp_path / "dl" - dest.mkdir() - (dest / "state-v4.tar.zst").write_bytes(b"cached bytes") + monkeypatch.delenv("TESTBED_STATE_REPO", raising=False) + repo_dir = tmp_path / "dl" / "NVIDIA-dev__NeMo-Optimizer" + repo_dir.mkdir(parents=True) + (repo_dir / "state-v4.tar.zst").write_bytes(b"cached bytes") monkeypatch.setattr(release, "_gh", lambda *args: pytest.fail("cached ref must not invoke gh")) - result = release.download_ref("state-v4", dest) - assert result == dest / "state-v4.tar.zst" + result = release.download_ref("state-v4", tmp_path / "dl") + assert result == repo_dir / "state-v4.tar.zst" assert result.read_bytes() == b"cached bytes" assert "using cached state-v4.tar.zst" in capsys.readouterr().out + + +def test_download_ref_cache_is_namespaced_by_state_repo(tmp_path, monkeypatch): + """A bundle cached from one repo must not satisfy the same ref from another repo.""" + monkeypatch.setenv("TESTBED_STATE_REPO", "owner/repository") + calls: list[tuple[str, ...]] = [] + monkeypatch.setattr(release, "_gh", lambda *args: calls.append(args) or "") + dest = tmp_path / "dl" + (dest / "NVIDIA-dev__NeMo-Optimizer").mkdir(parents=True) + (dest / "NVIDIA-dev__NeMo-Optimizer" / "state-v4.tar.zst").write_bytes(b"other repo's bytes") + result = release.download_ref("state-v4", dest) + assert result == dest / "owner__repository" / "state-v4.tar.zst" + assert len(calls) == 1 and ("--repo", "owner/repository") == calls[0][-2:]