cli: add hflow curate --dry-run and fix manifest: None rendering

src/hflow/cli.py

@@ -50,12 +50,18 @@ def _build_parser() -> argparse.ArgumentParser:
         default="./data/catalog",
         help="catalog directory or object-store prefix (default: ./data/catalog)",
     )
-    curate_parser.add_argument(
+    curate_output_group = curate_parser.add_mutually_exclusive_group()
+    curate_output_group.add_argument(
         "--output",
         "-o",
         default="./data/manifest.parquet",
         help="manifest path or object-store URL (default: ./data/manifest.parquet)",
     )
+    curate_output_group.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="run the query and report row count/coverage without writing a manifest",
+    )
 
     stale_parser = subparsers.add_parser(
         "stale",
@@ -477,7 +483,8 @@ def main(argv: list[str] | None = None) -> int:
             print("curate: pass exactly one of a SQL string or --sql-file", file=sys.stderr)
             return 2
         sql = arguments.sql if arguments.sql is not None else arguments.sql_file.read_text()
-        report = curate(arguments.catalog, sql, output=arguments.output)
+        output = None if arguments.dry_run else arguments.output
+        report = curate(arguments.catalog, sql, output=output)
         print(report.summary())
         return 0
     if arguments.command == "stale":

src/hflow/curation.py

@@ -74,8 +74,11 @@ class CurationReport:
     coverage: list[CheckCoverage]
 
     def summary(self) -> str:
+        manifest_text = (
+            "(not written; dry run)" if self.manifest_path is None else str(self.manifest_path)
+        )
         lines = [
-            f"manifest: {self.manifest_path} ({self.row_count} rows, "
+            f"manifest: {manifest_text} ({self.row_count} rows, "
             f"from {self.total_episodes} cataloged episodes)"
         ]
         lines.append("coverage (episodes each check ran on):")

tests/test_catalog_curation.py

@@ -274,6 +274,32 @@ def test_cli_curate_requires_exactly_one_sql_source(tmp_path: Path) -> None:
     assert cli_main(["curate", "--catalog", str(tmp_path)]) == 2
 
 
+def test_cli_curate_dry_run_does_not_write_manifest(
+    recorded_data_root: Path, tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    manifest = tmp_path / "should-not-exist.parquet"
+    exit_code = cli_main(
+        [
+            "curate",
+            "SELECT episode_id, task FROM episodes WHERE status != 'quarantined'",
+            "--catalog",
+            str(recorded_data_root / "catalog"),
+            "--dry-run",
+        ]
+    )
+    assert exit_code == 0
+    assert not manifest.is_file()
+    printed = capsys.readouterr().out
+    assert "(not written; dry run)" in printed
+    assert "None" not in printed
+    assert "1 rows" in printed
+
+
+def test_cli_curate_dry_run_rejects_output_flag(tmp_path: Path) -> None:
+    with pytest.raises(SystemExit):
+        cli_main(["curate", "SELECT 1", "--dry-run", "--output", "x.parquet"])
+
+
 def test_stale_episodes_lists_only_episodes_behind_the_current_versions(tmp_path: Path) -> None:
     catalog = Catalog(tmp_path / "catalog")
     stale_stamps = FAKE_STAMPS