feat(data): task_data validation gating: auto mode, misplaced-field check, repo drift test (#2827 cherry-pick into r0.6.0)

nemo_gym/task_data.py

@@ -13,10 +13,13 @@
 the row-format migration. Framework-owned keys (see ``RESERVED_ROW_KEYS``) are never part of
 ``TaskData``, and neither is a ``verifier_metadata`` wrapper: rows that still carry one have its
 contents spliced up by ``normalize_task_fields`` before validation, so one flat schema validates
-both today's rows and post-migration ``task_data`` contents. Fields that live inside
-``verifier_metadata`` on today's wire should be annotated
+both today's rows and post-migration ``task_data`` contents. Fields that today's wire reads
+EXCLUSIVELY from inside ``verifier_metadata`` should be annotated
 ``Field(..., json_schema_extra={"legacy_location": "verifier_metadata"})`` — that reverse map is
-what the row-format migration and the dispatch compatibility shim consume.
+what the row-format migration and the dispatch compatibility shim consume, and validation flags
+rows that carry such a field only top-level (the server would not see it). Servers whose wire
+accepts both placements (e.g. via a before-validator that nests top-level fields itself) must
+not carry the marker.
 
 This module is a dependency-light leaf: it may import only the standard library and Pydantic, and
 per-server ``task_data.py`` modules may import only the standard library, Pydantic, this module,
@@ -122,29 +125,31 @@ def load_task_data_schema(server_dir: Path) -> Optional[TypeAdapter]:
 
 
 LEGACY_METADATA_KEY = "verifier_metadata"
+TASK_DATA_ROW_KEY = "task_data"
 
 
 def normalize_task_fields(row: Dict[str, Any]) -> tuple[Dict[str, Any], List[str]]:
     """The task-owned subset of a dataset row, normalized to the flat end-state shape.
 
-    Drops framework keys, then splices the contents of a legacy ``verifier_metadata`` dict up to
-    the top level (the row-format migration renames that bucket into ``task_data``, so schemas are
-    written flat).
-    A key present in both places with the same value is a harmless duplicate; with different
+    Drops framework keys, then splices the contents of a legacy ``verifier_metadata`` dict and of
+    a migrated ``task_data`` dict up to the top level (schemas are written flat, so fields
+    validate the same whether a row is flat, legacy-nested, or migrated).
+    A key present in two places with the same value is a harmless duplicate; with different
     values it is ambiguous data and gets reported. Returns ``(fields, conflicts)``.
     """
     fields = {k: v for k, v in row.items() if k not in RESERVED_ROW_KEYS}
     conflicts: List[str] = []
-    legacy = fields.pop(LEGACY_METADATA_KEY, None)
-    if isinstance(legacy, dict):
-        for key, value in legacy.items():
-            if key in fields and fields[key] != value:
-                conflicts.append(key)
-                continue
-            fields[key] = value
-    elif legacy is not None:
-        # A non-dict verifier_metadata is malformed; surface it to the schema as-is.
-        fields[LEGACY_METADATA_KEY] = legacy
+    for container_key in (LEGACY_METADATA_KEY, TASK_DATA_ROW_KEY):
+        container = fields.pop(container_key, None)
+        if isinstance(container, dict):
+            for key, value in container.items():
+                if key in fields and fields[key] != value:
+                    conflicts.append(key)
+                    continue
+                fields[key] = value
+        elif container is not None:
+            # A non-dict container is malformed; surface it to the schema as-is.
+            fields[container_key] = container
     return fields, conflicts
 
 
@@ -159,12 +164,13 @@ class TaskDataValidationReport:
     errors: List[str] = field(default_factory=list)
     unknown_keys: Dict[str, int] = field(default_factory=dict)
     conflicting_keys: Dict[str, int] = field(default_factory=dict)
+    misplaced_keys: Dict[str, int] = field(default_factory=dict)
 
     MAX_RECORDED_ERRORS = 5
 
     @property
     def clean(self) -> bool:
-        return self.error_rows == 0 and not self.conflicting_keys
+        return self.error_rows == 0 and not self.conflicting_keys and not self.misplaced_keys and not self.unknown_keys
 
     def summary(self) -> str:
         parts = [
@@ -177,21 +183,59 @@ def summary(self) -> str:
         if self.conflicting_keys:
             keys = ", ".join(f"{k} ({n} rows)" for k, n in sorted(self.conflicting_keys.items()))
             parts.append(f"  keys with DIFFERENT values top-level vs verifier_metadata (ambiguous): {keys}")
+        if self.misplaced_keys:
+            keys = ", ".join(f"{k} ({n} rows)" for k, n in sorted(self.misplaced_keys.items()))
+            parts.append(
+                f"  keys this server's wire reads from verifier_metadata but found top-level "
+                f"(the server will not see them): {keys}"
+            )
         if self.unknown_keys:
             keys = ", ".join(f"{k} ({n} rows)" for k, n in sorted(self.unknown_keys.items()))
-            parts.append(f"  keys not declared by the schema (passed through unvalidated): {keys}")
+            parts.append(f"  keys not declared by the schema (typo, or missing schema field?): {keys}")
         return "\n".join(parts)
 
 
+def legacy_metadata_fields(adapter: TypeAdapter) -> frozenset:
+    """Schema fields annotated ``legacy_location: verifier_metadata`` (today's wire reads them there)."""
+    from typing import get_args
+
+    def models_of(tp, out):
+        if isinstance(tp, type) and issubclass(tp, BaseModel):
+            out.append(tp)
+            return out
+        for arg in get_args(tp):
+            models_of(arg, out)
+        return out
+
+    names = set()
+    for model in models_of(getattr(adapter, "_type", None), []):
+        for field_name, info in model.model_fields.items():
+            extra = info.json_schema_extra
+            if isinstance(extra, dict) and extra.get("legacy_location") == LEGACY_METADATA_KEY:
+                names.add(field_name)
+    return frozenset(names)
+
+
 class TaskDataValidator:
     """Validates dataset rows against a server's ``TaskData`` schema, accumulating a report."""
 
     def __init__(self, server_name: str, adapter: TypeAdapter, dataset_fpath: str):
         self._adapter = adapter
+        self._legacy_fields = legacy_metadata_fields(adapter)
         self.report = TaskDataValidationReport(server_name=server_name, dataset_fpath=dataset_fpath)
 
     def validate_row(self, row_index: int, row: Dict[str, Any]) -> None:
         self.report.rows += 1
+        # Misplacement: the schema says today's wire reads this field from inside
+        # verifier_metadata, but the row carries it only top-level. Validation would accept it
+        # (schemas are flat) while the server at runtime would never see it, so it is flagged.
+        # Rows already in the migrated format (a task_data key) are exempt: top-level inside
+        # task_data is the correct final position.
+        if self._legacy_fields and TASK_DATA_ROW_KEY not in row:
+            nested = row.get(LEGACY_METADATA_KEY)
+            nested_keys = set(nested) if isinstance(nested, dict) else set()
+            for key in (row.keys() & self._legacy_fields) - nested_keys:
+                self.report.misplaced_keys[key] = self.report.misplaced_keys.get(key, 0) + 1
         subject, conflicts = normalize_task_fields(row)
         for key in conflicts:
             self.report.conflicting_keys[key] = self.report.conflicting_keys.get(key, 0) + 1

nemo_gym/train_data_utils.py

@@ -88,16 +88,24 @@ class TrainDataProcessorConfig(BaseNeMoGymCLIConfig):
     overwrite_metrics_conflicts: bool = Field(
         default=False, description="Whether or not to overwrite metrics conflicts."
     )
-    task_data_validation: Literal["off", "warn", "error"] = Field(
-        default="warn",
+    task_data_validation: Literal["off", "warn", "error", "auto"] = Field(
+        default="auto",
         description=(
-            "Validate each dataset row against the owning resources server's task_data.py schema "
-            "during collation. 'warn' (default) prints a per-file report, 'error' fails collation "
-            "on schema violations, 'off' skips validation. Servers without a task_data.py are "
-            "always skipped."
+            "Validate each dataset row against the owning server's task_data.py schema during "
+            "collation. 'warn' prints a per-file report, 'error' fails collation on schema "
+            "violations, 'off' skips validation. The default 'auto' resolves to 'error' in "
+            "example_validation mode (the repo's PR gate, where all committed data is known "
+            "clean) and 'warn' in train_preparation mode (user datasets must not break "
+            "mid-pipeline). Servers without a task_data.py are always skipped."
         ),
     )
 
+    @property
+    def effective_task_data_validation(self) -> str:
+        if self.task_data_validation != "auto":
+            return self.task_data_validation
+        return "error" if self.mode == "example_validation" else "warn"
+
     @property
     def in_scope_dataset_types(self) -> List[DatasetType]:
         if self.mode == "train_preparation":
@@ -934,7 +942,7 @@ def collate_samples(
             paths_to_collate = self._collate_samples_single_type(
                 type=type,
                 server_instance_configs=server_instance_configs,
-                task_data_validation=config.task_data_validation,
+                task_data_validation=config.effective_task_data_validation,
             )
             collated_fpath = parent / f"{type}.jsonl"
             with open(collated_fpath, "wb") as outfile:

resources_servers/agentif/task_data.py

@@ -0,0 +1,55 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Task-data schema for the agentif server (AgentIF constraint following).
+
+Rows nest everything inside an untyped ``verifier_metadata`` bucket
+(``AgentIFRunRequest`` types it ``Optional[Dict[str, Any]]``, ``extra="allow"``); the schema is
+written flat with ``legacy_location`` annotations. Every field is Optional because the wire never
+422s on the bucket's contents. verify() reads only ``constraints`` (app.py:316) and scores each
+entry with the rule/judge checkers; the remaining fields ride along as provenance.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+_VM = {"legacy_location": "verifier_metadata"}
+
+
+class TaskData(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    constraints: Optional[List[Dict[str, Any]]] = Field(
+        default=None,
+        description=(
+            "Constraint specs scored by verify(): each has id, desc, other_info, and the "
+            "dimension/type keys the score breakdowns group by."
+        ),
+        json_schema_extra={"consumed_by": ["verify"], **_VM},
+    )
+    query_id: Optional[int] = Field(
+        default=None,
+        description="Source query identifier; provenance only.",
+        json_schema_extra={"consumed_by": ["provenance"], **_VM},
+    )
+    turn_id: Optional[int] = Field(
+        default=None,
+        description="Turn index within the source conversation; provenance only.",
+        json_schema_extra={"consumed_by": ["provenance"], **_VM},
+    )
+    domain: Optional[str] = Field(
+        default=None,
+        description="AgentIF domain the row came from, e.g. 'lawglm'; provenance only.",
+        json_schema_extra={"consumed_by": ["provenance"], **_VM},
+    )
+    agent_name: Optional[str] = Field(
+        default=None,
+        description="Source agent prompt name, e.g. 'Thought_prompt_lawglm'; provenance only.",
+        json_schema_extra={"consumed_by": ["provenance"], **_VM},
+    )
+    prompt_type: Optional[str] = Field(
+        default=None,
+        description="Source prompt family, e.g. 'Thought_prompt'; provenance only.",
+        json_schema_extra={"consumed_by": ["provenance"], **_VM},
+    )

resources_servers/instruction_following/task_data.py

@@ -6,8 +6,9 @@
 ``instruction_id_list``/``prompt``/``kwargs``/``grading_mode`` at the row top level, while the
 current format nests them under ``verifier_metadata`` (the server's ``_migrate_legacy_metadata``
 before-validator accepts both). This flat schema validates both generations unchanged, because
-core validation splices ``verifier_metadata`` contents up before checking; ``legacy_location``
-records the server's canonical nested wire placement. Required-ness mirrors
+core validation splices ``verifier_metadata`` contents up before checking. No field carries a
+``legacy_location`` marker: that marker means "the wire reads this field ONLY from inside
+verifier_metadata", and here the wire accepts both placements. Required-ness mirrors
 ``InstructionFollowingRunRequest``'s after-validator, which rejects rows missing
 ``instruction_id_list``/``prompt``/``kwargs`` from the merged metadata.
 """
@@ -29,14 +30,14 @@ class TaskData(BaseModel):
             "Registry keys of the verifiable instructions to check "
             "(e.g. 'length_constraints:nth_paragraph_first_word')."
         ),
-        json_schema_extra={"consumed_by": ["verify"], "legacy_location": "verifier_metadata"},
+        json_schema_extra={"consumed_by": ["verify"]},
     )
     prompt: str = Field(
         description=(
             "Original instruction prompt (duplicates the user turn in responses_create_params.input). "
             "Required by the wire after-validator but never read by verify()."
         ),
-        json_schema_extra={"consumed_by": ["provenance"], "legacy_location": "verifier_metadata"},
+        json_schema_extra={"consumed_by": ["provenance"]},
     )
     kwargs: List[Optional[Dict[str, Any]]] = Field(
         description=(
@@ -45,10 +46,10 @@ class TaskData(BaseModel):
             "first_word}, {N, relation}); entries may be null or {}, and None values inside a dict "
             "are filtered out by verify()."
         ),
-        json_schema_extra={"consumed_by": ["verify"], "legacy_location": "verifier_metadata"},
+        json_schema_extra={"consumed_by": ["verify"]},
     )
     grading_mode: Optional[str] = Field(
         default=None,
         description="'binary' (all instructions must pass) or 'fraction'; verify() defaults to 'binary'.",
-        json_schema_extra={"consumed_by": ["verify"], "legacy_location": "verifier_metadata"},
+        json_schema_extra={"consumed_by": ["verify"]},
     )

resources_servers/terminal_bench_2_1/task_data.py

@@ -0,0 +1,36 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Task-data schema for the terminal_bench_2_1 server.
+
+Rows carry the three sandbox coordinates top-level (no verifier_metadata):
+``TerminalBench21VerifyRequest`` (app.py:46) requires ``task_name``, ``docker_image``, and
+``task_folder`` as ``str``, so this schema does too. seed_session() starts the evaluation
+sandbox from ``docker_image`` (tagging it with ``task_name``); verify() uploads the task's
+``tests/`` (and, when config.is_verifying_golden_patch=true, ``solution/``) from
+``task_folder`` into the sandbox and runs the test script there.
+"""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TaskData(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    task_name: str = Field(
+        description=(
+            "Terminal-Bench 2.1 task id, e.g. 'terminal-bench/path-tracing'; also keys sandbox "
+            "metadata (instance_id) and is echoed in the verify response."
+        ),
+        json_schema_extra={"consumed_by": ["verify"]},
+    )
+    docker_image: str = Field(
+        description="Docker image the evaluation sandbox is started from.",
+        json_schema_extra={"consumed_by": ["verify"]},
+    )
+    task_folder: str = Field(
+        description=(
+            "Repo-relative path to the task directory (under benchmarks/terminal_bench_2_1/); "
+            "verify() uploads its tests/ (and solution/ in golden-patch mode) into the sandbox."
+        ),
+        json_schema_extra={"consumed_by": ["verify"]},
+    )

responses_api_agents/anyswe_agent/task_data.py

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Task-data schema for the self-contained anyswe_agent (no resources server).
+
+Rows are prompt-only: the committed example dataset carries nothing beyond
+``responses_create_params``. ``AnySweRunRequest`` (app.py) declares no task fields and is
+``extra="allow"``, so any additional row keys ride through the wire unread by the agent.
+"""
+
+from pydantic import BaseModel, ConfigDict
+
+
+class TaskData(BaseModel):
+    model_config = ConfigDict(extra="allow")

responses_api_agents/anyterminal_agent/task_data.py

@@ -0,0 +1,14 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Task-data schema for the self-contained anyterminal_agent (no resources server).
+
+Rows are prompt-only: the committed example dataset carries nothing beyond
+``responses_create_params``. ``AnyTerminalRunRequest`` (app.py) declares no task fields and is
+``extra="allow"``, so any additional row keys ride through the wire unread by the agent.
+"""
+
+from pydantic import BaseModel, ConfigDict
+
+
+class TaskData(BaseModel):
+    model_config = ConfigDict(extra="allow")

responses_api_agents/harbor_agent/task_data.py

@@ -0,0 +1,23 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Task-data schema for the self-contained harbor_agent (no resources server).
+
+``HarborAgentRunRequest`` (app.py) wire-requires ``instance_id``; run() splits it on ``::`` to
+pick the Harbor dataset alias and task, launches the Harbor trial, and names the output artifact
+after it. Benchmarks that bridge through Harbor (legal_agent_bench, terminal_bench_2_1 input
+sets) commit rows of exactly this shape.
+"""
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class TaskData(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    instance_id: str = Field(
+        description=(
+            "Harbor task coordinate in the form '<dataset_alias>::<task_name>'; the alias must "
+            "match a dataset configured on the agent, and the task name selects the trial."
+        ),
+        json_schema_extra={"consumed_by": ["run"]},
+    )