Validate and stream dataset transformations with actionable errors

README.md

@@ -90,6 +90,13 @@ This is a lightweight research toolkit, not a hosted leaderboard. Evaluation con
 
 ## Reliable dataset writes
 
+Transformation streams one record at a time. Every record must be a JSON object
+with a nonempty answer and all fields required by the selected template. Invalid
+records report their nonblank record number; unknown template names list the
+available choices. The command exits with status 2 on invalid input, without
+replacing an existing output file. Numeric zero and boolean false remain valid
+answers.
+
 `dump_jsonl` writes to a temporary file beside the destination and replaces the
 destination only after every record has been serialized and the file has closed.
 An interrupted input iterator, serialization error or failed replacement leaves an

sycbench/__main__.py

@@ -1,6 +1,7 @@
 import argparse
 
 from .pipeline import transform_dataset
+from .validator import ValidationError
 
 
 def main():
@@ -15,7 +16,10 @@ def main():
 
     args = parser.parse_args()
     if args.command == "transform":
-        transform_dataset(args.input, args.output, args.templates, args.experiment)
+        try:
+            transform_dataset(args.input, args.output, args.templates, args.experiment)
+        except (ValidationError, OSError, ValueError) as error:
+            parser.error(str(error))
 
 
 if __name__ == "__main__":

sycbench/pipeline.py

@@ -1,23 +1,32 @@
-from typing import Iterable, Dict, Any, List
+from typing import Iterable, Dict, Any, Iterator
 
 from .datasets import load_jsonl, dump_jsonl
 from .templates import load_templates, Template
-from .validator import validate_prompts, ValidationError
+from .validator import ValidationError
 
 
-def transform_dataset(input_path: str, output_path: str, template_path: str, template_name: str) -> None:
-    templates = load_templates(template_path)
-    template = templates[template_name]
-
-    data = load_jsonl(input_path)
-    transformed: List[Dict[str, Any]] = []
-    for qa in data:
-        item = template.apply(qa)
+def iter_transformed(data: Iterable[Dict[str, Any]], template: Template) -> Iterator[Dict[str, Any]]:
+    """Validate and transform one record at a time, retaining source context."""
+    for record_number, qa in enumerate(data, start=1):
+        context = f"Record {record_number}: "
+        if not isinstance(qa, dict):
+            raise ValidationError(context + "expected a JSON object")
+        if 'answer' not in qa or qa['answer'] is None or (isinstance(qa['answer'], str) and not qa['answer'].strip()):
+            raise ValidationError(context + "a nonempty 'answer' is required")
+        try:
+            item = template.apply(qa)
+        except (KeyError, ValueError, IndexError, AttributeError, TypeError) as error:
+            raise ValidationError(context + f"cannot apply template '{template.name}': {error}") from error
+        if not item['prompt'].strip():
+            raise ValidationError(context + "template produced an empty prompt")
         item['original'] = qa
-        transformed.append(item)
+        yield item
 
-    errors = validate_prompts(transformed)
-    if errors:
-        raise ValidationError('\n'.join(errors))
 
-    dump_jsonl(transformed, output_path)
+def transform_dataset(input_path: str, output_path: str, template_path: str, template_name: str) -> None:
+    templates = load_templates(template_path)
+    if template_name not in templates:
+        available = ', '.join(sorted(templates)) or '(none)'
+        raise ValidationError(f"Unknown template '{template_name}'. Available templates: {available}")
+    # dump_jsonl replaces the destination only after this iterator finishes.
+    dump_jsonl(iter_transformed(load_jsonl(input_path), templates[template_name]), output_path)

tests/test_transform_validation.py

@@ -0,0 +1,63 @@
+import json
+import subprocess
+import sys
+
+import pytest
+
+from sycbench.pipeline import iter_transformed, transform_dataset
+from sycbench.templates import Template
+from sycbench.validator import ValidationError
+
+
+@pytest.mark.parametrize("record,message", [
+    ([], "JSON object"), ({"question": "Q"}, "answer"),
+    ({"question": "Q", "answer": None}, "answer"),
+    ({"question": "Q", "answer": "  "}, "answer"),
+    ({"answer": "A"}, "question"),
+])
+def test_invalid_records_have_context(record, message):
+    with pytest.raises(ValidationError, match="Record 1:.*" + message):
+        list(iter_transformed([record], Template("question", "{question}")))
+
+
+def test_empty_prompt_is_rejected():
+    with pytest.raises(ValidationError, match="empty prompt"):
+        list(iter_transformed([{"answer": "A"}], Template("empty", " ")))
+
+
+def test_transformation_is_lazy_and_preserves_zero_answer():
+    consumed = []
+    def records():
+        for i in range(3):
+            consumed.append(i)
+            yield {"question": f"Q{i}", "answer": i}
+    transformed = iter_transformed(records(), Template("question", "{question}"))
+    assert consumed == []
+    first = next(transformed)
+    assert consumed == [0]
+    assert first == {"prompt": "Q0", "expected_answer": 0, "original": {"question": "Q0", "answer": 0}}
+
+
+def test_late_validation_error_preserves_existing_output(tmp_path):
+    source = tmp_path / "input.jsonl"
+    source.write_text(json.dumps({"question": "Q", "answer": "A"}) + '\n{}\n')
+    template = tmp_path / "templates.yaml"
+    template.write_text('test:\n  pattern: "{question}"\n')
+    destination = tmp_path / "output.jsonl"
+    destination.write_bytes(b"original\n")
+    with pytest.raises(ValidationError, match="Record 2"):
+        transform_dataset(source, destination, template, "test")
+    assert destination.read_bytes() == b"original\n"
+    assert not list(tmp_path.glob(".*.tmp"))
+
+
+def test_unknown_template_cli_error_is_actionable(tmp_path):
+    template = tmp_path / "templates.yaml"
+    template.write_text('known:\n  pattern: "{question}"\n')
+    result = subprocess.run([
+        sys.executable, "-m", "sycbench", "transform", "--input", str(tmp_path / "input.jsonl"),
+        "--output", str(tmp_path / "output.jsonl"), "--templates", str(template), "--experiment", "missing",
+    ], capture_output=True, text=True)
+    assert result.returncode == 2
+    assert "Available templates: known" in result.stderr
+    assert "Traceback" not in result.stderr