feat: suggest Insight Suite tasks for validation

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.",

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/<insight-slug>/artifacts/<sha256>/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://<workspace>/<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
 

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,
                 )
 

plugins/nemo-experimentalist/src/nemo_experimentalist_plugin/eval_author/materialization.py

@@ -10,17 +10,24 @@
 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
 from harbor.models.task.task import Task as HarborTask
 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,
+        )

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