diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/entities.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/entities.py index e6e7070fe1..a2346a70f7 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/entities.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/entities.py @@ -157,6 +157,18 @@ class Candidate(NemoEntity, entity_type="candidate"): default=None, description="Insight-suite trial results from the last evaluation run.", ) + insight_suite_identity: str | None = Field( + default=None, + description="Content identity of the Insight suite associated with insight_reward.", + ) + insight_suite_artifact_ref: str | None = Field( + default=None, + description="Portable reference to the immutable Insight suite associated with insight_reward.", + ) + insight_metric_keys: list[str] | None = Field( + default=None, + description="Validated runtime metric keys associated with insight_reward.", + ) validation_trajectory_reward: dict[str, float] | None = Field( default=None, description="Validation trajectory reward: aggregate + per-node scores.", diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/README.md b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/README.md index 810d0ae0ea..0f7319316a 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/README.md +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/README.md @@ -58,13 +58,36 @@ scoped to the Insight suite; the user's train and validation datasets remain unchanged. The authored verifiers must pass static Harbor validation before the local suite is returned to the optimization loop. +After authoring and validation, Eval Author hashes every task file and verifier +file, derives deterministic suite and scorer identities, and freezes the exact +content beneath: + +```text +eval-and-optimize/eval_author//artifacts//insight-suite/ +``` + +The returned dataset points at this immutable artifact and carries a portable +`nemo-experimentalist-insight-suite://.../sha256/...` reference. Candidate Insight +results persist the same suite identity and artifact reference. Resume reuses +those results only when the identity still matches; changed task or verifier +content is re-evaluated. + Task-template inputs may be local paths, `file://` URIs, or NeMo Platform `fileset:///` references. Fileset-backed templates are downloaded into the experiment-local staging directory before Harbor parses them. The staged template is refreshed on every invocation rather than reused. -`EvalAuthorResult.insight_suite` contains the experiment-local materialized -`Dataset` for immediate evaluation by the optimization loop. +`EvalAuthorResult.insight_suite` contains the finalized content-addressed +`Dataset` for immediate evaluation by the optimization loop. Its identity and +portable artifact reference are also available as +`EvalAuthorResult.insight_suite_identity` and +`EvalAuthorResult.insight_suite_artifact_ref`. + +Insight metrics remain adaptive development feedback. They may steer round +analysis, goal-tree updates, and proposals, but validation remains the direct +Pareto and winner-selection criterion. Promotion suggestions require complete +repeated baseline-to-winner improvement evidence, remain advisory, and never +mutate the canonical validation dataset. ## Intended Invocation diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/agent.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/agent.py index 6c13623ec2..af86c6c260 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/agent.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/agent.py @@ -366,10 +366,13 @@ async def _run( validation_feedback=str(exc), ) else: + artifact = insight_suite.finalize_artifact() return EvalAuthorResult( train_dataset=train_dataset, validation_dataset=validation_dataset, - insight_suite=materialized_dataset, + insight_suite=artifact.dataset, + insight_suite_identity=artifact.identity, + insight_suite_artifact_ref=artifact.ref, summary=summary, ) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/materialization.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/materialization.py index 96219b5cd7..dae2757e21 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/materialization.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/materialization.py @@ -10,8 +10,11 @@ import os import re import shutil +import tomllib from dataclasses import dataclass from pathlib import Path +from typing import cast +from urllib.parse import urlparse from uuid import uuid4 import tomlkit @@ -19,8 +22,12 @@ from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset from nemo_experimentalist_plugin.experimentalist.components.evaluator.models import Task, local_path_from_uri -_MANIFEST_SCHEMA_VERSION = 1 +_MANIFEST_SCHEMA_VERSION = 2 +_CONTENT_HASH_SCHEMA_VERSION = 1 +_METRIC_CONTRACT_VERSION = 1 +_ARTIFACT_SCHEME = "nemo-experimentalist-insight-suite" _SLUG_RE = re.compile(r"[^a-z0-9]+") +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") def _slug(value: str, *, fallback: str, max_length: int = 48) -> str: @@ -32,6 +39,104 @@ def _digest(value: str, length: int = 10) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest()[:length] +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _canonical_digest(value: object) -> str: + return _sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")) + + +def _file_hashes(root: Path) -> dict[str, str]: + return { + path.relative_to(root).as_posix(): _sha256_bytes(path.read_bytes()) + for path in sorted(root.rglob("*")) + if path.is_file() + } + + +def _verifier_dir(task_dir: Path) -> Path: + config = tomllib.loads((task_dir / "task.toml").read_text(encoding="utf-8")) + verifier = config.get("verifier") + if isinstance(verifier, dict): + configured = verifier.get("directory") + if isinstance(configured, str) and configured.strip(): + path = Path(configured) + return path if path.is_absolute() else task_dir / path + for name in ("tests", "test"): + path = task_dir / name + if path.is_dir(): + return path + raise ValueError(f"Materialized task has no verifier directory: {task_dir}") + + +def _content_provenance(suite_dir: Path, manifest: dict[str, object]) -> tuple[list[dict[str, object]], str, str]: + raw_tasks = manifest.get("tasks") + if not isinstance(raw_tasks, list): + raise ValueError(f"Insight suite manifest has invalid tasks: {suite_dir / 'manifest.json'}") + + tasks: list[dict[str, object]] = [] + scorer_inputs: list[dict[str, str]] = [] + suite_inputs: list[dict[str, str]] = [] + for raw_task in raw_tasks: + if not isinstance(raw_task, dict): + raise ValueError(f"Insight suite manifest has invalid task entry: {raw_task!r}") + if not all(isinstance(key, str) for key in raw_task): + raise ValueError(f"Insight suite manifest task has invalid keys: {raw_task!r}") + task_entry = cast(dict[str, object], raw_task) + relative_path = task_entry.get("path") + if not isinstance(relative_path, str) or not relative_path: + raise ValueError(f"Insight suite manifest task has invalid path: {relative_path!r}") + task_dir = (suite_dir / relative_path).resolve() + try: + task_dir.relative_to(suite_dir.resolve()) + except ValueError as exc: + raise ValueError(f"Insight suite manifest task escapes the suite: {relative_path!r}") from exc + if not task_dir.is_dir(): + raise ValueError(f"Insight suite manifest task path is missing: {task_dir}") + + files = _file_hashes(task_dir) + verifier_dir = _verifier_dir(task_dir).resolve() + try: + verifier_path = verifier_dir.relative_to(task_dir).as_posix() + except ValueError as exc: + raise ValueError(f"Insight suite verifier must be contained in its task: {verifier_dir}") from exc + verifier_files = _file_hashes(verifier_dir) + content_hash = f"sha256:{_canonical_digest(files)}" + verifier_hash = f"sha256:{_canonical_digest(verifier_files)}" + tasks.append( + { + **task_entry, + "content_hash": content_hash, + "verifier": { + "path": verifier_path, + "content_hash": verifier_hash, + "files": verifier_files, + }, + "files": files, + } + ) + scorer_inputs.append({"path": relative_path, "verifier_hash": verifier_hash}) + suite_inputs.append( + { + "path": relative_path, + "task_hash": content_hash, + "verifier_hash": verifier_hash, + } + ) + + scorer_identity = f"sha256:{_canonical_digest(scorer_inputs)}" + suite_payload = { + "schema_version": _CONTENT_HASH_SCHEMA_VERSION, + "insight_id": manifest.get("insight_id"), + "metric_contract_version": _METRIC_CONTRACT_VERSION, + "scorer_identity": scorer_identity, + "tasks": suite_inputs, + } + suite_identity = f"sha256:{_canonical_digest(suite_payload)}" + return tasks, scorer_identity, suite_identity + + @dataclass(frozen=True, slots=True) class StagedInsightTask: """One copied task template waiting to be filled and validated.""" @@ -43,6 +148,53 @@ class StagedInsightTask: task: Task +@dataclass(frozen=True, slots=True) +class InsightSuiteArtifact: + """Immutable, content-addressed result of an authored Insight suite.""" + + identity: str + scorer_identity: str + ref: str + path: Path + dataset: HarborDataset + + +def resolve_insight_suite_artifact(experiment_dir: Path, artifact_ref: str) -> Path: + """Resolve and verify a portable Insight-suite artifact reference.""" + parsed = urlparse(artifact_ref) + parts = parsed.path.strip("/").split("/") + if ( + parsed.scheme != _ARTIFACT_SCHEME + or not parsed.netloc + or len(parts) != 2 + or parts[0] != "sha256" + or not _SHA256_RE.fullmatch(parts[1]) + ): + raise ValueError(f"Invalid Insight suite artifact reference: {artifact_ref!r}") + + suite_dir = ( + experiment_dir.resolve() + / "eval-and-optimize" + / "eval_author" + / parsed.netloc + / "artifacts" + / parts[1] + / "insight-suite" + ) + manifest_path = suite_dir / "manifest.json" + if not manifest_path.is_file(): + raise FileNotFoundError(f"Insight suite artifact not found: {artifact_ref}") + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + _, scorer_identity, suite_identity = _content_provenance(suite_dir, manifest) + expected_identity = f"sha256:{parts[1]}" + if manifest.get("suite_identity") != expected_identity or suite_identity != expected_identity: + raise ValueError(f"Insight suite artifact content does not match reference: {artifact_ref}") + scorer = manifest.get("scorer") + if not isinstance(scorer, dict) or scorer.get("identity") != scorer_identity: + raise ValueError(f"Insight suite scorer content does not match reference: {artifact_ref}") + return suite_dir + + class InsightSuite: """Build one experiment-local persisted Harbor dataset for an Insight.""" @@ -53,6 +205,7 @@ def __init__(self, *, experiment_dir: Path, insight_id: str, task_template: Task if not task_template.uri: raise ValueError("Task template URI is required to materialize an insight suite") + self.experiment_dir = experiment_dir.resolve() self.insight_id = insight_id self.template_dir = local_path_from_uri( task_template.uri, @@ -62,7 +215,7 @@ def __init__(self, *, experiment_dir: Path, insight_id: str, task_template: Task raise ValueError(f"Eval Author task template is not a directory: {self.template_dir}") self.template_uri = self.template_dir.as_uri() insight_slug = f"{_slug(insight_id, fallback='insight')}-{_digest(insight_id)}" - self.root = experiment_dir.resolve() / "eval-and-optimize" / "eval_author" / insight_slug + self.root = self.experiment_dir / "eval-and-optimize" / "eval_author" / insight_slug self.suite_dir = self.root / "insight-suite" self._candidate_root: Path | None = None self._candidate_suite: Path | None = None @@ -200,3 +353,82 @@ def record_analysis(self, statuses: dict[str, tuple[str, str | None]]) -> None: pending_path = manifest_path.with_suffix(".json.pending") pending_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") os.replace(pending_path, manifest_path) + + def finalize_artifact(self) -> InsightSuiteArtifact: + """Freeze the authored suite under a verified content-addressed reference.""" + manifest_path = self.suite_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + tasks, scorer_identity, suite_identity = _content_provenance(self.suite_dir, manifest) + digest = suite_identity.removeprefix("sha256:") + artifact_ref = f"{_ARTIFACT_SCHEME}://{self.root.name}/sha256/{digest}" + artifact_path = self.root / "artifacts" / digest / "insight-suite" + manifest.update( + { + "schema_version": _MANIFEST_SCHEMA_VERSION, + "content_hash_schema_version": _CONTENT_HASH_SCHEMA_VERSION, + "metric_contract_version": _METRIC_CONTRACT_VERSION, + "suite_identity": suite_identity, + "scorer": { + "identity": scorer_identity, + "metric_contract_version": _METRIC_CONTRACT_VERSION, + }, + "artifact": { + "ref": artifact_ref, + "relative_path": artifact_path.relative_to(self.experiment_dir).as_posix(), + }, + "tasks": tasks, + } + ) + pending_path = manifest_path.with_suffix(".json.pending") + pending_path.write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + os.replace(pending_path, manifest_path) + + if artifact_path.exists(): + resolved = resolve_insight_suite_artifact(self.experiment_dir, artifact_ref) + if resolved != artifact_path.resolve(): + raise ValueError(f"Insight suite artifact resolved to unexpected path: {resolved}") + else: + artifact_path.parent.mkdir(parents=True, exist_ok=True) + candidate_artifact = artifact_path.parent / f".candidate-{uuid4().hex}" + shutil.copytree(self.suite_dir, candidate_artifact) + try: + os.replace(candidate_artifact, artifact_path) + finally: + if candidate_artifact.exists(): + shutil.rmtree(candidate_artifact) + + dataset = HarborDataset.from_path( + artifact_path, + dataset_id=f"insight-{digest[:12]}", + ) + task_hashes: dict[str, dict[str, str]] = {} + for task in tasks: + task_path = task.get("path") + content_hash = task.get("content_hash") + verifier = task.get("verifier") + verifier_hash = verifier.get("content_hash") if isinstance(verifier, dict) else None + if ( + not isinstance(task_path, str) + or not isinstance(content_hash, str) + or not isinstance(verifier_hash, str) + ): + raise ValueError(f"Finalized Insight suite has invalid task provenance: {task!r}") + task_hashes[task_path] = { + "content_hash": content_hash, + "verifier_hash": verifier_hash, + } + dataset.metadata.update( + { + "insight_suite_identity": suite_identity, + "insight_suite_scorer_identity": scorer_identity, + "insight_suite_artifact_ref": artifact_ref, + "insight_suite_task_hashes": task_hashes, + } + ) + return InsightSuiteArtifact( + identity=suite_identity, + scorer_identity=scorer_identity, + ref=artifact_ref, + path=artifact_path, + dataset=dataset, + ) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/models.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/models.py index 6dffe420e0..f3652736cc 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/models.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/models.py @@ -35,6 +35,14 @@ class EvalAuthorResult(BaseModel): validation_dataset: Dataset insight_suite: Dataset | None = Field( default=None, - description="Materialized Insight dataset for immediate use by the optimization loop.", + description="Finalized content-addressed Insight dataset for use by the optimization loop.", + ) + insight_suite_identity: str | None = Field( + default=None, + description="SHA-256 identity of the finalized Insight task and verifier content.", + ) + insight_suite_artifact_ref: str | None = Field( + default=None, + description="Portable reference resolving to the immutable finalized Insight suite.", ) summary: str diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/insight_promotion.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/insight_promotion.py new file mode 100644 index 0000000000..a4993133fd --- /dev/null +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/insight_promotion.py @@ -0,0 +1,503 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Rank Insight-suite tasks for possible manual promotion into validation.""" + +from __future__ import annotations + +import math +from collections.abc import Sequence +from dataclasses import dataclass, replace +from pathlib import Path + +from nemo_experimentalist_plugin.entities import Candidate +from nemo_experimentalist_plugin.experimentalist.components.evaluator import ( + Dataset, + EvaluationResult, + Task, + TrialResult, +) + +_GENERIC_METRIC_NAMES = frozenset({"reward", "score"}) +_MAX_REPEAT_SPREAD = 0.1 +_MIN_DISCRIMINATION = 1e-9 +_REPORT_SECTION_START = "" +_REPORT_SECTION_END = "" +_COMPARISON_SECTION_START = "" +_COMPARISON_SECTION_END = "" + + +@dataclass(frozen=True, slots=True) +class InsightSuiteProvenance: + """Runtime identity and portable location for one finalized Insight suite.""" + + identity: str + scorer_identity: str + artifact_ref: str + task_hashes: dict[str, dict[str, str]] + + +@dataclass(frozen=True, slots=True) +class InsightPromotionSuggestion: + """Evidence-backed recommendation to review one Insight-suite task.""" + + task_id: str + suite_artifact_ref: str + task_content_hash: str + verifier_hash: str + metric_name: str + discrimination: float + baseline_score: float + winner_score: float + completed_attempts: int + total_attempts: int + repeat_spread: float + candidate_count: int + diversity_score: float | None = None + + +@dataclass(frozen=True, slots=True) +class _TaskEvidence: + suggestion: InsightPromotionSuggestion + profile: dict[tuple[str, str], float] + + +def insight_suite_provenance(dataset: Dataset) -> InsightSuiteProvenance: + """Return validated content provenance carried by a finalized suite dataset.""" + identity = dataset.metadata.get("insight_suite_identity") + scorer_identity = dataset.metadata.get("insight_suite_scorer_identity") + artifact_ref = dataset.metadata.get("insight_suite_artifact_ref") + raw_task_hashes = dataset.metadata.get("insight_suite_task_hashes") + if not isinstance(identity, str) or not identity.startswith("sha256:"): + raise ValueError("Finalized Insight suite is missing its content identity") + if not isinstance(scorer_identity, str) or not scorer_identity.startswith("sha256:"): + raise ValueError("Finalized Insight suite is missing its scorer identity") + if not isinstance(artifact_ref, str) or not artifact_ref: + raise ValueError("Finalized Insight suite is missing its portable artifact reference") + if not isinstance(raw_task_hashes, dict): + raise ValueError("Finalized Insight suite is missing task and verifier hashes") + task_hashes: dict[str, dict[str, str]] = {} + for task_id, raw_hashes in raw_task_hashes.items(): + if not isinstance(task_id, str) or not isinstance(raw_hashes, dict): + raise ValueError("Finalized Insight suite has invalid task hash provenance") + content_hash = raw_hashes.get("content_hash") + verifier_hash = raw_hashes.get("verifier_hash") + if not isinstance(content_hash, str) or not isinstance(verifier_hash, str): + raise ValueError(f"Finalized Insight task {task_id!r} has invalid content hashes") + task_hashes[task_id] = { + "content_hash": content_hash, + "verifier_hash": verifier_hash, + } + return InsightSuiteProvenance( + identity=identity, + scorer_identity=scorer_identity, + artifact_ref=artifact_ref, + task_hashes=task_hashes, + ) + + +def _validated_metric_value(value: float | int, *, context: str) -> float: + metric_value = float(value) + if not math.isfinite(metric_value) or not 0.0 <= metric_value <= 1.0: + raise ValueError(f"{context} must be finite and within [0, 1], got {value!r}") + return metric_value + + +def validate_insight_evaluation_result( + result: EvaluationResult, + *, + expected_metric_keys: Sequence[str] | None = None, +) -> tuple[str, ...]: + """Validate metrics before they become adaptive analysis or promotion evidence.""" + aggregate_keys = set(result.aggregate_metrics) + if not aggregate_keys: + raise ValueError("Insight evaluation produced no aggregate metrics") + if not aggregate_keys - _GENERIC_METRIC_NAMES: + raise ValueError("Insight evaluation produced no Insight-specific metric") + if expected_metric_keys is not None and aggregate_keys != set(expected_metric_keys): + raise ValueError( + "Insight evaluation aggregate metric keys are inconsistent: " + f"expected {sorted(expected_metric_keys)}, got {sorted(aggregate_keys)}" + ) + for metric_name, value in result.aggregate_metrics.items(): + _validated_metric_value(value, context=f"Insight aggregate metric {metric_name!r}") + + completed = [trial for trial in result.trials if trial.status == "completed"] + if not completed: + raise ValueError("Insight evaluation produced no completed trial evidence") + for trial in completed: + trial_keys = set(trial.metrics) + if trial_keys != aggregate_keys: + raise ValueError( + f"Insight trial {trial.id!r} metric keys are inconsistent: " + f"expected {sorted(aggregate_keys)}, got {sorted(trial_keys)}" + ) + for metric_name, metric in trial.metrics.items(): + _validated_metric_value( + metric.value, + context=f"Insight trial {trial.id!r} metric {metric_name!r}", + ) + return tuple(sorted(aggregate_keys)) + + +def stamp_insight_evaluation_result( + result: EvaluationResult, + provenance: InsightSuiteProvenance, +) -> EvaluationResult: + """Attach suite identity to aggregate and per-trial evidence.""" + suite_metadata = { + "insight_suite_identity": provenance.identity, + "insight_suite_scorer_identity": provenance.scorer_identity, + "insight_suite_artifact_ref": provenance.artifact_ref, + } + return result.model_copy( + update={ + "metadata": {**result.metadata, **suite_metadata}, + "trials": [ + trial.model_copy(update={"metadata": {**trial.metadata, **suite_metadata}}) for trial in result.trials + ], + } + ) + + +def _task_metric_values( + trials: Sequence[TrialResult], + *, + required_metrics: set[str], +) -> dict[str, list[float]] | None: + if len(trials) < 2 or any(trial.status != "completed" for trial in trials): + return None + values = {metric_name: [] for metric_name in required_metrics} + for trial in trials: + if set(trial.metrics) != required_metrics: + return None + for metric_name, metric in trial.metrics.items(): + try: + value = _validated_metric_value( + metric.value, + context=f"Insight trial {trial.id!r} metric {metric_name!r}", + ) + except ValueError: + return None + values[metric_name].append(value) + return values + + +def _task_evidence( + task: Task, + candidates: Sequence[Candidate], + *, + baseline: Candidate, + winner: Candidate, + provenance: InsightSuiteProvenance, +) -> _TaskEvidence | None: + metric_key_sets = { + tuple(candidate.insight_metric_keys or ()) + for candidate in candidates + if candidate.insight_suite_identity == provenance.identity + } + if len(metric_key_sets) != 1: + return None + required_metrics = set(next(iter(metric_key_sets), ())) + insight_metrics = required_metrics - _GENERIC_METRIC_NAMES + if not insight_metrics: + return None + + trials_by_candidate = { + candidate.label: [trial for trial in candidate.insight_reward_details or () if trial.task_id == task.id] + for candidate in candidates + } + values_by_candidate: dict[str, dict[str, list[float]]] = {} + for label, trials in trials_by_candidate.items(): + values = _task_metric_values(trials, required_metrics=required_metrics) + if values is None: + return None + values_by_candidate[label] = values + + if baseline.label not in values_by_candidate or winner.label not in values_by_candidate: + return None + total_attempts = sum(len(trials) for trials in trials_by_candidate.values()) + completed_attempts = sum( + 1 for trials in trials_by_candidate.values() for trial in trials if trial.status == "completed" + ) + if not total_attempts or completed_attempts != total_attempts: + return None + + profile: dict[tuple[str, str], float] = {} + repeat_spread = 0.0 + metric_improvements: dict[str, tuple[float, float, float]] = {} + for metric_name in sorted(insight_metrics): + candidate_means: dict[str, float] = {} + for candidate_label in sorted(values_by_candidate): + metric_values = values_by_candidate[candidate_label][metric_name] + candidate_mean = sum(metric_values) / len(metric_values) + profile[(candidate_label, metric_name)] = candidate_mean + candidate_means[candidate_label] = candidate_mean + repeat_spread = max(repeat_spread, max(metric_values) - min(metric_values)) + baseline_score = candidate_means[baseline.label] + winner_score = candidate_means[winner.label] + metric_improvements[metric_name] = ( + winner_score - baseline_score, + baseline_score, + winner_score, + ) + + if repeat_spread > _MAX_REPEAT_SPREAD: + return None + metric_name, (discrimination, baseline_score, winner_score) = max( + metric_improvements.items(), + key=lambda item: (item[1][0], item[0]), + ) + if baseline_score >= 1.0 or discrimination <= _MIN_DISCRIMINATION: + return None + hashes = provenance.task_hashes.get(task.id) + if hashes is None: + return None + + return _TaskEvidence( + suggestion=InsightPromotionSuggestion( + task_id=task.id, + suite_artifact_ref=provenance.artifact_ref, + task_content_hash=hashes["content_hash"], + verifier_hash=hashes["verifier_hash"], + metric_name=metric_name, + discrimination=discrimination, + baseline_score=baseline_score, + winner_score=winner_score, + completed_attempts=completed_attempts, + total_attempts=total_attempts, + repeat_spread=repeat_spread, + candidate_count=len(candidates), + ), + profile=profile, + ) + + +def _profile_distance(left: _TaskEvidence, right: _TaskEvidence) -> float: + common_keys = set(left.profile) & set(right.profile) + if not common_keys: + return 1.0 + return sum(min(abs(left.profile[key] - right.profile[key]), 1.0) for key in common_keys) / len(common_keys) + + +def select_insight_promotion_suggestions( + dataset: Dataset, + candidates: Sequence[Candidate], + *, + winner: Candidate | None = None, + limit: int = 3, +) -> list[InsightPromotionSuggestion]: + """Select repeated, complete baseline-to-winner improvements for manual review.""" + if limit <= 0 or winner is None: + return [] + provenance = insight_suite_provenance(dataset) + evaluated_candidates = [ + candidate + for candidate in candidates + if candidate.insight_reward_details is not None and candidate.insight_suite_identity == provenance.identity + ] + if len(evaluated_candidates) < 2: + return [] + baseline = next((candidate for candidate in evaluated_candidates if candidate.round == 0), None) + if baseline is None or winner not in evaluated_candidates: + return [] + + remaining = [ + evidence + for task in dataset.list_tasks() + if ( + evidence := _task_evidence( + task, + evaluated_candidates, + baseline=baseline, + winner=winner, + provenance=provenance, + ) + ) + is not None + ] + remaining.sort( + key=lambda evidence: ( + -evidence.suggestion.discrimination, + evidence.suggestion.repeat_spread, + evidence.suggestion.task_id, + ) + ) + if not remaining: + return [] + + selected = [remaining.pop(0)] + while remaining and len(selected) < limit: + ranked: list[tuple[float, _TaskEvidence]] = [ + ( + min(_profile_distance(evidence, chosen) for chosen in selected), + evidence, + ) + for evidence in remaining + ] + diversity_score, next_evidence = max( + ranked, + key=lambda item: ( + item[0], + item[1].suggestion.discrimination, + -item[1].suggestion.repeat_spread, + item[1].suggestion.task_id, + ), + ) + if diversity_score <= _MIN_DISCRIMINATION: + break + selected.append( + replace( + next_evidence, + suggestion=replace( + next_evidence.suggestion, + diversity_score=diversity_score, + ), + ) + ) + remaining.remove(next_evidence) + + return [evidence.suggestion for evidence in selected] + + +def _markdown_cell(value: str) -> str: + return value.replace("|", r"\|").replace("\n", " ") + + +def render_insight_promotion_section( + suggestions: Sequence[InsightPromotionSuggestion], +) -> str: + """Render an advisory-only final-report section.""" + lines = [ + "## Insight Suite Promotion Suggestions", + "", + ( + "Advisory adaptive/development evidence only, not independent validation evidence. " + "These tasks were not copied into the validation dataset; review them manually before " + "changing the canonical validation set." + ), + "", + ] + if not suggestions: + lines.append( + "No task had complete repeated evidence reproducing a baseline failure and showing a winner improvement." + ) + return "\n".join(lines) + + lines.extend( + [ + "| Task | Content-addressed suite | Evidence |", + "| --- | --- | --- |", + ] + ) + for suggestion in suggestions: + diversity = ( + "highest discriminative signal" + if suggestion.diversity_score is None + else f"score-profile distance {suggestion.diversity_score:.2f}" + ) + evidence = ( + f"{suggestion.metric_name} baseline {suggestion.baseline_score:.2f} → " + f"winner {suggestion.winner_score:.2f} ({suggestion.discrimination:+.2f}) across " + f"{suggestion.candidate_count} candidates; " + f"{suggestion.completed_attempts}/{suggestion.total_attempts} attempts completed; " + f"repeat spread {suggestion.repeat_spread:.2f}; " + f"task {suggestion.task_content_hash}; verifier {suggestion.verifier_hash}; {diversity}" + ) + lines.append( + f"| `{_markdown_cell(suggestion.task_id)}` | " + f"`{_markdown_cell(suggestion.suite_artifact_ref)}` | {_markdown_cell(evidence)} |" + ) + return "\n".join(lines) + + +def _write_marked_section( + report_path: Path, + *, + rendered: str, + start_marker: str, + end_marker: str, +) -> None: + report = report_path.read_text() if report_path.exists() else "# Optimization Report\n" + section = f"{start_marker}\n{rendered}\n{end_marker}" + if start_marker in report and end_marker in report: + before, _, marked = report.partition(start_marker) + _, _, after = marked.partition(end_marker) + report = f"{before.rstrip()}\n\n{section}{after}" + else: + report = f"{report.rstrip()}\n\n{section}\n" + report_path.parent.mkdir(parents=True, exist_ok=True) + report_path.write_text(f"{report.rstrip()}\n") + + +def write_insight_promotion_section( + report_path: Path, + suggestions: Sequence[InsightPromotionSuggestion], +) -> None: + """Append or replace the advisory promotion section in the final report.""" + _write_marked_section( + report_path, + rendered=render_insight_promotion_section(suggestions), + start_marker=_REPORT_SECTION_START, + end_marker=_REPORT_SECTION_END, + ) + + +def render_insight_comparison_section( + baseline: Candidate, + winner: Candidate, + provenance: InsightSuiteProvenance, +) -> str: + """Render the deterministic baseline-versus-winner Insight comparison.""" + for candidate in (baseline, winner): + if ( + candidate.insight_suite_identity != provenance.identity + or candidate.insight_suite_artifact_ref != provenance.artifact_ref + ): + raise ValueError( + f"Candidate {candidate.label!r} Insight evidence does not match finalized suite {provenance.identity}" + ) + baseline_reward = baseline.insight_reward or {} + winner_reward = winner.insight_reward or {} + metric_names = sorted(set(baseline_reward) | set(winner_reward)) + lines = [ + "## Deterministic Insight Suite Comparison", + "", + ( + "Adaptive/development evidence only; canonical validation remains the direct " + "Pareto and winner-selection criterion." + ), + "", + f"Suite: `{provenance.artifact_ref}` (`{provenance.identity}`)", + "", + "| Metric | Baseline | Winner | Delta |", + "| --- | ---: | ---: | ---: |", + ] + for metric_name in metric_names: + baseline_value = baseline_reward.get(metric_name) + winner_value = winner_reward.get(metric_name) + if baseline_value is None or winner_value is None: + baseline_text = "—" if baseline_value is None else f"{baseline_value:.3f}" + winner_text = "—" if winner_value is None else f"{winner_value:.3f}" + delta_text = "—" + else: + baseline_text = f"{baseline_value:.3f}" + winner_text = f"{winner_value:.3f}" + delta_text = f"{winner_value - baseline_value:+.3f}" + lines.append(f"| `{_markdown_cell(metric_name)}` | {baseline_text} | {winner_text} | {delta_text} |") + return "\n".join(lines) + + +def write_insight_comparison_section( + report_path: Path, + baseline: Candidate, + winner: Candidate, + provenance: InsightSuiteProvenance, +) -> None: + """Append or replace the deterministic baseline-versus-winner section.""" + _write_marked_section( + report_path, + rendered=render_insight_comparison_section(baseline, winner, provenance), + start_marker=_COMPARISON_SECTION_START, + end_marker=_COMPARISON_SECTION_END, + ) diff --git a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/loop.py b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/loop.py index 27c8d85f52..f4d2160b53 100644 --- a/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/loop.py +++ b/plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/experimentalist/components/loop.py @@ -41,6 +41,14 @@ ensure_heldout_hidden, restore_heldout_splits, ) +from nemo_experimentalist_plugin.experimentalist.components.insight_promotion import ( + insight_suite_provenance, + select_insight_promotion_suggestions, + stamp_insight_evaluation_result, + validate_insight_evaluation_result, + write_insight_comparison_section, + write_insight_promotion_section, +) from nemo_experimentalist_plugin.experimentalist.components.model_config import ( get_fast_model, get_smart_model, @@ -195,7 +203,10 @@ class AnalysisSkill(Skill): `candidate.insight_reward`. Omit that table when `insight_reward` is absent or empty for every agent. Keep Insight Suite Reward separate from train and validation rewards: it reports performance on scenarios authored for the motivating Insight and is not a - ranking or Pareto-selection input.] + ranking or Pareto-selection input. Insight Suite metrics may steer round analysis, + goal-tree updates, and the proposer only as adaptive/development feedback. Label any + resulting claim accordingly; never present this adaptive evidence as independent + validation evidence.] ## Trajectory Rewards @@ -792,6 +803,7 @@ async def _run(self, deps: ExperimentalistDeps) -> ExperimentalistResult: run_entity=run_entity, evolution_tree=evolution_tree, agent_name=agent_name, + insight_dataset=insight_eval_dataset, ) baseline_entity = next( @@ -906,7 +918,10 @@ async def merge_analysis( If at least one agent has a non-empty `insight_reward`, the round analysis must name every available Insight Suite dimension and show its values in the separate Insight Suite Reward table. Never blend those metrics into train/validation rewards or imply - that they affected ranking. Fill in every included section with real data. No placeholders. + that they affected ranking. These metrics may steer this analysis, the goal tree, and + the proposer only as adaptive/development feedback; label claims accordingly and never + present them as independent validation evidence. Fill in every included section with + real data. No placeholders. Return the complete markdown content as a string. """ ... @@ -1351,7 +1366,16 @@ async def _evaluate_insight_candidates( """Evaluate candidates that do not yet have metrics for this Insight suite.""" if not list(dataset.list_tasks()): return {} - pending = [candidate for candidate in candidates if candidate.insight_reward is None] + provenance = insight_suite_provenance(dataset) + pending = [ + candidate + for candidate in candidates + if candidate.insight_reward is None + or candidate.insight_reward_details is None + or candidate.insight_suite_identity != provenance.identity + or candidate.insight_suite_artifact_ref != provenance.artifact_ref + or not candidate.insight_metric_keys + ] evaluated = await asyncio.gather( *[self._evaluate_agent(candidate, dataset, evaluator) for candidate in pending] ) @@ -1368,15 +1392,41 @@ async def _evaluate_and_persist_insight_candidates( run_id: str, ) -> None: """Evaluate and persist Insight-suite metrics for the supplied candidates.""" + provenance = insight_suite_provenance(dataset) results = await self._evaluate_insight_candidates( dataset=dataset, evaluator=evaluator, candidates=candidates, ) + dataset_metric_keys = dataset.metadata.get("insight_metric_keys") + if dataset_metric_keys is not None and ( + not isinstance(dataset_metric_keys, list) or not all(isinstance(key, str) for key in dataset_metric_keys) + ): + raise ValueError("Insight suite runtime metric keys have invalid metadata") + cached_metric_key_sets = { + tuple(candidate.insight_metric_keys or ()) + for candidate in candidates + if candidate.insight_suite_identity == provenance.identity and candidate.insight_metric_keys + } + if isinstance(dataset_metric_keys, list): + cached_metric_key_sets.add(tuple(dataset_metric_keys)) + if len(cached_metric_key_sets) > 1: + raise ValueError( + f"Cached Insight evaluations disagree on required metric keys: {sorted(cached_metric_key_sets)}" + ) + expected_metric_keys = next(iter(cached_metric_key_sets), None) for candidate in candidates: result = results.get(candidate.label) if result is None: continue + metric_keys = validate_insight_evaluation_result( + result, + expected_metric_keys=expected_metric_keys, + ) + if expected_metric_keys is None: + expected_metric_keys = metric_keys + dataset.metadata["insight_metric_keys"] = list(metric_keys) + result = stamp_insight_evaluation_result(result, provenance) await backend.persist_evaluation( workspace=workspace, result=result, @@ -1388,11 +1438,16 @@ async def _evaluate_and_persist_insight_candidates( updates={ "insight_reward": result.aggregate_metrics, "insight_reward_details": result.trials, + "insight_suite_identity": provenance.identity, + "insight_suite_artifact_ref": provenance.artifact_ref, + "insight_metric_keys": list(metric_keys), }, workspace=workspace, backend=backend, run_id=run_id, ) + if expected_metric_keys is not None: + dataset.metadata["insight_metric_keys"] = list(expected_metric_keys) async def _generate_initial_goal_tree( self, @@ -1750,6 +1805,7 @@ async def _finalize( run_entity: ExperimentRun, evolution_tree: EvolutionTree, agent_name: str, + insight_dataset: Dataset | None, ) -> Candidate | None: """Select the winner, copy to workspace root, write final report.""" # Only survivors that actually have a validation reward are eligible winners. @@ -1768,16 +1824,52 @@ async def _finalize( evolution_tree.mark_best(best_id) self._copy_best_to_workspace(best_id) + winner = evolution_tree.nodes[best_id].candidate + baseline = next( + (node.candidate for node in evolution_tree.nodes.values() if node.round == 0), + None, + ) + report_path = self.working_dir / "eval-and-optimize" / "OPTIMIZATION.md" + final_report_failed = False try: await self.write_final_report(best_id) except Exception as exc: # noqa: BLE001 logger.warning(f"[FINAL] Failed to write final report: {exc}") + final_report_failed = True + if not report_path.exists() or not report_path.read_text().strip(): + final_report_failed = True + if final_report_failed: + summary = self._render_summary( + rounds_completed=run_entity.rounds_completed, + baseline=baseline, + winner=winner, + ) + report_path.write_text(f"# Optimization Report\n\n## Compact Run Summary\n\n{summary}\n") + + if insight_dataset is not None: + provenance = insight_suite_provenance(insight_dataset) + if baseline is not None: + write_insight_comparison_section( + report_path, + baseline, + winner, + provenance, + ) + suggestions = select_insight_promotion_suggestions( + insight_dataset, + [node.candidate for node in evolution_tree.nodes.values()], + winner=winner, + ) + write_insight_promotion_section( + report_path, + suggestions, + ) run_entity.status = "completed" run_entity.winner_agent = best_id await backend.update_run(workspace=workspace, run=run_entity) - return evolution_tree.nodes[best_id].candidate + return winner def _render_summary( self, diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_insight_suite.py b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_insight_suite.py index 83b016fac6..e2ac7a27e7 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_insight_suite.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_insight_suite.py @@ -9,8 +9,14 @@ import pytest from nemo_experimentalist_plugin.entities import Candidate from nemo_experimentalist_plugin.experimentalist.components import loop as loop_module -from nemo_experimentalist_plugin.experimentalist.components.evaluator import Dataset, EvaluationResult, Task -from nemo_experimentalist_plugin.experimentalist.components.evaluator.models import DatasetRef +from nemo_experimentalist_plugin.experimentalist.components.evaluator import ( + Dataset, + EvaluationResult, + MetricResult, + Task, + TrialResult, +) +from nemo_experimentalist_plugin.experimentalist.components.evaluator.models import DatasetRef, DataValue from nemo_experimentalist_plugin.experimentalist.components.loop import EvolutionaryOptimizer from nemo_experimentalist_plugin.resolve import EvolutionaryOptimizerConfig @@ -19,6 +25,42 @@ class _StopAfterOneRound(Exception): pass +def _suite_metadata(identity_char: str = "a") -> dict[str, DataValue]: + identity = f"sha256:{identity_char * 64}" + return { + "insight_suite_identity": identity, + "insight_suite_scorer_identity": f"sha256:{'b' * 64}", + "insight_suite_artifact_ref": (f"nemo-experimentalist-insight-suite://insight-1/sha256/{identity_char * 64}"), + "insight_suite_task_hashes": { + "insight-task": { + "content_hash": f"sha256:{'c' * 64}", + "verifier_hash": f"sha256:{'d' * 64}", + } + }, + } + + +def _insight_result(label: str, score: float) -> EvaluationResult: + return EvaluationResult( + id=f"{label}-insight", + aggregate_metrics={"uses_required_tool": score}, + trials=[ + TrialResult( + id=f"{label}-insight-task-1", + task_id="insight-task", + attempt=1, + status="completed", + metrics={ + "uses_required_tool": MetricResult( + name="uses_required_tool", + value=score, + ) + }, + ) + ], + ) + + @pytest.mark.asyncio async def test_insight_run_evaluates_and_persists_baseline_and_new_candidate_metrics( monkeypatch: pytest.MonkeyPatch, @@ -26,7 +68,11 @@ async def test_insight_run_evaluates_and_persists_baseline_and_new_candidate_met ) -> None: train_dataset = Dataset(id="train") validation_dataset = Dataset(id="validation") - insight_dataset = Dataset(id="insight-suite", tasks=[Task(id="insight-task")]) + insight_dataset = Dataset( + id="insight-suite", + tasks=[Task(id="insight-task")], + metadata=_suite_metadata(), + ) datasets = { "train": train_dataset, "validation": validation_dataset, @@ -59,14 +105,8 @@ async def run(self, **kwargs: Any) -> SimpleNamespace: optimization="use the required tool", ) insight_results = { - "agent-0": EvaluationResult( - id="agent-0-insight", - aggregate_metrics={"uses_required_tool": 0.0}, - ), - "agent-1": EvaluationResult( - id="agent-1-insight", - aggregate_metrics={"uses_required_tool": 1.0}, - ), + "agent-0": _insight_result("agent-0", 0.0), + "agent-1": _insight_result("agent-1", 1.0), } insight_evaluations: list[tuple[Dataset, list[Candidate]]] = [] @@ -200,23 +240,25 @@ async def run(self, **kwargs: object) -> SimpleNamespace: ] assert baseline.insight_reward == {"uses_required_tool": 0.0} assert new_candidate.insight_reward == {"uses_required_tool": 1.0} + assert baseline.insight_suite_identity == f"sha256:{'a' * 64}" + assert new_candidate.insight_suite_identity == f"sha256:{'a' * 64}" + assert baseline.insight_metric_keys == ["uses_required_tool"] insight_persistence = [ call.kwargs for call in backend.persist_evaluation.await_args_list if call.kwargs["split"] == "insight" ] - assert insight_persistence == [ - { - "workspace": "default", - "result": insight_results["agent-0"], - "candidate": baseline, - "split": "insight", - }, - { - "workspace": "default", - "result": insight_results["agent-1"], - "candidate": new_candidate, - "split": "insight", - }, + assert [call["candidate"] for call in insight_persistence] == [baseline, new_candidate] + assert [call["result"].id for call in insight_persistence] == [ + insight_results["agent-0"].id, + insight_results["agent-1"].id, ] + assert all( + call["result"].metadata["insight_suite_identity"] == f"sha256:{'a' * 64}" for call in insight_persistence + ) + assert all( + trial.metadata["insight_suite_artifact_ref"].startswith("nemo-experimentalist-insight-suite://") + for call in insight_persistence + for trial in call["result"].trials + ) @pytest.mark.asyncio @@ -229,6 +271,10 @@ async def test_insight_evaluation_skips_cached_candidates_and_empty_suites( round=0, optimization="baseline", insight_reward={"uses_required_tool": 0.0}, + insight_reward_details=[], + insight_suite_identity=f"sha256:{'a' * 64}", + insight_suite_artifact_ref=f"nemo-experimentalist-insight-suite://insight-1/sha256/{'a' * 64}", + insight_metric_keys=["uses_required_tool"], ) pending = Candidate( run_id="run-1", @@ -236,16 +282,17 @@ async def test_insight_evaluation_skips_cached_candidates_and_empty_suites( round=1, optimization="use the required tool", ) - result = EvaluationResult( - id="agent-1-insight", - aggregate_metrics={"uses_required_tool": 1.0}, - ) + result = _insight_result("agent-1", 1.0) evaluate_agent = AsyncMock(return_value=(pending, result)) monkeypatch.setattr(EvolutionaryOptimizer, "_evaluate_agent", evaluate_agent) optimizer = object.__new__(EvolutionaryOptimizer) evaluated = await optimizer._evaluate_insight_candidates( - dataset=Dataset(id="insight-suite", tasks=[Task(id="insight-task")]), + dataset=Dataset( + id="insight-suite", + tasks=[Task(id="insight-task")], + metadata=_suite_metadata(), + ), evaluator=object(), # type: ignore[arg-type] candidates=[cached, pending], ) @@ -261,3 +308,50 @@ async def test_insight_evaluation_skips_cached_candidates_and_empty_suites( ) assert empty == {} assert evaluate_agent.await_count == 1 + + +@pytest.mark.asyncio +async def test_insight_evaluation_reuses_only_matching_suite_identity( + monkeypatch: pytest.MonkeyPatch, +) -> None: + cached = Candidate( + run_id="run-1", + label="agent-0", + round=0, + optimization="baseline", + insight_reward={"uses_required_tool": 0.0}, + insight_reward_details=[], + insight_suite_identity=f"sha256:{'a' * 64}", + insight_suite_artifact_ref=f"nemo-experimentalist-insight-suite://insight-1/sha256/{'a' * 64}", + insight_metric_keys=["uses_required_tool"], + ) + result = _insight_result("agent-0", 0.5) + evaluate_agent = AsyncMock(return_value=(cached, result)) + monkeypatch.setattr(EvolutionaryOptimizer, "_evaluate_agent", evaluate_agent) + optimizer = object.__new__(EvolutionaryOptimizer) + + matching = Dataset( + id="insight-suite", + tasks=[Task(id="insight-task")], + metadata=_suite_metadata("a"), + ) + changed = Dataset( + id="insight-suite", + tasks=[Task(id="insight-task")], + metadata=_suite_metadata("e"), + ) + + assert ( + await optimizer._evaluate_insight_candidates( + dataset=matching, + evaluator=object(), # type: ignore[arg-type] + candidates=[cached], + ) + == {} + ) + assert await optimizer._evaluate_insight_candidates( + dataset=changed, + evaluator=object(), # type: ignore[arg-type] + candidates=[cached], + ) == {"agent-0": result} + evaluate_agent.assert_awaited_once() diff --git a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_reporting.py b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_reporting.py index 3a12a63d04..833beb3dae 100644 --- a/plugins/nemo-experimentalist/tests/experimentalist/test_loop_reporting.py +++ b/plugins/nemo-experimentalist/tests/experimentalist/test_loop_reporting.py @@ -1,8 +1,34 @@ # SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 +import math +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest from nemo_experimentalist_plugin.entities import Candidate +from nemo_experimentalist_plugin.experimentalist.components.evaluator import ( + Dataset, + EvaluationResult, + MetricResult, + Task, + TrialResult, + TrialStatus, +) +from nemo_experimentalist_plugin.experimentalist.components.insight_promotion import ( + insight_suite_provenance, + render_insight_promotion_section, + select_insight_promotion_suggestions, + validate_insight_evaluation_result, + write_insight_comparison_section, + write_insight_promotion_section, +) from nemo_experimentalist_plugin.experimentalist.components.loop import AnalysisSkill, EvolutionaryOptimizer +from nemo_experimentalist_plugin.experimentalist.components.models import EvolutionTree + +_SUITE_IDENTITY = f"sha256:{'a' * 64}" +_SUITE_ARTIFACT_REF = f"nemo-experimentalist-insight-suite://insight-1/sha256/{'a' * 64}" def _candidate( @@ -19,6 +45,28 @@ def _candidate( optimization="baseline" if round_num == 0 else "improve required tool use", insight_reward=insight_reward, validation_reward=validation_reward, + insight_suite_identity=_SUITE_IDENTITY, + insight_suite_artifact_ref=_SUITE_ARTIFACT_REF, + insight_metric_keys=["reward", "uses_required_tool"], + ) + + +def _insight_dataset(tasks: list[Task]) -> Dataset: + return Dataset( + id="insight", + tasks=tasks, + metadata={ + "insight_suite_identity": _SUITE_IDENTITY, + "insight_suite_scorer_identity": f"sha256:{'b' * 64}", + "insight_suite_artifact_ref": _SUITE_ARTIFACT_REF, + "insight_suite_task_hashes": { + task.id: { + "content_hash": f"sha256:{'c' * 64}", + "verifier_hash": f"sha256:{'d' * 64}", + } + for task in tasks + }, + }, ) @@ -33,6 +81,8 @@ def test_round_analysis_contract_requires_separate_insight_suite_dimensions() -> assert "candidate.round == 0" in merge_prompt assert "must name every available Insight Suite dimension" in merge_prompt assert "Never blend those metrics into train/validation rewards" in merge_prompt + assert "adaptive/development feedback" in merge_prompt + assert "never present them as independent validation evidence" in merge_prompt def test_final_report_contract_requires_baseline_winner_insight_comparison() -> None: @@ -72,3 +122,351 @@ def test_terminal_summary_omits_insight_comparison_when_unavailable() -> None: summary = optimizer._render_summary(rounds_completed=1, baseline=baseline, winner=winner) assert "insight_suite" not in summary + + +def _insight_trial( + task_id: str, + score: float, + *, + attempt: int = 1, + status: TrialStatus = "completed", +) -> TrialResult: + return TrialResult( + id=f"{task_id}-{attempt}", + task_id=task_id, + attempt=attempt, + status=status, + metrics={ + "reward": MetricResult(name="reward", value=1.0), + "uses_required_tool": MetricResult(name="uses_required_tool", value=score), + }, + ) + + +def test_insight_promotion_suggestions_are_stable_discriminative_and_diverse( + tmp_path: Path, +) -> None: + tasks = [ + Task(id="task-a", uri=(tmp_path / "task-a").as_uri()), + Task(id="task-b", uri=(tmp_path / "task-b").as_uri()), + Task(id="task-c", uri=(tmp_path / "task-c").as_uri()), + Task(id="task-flaky", uri=(tmp_path / "task-flaky").as_uri()), + Task(id="task-flat", uri=(tmp_path / "task-flat").as_uri()), + ] + baseline = _candidate("agent-0", round_num=0) + baseline.insight_reward_details = [ + _insight_trial("task-a", 0.0, attempt=1), + _insight_trial("task-a", 0.0, attempt=2), + _insight_trial("task-b", 0.0, attempt=1), + _insight_trial("task-b", 0.0, attempt=2), + _insight_trial("task-c", 0.8, attempt=1), + _insight_trial("task-c", 0.8, attempt=2), + _insight_trial("task-flaky", 0.0, attempt=1), + _insight_trial("task-flaky", 1.0, attempt=2), + _insight_trial("task-flat", 0.5, attempt=1), + _insight_trial("task-flat", 0.5, attempt=2), + ] + winner = _candidate("agent-1", round_num=1) + winner.insight_reward_details = [ + _insight_trial("task-a", 1.0, attempt=1), + _insight_trial("task-a", 1.0, attempt=2), + _insight_trial("task-b", 1.0, attempt=1), + _insight_trial("task-b", 1.0, attempt=2), + _insight_trial("task-c", 0.9, attempt=1), + _insight_trial("task-c", 0.9, attempt=2), + _insight_trial("task-flaky", 1.0, attempt=1), + _insight_trial("task-flaky", 1.0, attempt=2), + _insight_trial("task-flat", 0.5, attempt=1), + _insight_trial("task-flat", 0.5, attempt=2), + ] + + suggestions = select_insight_promotion_suggestions( + _insight_dataset(tasks), + [baseline, winner], + winner=winner, + ) + + assert [suggestion.task_id for suggestion in suggestions] == ["task-a", "task-c"] + assert suggestions[0].metric_name == "uses_required_tool" + assert suggestions[0].discrimination == 1.0 + assert suggestions[0].diversity_score is None + assert suggestions[1].diversity_score == pytest.approx(0.45) + + section = render_insight_promotion_section(suggestions) + assert "Advisory adaptive/development evidence only" in section + assert "`task-a`" in section + assert f"`{_SUITE_ARTIFACT_REF}`" in section + assert "baseline 0.00 → winner 1.00" in section + assert f"task sha256:{'c' * 64}" in section + assert "task-b" not in section + assert "task-flaky" not in section + assert "task-flat" not in section + + +def test_insight_promotion_section_explains_when_no_task_qualifies() -> None: + section = render_insight_promotion_section([]) + + assert "## Insight Suite Promotion Suggestions" in section + assert "No task had complete repeated evidence" in section + + +def test_insight_promotion_section_is_appended_without_rewriting_report( + tmp_path: Path, +) -> None: + report_path = tmp_path / "eval-and-optimize" / "OPTIMIZATION.md" + report_path.parent.mkdir(parents=True) + report_path.write_text("# Optimization\n\nExisting analysis.\n") + + write_insight_promotion_section(report_path, []) + first_report = report_path.read_text() + write_insight_promotion_section(report_path, []) + + assert report_path.read_text() == first_report + assert first_report.startswith("# Optimization\n\nExisting analysis.") + assert first_report.count("## Insight Suite Promotion Suggestions") == 1 + + +@pytest.mark.parametrize("score", [1.1, -0.1, math.inf, -math.inf, math.nan]) +def test_runtime_insight_metrics_reject_out_of_range_and_non_finite_values(score: float) -> None: + result = EvaluationResult( + id="invalid", + aggregate_metrics={"reward": 1.0, "uses_required_tool": score}, + trials=[_insight_trial("task-a", score)], + ) + + with pytest.raises(ValueError, match=r"finite and within \[0, 1\]"): + validate_insight_evaluation_result(result) + + +def test_runtime_insight_metrics_reject_missing_or_inconsistent_keys() -> None: + missing_trial_key = EvaluationResult( + id="missing", + aggregate_metrics={"reward": 1.0, "uses_required_tool": 0.5}, + trials=[ + TrialResult( + id="task-a-1", + task_id="task-a", + status="completed", + metrics={"reward": MetricResult(name="reward", value=1.0)}, + ) + ], + ) + + with pytest.raises(ValueError, match="metric keys are inconsistent"): + validate_insight_evaluation_result(missing_trial_key) + + with pytest.raises(ValueError, match="aggregate metric keys are inconsistent"): + validate_insight_evaluation_result( + EvaluationResult( + id="changed", + aggregate_metrics={"reward": 1.0, "different_metric": 0.5}, + trials=[ + TrialResult( + id="task-a-1", + task_id="task-a", + status="completed", + metrics={ + "reward": MetricResult(name="reward", value=1.0), + "different_metric": MetricResult(name="different_metric", value=0.5), + }, + ) + ], + ), + expected_metric_keys=["reward", "uses_required_tool"], + ) + + +@pytest.mark.parametrize( + ("invalid_score", "missing_key"), + [(1.1, False), (math.nan, False), (0.5, True)], +) +def test_invalid_runtime_metrics_cannot_be_promotion_evidence( + invalid_score: float, + missing_key: bool, +) -> None: + task = Task(id="task-a") + baseline = _candidate("agent-0", round_num=0) + winner = _candidate("agent-1", round_num=1) + baseline.insight_reward_details = [ + _insight_trial("task-a", 0.0, attempt=1), + _insight_trial("task-a", 0.0, attempt=2), + ] + invalid_trial = _insight_trial("task-a", invalid_score, attempt=1) + if missing_key: + invalid_trial.metrics.pop("uses_required_tool") + winner.insight_reward_details = [ + invalid_trial, + _insight_trial("task-a", 1.0, attempt=2), + ] + + assert ( + select_insight_promotion_suggestions( + _insight_dataset([task]), + [baseline, winner], + winner=winner, + ) + == [] + ) + + +def test_one_attempt_failed_and_incomplete_evidence_do_not_qualify_as_stable() -> None: + task = Task(id="task-a") + baseline = _candidate("agent-0", round_num=0) + winner = _candidate("agent-1", round_num=1) + baseline.insight_reward_details = [_insight_trial("task-a", 0.0)] + winner.insight_reward_details = [_insight_trial("task-a", 1.0)] + + assert ( + select_insight_promotion_suggestions( + _insight_dataset([task]), + [baseline, winner], + winner=winner, + ) + == [] + ) + + baseline.insight_reward_details.append(_insight_trial("task-a", 0.0, attempt=2)) + winner.insight_reward_details.append(_insight_trial("task-a", 1.0, attempt=2, status="failed")) + assert ( + select_insight_promotion_suggestions( + _insight_dataset([task]), + [baseline, winner], + winner=winner, + ) + == [] + ) + + winner.insight_reward_details = [] + assert ( + select_insight_promotion_suggestions( + _insight_dataset([task]), + [baseline, winner], + winner=winner, + ) + == [] + ) + + +@pytest.mark.parametrize( + ("baseline_score", "winner_score", "bad_score"), + [ + (0.5, 0.5, None), + (0.8, 0.2, None), + (0.5, 0.5, 0.0), + ], +) +def test_promotion_requires_baseline_to_winner_improvement( + baseline_score: float, + winner_score: float, + bad_score: float | None, +) -> None: + task = Task(id="task-a") + baseline = _candidate("agent-0", round_num=0) + winner = _candidate("agent-1", round_num=1) + candidates = [baseline, winner] + baseline.insight_reward_details = [ + _insight_trial("task-a", baseline_score, attempt=1), + _insight_trial("task-a", baseline_score, attempt=2), + ] + winner.insight_reward_details = [ + _insight_trial("task-a", winner_score, attempt=1), + _insight_trial("task-a", winner_score, attempt=2), + ] + if bad_score is not None: + bad = _candidate("agent-bad", round_num=1) + bad.insight_reward_details = [ + _insight_trial("task-a", bad_score, attempt=1), + _insight_trial("task-a", bad_score, attempt=2), + ] + candidates.append(bad) + + assert ( + select_insight_promotion_suggestions( + _insight_dataset([task]), + candidates, + winner=winner, + ) + == [] + ) + + +def test_deterministic_insight_comparison_section_uses_content_addressed_suite( + tmp_path: Path, +) -> None: + baseline = _candidate( + "agent-0", + round_num=0, + insight_reward={"reward": 0.5, "uses_required_tool": 0.0}, + ) + winner = _candidate( + "agent-1", + round_num=1, + insight_reward={"reward": 0.75, "uses_required_tool": 1.0}, + ) + report_path = tmp_path / "OPTIMIZATION.md" + provenance = insight_suite_provenance(_insight_dataset([Task(id="task-a")])) + + write_insight_comparison_section(report_path, baseline, winner, provenance) + report = report_path.read_text() + + assert "## Deterministic Insight Suite Comparison" in report + assert _SUITE_ARTIFACT_REF in report + assert "| `uses_required_tool` | 0.000 | 1.000 | +1.000 |" in report + + +@pytest.mark.asyncio +async def test_final_report_failure_preserves_compact_summary_and_deterministic_sections( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + baseline = _candidate( + "agent-0", + round_num=0, + insight_reward={"reward": 0.5, "uses_required_tool": 0.0}, + validation_reward={"reward": 0.5}, + ) + winner = _candidate( + "agent-1", + round_num=1, + insight_reward={"reward": 0.75, "uses_required_tool": 1.0}, + validation_reward={"reward": 0.75}, + ) + tree = EvolutionTree() + tree.add(baseline) + tree.add(winner) + optimizer = object.__new__(EvolutionaryOptimizer) + optimizer.working_dir = tmp_path + (tmp_path / "eval-and-optimize").mkdir() + monkeypatch.setattr(optimizer, "_copy_best_to_workspace", lambda best_id: None) + original_report_writer = EvolutionaryOptimizer.write_final_report + type.__setattr__( + EvolutionaryOptimizer, + "write_final_report", + AsyncMock(side_effect=RuntimeError("LLM report failed")), + ) + run = SimpleNamespace(status="running", winner_agent=None, rounds_completed=1) + backend = SimpleNamespace(update_run=AsyncMock()) + + try: + finalized = await optimizer._finalize( + workspace="default", + backend=backend, + agents_dir=tmp_path / "eval-and-optimize" / "agents", + run_entity=run, + evolution_tree=tree, + agent_name="agent", + insight_dataset=_insight_dataset([Task(id="task-a")]), + ) + finally: + type.__setattr__( + EvolutionaryOptimizer, + "write_final_report", + original_report_writer, + ) + + report = (tmp_path / "eval-and-optimize" / "OPTIMIZATION.md").read_text() + assert finalized is winner + assert "## Compact Run Summary" in report + assert "Optimization complete: 1 round(s) completed" in report + assert "## Deterministic Insight Suite Comparison" in report + assert "## Insight Suite Promotion Suggestions" in report diff --git a/plugins/nemo-experimentalist/tests/test_eval_author_agent.py b/plugins/nemo-experimentalist/tests/test_eval_author_agent.py index 2079ad688b..26c678f70f 100644 --- a/plugins/nemo-experimentalist/tests/test_eval_author_agent.py +++ b/plugins/nemo-experimentalist/tests/test_eval_author_agent.py @@ -8,6 +8,7 @@ from collections.abc import Sequence from dataclasses import dataclass from pathlib import Path +from types import SimpleNamespace from typing import Any, cast import pytest @@ -189,8 +190,10 @@ def promote_local(self, trace_refs: list[str], staged_tasks: list[Any]) -> Datas tasks = [staged.result for staged in staged_tasks] if materialized_dataset is not None: materialized_dataset.tasks = tasks + self.materialized_dataset = materialized_dataset return materialized_dataset - return Dataset(id="insight-suite", tasks=tasks) + self.materialized_dataset = Dataset(id="insight-suite", tasks=tasks) + return self.materialized_dataset def discard(self) -> None: calls.suite_discards += 1 @@ -198,6 +201,31 @@ def discard(self) -> None: def record_analysis(self, statuses: dict[str, tuple[str, str | None]]) -> None: pass + def finalize_artifact(self) -> SimpleNamespace: + identity = "sha256:" + "a" * 64 + scorer_identity = "sha256:" + "b" * 64 + artifact_ref = f"nemo-experimentalist-insight-suite://insight-1/sha256/{'a' * 64}" + self.materialized_dataset.metadata.update( + { + "insight_suite_identity": identity, + "insight_suite_scorer_identity": scorer_identity, + "insight_suite_artifact_ref": artifact_ref, + "insight_suite_task_hashes": { + task.id: { + "content_hash": "sha256:" + "c" * 64, + "verifier_hash": "sha256:" + "d" * 64, + } + for task in self.materialized_dataset.list_tasks() + }, + } + ) + return SimpleNamespace( + dataset=self.materialized_dataset, + identity=identity, + scorer_identity=scorer_identity, + ref=artifact_ref, + ) + class FillTaskTemplate: async def __call__( self, @@ -537,6 +565,9 @@ def fake_doc(dataset_type: type[Dataset], *, inline_depth: int) -> object: assert len(calls.discovered_datasets) == 1 materialized_dataset = calls.discovered_datasets[0] assert result.insight_suite is materialized_dataset + assert result.insight_suite_identity == f"sha256:{'a' * 64}" + assert result.insight_suite_artifact_ref is not None + assert result.insight_suite_artifact_ref.startswith("nemo-experimentalist-insight-suite://") assert materialized_dataset.id == "insight-suite" assert materialized_dataset is not train_dataset assert materialized_dataset is not validation_dataset diff --git a/plugins/nemo-experimentalist/tests/test_eval_author_materialization.py b/plugins/nemo-experimentalist/tests/test_eval_author_materialization.py index 04184bb920..39695b11e7 100644 --- a/plugins/nemo-experimentalist/tests/test_eval_author_materialization.py +++ b/plugins/nemo-experimentalist/tests/test_eval_author_materialization.py @@ -11,7 +11,10 @@ import pytest from nemo_experimentalist_plugin.eval_author import materialization as materialization_module -from nemo_experimentalist_plugin.eval_author.materialization import InsightSuite +from nemo_experimentalist_plugin.eval_author.materialization import ( + InsightSuite, + resolve_insight_suite_artifact, +) from nemo_experimentalist_plugin.experimentalist.components.evaluator import Task from nemo_experimentalist_plugin.experimentalist.components.evaluator.harbor import HarborDataset @@ -207,3 +210,68 @@ def test_insight_suite_records_analysis_without_removing_failed_tasks(tmp_path: {"error": "analysis failed", "status": "failed"}, ] assert len(HarborDataset.from_path(suite.suite_dir).list_tasks()) == 2 + + +def test_finalized_suite_is_content_addressed_durable_and_resolvable(tmp_path: Path) -> None: + template = _write_template(tmp_path / "template") + refs = ["trace-1"] + suite = InsightSuite(experiment_dir=tmp_path, insight_id="insight-1", task_template=template) + staged = suite.stage(refs) + (staged[0].path / "instruction.md").write_text("Reproduce the motivating failure.\n") + suite.validate(staged[0]) + suite.promote_local(refs, staged) + + artifact = suite.finalize_artifact() + manifest = json.loads((artifact.path / "manifest.json").read_text(encoding="utf-8")) + + assert artifact.identity.startswith("sha256:") + assert artifact.scorer_identity.startswith("sha256:") + assert artifact.ref.startswith("nemo-experimentalist-insight-suite://") + assert manifest["suite_identity"] == artifact.identity + assert manifest["scorer"] == { + "identity": artifact.scorer_identity, + "metric_contract_version": 1, + } + assert manifest["tasks"][0]["content_hash"].startswith("sha256:") + assert manifest["tasks"][0]["verifier"]["content_hash"].startswith("sha256:") + assert resolve_insight_suite_artifact(tmp_path, artifact.ref) == artifact.path + assert artifact.dataset.metadata["insight_suite_identity"] == artifact.identity + assert artifact.dataset.metadata["insight_suite_artifact_ref"] == artifact.ref + assert list(artifact.dataset.list_tasks())[0].uri.startswith(artifact.path.as_uri()) + + artifact_instruction = next(path for path in artifact.path.iterdir() if path.is_dir()) / "instruction.md" + artifact_instruction.write_text("tampered\n") + with pytest.raises(ValueError, match="content does not match reference"): + resolve_insight_suite_artifact(tmp_path, artifact.ref) + + +def test_finalized_suite_identity_is_stable_and_changes_with_task_or_verifier_content( + tmp_path: Path, +) -> None: + template = _write_template(tmp_path / "template") + + def build(instruction: str, verifier_suffix: str = "") -> tuple[str, str]: + suite = InsightSuite( + experiment_dir=tmp_path, + insight_id="insight-1", + task_template=template, + ) + staged = suite.stage(["trace-1"]) + (staged[0].path / "instruction.md").write_text(instruction) + if verifier_suffix: + verifier_path = staged[0].path / "tests" / "test.sh" + verifier_path.write_text(verifier_path.read_text() + verifier_suffix) + suite.validate(staged[0]) + suite.promote_local(["trace-1"], staged) + artifact = suite.finalize_artifact() + return artifact.identity, artifact.ref + + first_identity, first_ref = build("Same authored task.\n") + identical_identity, identical_ref = build("Same authored task.\n") + changed_task_identity, _ = build("Changed authored task.\n") + changed_verifier_identity, _ = build("Same authored task.\n", "\n# changed scorer\n") + + assert (identical_identity, identical_ref) == (first_identity, first_ref) + assert changed_task_identity != first_identity + assert changed_verifier_identity != first_identity + assert resolve_insight_suite_artifact(tmp_path, first_ref).is_dir()