cowork: schema command error-path hardening + honest sample footer

src/datamorph/cli.py

@@ -118,7 +118,7 @@ def batch_cmd(
     if csv_delimiter != ",":
         writer_kwargs["delimiter"] = csv_delimiter
 
-    results = convert_batch(
+    batch = convert_batch(
         input_dir,
         output_dir,
         from_format,
@@ -127,11 +127,25 @@ def batch_cmd(
         recursive=recursive,
         **writer_kwargs,
     )
+    results = batch.results
 
     success = [r for r in results if not r.errors]
     failed = [r for r in results if r.errors]
 
     console.print("\n[bold]Batch Conversion Complete[/bold]")
+
+    # Files that matched the pattern but were skipped due to format mismatch
+    # are surfaced explicitly instead of vanishing silently.
+    if batch.skipped:
+        console.print(
+            f"  Skipped (format mismatch): {len(batch.skipped)}"
+            " - did not match --from format"
+        )
+        for item in batch.skipped:
+            err_console.print(
+                f"  [yellow]SKIPPED[/yellow] {item['file']} "
+                f"(detected: {item['detected_format']}, expected: {from_format})"
+            )
     console.print(f"  Files: {len(success)} converted, {len(failed)} failed")
 
     if failed:
@@ -176,8 +190,18 @@ def schema_cmd(
         err_console.print(f"[red]Could not detect format for: {file}[/red]")
         sys.exit(1)
 
-    reader = get_reader(fmt)
-    schema = reader.infer_schema(file, sample_size=sample)
+    try:
+        reader = get_reader(fmt)
+    except ValueError as e:
+        err_console.print(f"[red]ERROR:[/red] {e}")
+        sys.exit(1)
+    try:
+        schema = reader.infer_schema(file, sample_size=sample)
+    except Exception as e:
+        err_console.print(
+            f"[red]ERROR:[/red] Could not infer schema from {file}: {e}"
+        )
+        sys.exit(1)
 
     if json_output:
         console.print(json.dumps(schema, indent=2))
@@ -192,7 +216,7 @@ def schema_cmd(
 
     console.print(f"\nDetected format: [bold]{fmt}[/bold]")
     console.print(table)
-    console.print(f"[dim]Inferred from {sample}+ rows[/dim]")
+    console.print(f"[dim]Inferred from a sample of up to {sample} rows[/dim]")
 
 
 # ── formats ──────────────────────────────────────────────────────────
@@ -265,8 +289,28 @@ def validate_cmd(
     # Load expected schema if provided
     expected_schema = None
     if schema_file:
-        with open(schema_file, "r", encoding="utf-8") as f:
-            expected_schema = json.load(f)
+        try:
+            with open(schema_file, "r", encoding="utf-8") as f:
+                expected_schema = json.load(f)
+        except (OSError, json.JSONDecodeError) as e:
+            err_console.print(
+                f"[red]ERROR:[/red] Could not load schema file {schema_file}: {e}"
+            )
+            sys.exit(1)
+        if (
+            not isinstance(expected_schema, list)
+            or not expected_schema
+            or not all(
+                isinstance(f_, dict) and "name" in f_ and "type" in f_
+                for f_ in expected_schema
+            )
+        ):
+            err_console.print(
+                "[red]ERROR:[/red] Schema file must be a non-empty JSON list of "
+                'objects with "name" and "type" keys '
+                '(generate one with: datamorph schema data.csv --json-output)'
+            )
+            sys.exit(1)
 
     result = validate(
         file,

src/datamorph/converters.py

@@ -399,6 +399,12 @@ def write_stream(self, rows: RowStream, path: str | Path) -> int:
 
         rows_list = list(rows)
         if not rows_list:
+            # Zero-in is legitimate, but the output file must still exist:
+            # write a valid empty Avro container (record with no fields)
+            # instead of silently producing no artifact.
+            empty_schema = {"type": "record", "name": "Record", "fields": []}
+            with open(path, "wb") as f:
+                fastavro.writer(f, empty_schema, [])
             return 0
 
         # Infer schema across all rows for proper type detection
@@ -594,6 +600,21 @@ def _counting(stream: RowStream) -> RowStream:
     return result
 
 
+@dataclass
+class BatchConversionResult:
+    """Outcome of a batch conversion, including files that were skipped.
+
+    ``skipped`` records every file that matched ``pattern`` but was NOT
+    converted because its detected format differed from ``input_format``
+    (or the format could not be detected at all). Surfacing these prevents
+    the classic silent failure where a directory conversion quietly drops
+    mislabeled or foreign-format files while reporting success.
+    """
+
+    results: list[ConversionResult] = field(default_factory=list)
+    skipped: list[dict[str, str]] = field(default_factory=list)
+
+
 def convert_batch(
     input_dir: str | Path,
     output_dir: str | Path,
@@ -602,19 +623,26 @@ def convert_batch(
     pattern: str = "*",
     recursive: bool = False,
     **writer_kwargs: Any,
-) -> list[ConversionResult]:
+) -> BatchConversionResult:
     """Convert all matching files in a directory."""
     input_dir = Path(input_dir)
     output_dir = Path(output_dir)
     output_dir.mkdir(parents=True, exist_ok=True)
 
     glob_pattern = f"**/{pattern}" if recursive else pattern
-    results: list[ConversionResult] = []
+    batch = BatchConversionResult()
 
     for input_path in sorted(input_dir.glob(glob_pattern)):
         if input_path.is_dir():
             continue
-        if detect_format(str(input_path)) != input_format:
+        detected = detect_format(str(input_path))
+        if detected != input_format:
+            batch.skipped.append(
+                {
+                    "file": str(input_path),
+                    "detected_format": detected or "unknown",
+                }
+            )
             continue
 
         # Preserve relative path structure
@@ -631,9 +659,9 @@ def convert_batch(
             output_format,
             **writer_kwargs,
         )
-        results.append(result)
+        batch.results.append(result)
 
-    return results
+    return batch
 
 
 def _format_to_extension(fmt: str) -> str:

tests/test_cli_error_paths.py

@@ -31,3 +31,35 @@ def test_convert_nonexistent_file(self):
         """convert subcommand with nonexistent file shows error."""
         result = runner.invoke(cli, ["convert", "/nonexistent/file.json"])
         assert result.exit_code != 0
+
+
+class TestSchemaCmdErrorPaths:
+    """Tests for the `schema` subcommand error paths (silent-failure class)."""
+
+    def test_schema_unsupported_format_exits_cleanly(self, tmp_path):
+        """--format with an unsupported name errors instead of traceback."""
+        f = tmp_path / "data.txt"
+        f.write_text("hello")
+        result = runner.invoke(cli, ["schema", str(f), "--format", "nope"])
+        assert result.exit_code == 1
+        assert "Unsupported format" in result.output
+        assert "Traceback" not in result.output
+
+    def test_schema_malformed_json_exits_cleanly(self, tmp_path):
+        """Malformed input yields a clean error, not an unhandled traceback."""
+        f = tmp_path / "broken.json"
+        f.write_text("{not valid json!!!")
+        result = runner.invoke(cli, ["schema", str(f)])
+        assert result.exit_code == 1
+        assert "Could not infer schema" in result.output
+        assert "Traceback" not in result.output
+
+    def test_schema_sample_message_is_honest(self, tmp_path):
+        """Footer reports the sample cap, not '{sample}+ rows'."""
+        import json as _json
+
+        f = tmp_path / "rows.json"
+        f.write_text(_json.dumps([{"a": 1}, {"a": 2}]))
+        result = runner.invoke(cli, ["schema", str(f), "--sample", "100"])
+        assert result.exit_code == 0
+        assert "up to 100 rows" in result.output

tests/test_converters.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from pathlib import Path
 
 import pytest
 import yaml
@@ -819,9 +820,10 @@ def test_batch_single_file(self, sample_csv, tmp_path):
             "json",
             pattern="test.csv",
         )
-        assert len(results) >= 1
-        assert not results[0].errors
-        assert results[0].rows_written == 3
+        assert len(results.results) >= 1
+        assert results.skipped == []
+        assert not results.results[0].errors
+        assert results.results[0].rows_written == 3
         assert (output_dir / "test.json").exists()
 
     def test_batch_no_matches(self, tmp_path):
@@ -835,7 +837,27 @@ def test_batch_no_matches(self, tmp_path):
             "json",
             pattern="*.csv",
         )
-        assert results == []
+        assert results.results == []
+        assert results.skipped == []
+
+
+class TestBatchSkippedReporting:
+    def test_format_mismatch_is_reported_not_silent(self, tmp_path):
+        """A file matching the pattern but with a different detected format is
+        recorded in ``skipped`` instead of vanishing without a trace."""
+        input_dir = tmp_path / "in"
+        input_dir.mkdir()
+        (input_dir / "data.csv").write_text("a,b\n1,2\n", encoding="utf-8")
+        # Mislabeled: .csv extension but JSONL content -> detect_format sees jsonl? No:
+        # detection is extension-based, so use a foreign extension instead.
+        (input_dir / "notes.txt").write_text("hello", encoding="utf-8")
+        output_dir = tmp_path / "out"
+        results = convert_batch(str(input_dir), str(output_dir), "csv", "json")
+        names = [Path(item["file"]).name for item in results.skipped]
+        assert names == ["notes.txt"]
+        assert results.skipped[0]["detected_format"] in ("txt", "unknown")
+        # The real csv was still converted.
+        assert [r.rows_written for r in results.results] == [1]
 
 
 # ── Type inference ────────────────────────────────────────────────────
@@ -993,3 +1015,4 @@ def test_scalar_bool_root_round_trip(self, tmp_path):
         assert result.rows_read == 1
         assert result.rows_written == 1
         assert json.loads(out.read_text(encoding="utf-8")) == [{"data": True}]
+

tests/test_cowork_hardening.py

@@ -0,0 +1,57 @@
+"""Regression tests: zero-row Avro output artifact + CLI schema-file hardening."""
+from __future__ import annotations
+
+import json
+
+from click.testing import CliRunner
+
+from datamorph.cli import cli
+from datamorph.converters import convert
+
+
+def _csv(tmp_path, rows="name,age\nalice,30\n"):
+    p = tmp_path / "in.csv"
+    p.write_text(rows, encoding="utf-8")
+    return str(p)
+
+
+def test_zero_row_avro_output_file_exists_and_is_valid(tmp_path):
+    src = _csv(tmp_path, rows="name,age\n")  # header only -> zero data rows
+    out = tmp_path / "out.avro"
+    result = convert(src, out)
+    assert not result.errors
+    assert result.rows_written == 0
+    assert out.exists(), "empty conversion must still create the output file"
+    import fastavro
+
+    with open(out, "rb") as f:
+        rows = list(fastavro.reader(f))
+    assert rows == []
+
+
+def test_validate_cmd_bad_json_schema_file_clean_exit(tmp_path):
+    data = _csv(tmp_path)
+    bad = tmp_path / "schema.json"
+    bad.write_text("{not valid json", encoding="utf-8")
+    r = CliRunner().invoke(cli, ["validate", data, "--schema", str(bad)])
+    assert r.exit_code == 1
+    assert "Could not load schema file" in r.output
+
+
+def test_validate_cmd_wrong_shape_schema_file_clean_exit(tmp_path):
+    data = _csv(tmp_path)
+    bad = tmp_path / "schema.json"
+    bad.write_text(json.dumps({"name": "x"}), encoding="utf-8")
+    r = CliRunner().invoke(cli, ["validate", data, "--schema", str(bad)])
+    assert r.exit_code == 1
+    assert "non-empty JSON list" in r.output
+
+
+def test_validate_cmd_good_schema_file_still_works(tmp_path):
+    data = _csv(tmp_path)
+    schema = [{"name": "name", "type": "string"}, {"name": "age", "type": "string"}]
+    good = tmp_path / "schema.json"
+    good.write_text(json.dumps(schema), encoding="utf-8")
+    r = CliRunner().invoke(cli, ["validate", data, "--schema", str(good)])
+    assert r.exit_code == 0
+    assert "VALID" in r.output