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

src/datamorph/cli.py

@@ -176,8 +176,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 +202,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 +275,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

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_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