From f7c67f7a1666b3e7861d307eb49a9d9639cecd5a Mon Sep 17 00:00:00 2001 From: Nico Tonozzi Date: Tue, 18 Aug 2026 17:10:45 -0700 Subject: [PATCH 1/4] feat(insights): add experiment-scoped trace fixtures Capture complete experiment traces and replay provider-neutral spans so large Analyst evaluation datasets can be published and reused faithfully. Signed-off-by: Nico Tonozzi --- .../nemo-intake/references/ingest-formats.md | 2 + plugins/nemo-insights/evaluation/README.md | 7 +- plugins/nemo-insights/evaluation/artifact.py | 19 +- .../nemo-insights/evaluation/evaluations.toml | 7 + plugins/nemo-insights/evaluation/export.py | 229 ++++++++++++- plugins/nemo-insights/evaluation/reingest.py | 309 +++++++++++++----- plugins/nemo-insights/evaluation/release.py | 29 +- .../tests/evaluation/test_export.py | 118 ++++++- .../tests/evaluation/test_reingest.py | 131 +++++--- .../tests/evaluation/test_release.py | 43 +-- .../nemo-intake/references/ingest-formats.md | 2 + .../nmp/intake/repository/clickhouse/span.py | 2 +- .../src/nmp/intake/spans/ingest/spans.py | 14 +- .../spans/test_direct_span_ingest.py | 7 +- 14 files changed, 740 insertions(+), 179 deletions(-) diff --git a/packages/nemo_platform_ext/src/nemo_platform_ext/skills/nemo-intake/references/ingest-formats.md b/packages/nemo_platform_ext/src/nemo_platform_ext/skills/nemo-intake/references/ingest-formats.md index c014fc347c..59b0b14457 100644 --- a/packages/nemo_platform_ext/src/nemo_platform_ext/skills/nemo-intake/references/ingest-formats.md +++ b/packages/nemo_platform_ext/src/nemo_platform_ext/skills/nemo-intake/references/ingest-formats.md @@ -197,6 +197,8 @@ attributes retain native JSON types. Known semantic attributes populate queryabl attributes appear in detailed reads under `raw_attributes`. Reposting the same `(source, trace_id, span_id)` updates the existing logical span. +String inputs and outputs are stored verbatim; objects and arrays are serialized as JSON. + The default ClickHouse TTL is 90 days from `started_at`. If any span is outside that window, Intake returns `422` before writing the batch and instructs the operator to increase the `spans` and `trace_index` TTLs. Provider timestamps are never rewritten by the endpoint. diff --git a/plugins/nemo-insights/evaluation/README.md b/plugins/nemo-insights/evaluation/README.md index 5d72cca524..087491260b 100644 --- a/plugins/nemo-insights/evaluation/README.md +++ b/plugins/nemo-insights/evaluation/README.md @@ -158,7 +158,10 @@ uv run python -m evaluation publish evaluation/tmp/glamr.tar.zst --base http://l ``` `snapshot` drains the subject's workspaces (benchmark subjects: realistic + -`-oracle` twin) into JSONL + manifest — no ClickHouse, no Docker. `publish` +`-oracle` twin) into JSONL + manifest — no ClickHouse, no Docker. An intake +subject can set `experiment = ""` to capture only that Experiment's +complete traces; membership comes from root spans, then every child is exported +by trace ID. `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` only after running `roundtrip` separately and confirming it @@ -190,6 +193,8 @@ What restore touches: - Accepted losses: annotation/evaluator-result `created_at`/`created_by` are server-stamped at restore (the write APIs reject client values); a running platform is required to analyze. +- Non-OTLP spans, including ATIF, restore through the provider-neutral direct + ingest API so their source names and arbitrary span IDs remain unchanged. - Legacy `state-v1..v5` tars are not present in CSS (the v4 corpus lives on as `state-v6`); a stray local copy restores only from a checkout predating the v6 migration. diff --git a/plugins/nemo-insights/evaluation/artifact.py b/plugins/nemo-insights/evaluation/artifact.py index bd452b9234..042e4ad42d 100644 --- a/plugins/nemo-insights/evaluation/artifact.py +++ b/plugins/nemo-insights/evaluation/artifact.py @@ -93,6 +93,8 @@ def build_export_manifest( "platform_revision": (platform_info or {}).get("revision"), "records": [r.name for r in records], } + if stats.get("selections"): + manifest["selections"] = stats["selections"] env = env or {} manifest |= {key: env[var] for key, var in _LINEAGE_ENV.items() if env.get(var)} return manifest @@ -222,12 +224,19 @@ def snapshot_export( CLI's ``--base`` override rewrites every stanza before this is called). """ subject_workspaces: list[tuple[Subject, list[str]]] = [] - claimed_workspaces: set[str] = set() + claimed_workspaces: dict[str, Subject] = {} for subject in subjects: + for workspace in workspaces_for_subject(subject): + owner = claimed_workspaces.get(workspace) + if owner is not None and (owner.config.get("experiment") or subject.config.get("experiment")): + sys.exit( + f"snapshot: workspace '{workspace}' is shared by experiment-scoped subject " + f"'{owner.name if owner.config.get('experiment') else subject.name}' and another subject" + ) workspaces = [workspace for workspace in workspaces_for_subject(subject) if workspace not in claimed_workspaces] if workspaces: subject_workspaces.append((subject, workspaces)) - claimed_workspaces.update(workspaces) + claimed_workspaces.update({workspace: subject for workspace in workspaces}) # Every selected subject must carry a base_url: a partial miss would silently # let the agreement check pass on the configured subset and export the # unconfigured subject from the others' platform. @@ -256,6 +265,7 @@ def snapshot_export( state, since=since, client=_basic_auth_intake_client_for(subject, source_url), + experiment=str(subject.config["experiment"]) if subject.config.get("experiment") else None, ) for subject, workspaces in subject_workspaces ] @@ -275,6 +285,11 @@ def snapshot_export( }, "min_start_time": min(min_bounds).isoformat() if min_bounds else None, "max_start_time": max(max_bounds).isoformat() if max_bounds else None, + "selections": { + workspace: selection + for result in exported + for workspace, selection in result.get("selections", {}).items() + }, } for rec in records: shutil.copy2(rec, state / "tmp" / rec.name) diff --git a/plugins/nemo-insights/evaluation/evaluations.toml b/plugins/nemo-insights/evaluation/evaluations.toml index 48f076ab1e..b64193ea74 100644 --- a/plugins/nemo-insights/evaluation/evaluations.toml +++ b/plugins/nemo-insights/evaluation/evaluations.toml @@ -14,6 +14,13 @@ agent = "content-dedup" workspace = "nvq" state = "state-v7" +[kernel-factory] +type = "intake" +agent = "solswarm-campaign" +workspace = "kf-prod-evals" +experiment = "prod-latest-completed" +state = "state-v11" + [glamr] type = "intake" agent = "glamr" # the main GLAMR agent (ux-agent); the only agent_name-tagged one diff --git a/plugins/nemo-insights/evaluation/export.py b/plugins/nemo-insights/evaluation/export.py index 990f4d5764..278b5b11a5 100644 --- a/plugins/nemo-insights/evaluation/export.py +++ b/plugins/nemo-insights/evaluation/export.py @@ -14,17 +14,20 @@ import asyncio import json +import shutil from collections.abc import Callable +from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path -from typing import Any, cast +from typing import Any, TextIO, cast from urllib.parse import urlparse from nemo_platform import AsyncNeMoPlatform from nemo_platform.config.config import Config EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) -PAGE_SIZE = 200 # generous pages: drain-all in few round-trips +PAGE_SIZE = 1000 +TRACE_EXPORT_CONCURRENCY = 8 _LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "0.0.0.0"}) @@ -45,17 +48,45 @@ def _dump(item) -> dict: async def _drain_to_jsonl(paginator, path: Path, *, on_doc: Callable[[dict], None] | None = None) -> int: """Write every paginated doc to *path*, one JSON document per line; return the count.""" - count = 0 with path.open("w", encoding="utf-8") as fh: - async for item in paginator: - doc = _dump(item) - fh.write(json.dumps(doc, ensure_ascii=False) + "\n") - if on_doc is not None: - on_doc(doc) - count += 1 + return await _drain_to_stream(paginator, fh, on_doc=on_doc) + + +async def _drain_to_stream(paginator, stream: TextIO, *, on_doc: Callable[[dict], None] | None = None) -> int: + """Append every paginated doc to an open JSONL stream.""" + count = 0 + async for item in paginator: + doc = _dump(item) + stream.write(json.dumps(doc, ensure_ascii=False) + "\n") + if on_doc is not None: + on_doc(doc) + count += 1 return count +@dataclass(frozen=True) +class ExperimentScope: + """The immutable trace/session membership selected from one Experiment.""" + + experiment: str + experiment_id: str + evaluation_names: list[str] + trace_ids: list[str] + session_ids: list[str] + expected_spans: int + + def manifest(self) -> dict: + return { + "kind": "experiment", + "experiment": self.experiment, + "experiment_id": self.experiment_id, + "evaluation_names": self.evaluation_names, + "trace_ids": self.trace_ids, + "session_ids": self.session_ids, + "expected_spans": self.expected_spans, + } + + class _StartBounds: """Min/max span ``started_at`` across everything exported (manifest time bounds).""" @@ -83,6 +114,8 @@ def export_workspaces( out_dir: Path, *, since: datetime | None, + experiment: str | None = None, + selection: dict | None = None, client: AsyncNeMoPlatform | None = None, ) -> dict: """Drain spans/annotations/evaluator-results per workspace into JSONL files. @@ -92,7 +125,140 @@ def export_workspaces( "max_start_time": ...}`` (time bounds from span ``started_at``; ISO strings or None when no spans matched). """ - return asyncio.run(_export_workspaces(base_url, workspaces, out_dir, since=since, client=client)) + return asyncio.run( + _export_workspaces( + base_url, + workspaces, + out_dir, + since=since, + experiment=experiment, + selection=selection, + client=client, + ) + ) + + +async def _resolve_experiment_scope( + client: AsyncNeMoPlatform, + *, + workspace: str, + experiment_name: str, + lower: str, +) -> ExperimentScope: + """Resolve an Experiment to complete traces and their expected span count.""" + experiment = await client.experiments.retrieve(experiment_name, workspace=workspace) + evaluation_names = sorted( + [ + evaluation.name + async for evaluation in client.evaluations.list( + workspace=workspace, + filter=cast(Any, {"experiment_id": experiment.id}), + page_size=PAGE_SIZE, + sort="name", + ) + ] + ) + if experiment.evaluation_count is not None and len(evaluation_names) != experiment.evaluation_count: + raise RuntimeError( + f"{workspace}/{experiment_name}: Experiment reports {experiment.evaluation_count} evaluations " + f"but the membership query returned {len(evaluation_names)}" + ) + + traces: dict[str, tuple[str, int]] = {} + for evaluation_name in evaluation_names: + paginator = client.intake.traces.list( + workspace=workspace, + page_size=PAGE_SIZE, + mode="preview", + sort="started_at", + filter=cast(Any, {"evaluation_id": evaluation_name, "started_at": {"gte": lower}}), + ) + async for trace in paginator: + if trace.span_count is None: + raise RuntimeError(f"{workspace}/{trace.id}: preview response omitted span_count") + value = (trace.session_id, trace.span_count) + previous = traces.setdefault(trace.id, value) + if previous != value: + raise RuntimeError(f"{workspace}/{trace.id}: conflicting trace membership metadata") + if not traces: + raise RuntimeError(f"{workspace}/{experiment_name}: no traces matched the Experiment") + + trace_ids = sorted(traces) + return ExperimentScope( + experiment=experiment_name, + experiment_id=experiment.id, + evaluation_names=evaluation_names, + trace_ids=trace_ids, + session_ids=sorted({traces[trace_id][0] for trace_id in trace_ids}), + expected_spans=sum(traces[trace_id][1] for trace_id in trace_ids), + ) + + +async def _export_scoped_workspace( + client: AsyncNeMoPlatform, + *, + workspace: str, + ws_dir: Path, + scope: ExperimentScope, + bounds: _StartBounds, +) -> dict[str, int]: + """Export complete selected traces and their session-level auxiliary rows.""" + epoch = EPOCH.isoformat() + parts = ws_dir / ".span-parts" + parts.mkdir() + semaphore = asyncio.Semaphore(TRACE_EXPORT_CONCURRENCY) + + async def export_trace(index: int, trace_id: str) -> tuple[Path, int]: + path = parts / f"{index:06d}.jsonl" + async with semaphore: + count = await _drain_to_jsonl( + client.intake.spans.list( + workspace=workspace, + page_size=PAGE_SIZE, + mode="detailed", + sort="started_at", + filter=cast(Any, {"trace_id": trace_id, "started_at": {"gte": epoch}}), + ), + path, + on_doc=bounds.note, + ) + return path, count + + try: + exports = await asyncio.gather( + *(export_trace(index, trace_id) for index, trace_id in enumerate(scope.trace_ids)) + ) + with (ws_dir / "spans.jsonl").open("wb") as stream: + for path, _count in exports: + with path.open("rb") as part: + shutil.copyfileobj(part, stream) + finally: + shutil.rmtree(parts) + n_spans = sum(count for _path, count in exports) + if n_spans != scope.expected_spans: + raise RuntimeError( + f"{workspace}/{scope.experiment}: selected traces reported {scope.expected_spans} spans " + f"but export returned {n_spans}; the source changed during capture" + ) + + async def drain_sessions(collection, path: Path) -> int: + count = 0 + with path.open("w", encoding="utf-8") as stream: + for session_id in scope.session_ids: + count += await _drain_to_stream( + collection.list( + workspace=workspace, + page_size=PAGE_SIZE, + sort="created_at", + filter=cast(Any, {"session_id": session_id, "created_at": {"gte": epoch}}), + ), + stream, + ) + return count + + n_annotations = await drain_sessions(client.intake.annotations, ws_dir / "annotations.jsonl") + n_results = await drain_sessions(client.intake.evaluator_results, ws_dir / "evaluator_results.jsonl") + return {"spans": n_spans, "annotations": n_annotations, "evaluator_results": n_results} async def _export_workspaces( @@ -101,16 +267,58 @@ async def _export_workspaces( out_dir: Path, *, since: datetime | None, + experiment: str | None = None, + selection: dict | None = None, client: AsyncNeMoPlatform | None = None, ) -> dict: + if (experiment is not None or selection is not None) and len(workspaces) != 1: + raise ValueError("experiment-scoped export requires exactly one workspace") + if experiment is not None and selection is not None: + raise ValueError("pass experiment or selection, not both") lower = (since or EPOCH).isoformat() bounds = _StartBounds() counts: dict[str, dict[str, int]] = {} + selections: dict[str, dict] = {} client = client if client is not None else make_client(base_url) try: for workspace in workspaces: ws_dir = out_dir / "export" / workspace ws_dir.mkdir(parents=True, exist_ok=True) + scope = ( + await _resolve_experiment_scope( + client, + workspace=workspace, + experiment_name=experiment, + lower=lower, + ) + if experiment is not None + else ExperimentScope( + experiment=str(selection["experiment"]), + experiment_id=str(selection["experiment_id"]), + evaluation_names=list(selection["evaluation_names"]), + trace_ids=list(selection["trace_ids"]), + session_ids=list(selection["session_ids"]), + expected_spans=int(selection["expected_spans"]), + ) + if selection is not None + else None + ) + if scope is not None: + counts[workspace] = await _export_scoped_workspace( + client, + workspace=workspace, + ws_dir=ws_dir, + scope=scope, + bounds=bounds, + ) + selections[workspace] = scope.manifest() + workspace_counts = counts[workspace] + print( + f"exported {workspace}/{scope.experiment}: {workspace_counts['spans']} spans, " + f"{workspace_counts['annotations']} annotations, " + f"{workspace_counts['evaluator_results']} evaluator results" + ) + continue spans = client.intake.spans.list( workspace=workspace, page_size=PAGE_SIZE, @@ -145,4 +353,5 @@ async def _export_workspaces( "workspaces": counts, "min_start_time": bounds.min.isoformat() if bounds.min else None, "max_start_time": bounds.max.isoformat() if bounds.max else None, + "selections": selections, } diff --git a/plugins/nemo-insights/evaluation/reingest.py b/plugins/nemo-insights/evaluation/reingest.py index e996f47221..0c12bae187 100644 --- a/plugins/nemo-insights/evaluation/reingest.py +++ b/plugins/nemo-insights/evaluation/reingest.py @@ -56,9 +56,9 @@ import sys import tempfile import time -from collections import Counter -from collections.abc import Callable, Mapping +from collections.abc import Callable, Iterator, Mapping from datetime import datetime, timedelta, timezone +from itertools import groupby from pathlib import Path from types import ModuleType from typing import Any @@ -76,6 +76,8 @@ _STATUS_CODE_ERROR = 2 # opentelemetry.proto.trace.v1.Status.STATUS_CODE_ERROR OTLP_REQUEST_MAX_BYTES = 4 * 1024 * 1024 OTLP_REQUEST_MAX_SPANS = 100 +DIRECT_REQUEST_MAX_SPANS = 1000 +DIRECT_REQUEST_MAX_BYTES = 4 * 1024 * 1024 # Where the attribute catalog lives inside a nemo-platform checkout. CATALOG_RELPATH = Path("services/intake/src/nmp/intake/spans/span_attribute_catalog.py") @@ -254,17 +256,25 @@ def _iso_to_ns(value: str) -> int: return ((dt - _EPOCH) // timedelta(microseconds=1)) * 1000 -def doc_to_otlp(doc: dict, catalog) -> dict: - """One detailed span doc -> a plain OTLP span dict (no protobuf, no network). - - Returns ``{trace_id, span_id, parent_span_id, name, start_ns, end_ns, - status_error, scope, attributes}`` — :func:`build_trace_request` turns a - batch of these into one protobuf request. ``catalog`` is the platform's - ``span_attribute_catalog`` module (see :func:`load_catalog`); the semantic - doc columns are inverted through it mechanically, so a catalog change on - the platform side changes the inversion with it. - """ +def _doc_attributes(doc: dict, catalog) -> dict[str, Any]: + """Rebuild source attributes from detailed read-API fields.""" attrs: dict[str, Any] = json.loads(doc.get("raw_attributes") or "{}") + for spec in catalog.ATTRIBUTE_SPECS: + if any(key in attrs for key in spec.source_keys): + continue # raw passthrough already carries a source alias — let it win, as it did originally + value = _doc_value(doc, spec.field.value) + if value is not None: + attrs[spec.source_keys[0]] = value + metadata = (doc.get("evaluation_context") or {}).get("metadata") + if metadata and "nemo.experiment.metadata" not in attrs: + # Stored in the string bag but excluded from raw_attributes; resurfaces as evaluation_context.metadata. + attrs["nemo.experiment.metadata"] = json.dumps(metadata, separators=(",", ":"), ensure_ascii=False) + return attrs + + +def doc_to_otlp(doc: dict, catalog) -> dict: + """One detailed span doc -> a plain OTLP span dict (no protobuf, no network).""" + attrs = _doc_attributes(doc, catalog) # Ingest re-stamps otel.scope from the protobuf scope on every span, so the exported value # must travel as the scope itself (build_trace_request groups spans by it), not as an attribute. scope = json.loads(attrs.pop("otel.scope", "null") or "null") @@ -277,16 +287,6 @@ def doc_to_otlp(doc: dict, catalog) -> dict: attrs["status"] = "cancelled" # source-only; the only OTLP route to a cancelled row if "openinference.span.kind" not in attrs and doc.get("kind") not in (None, "UNKNOWN"): attrs["openinference.span.kind"] = doc["kind"] - for spec in catalog.ATTRIBUTE_SPECS: - if any(key in attrs for key in spec.source_keys): - continue # raw passthrough already carries a source alias — let it win, as it did originally - value = _doc_value(doc, spec.field.value) - if value is not None: - attrs[spec.source_keys[0]] = value - metadata = (doc.get("evaluation_context") or {}).get("metadata") - if metadata and "nemo.experiment.metadata" not in attrs: - # Stored in the string bag but excluded from raw_attributes; resurfaces as evaluation_context.metadata. - attrs["nemo.experiment.metadata"] = json.dumps(metadata, separators=(",", ":"), ensure_ascii=False) if not doc.get("trace_id"): raise ValueError(f"span {doc.get('span_id')}: no trace_id — cannot rebuild an OTLP span") return { @@ -302,6 +302,38 @@ def doc_to_otlp(doc: dict, catalog) -> dict: } +def _iso_utc(value: str) -> str: + """Read-API timestamp (naive means UTC) -> offset-aware UTC ISO string.""" + timestamp = datetime.fromisoformat(value) + if timestamp.tzinfo is None: + timestamp = timestamp.replace(tzinfo=timezone.utc) + return timestamp.astimezone(timezone.utc).isoformat() + + +def doc_to_direct_span(doc: dict, catalog) -> tuple[str, dict]: + """One detailed span doc -> its exact provider-neutral direct-ingest representation.""" + source = doc.get("source") + if not isinstance(source, str) or not source: + raise ValueError(f"span {doc.get('span_id')}: no source") + if not doc.get("trace_id"): + raise ValueError(f"span {doc.get('span_id')}: no trace_id") + body = { + "span_id": doc["span_id"], + "trace_id": doc["trace_id"], + "session_id": doc["session_id"], + "parent_span_id": doc.get("parent_span_id"), + "name": doc.get("name") or "", + "kind": doc.get("kind") or "UNKNOWN", + "status": doc.get("status") or "unknown", + "started_at": _iso_utc(doc["started_at"]), + "ended_at": _iso_utc(doc["ended_at"]) if doc.get("ended_at") else None, + "input": doc.get("input"), + "output": doc.get("output"), + "attributes": _doc_attributes(doc, catalog), + } + return source, {key: value for key, value in body.items() if value is not None} + + def build_trace_request(otlp_spans: list[dict]) -> ExportTraceServiceRequest: """Batch OTLP span dicts into one ``ExportTraceServiceRequest``. @@ -430,8 +462,8 @@ def _require_zero(workspace: str, collection: str, count: int) -> None: ) -def _collection_outcome(documents: list[dict], ingested: bool) -> dict[str, int]: - count = len(documents) +def _collection_outcome(documents: list[dict] | int, ingested: bool) -> dict[str, int]: + count = documents if isinstance(documents, int) else len(documents) if ingested: return {"ingested": count, "skipped": 0} return {"ingested": 0, "skipped": count} @@ -449,6 +481,67 @@ def _read_jsonl(path: Path) -> list[dict]: return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] +def _iter_jsonl(path: Path) -> Iterator[dict]: + if not path.is_file(): + return + with path.open(encoding="utf-8") as stream: + for line in stream: + if line.strip(): + yield json.loads(line) + + +def _scan_span_file(path: Path, catalog) -> tuple[int, set[str], set[str]]: + """Check direct inversion and return count, sources, and earliest span IDs.""" + count = 0 + sources: set[str] = set() + earliest: datetime | None = None + earliest_ids: set[str] = set() + for doc in _iter_jsonl(path): + source, direct = doc_to_direct_span(doc, catalog) + sources.add(source) + started_at = datetime.fromisoformat(direct["started_at"]) + if earliest is None or started_at < earliest: + earliest = started_at + earliest_ids = {doc["span_id"]} + elif started_at == earliest: + earliest_ids.add(doc["span_id"]) + count += 1 + return count, sources, earliest_ids + + +def _ingest_direct_span_file( + base_url: str, + workspace: str, + path: Path, + catalog, + *, + client: httpx.Client, +) -> None: + """Stream detailed span docs into bounded provider-neutral batches.""" + batches: dict[str, list[dict]] = {} + batch_bytes: dict[str, int] = {} + url = f"{base_url.rstrip('/')}/apis/intake/v2/workspaces/{workspace}/ingest/spans" + + def flush(source: str) -> None: + spans = batches.get(source) + if not spans: + return + _post_created(client, url, {"source": source, "spans": spans}) + spans.clear() + batch_bytes[source] = 0 + + for doc in _iter_jsonl(path): + source, direct = doc_to_direct_span(doc, catalog) + batch = batches.setdefault(source, []) + size = len(json.dumps(direct, ensure_ascii=False).encode()) + if batch and (len(batch) == DIRECT_REQUEST_MAX_SPANS or batch_bytes[source] + size > DIRECT_REQUEST_MAX_BYTES): + flush(source) + batch.append(direct) + batch_bytes[source] = batch_bytes.get(source, 0) + size + for source in batches: + flush(source) + + def _wait_for_spans( base_url: str, workspace: str, @@ -582,13 +675,13 @@ 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: +def _assert_same_first_span( + base_url: str, + workspace: str, + expected: set[str], + *, + client: httpx.Client, +) -> None: """Harden the count-only skip guard: matching counts can still be a different corpus. A re-minted ref (same workspaces, same counts, fresh ids — e.g. re-published @@ -601,8 +694,6 @@ 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(_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} " @@ -678,31 +769,13 @@ def ingest_bundle( satisfied. Returns ``{source_ws: {"workspace": target, : {"ingested": n, "skipped": n}}}``. - Only otel-sourced corpora are restorable: ATIF/chat-completions span docs - carry non-hex ids (``span-``, ``chatcmpl-``) that the OTLP - inversion cannot encode, and re-ingest would re-stamp their ``source`` as - otel. Every workspace's docs are scanned up front so a bad bundle errors - before ANYTHING is ingested (all-or-nothing). + OTLP-only workspaces retain the protobuf path. Other source formats use the + provider-neutral direct-span API so arbitrary ids and source names survive. + Every span file is checked for invertibility in a bounded-memory pass before any write. """ - spans_by_ws: dict[str, list[dict]] = {} - foreign_sources: dict[str, Counter] = {} + span_info: dict[str, tuple[int, set[str], set[str]]] = {} for source_ws in manifest["workspaces"]: - docs = _read_jsonl(Path(export_dir) / source_ws / "spans.jsonl") - spans_by_ws[source_ws] = docs - bad = Counter(doc.get("source") for doc in docs if doc.get("source") != "otel") - if bad: - foreign_sources[source_ws] = bad - if foreign_sources: - detail = "; ".join( - f"workspace '{ws}': " - + ", ".join(f"{src or ''}={n}" for src, n in sorted(bad.items(), key=lambda kv: str(kv[0]))) - for ws, bad in foreign_sources.items() - ) - raise RuntimeError( - f"bundle contains non-otel span docs ({detail}) — their ids are not OTLP hex, so re-ingest " - "would crash mid-batch (partial restore) and any doc that survived would be silently " - "re-stamped as otel. Only otel-sourced corpora are restorable today; nothing was ingested." - ) + span_info[source_ws] = _scan_span_file(Path(export_dir) / source_ws / "spans.jsonl", catalog) if warn_if_stale(manifest) and _is_loopback(base_url): # CI's stack freezes TTL merges unconditionally; a laptop restore of a >90d bundle # would otherwise lose its spans on the next TTL merge with only an ignorable warning. @@ -718,13 +791,18 @@ def ingest_bundle( if not _WS_OK.fullmatch(target): raise RuntimeError(f"target workspace '{target}' violates the platform naming rule ({_WS_OK.pattern})") ws_dir = Path(export_dir) / source_ws - spans = spans_by_ws[source_ws] + spans_path = ws_dir / "spans.jsonl" + span_file_count, sources, earliest_span_ids = span_info[source_ws] annotations = _read_jsonl(ws_dir / "annotations.jsonl") results = _read_jsonl(ws_dir / "evaluator_results.jsonl") ws_counts = counts.get(source_ws) or {} - expected_spans = int(ws_counts.get("spans", len(spans))) + expected_spans = int(ws_counts.get("spans", span_file_count)) expected_ann = int(ws_counts.get("annotations", len(annotations))) expected_res = int(ws_counts.get("evaluator_results", len(results))) + if expected_spans != span_file_count: + raise RuntimeError( + f"{source_ws}: manifest expects {expected_spans} spans but spans.jsonl contains {span_file_count}" + ) ensure_workspace(base_url, target, client=client) have_spans = span_count(base_url, target, client=client) @@ -737,15 +815,15 @@ def ingest_bundle( ("evaluator results", have_res), ): _require_zero(target, collection, count) - ingest_spans = bool(spans) + ingest_spans = bool(span_file_count) post_annotations = bool(annotations) post_results = bool(results) else: if have_spans == expected_spans: ingest_spans = False - if expected_spans and spans: + if expected_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) + _assert_same_first_span(base_url, target, earliest_span_ids, client=client) elif have_spans == 0: ingest_spans = True else: @@ -782,7 +860,7 @@ def ingest_bundle( print(f"{target}: already restored ({have_spans} spans) — skipping") outcome[source_ws] = { "workspace": target, - "spans": _collection_outcome(spans, False), + "spans": _collection_outcome(span_file_count, False), "annotations": _collection_outcome(annotations, False), "evaluator_results": _collection_outcome(results, False), } @@ -790,14 +868,23 @@ def ingest_bundle( # Healing = posting into a workspace whose spans already landed (interrupted restore). healing = expected_spans > 0 and not ingest_spans if ingest_spans: - print(f"ingesting {len(spans)} spans into {target}") - # Materialize every request before posting: higher memory use buys up-front - # validation (including oversized spans) and prevents a partial restore. - trace_requests = build_trace_requests(spans, catalog) - for request in trace_requests: - export_trace_request(base_url, target, request, client=client) - if spans: - _wait_for_spans(base_url, target, expected_spans or len(spans), client=client, sleep=sleep) + print(f"ingesting {span_file_count} spans into {target}") + if sources == {"otel"}: + spans = _read_jsonl(spans_path) + trace_requests = build_trace_requests(spans, catalog) + for request in trace_requests: + export_trace_request(base_url, target, request, client=client) + else: + _ingest_direct_span_file(base_url, target, spans_path, catalog, client=client) + if span_file_count: + _wait_for_spans( + base_url, + target, + expected_spans, + client=client, + timeout_s=1800.0, + sleep=sleep, + ) root = f"{base_url.rstrip('/')}/apis/intake/v2/workspaces/{target}" if post_annotations: if healing: @@ -813,7 +900,7 @@ def ingest_bundle( _post_created(client, f"{root}/evaluator-results", body) outcome[source_ws] = { "workspace": target, - "spans": _collection_outcome(spans, ingest_spans), + "spans": _collection_outcome(span_file_count, ingest_spans), "annotations": _collection_outcome(annotations, post_annotations), "evaluator_results": _collection_outcome(results, post_results), } @@ -849,6 +936,37 @@ def _diff_collection(original: list[dict], restored: list[dict], *, workspace: s return mismatches +def _diff_grouped_jsonl(original: Path, restored: Path, *, workspace: str, collection: str) -> list[str]: + """Streaming exact diff for trace-grouped scoped exports.""" + left = groupby(_iter_jsonl(original), key=lambda doc: doc["trace_id"]) + right = groupby(_iter_jsonl(restored), key=lambda doc: doc["trace_id"]) + while True: + group_a = next(left, None) + group_b = next(right, None) + if group_a is None or group_b is None: + if group_a is group_b: + return [] + return [f"{workspace}/{collection}: trace count differs"] + trace_a, documents_a = group_a + trace_b, documents_b = group_b + if trace_a != trace_b: + return [f"{workspace}/{collection}: trace {trace_a!r} != {trace_b!r}"] + normalized_a = {(doc["source"], doc["span_id"]): _normalize(doc, collection) for doc in documents_a} + normalized_b = {(doc["source"], doc["span_id"]): _normalize(doc, collection) for doc in documents_b} + if normalized_a.keys() != normalized_b.keys(): + return [f"{workspace}/{collection} trace {trace_a}: span identities differ"] + for identity, doc_a in normalized_a.items(): + doc_b = normalized_b[identity] + if doc_a == doc_b: + continue + label = identity[1] + return [ + f"{workspace}/{collection} {label}.{key}: {doc_a.get(key)!r} != {doc_b.get(key)!r}" + for key in sorted(set(doc_a) | set(doc_b)) + if doc_a.get(key) != doc_b.get(key) + ] + + def cleanup_scratch(base_url: str, workspaces: list[str]) -> None: """Best-effort scratch-workspace cleanup after a round-trip check. @@ -858,10 +976,10 @@ def cleanup_scratch(base_url: str, workspaces: list[str]) -> None: publishing that URL's port. The DELETE mutations are scoped to an exact IN-list of the scratch names. When the binding cannot be verified NOTHING is deleted — docker-execing another container would purge the wrong ClickHouse, - and deleting just the workspace *record* would orphan rows behind it. Leaving - the record intact means a later collision hits the loud foreign-data guard; - leftovers are printed for manual cleanup. Failures are reported, never - raised: on CI the whole platform is ephemeral, so residue is moot. + and deleting just the workspace *record* would orphan rows behind it. Workspace + records are retained after successful row cleanup so the same content-scoped + check can reuse them. Failures are reported, never raised: on CI the whole + platform is ephemeral, so residue is moot. """ if not workspaces: return @@ -913,11 +1031,6 @@ def cleanup_scratch(base_url: str, workspaces: list[str]) -> None: file=sys.stderr, ) return - with httpx.Client(timeout=30.0) as client: - for ws in workspaces: - resp = client.delete(f"{base_url.rstrip('/')}/apis/entities/v2/workspaces/{ws}") - if resp.status_code not in (200, 204, 404): - print(f"cleanup: workspace record {ws} not deleted ({resp.status_code})", file=sys.stderr) def _scratch_digest(manifest: dict, content_key: str | None) -> str: @@ -990,12 +1103,40 @@ def roundtrip_diff( TEMP_ROOT.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(dir=TEMP_ROOT) as tmp: re_dir = Path(tmp) - export.export_workspaces(base_url, list(workspace_map.values()), re_dir, since=None) + selections = manifest.get("selections") or {} + if selections: + for source_ws, scratch_ws in workspace_map.items(): + export.export_workspaces( + base_url, + [scratch_ws], + re_dir, + since=None, + selection=selections.get(source_ws), + ) + else: + export.export_workspaces(base_url, list(workspace_map.values()), re_dir, since=None) for source_ws, scratch_ws in workspace_map.items(): for collection in ("spans", "annotations", "evaluator_results"): - original = _read_jsonl(Path(export_dir) / source_ws / f"{collection}.jsonl") - restored = _read_jsonl(re_dir / "export" / scratch_ws / f"{collection}.jsonl") - mismatches.extend(_diff_collection(original, restored, workspace=source_ws, collection=collection)) + original_path = Path(export_dir) / source_ws / f"{collection}.jsonl" + restored_path = re_dir / "export" / scratch_ws / f"{collection}.jsonl" + if source_ws in selections and collection == "spans": + mismatches.extend( + _diff_grouped_jsonl( + original_path, + restored_path, + workspace=source_ws, + collection=collection, + ) + ) + else: + mismatches.extend( + _diff_collection( + _read_jsonl(original_path), + _read_jsonl(restored_path), + workspace=source_ws, + collection=collection, + ) + ) finally: cleanup_scratch(base_url, list(workspace_map.values())) return mismatches diff --git a/plugins/nemo-insights/evaluation/release.py b/plugins/nemo-insights/evaluation/release.py index 0432be0a8b..3ff86c1181 100644 --- a/plugins/nemo-insights/evaluation/release.py +++ b/plugins/nemo-insights/evaluation/release.py @@ -155,12 +155,9 @@ def download_ref(ref: str, dest_dir: Path, *, store: StateStore | None = None) - try: _aws( store, - "s3api", - "get-object", - "--bucket", - store.bucket, - "--key", - f"{ref}.tar.zst", + "s3", + "cp", + f"s3://{store.bucket}/{ref}.tar.zst", str(partial), ) os.replace(partial, dest) @@ -179,16 +176,11 @@ def upload_ref( """Upload one immutable state bundle to CSS S3.""" store = store or state_store() args = [ - "s3api", - "put-object", - "--bucket", - store.bucket, - "--key", - f"{ref}.tar.zst", - "--body", + "s3", + "cp", str(bundle), - "--if-none-match", - "*", + f"s3://{store.bucket}/{ref}.tar.zst", + "--no-overwrite", ] if metadata: args += ["--metadata", json.dumps(dict(metadata), separators=(",", ":"))] @@ -199,3 +191,10 @@ def upload_ref( if any(marker in stderr for marker in ("PreconditionFailed", "ConditionalRequestConflict", "412", "409")): raise StateRefConflict(ref) from exc raise + expected_sha256 = (metadata or {}).get("sha256") + if expected_sha256 is not None: + head = json.loads( + _aws(store, "s3api", "head-object", "--bucket", store.bucket, "--key", f"{ref}.tar.zst", "--output", "json") + ) + if head.get("Metadata", {}).get("sha256") != expected_sha256: + raise StateRefConflict(ref) diff --git a/plugins/nemo-insights/tests/evaluation/test_export.py b/plugins/nemo-insights/tests/evaluation/test_export.py index 802187959c..47590651cd 100644 --- a/plugins/nemo-insights/tests/evaluation/test_export.py +++ b/plugins/nemo-insights/tests/evaluation/test_export.py @@ -29,6 +29,9 @@ def model_dump(self, mode="python", exclude_none=False): assert exclude_none, "export must dump SDK models with exclude_none=True" return {k: v for k, v in self.payload.items() if v is not None} + def __getattr__(self, name): + return self.payload[name] + def _paginator(items): async def gen(): @@ -53,14 +56,27 @@ def __init__(self, name): def list(self, **kwargs): outer.calls.append((self.name, kwargs)) - return _paginator(outer.docs.get((self.name, kwargs["workspace"]), [])) + items = outer.docs.get((self.name, kwargs["workspace"]), []) + filters = kwargs.get("filter") or {} + for key in ("evaluation_id", "trace_id", "session_id"): + if key in filters: + items = [item for item in items if item.payload.get(key) == filters[key]] + return _paginator(items) class _Intake: spans = _Collection("spans") annotations = _Collection("annotations") evaluator_results = _Collection("evaluator_results") + traces = _Collection("traces") + + class _Experiments: + async def retrieve(self, name, **kwargs): + outer.calls.append(("experiments", {"name": name, **kwargs})) + return outer.docs[("experiments", kwargs["workspace"])][0] self.intake = _Intake() + self.experiments = _Experiments() + self.evaluations = _Collection("evaluations") async def close(self): self.closed = True @@ -191,6 +207,66 @@ def test_export_closes_injected_client(tmp_path): assert client.closed +def test_export_experiment_scope_includes_complete_traces(tmp_path, monkeypatch): + docs = { + ("experiments", "ws-a"): [Doc(id="experiment-id", evaluation_count=2)], + ("evaluations", "ws-a"): [Doc(name="eval-b"), Doc(name="eval-a")], + ("traces", "ws-a"): [ + Doc(id="trace-a", session_id="session-a", evaluation_id="eval-a", span_count=2), + Doc(id="trace-b", session_id="session-b", evaluation_id="eval-b", span_count=1), + ], + ("spans", "ws-a"): [ + Doc(span_id="root-a", trace_id="trace-a", started_at="2026-07-01T10:00:00Z"), + Doc(span_id="child-a", trace_id="trace-a", started_at="2026-07-01T10:00:01Z"), + Doc(span_id="root-b", trace_id="trace-b", started_at="2026-07-02T10:00:00Z"), + Doc(span_id="unrelated", trace_id="other", started_at="2026-07-03T10:00:00Z"), + ], + ("annotations", "ws-a"): [ + Doc(annotation_id="a", session_id="session-a"), + Doc(annotation_id="other", session_id="other"), + ], + ("evaluator_results", "ws-a"): [Doc(evaluator_result_id="e", session_id="session-b")], + } + client = _install_fake_client(monkeypatch, docs) + + stats = export.export_workspaces( + "http://localhost:8080", + ["ws-a"], + tmp_path, + since=None, + experiment="prod-latest-completed", + ) + + assert stats["workspaces"]["ws-a"] == {"spans": 3, "annotations": 1, "evaluator_results": 1} + selection = stats["selections"]["ws-a"] + assert selection["evaluation_names"] == ["eval-a", "eval-b"] + assert selection["trace_ids"] == ["trace-a", "trace-b"] + assert selection["expected_spans"] == 3 + assert all("trace_id" in call["filter"] for call in client.calls_for("spans")) + assert all("evaluation_id" not in call["filter"] for call in client.calls_for("spans")) + + +def test_export_experiment_scope_rejects_source_drift(tmp_path, monkeypatch): + docs = { + ("experiments", "ws-a"): [Doc(id="experiment-id", evaluation_count=1)], + ("evaluations", "ws-a"): [Doc(name="eval-a")], + ("traces", "ws-a"): [ + Doc(id="trace-a", session_id="session-a", evaluation_id="eval-a", span_count=2), + ], + ("spans", "ws-a"): [Doc(span_id="root-a", trace_id="trace-a", started_at="2026-07-01T10:00:00Z")], + } + _install_fake_client(monkeypatch, docs) + + with pytest.raises(RuntimeError, match="source changed during capture"): + export.export_workspaces( + "http://localhost:8080", + ["ws-a"], + tmp_path, + since=None, + experiment="prod-latest-completed", + ) + + # --------------------------------------------------------------------------- # # subject scoping # --------------------------------------------------------------------------- # @@ -356,10 +432,11 @@ def boom(url, timeout): def _fake_export(seen): - def fake(base_url, workspaces, out_dir, *, since, client=None): + def fake(base_url, workspaces, out_dir, *, since, experiment=None, client=None): seen["base_url"] = base_url seen["workspaces"] = list(workspaces) seen["since"] = since + seen["experiment"] = experiment seen["client"] = client for ws in workspaces: ws_dir = out_dir / "export" / ws @@ -367,11 +444,14 @@ def fake(base_url, workspaces, out_dir, *, since, client=None): (ws_dir / "spans.jsonl").write_text('{"span_id": "s1"}\n', encoding="utf-8") (ws_dir / "annotations.jsonl").write_text("", encoding="utf-8") (ws_dir / "evaluator_results.jsonl").write_text("", encoding="utf-8") - return { + result = { "workspaces": {ws: {"spans": 1, "annotations": 0, "evaluator_results": 0} for ws in workspaces}, "min_start_time": "2026-07-01T00:00:00+00:00", "max_start_time": "2026-07-02T00:00:00+00:00", } + if experiment: + result["selections"] = {workspaces[0]: {"kind": "experiment", "experiment": experiment}} + return result return fake @@ -452,6 +532,29 @@ def test_snapshot_export_passes_since_through(tmp_path, monkeypatch): assert seen["since"] == since +def test_snapshot_export_passes_experiment_scope(tmp_path, monkeypatch): + seen: dict = {} + monkeypatch.setattr(artifact.export, "export_workspaces", _fake_export(seen)) + monkeypatch.setattr(artifact, "fetch_platform_info", lambda url: None) + subject = _subject( + "kernel-factory", + "intake", + workspace="kf-prod-evals", + base_url="u", + experiment="prod-latest-completed", + ) + (tmp_path / "tmp").mkdir() + + artifact.snapshot_export([subject], tmp_path / "b.tar.zst", tmp_path / "tmp", since=None) + + assert seen["experiment"] == "prod-latest-completed" + extract = tmp_path / "extract" + extract.mkdir() + subprocess.run(["tar", "--zstd", "-xf", str(tmp_path / "b.tar.zst"), "-C", str(extract)], check=True) + manifest = json.loads((extract / "state" / "manifest.json").read_text()) + assert manifest["selections"]["kf-prod-evals"]["experiment"] == "prod-latest-completed" + + def test_snapshot_export_dedupes_workspaces_across_subjects(tmp_path, monkeypatch): seen: dict = {} monkeypatch.setattr(artifact.export, "export_workspaces", _fake_export(seen)) @@ -463,6 +566,15 @@ def test_snapshot_export_dedupes_workspaces_across_subjects(tmp_path, monkeypatc assert seen["workspaces"] == ["shared"] +def test_snapshot_export_rejects_shared_experiment_workspace(tmp_path, monkeypatch): + monkeypatch.setattr(artifact.export, "export_workspaces", _fake_export({})) + a = _subject("a", "intake", workspace="shared", base_url="u", experiment="run-a") + b = _subject("b", "intake", workspace="shared", base_url="u") + (tmp_path / "tmp").mkdir() + with pytest.raises(SystemExit, match="shared"): + artifact.snapshot_export([a, b], tmp_path / "b.tar.zst", tmp_path / "tmp", since=None) + + def test_snapshot_export_conflicting_base_urls_exit(tmp_path, monkeypatch): monkeypatch.setattr(artifact.export, "export_workspaces", _fake_export({})) a = _subject("a", "intake", workspace="wa", base_url="http://one:8080") diff --git a/plugins/nemo-insights/tests/evaluation/test_reingest.py b/plugins/nemo-insights/tests/evaluation/test_reingest.py index d3327d9527..6583ab7956 100644 --- a/plugins/nemo-insights/tests/evaluation/test_reingest.py +++ b/plugins/nemo-insights/tests/evaluation/test_reingest.py @@ -150,6 +150,27 @@ def test_llm_doc_inverts_model_and_sets_parent(): assert "input.value" not in otlp["attributes"] # doc had no input +def test_atif_doc_inverts_to_exact_direct_span(): + doc = { + **AGENT_DOC, + "source": "atif", + "span_id": "span-abc123", + "trace_id": "campaign-session", + "session_id": "campaign-session", + } + + source, direct = reingest.doc_to_direct_span(doc, CATALOG) + + assert source == "atif" + assert direct["span_id"] == "span-abc123" + assert direct["trace_id"] == "campaign-session" + assert direct["started_at"].endswith("+00:00") + assert direct["input"] == "do 1" + assert direct["output"] == "ok" + assert direct["attributes"]["otel.scope"] == '{"name":"evaluation","version":"1.0.0"}' + assert direct["attributes"]["nemo.evaluation.name"] == "smoke-20260626-121437-5559-20260626-121438-a833" + + def test_raw_alias_wins_over_inversion(): doc = dict(LLM_DOC) doc["raw_attributes"] = json.dumps({"llm.model_name": "other"}) @@ -921,46 +942,43 @@ def test_ingest_bundle_validates_target_names(tmp_path, quiet_platform): ) -# --- source guard: only otel-sourced corpora are restorable (B1) --- +# --- provider-neutral restore for non-OTLP sources --- -def test_ingest_bundle_rejects_non_otel_sources(tmp_path, quiet_platform): - """ATIF/chat-completions span ids are not OTLP hex — bytes.fromhex would crash mid-batch - (partial restore) and any doc that survived would silently mutate (source_format -> otel). - The guard must fire BEFORE any network I/O, naming workspace + offending sources.""" +def test_ingest_bundle_restores_non_otel_sources_directly(tmp_path, quiet_platform): atif = {**AGENT_DOC, "source": "atif", "span_id": "span-abc123"} chat = {**LLM_DOC, "source": "chat-completions", "span_id": "chatcmpl-xyz"} export_dir = _write_export(tmp_path, "ws-a", [atif, chat, AGENT_DOC], [ANNOTATION_DOC]) - with pytest.raises(RuntimeError) as exc: - reingest.ingest_bundle( - "http://x", - export_dir, - _manifest("ws-a", 3, 1), - workspace_map={"ws-a": "ws-b"}, - catalog=CATALOG, - ) - message = str(exc.value) - assert "ws-a" in message # names the workspace - assert "atif" in message and "chat-completions" in message # offending source values - assert "otel" in message # states only otel-sourced corpora are restorable - assert quiet_platform["requests"] == [] - assert quiet_platform["posts"] == [] - assert quiet_platform["ensured"] == [] # nothing touched the platform at all + quiet_platform["span_counts"] = [0, 3] + quiet_platform["annotation_counts"] = [0] + quiet_platform["result_counts"] = [0] + + outcome = reingest.ingest_bundle( + "http://x", + export_dir, + _manifest("ws-a", 3, 1), + workspace_map={"ws-a": "ws-b"}, + catalog=CATALOG, + sleep=lambda _: None, + ) + + direct = [body for url, body in quiet_platform["posts"] if url.endswith("/ingest/spans")] + assert {body["source"] for body in direct} == {"atif", "chat-completions", "otel"} + assert outcome["ws-a"]["spans"] == {"ingested": 3, "skipped": 0} -def test_ingest_bundle_non_otel_anywhere_blocks_every_workspace(tmp_path, quiet_platform): - """The scan covers ALL bundle workspaces before ingesting ANY — all-or-nothing.""" +def test_ingest_bundle_scans_all_span_files_before_writes(tmp_path, quiet_platform): _write_export(tmp_path, "ws-a", [AGENT_DOC]) ws_dir = tmp_path / "export" / "ws-c" ws_dir.mkdir(parents=True) - bad = {**AGENT_DOC, "source": "atif"} + bad = {key: value for key, value in AGENT_DOC.items() if key != "source"} (ws_dir / "spans.jsonl").write_text(json.dumps(bad) + "\n", encoding="utf-8") manifest = { "workspaces": ["ws-a", "ws-c"], "counts": {"ws-a": {"spans": 1}, "ws-c": {"spans": 1}}, "min_start_time": datetime.now(timezone.utc).isoformat(), } - with pytest.raises(RuntimeError, match="ws-c"): + with pytest.raises(ValueError, match="no source"): reingest.ingest_bundle( "http://x", tmp_path / "export", @@ -1329,6 +1347,18 @@ def test_diff_reports_value_changes_and_count_drift(): assert mismatches == ["ws-a/spans: 1 exported vs 0 restored"] +def test_grouped_jsonl_diff_ignores_order_within_trace(tmp_path): + original = tmp_path / "original.jsonl" + restored = tmp_path / "restored.jsonl" + original.write_text("\n".join(json.dumps(doc) for doc in (AGENT_DOC, LLM_DOC)) + "\n", encoding="utf-8") + restored_docs = [ + {**doc, "workspace": "scratch-rt", "ingested_at": "2026-08-18T00:00:00Z"} for doc in (LLM_DOC, AGENT_DOC) + ] + restored.write_text("\n".join(json.dumps(doc) for doc in restored_docs) + "\n", encoding="utf-8") + + assert reingest._diff_grouped_jsonl(original, restored, workspace="ws-a", collection="spans") == [] + + def test_diff_normalizes_annotation_server_fields(): restored = { **ANNOTATION_DOC, @@ -1388,6 +1418,38 @@ def fake_export(base_url, workspaces, out_dir, *, since): assert seen["cleaned"] == ["scratch-reingest-cafef00d-ws-a"] +def test_roundtrip_diff_handles_mixed_scoped_workspaces(tmp_path, monkeypatch): + manifest = { + "workspaces": ["ws-a", "ws-b"], + "counts": {}, + "selections": {"ws-a": {"experiment": "run-a"}}, + } + exports = [] + + def fake_export(base_url, workspaces, out_dir, *, since, selection=None): + exports.append((workspaces[0], selection)) + return {"workspaces": {}} + + monkeypatch.setattr(reingest, "ingest_bundle", lambda *args, **kwargs: {}) + monkeypatch.setattr(reingest.export, "export_workspaces", fake_export) + monkeypatch.setattr(reingest, "_diff_grouped_jsonl", lambda *args, **kwargs: []) + monkeypatch.setattr(reingest, "cleanup_scratch", lambda *args: None) + + mismatches = reingest.roundtrip_diff( + "http://x", + tmp_path, + manifest, + scratch_prefix="scratch-reingest-", + catalog=CATALOG, + content_key="cafef00d", + ) + assert mismatches == [] + assert exports == [ + ("scratch-reingest-cafef00d-ws-a", {"experiment": "run-a"}), + ("scratch-reingest-cafef00d-ws-b", None), + ] + + def test_roundtrip_diff_cleans_up_even_when_ingest_fails(tmp_path, monkeypatch): export_dir = _write_export(tmp_path / "src", "ws-a", [AGENT_DOC]) cleaned = [] @@ -1468,29 +1530,16 @@ def __init__(self, *args, **kwargs): assert "manual" in err.lower() -def test_cleanup_scratch_loopback_deletes_rows_and_records(monkeypatch, fake_docker): - deleted: list[str] = [] - - class _FakeClient: +def test_cleanup_scratch_loopback_deletes_rows_and_keeps_reusable_record(monkeypatch, fake_docker): + class _NoHTTP: def __init__(self, *args, **kwargs): - pass - - def __enter__(self): - return self + raise AssertionError("row cleanup must keep the active workspace record") - def __exit__(self, *args): - return False - - def delete(self, url): - deleted.append(url) - return httpx.Response(204) - - monkeypatch.setattr(reingest.httpx, "Client", _FakeClient) + monkeypatch.setattr(reingest.httpx, "Client", _NoHTTP) monkeypatch.setattr(reingest, "_local_clickhouse_container", lambda: "nmp-intake-clickhouse-a1b2c3d4e5f6") reingest.cleanup_scratch("http://127.0.0.1:8080", ["scratch-rt-abc12345-ws-a"]) tables = {cmd[-1].split("FROM intake.")[1].split(" ")[0] for cmd in fake_docker} assert tables == {"spans", "annotations", "evaluator_results", "trace_index"} - assert len(deleted) == 1 and deleted[0].endswith("/apis/entities/v2/workspaces/scratch-rt-abc12345-ws-a") def test_cleanup_scratch_keeps_workspace_record_when_container_is_ambiguous(monkeypatch, capsys): diff --git a/plugins/nemo-insights/tests/evaluation/test_release.py b/plugins/nemo-insights/tests/evaluation/test_release.py index 723bf73252..18accdb1e4 100644 --- a/plugins/nemo-insights/tests/evaluation/test_release.py +++ b/plugins/nemo-insights/tests/evaluation/test_release.py @@ -187,12 +187,9 @@ def fake_aws(store, *args): partial = Path(calls[0][-1]) assert calls == [ ( - "s3api", - "get-object", - "--bucket", - STORE.bucket, - "--key", - "state-v4.tar.zst", + "s3", + "cp", + f"s3://{STORE.bucket}/state-v4.tar.zst", str(partial), ) ] @@ -243,26 +240,34 @@ def test_upload_ref_uses_key_and_metadata(tmp_path, monkeypatch): bundle = tmp_path / "state-v11.tar.zst" bundle.write_bytes(b"fixture") calls = [] - monkeypatch.setattr(release, "_aws", lambda store, *args: calls.append(args) or "") - release.upload_ref("state-v11", bundle, metadata={"reason": "new data"}, store=STORE) + monkeypatch.setattr( + release, + "_aws", + lambda store, *args: calls.append(args) or '{"Metadata":{"sha256":"abc"}}', + ) + release.upload_ref("state-v11", bundle, metadata={"reason": "new data", "sha256": "abc"}, store=STORE) assert calls == [ ( - "s3api", - "put-object", - "--bucket", - STORE.bucket, - "--key", - "state-v11.tar.zst", - "--body", + "s3", + "cp", str(bundle), - "--if-none-match", - "*", + f"s3://{STORE.bucket}/state-v11.tar.zst", + "--no-overwrite", "--metadata", - '{"reason":"new data"}', - ) + '{"reason":"new data","sha256":"abc"}', + ), + ("s3api", "head-object", "--bucket", STORE.bucket, "--key", "state-v11.tar.zst", "--output", "json"), ] +def test_upload_ref_detects_no_overwrite_conflict(tmp_path, monkeypatch): + bundle = tmp_path / "state-v11.tar.zst" + bundle.write_bytes(b"fixture") + monkeypatch.setattr(release, "_aws", lambda *args: '{"Metadata":{"sha256":"different"}}') + with pytest.raises(release.StateRefConflict): + release.upload_ref("state-v11", bundle, metadata={"sha256": "expected"}, store=STORE) + + @pytest.mark.parametrize("stderr", ["PreconditionFailed", "ConditionalRequestConflict", "HTTP 412", "HTTP 409"]) def test_upload_ref_reports_conditional_conflict(tmp_path, monkeypatch, stderr): bundle = tmp_path / "state-v11.tar.zst" diff --git a/sdk/python/nemo-platform/src/nemo_platform/skills/nemo-intake/references/ingest-formats.md b/sdk/python/nemo-platform/src/nemo_platform/skills/nemo-intake/references/ingest-formats.md index c014fc347c..59b0b14457 100644 --- a/sdk/python/nemo-platform/src/nemo_platform/skills/nemo-intake/references/ingest-formats.md +++ b/sdk/python/nemo-platform/src/nemo_platform/skills/nemo-intake/references/ingest-formats.md @@ -197,6 +197,8 @@ attributes retain native JSON types. Known semantic attributes populate queryabl attributes appear in detailed reads under `raw_attributes`. Reposting the same `(source, trace_id, span_id)` updates the existing logical span. +String inputs and outputs are stored verbatim; objects and arrays are serialized as JSON. + The default ClickHouse TTL is 90 days from `started_at`. If any span is outside that window, Intake returns `422` before writing the batch and instructs the operator to increase the `spans` and `trace_index` TTLs. Provider timestamps are never rewritten by the endpoint. diff --git a/services/intake/src/nmp/intake/repository/clickhouse/span.py b/services/intake/src/nmp/intake/repository/clickhouse/span.py index e7f7b505f6..0c56d4b778 100644 --- a/services/intake/src/nmp/intake/repository/clickhouse/span.py +++ b/services/intake/src/nmp/intake/repository/clickhouse/span.py @@ -441,6 +441,6 @@ def _row_to_span( def _none_if_zero_datetime(value: Any) -> datetime | None: if value is None: return None - if value.timestamp() == 0: + if value == _ZERO_DATETIME or (value.tzinfo is None and value == _ZERO_DATETIME.replace(tzinfo=None)): return None return value diff --git a/services/intake/src/nmp/intake/spans/ingest/spans.py b/services/intake/src/nmp/intake/spans/ingest/spans.py index 322ed92e12..fcb8ae42ea 100644 --- a/services/intake/src/nmp/intake/spans/ingest/spans.py +++ b/services/intake/src/nmp/intake/spans/ingest/spans.py @@ -167,8 +167,18 @@ def direct_span_to_domain( attributes_string=attribute_bags.string, attributes_number=attribute_bags.number, attributes_bool=attribute_bags.boolean, - input="" if span.input is None else json_dumps_preserve(span.input), - output="" if span.output is None else json_dumps_preserve(span.output), + input="" + if span.input is None + else span.input + if isinstance(span.input, str) + else json_dumps_preserve(span.input), + output=( + "" + if span.output is None + else span.output + if isinstance(span.output, str) + else json_dumps_preserve(span.output) + ), event_ts=ingested_at, ) return intake_span, semantic_attributes diff --git a/services/intake/tests/integration/spans/test_direct_span_ingest.py b/services/intake/tests/integration/spans/test_direct_span_ingest.py index a11fc3e7c5..9a589d2f76 100644 --- a/services/intake/tests/integration/spans/test_direct_span_ingest.py +++ b/services/intake/tests/integration/spans/test_direct_span_ingest.py @@ -58,7 +58,9 @@ def test_direct_span_ingest_round_trips_batch_and_raw_json(client: TestClient): name="model-call", kind="LLM", started_at=_timestamp(1), - ended_at=_timestamp(2), + ended_at=None, + input="plain input", + output='{"already":"serialized"}', attributes={ "llm.model_name": "child-model", "llm.token_count.prompt": 12, @@ -97,6 +99,9 @@ def test_direct_span_ingest_round_trips_batch_and_raw_json(client: TestClient): assert child["parent_span_id"] == "root-span" assert child["input_tokens"] == 12 assert child["output_tokens"] == 3 + assert child["input"] == "plain input" + assert child["output"] == '{"already":"serialized"}' + assert "ended_at" not in child assert json.loads(child["raw_attributes"])["provider.raw"]["events"][0]["value"] is None From 9146cd88cf063fb398c0eb493ae6f243ee00d985 Mon Sep 17 00:00:00 2001 From: Nico Tonozzi Date: Tue, 18 Aug 2026 17:50:40 -0700 Subject: [PATCH 2/4] test(insights): cover kernel fixture registry Keep the full evaluation suite aligned with the new pinned subject and scoped export argument. Signed-off-by: Nico Tonozzi --- .../tests/evaluation/test_artifact_snapshot_auth.py | 1 + plugins/nemo-insights/tests/evaluation/test_registry.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/plugins/nemo-insights/tests/evaluation/test_artifact_snapshot_auth.py b/plugins/nemo-insights/tests/evaluation/test_artifact_snapshot_auth.py index 05ad933bc3..7c912b2eb2 100644 --- a/plugins/nemo-insights/tests/evaluation/test_artifact_snapshot_auth.py +++ b/plugins/nemo-insights/tests/evaluation/test_artifact_snapshot_auth.py @@ -149,6 +149,7 @@ def fake_export( *, since: object, client: object | None, + experiment: str | None = None, ) -> dict: exports.append((workspaces, client)) return { diff --git a/plugins/nemo-insights/tests/evaluation/test_registry.py b/plugins/nemo-insights/tests/evaluation/test_registry.py index bf58558e4c..499d7cce49 100644 --- a/plugins/nemo-insights/tests/evaluation/test_registry.py +++ b/plugins/nemo-insights/tests/evaluation/test_registry.py @@ -30,6 +30,7 @@ def test_registry_contains_only_expected_analyzable_subjects() -> None: assert set(subjects) == { "glamr", + "kernel-factory", "nemo-oo-airline", "nvq", "tau2-airline", @@ -67,6 +68,7 @@ def test_tau2_telecom_uses_small_split() -> None: def test_every_analyzable_subject_has_expected_state_pin() -> None: expected = { "glamr": "state-v8", + "kernel-factory": "state-v11", "nemo-oo-airline": "state-v9", "nvq": "state-v7", "tau2-airline": "state-v6", From d84527ac114a0be064776836d05c4a6d8e45a1e3 Mon Sep 17 00:00:00 2001 From: Nico Tonozzi Date: Tue, 18 Aug 2026 17:54:18 -0700 Subject: [PATCH 3/4] fix(insights): validate snapshot prerequisites Require the AWS CLI version that supports immutable multipart uploads and correct the experiment membership documentation. Signed-off-by: Nico Tonozzi --- plugins/nemo-insights/evaluation/README.md | 6 +++--- plugins/nemo-insights/evaluation/cli.py | 17 +++++++++++++++-- .../tests/evaluation/test_cli.py | 19 ++++++++++++++++++- 3 files changed, 36 insertions(+), 6 deletions(-) diff --git a/plugins/nemo-insights/evaluation/README.md b/plugins/nemo-insights/evaluation/README.md index 087491260b..90e929b6e8 100644 --- a/plugins/nemo-insights/evaluation/README.md +++ b/plugins/nemo-insights/evaluation/README.md @@ -12,7 +12,7 @@ wheel. - Run every command from `plugins/nemo-insights`; the `evaluation` package is not installed in the plugin wheel. -- Install the AWS CLI and configure the CSS S3 credentials described in +- Install AWS CLI 2.33.0 or newer and configure the CSS S3 credentials described in [State bundles](#state-bundles). - Start a local NeMo Platform at `http://localhost:8080` before using pinned analysis, restore, roundtrip, or guarded publish commands. @@ -160,8 +160,8 @@ uv run python -m evaluation publish evaluation/tmp/glamr.tar.zst --base http://l `snapshot` drains the subject's workspaces (benchmark subjects: realistic + `-oracle` twin) into JSONL + manifest — no ClickHouse, no Docker. An intake subject can set `experiment = ""` to capture only that Experiment's -complete traces; membership comes from root spans, then every child is exported -by trace ID. `publish` +complete traces; membership comes from the Experiment's evaluations and their +traces, then every span in each trace is exported by trace ID. `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` only after running `roundtrip` separately and confirming it diff --git a/plugins/nemo-insights/evaluation/cli.py b/plugins/nemo-insights/evaluation/cli.py index 2a2069468c..ac84ba6c98 100644 --- a/plugins/nemo-insights/evaluation/cli.py +++ b/plugins/nemo-insights/evaluation/cli.py @@ -95,17 +95,30 @@ def _load_dotenv(path: Path = ENV_PATH) -> None: os.environ.setdefault(key, value) +def _aws_cli_version(executable: str) -> tuple[int, int, int] | None: + result = subprocess.run([executable, "--version"], capture_output=True, text=True) + match = re.search(r"aws-cli/(\d+)\.(\d+)\.(\d+)", result.stdout + result.stderr) + if not match: + return None + major, minor, patch = match.groups() + return int(major), int(minor), int(patch) + + def _doctor(subjects: dict[str, Subject], name: str | None) -> None: """Print a readiness checklist for one subject (or all): what's set up, what's not.""" names = [name] if name else sorted(subjects) + aws = shutil.which("aws") + aws_version = _aws_cli_version(aws) if aws else None for subject_name in names: subject = subjects.get(subject_name) if subject is None: print(f"✗ {subject_name}: unknown subject") continue unmet: list[str] = [] - if shutil.which("aws") is None: - unmet.append("AWS CLI (needed for pinned/--state analyze; install with `brew install awscli`)") + if aws is None: + unmet.append("AWS CLI 2.33.0+ (needed for pinned/--state analyze; install with `brew install awscli`)") + elif (aws_version or (0, 0, 0)) < (2, 33, 0): + unmet.append("AWS CLI 2.33.0+ (required for immutable multipart state uploads)") if not os.environ.get(release.ACCESS_KEY_ENV) or not os.environ.get(release.SECRET_KEY_ENV): unmet.append("CSS S3 credentials in evaluation/.env (CSS Portal → Auth Info)") unmet += build_adapter(subject).check() diff --git a/plugins/nemo-insights/tests/evaluation/test_cli.py b/plugins/nemo-insights/tests/evaluation/test_cli.py index b6d6010989..805780a36e 100644 --- a/plugins/nemo-insights/tests/evaluation/test_cli.py +++ b/plugins/nemo-insights/tests/evaluation/test_cli.py @@ -863,10 +863,17 @@ def test_load_dotenv_missing_file_is_noop(tmp_path, monkeypatch): assert os.environ.get("NOPE_SENTINEL") is None # and sets nothing +def test_aws_cli_version(monkeypatch): + result = subprocess.CompletedProcess([], 0, stdout="aws-cli/2.36.24 Python/3.13.11", stderr="") + monkeypatch.setattr(cli.subprocess, "run", lambda *args, **kwargs: result) + assert cli._aws_cli_version("/usr/bin/aws") == (2, 36, 24) + + def test_doctor_ready(monkeypatch, capsys): monkeypatch.setattr(cli, "_load_dotenv", lambda *a, **k: None) monkeypatch.setattr("evaluation.adapters.IntakeAdapter.check", lambda self: []) monkeypatch.setattr(cli.shutil, "which", lambda _name: "/usr/bin/aws") + monkeypatch.setattr(cli, "_aws_cli_version", lambda _executable: (2, 33, 0)) monkeypatch.setenv(cli.release.ACCESS_KEY_ENV, "team-test") monkeypatch.setenv(cli.release.SECRET_KEY_ENV, "secret") monkeypatch.setattr(sys, "argv", ["evaluation", "doctor", "nvq"]) @@ -889,7 +896,17 @@ def test_doctor_flags_missing_aws(monkeypatch, capsys): cli.main() out = capsys.readouterr().out assert "✗" in out - assert "AWS CLI (needed for pinned/--state analyze; install with `brew install awscli`)" in out + assert "AWS CLI 2.33.0+ (needed for pinned/--state analyze; install with `brew install awscli`)" in out + + +def test_doctor_flags_old_aws(monkeypatch, capsys): + monkeypatch.setattr("evaluation.adapters.IntakeAdapter.check", lambda self: []) + monkeypatch.setattr(cli.shutil, "which", lambda _name: "/usr/bin/aws") + monkeypatch.setattr(cli, "_aws_cli_version", lambda _executable: (2, 32, 34)) + monkeypatch.setenv(cli.release.ACCESS_KEY_ENV, "team-test") + monkeypatch.setenv(cli.release.SECRET_KEY_ENV, "secret") + cli._doctor(cli.load_registry(cli.REGISTRY_PATH), "nvq") + assert "AWS CLI 2.33.0+" in capsys.readouterr().out def test_doctor_lists_unmet(monkeypatch, capsys): From 83da1a07c8c327f2bfcacef507b781f55c3c8160 Mon Sep 17 00:00:00 2001 From: Nico Tonozzi Date: Tue, 18 Aug 2026 18:24:15 -0700 Subject: [PATCH 4/4] fix(insights): bound AWS version probe Treat failed or timed-out AWS CLI probes as an unknown version so doctor reports the actual prerequisite failure. Signed-off-by: Nico Tonozzi --- plugins/nemo-insights/evaluation/cli.py | 11 +++++++-- .../tests/evaluation/test_cli.py | 24 ++++++++++++++++--- 2 files changed, 30 insertions(+), 5 deletions(-) diff --git a/plugins/nemo-insights/evaluation/cli.py b/plugins/nemo-insights/evaluation/cli.py index ac84ba6c98..c65f22707c 100644 --- a/plugins/nemo-insights/evaluation/cli.py +++ b/plugins/nemo-insights/evaluation/cli.py @@ -96,7 +96,12 @@ def _load_dotenv(path: Path = ENV_PATH) -> None: def _aws_cli_version(executable: str) -> tuple[int, int, int] | None: - result = subprocess.run([executable, "--version"], capture_output=True, text=True) + try: + result = subprocess.run([executable, "--version"], capture_output=True, text=True, timeout=5) + except (OSError, subprocess.TimeoutExpired): + return None + if result.returncode: + return None match = re.search(r"aws-cli/(\d+)\.(\d+)\.(\d+)", result.stdout + result.stderr) if not match: return None @@ -117,7 +122,9 @@ def _doctor(subjects: dict[str, Subject], name: str | None) -> None: unmet: list[str] = [] if aws is None: unmet.append("AWS CLI 2.33.0+ (needed for pinned/--state analyze; install with `brew install awscli`)") - elif (aws_version or (0, 0, 0)) < (2, 33, 0): + elif aws_version is None: + unmet.append("could not determine AWS CLI version (2.33.0+ required)") + elif aws_version < (2, 33, 0): unmet.append("AWS CLI 2.33.0+ (required for immutable multipart state uploads)") if not os.environ.get(release.ACCESS_KEY_ENV) or not os.environ.get(release.SECRET_KEY_ENV): unmet.append("CSS S3 credentials in evaluation/.env (CSS Portal → Auth Info)") diff --git a/plugins/nemo-insights/tests/evaluation/test_cli.py b/plugins/nemo-insights/tests/evaluation/test_cli.py index 805780a36e..da2d67121b 100644 --- a/plugins/nemo-insights/tests/evaluation/test_cli.py +++ b/plugins/nemo-insights/tests/evaluation/test_cli.py @@ -869,6 +869,20 @@ def test_aws_cli_version(monkeypatch): assert cli._aws_cli_version("/usr/bin/aws") == (2, 36, 24) +@pytest.mark.parametrize( + "failure", + [OSError(), subprocess.TimeoutExpired("/usr/bin/aws", 5), subprocess.CompletedProcess([], 1)], +) +def test_aws_cli_version_probe_failure(monkeypatch, failure): + def fail(*args, **kwargs): + if isinstance(failure, BaseException): + raise failure + return failure + + monkeypatch.setattr(cli.subprocess, "run", fail) + assert cli._aws_cli_version("/usr/bin/aws") is None + + def test_doctor_ready(monkeypatch, capsys): monkeypatch.setattr(cli, "_load_dotenv", lambda *a, **k: None) monkeypatch.setattr("evaluation.adapters.IntakeAdapter.check", lambda self: []) @@ -899,14 +913,18 @@ def test_doctor_flags_missing_aws(monkeypatch, capsys): assert "AWS CLI 2.33.0+ (needed for pinned/--state analyze; install with `brew install awscli`)" in out -def test_doctor_flags_old_aws(monkeypatch, capsys): +@pytest.mark.parametrize( + ("version", "message"), + [(None, "could not determine AWS CLI version"), ((2, 32, 34), "AWS CLI 2.33.0+")], +) +def test_doctor_flags_unsupported_aws(monkeypatch, capsys, version, message): monkeypatch.setattr("evaluation.adapters.IntakeAdapter.check", lambda self: []) monkeypatch.setattr(cli.shutil, "which", lambda _name: "/usr/bin/aws") - monkeypatch.setattr(cli, "_aws_cli_version", lambda _executable: (2, 32, 34)) + monkeypatch.setattr(cli, "_aws_cli_version", lambda _executable: version) monkeypatch.setenv(cli.release.ACCESS_KEY_ENV, "team-test") monkeypatch.setenv(cli.release.SECRET_KEY_ENV, "secret") cli._doctor(cli.load_registry(cli.REGISTRY_PATH), "nvq") - assert "AWS CLI 2.33.0+" in capsys.readouterr().out + assert message in capsys.readouterr().out def test_doctor_lists_unmet(monkeypatch, capsys):