From a3b542dba81ce4ddf5a53791bd8e1a87e016ea3a Mon Sep 17 00:00:00 2001 From: Yiorgis Gozadinos Date: Mon, 22 Jun 2026 11:31:01 +0300 Subject: [PATCH] Add run-batch dry-run manifest output --- haiku_rag_slim/haiku/rag/ingester/cli.py | 53 +++++++++++- tests/ingester/test_cli.py | 102 +++++++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) diff --git a/haiku_rag_slim/haiku/rag/ingester/cli.py b/haiku_rag_slim/haiku/rag/ingester/cli.py index 82f758f1..7ecb1941 100644 --- a/haiku_rag_slim/haiku/rag/ingester/cli.py +++ b/haiku_rag_slim/haiku/rag/ingester/cli.py @@ -1,8 +1,10 @@ import asyncio import sys +from datetime import UTC, datetime from pathlib import Path import typer +import yaml from dotenv import find_dotenv, load_dotenv from sqlalchemy import make_url @@ -17,6 +19,7 @@ from haiku.rag.config import ( # noqa: E402 set_config, ) from haiku.rag.ingester.app import IngesterApp # noqa: E402 +from haiku.rag.ingester.batch import BatchManifest # noqa: E402 from haiku.rag.ingester.queue.migrations import open_queue # noqa: E402 from haiku.rag.logging import configure_cli_logging # noqa: E402 from haiku.rag.store.exceptions import ( # noqa: E402 @@ -137,6 +140,16 @@ def _resolve_db_path(config: AppConfig, override: Path | None) -> Path: return override or (config.storage.data_dir / "haiku.rag.lancedb") +def _default_manifest_path() -> Path: + datestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%SZ") + return Path(f"manifest-{datestamp}.yaml") + + +def _write_manifest(manifest: BatchManifest, path: Path) -> None: + data = manifest.model_dump(mode="json") + path.write_text(yaml.safe_dump(data, sort_keys=False), encoding="utf-8") + + @_cli.command("serve") def serve( db: Path | None = typer.Option( @@ -188,17 +201,53 @@ def run_batch( "--db", help="LanceDB path (overrides config.storage.data_dir).", ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Discover planned changes and write a YAML manifest without ingesting.", + ), + output: Path | None = typer.Option( + None, + "--output", + "-o", + help="Dry-run manifest path (defaults to manifest-.yaml).", + ), ) -> None: """Run one discover sweep across every configured source, drain the queue, then exit. New and changed resources are ingested, resources that vanished from a source are deleted. Exits non-zero if any job dead-letters or a source's sweep does not complete.""" - asyncio.run(_run_batch(get_config(), db)) + asyncio.run(_run_batch(get_config(), db, dry_run=dry_run, output=output)) -async def _run_batch(app_config: AppConfig, db_path: Path | None) -> None: +async def _run_batch( + app_config: AppConfig, + db_path: Path | None, + *, + dry_run: bool = False, + output: Path | None = None, +) -> None: db = _resolve_db_path(app_config, db_path) app = IngesterApp(config=app_config, db_path=db) + if dry_run: + report = await app.run_batch_dry_run() + if report.failed_sweeps: + typer.echo( + f"Sources that failed to sweep: {', '.join(report.failed_sweeps)}" + ) + raise typer.Exit(1) + manifest_path = output or _default_manifest_path() + _write_manifest(report.manifest, manifest_path) + upserts = sum(source.upsert_count for source in report.manifest.sources) + deletes = sum(source.delete_count for source in report.manifest.sources) + unchanged = sum(source.unchanged_count for source in report.manifest.sources) + typer.echo( + "Dry run complete: " + f"{upserts} upsert, {deletes} delete, {unchanged} unchanged " + f"-> {manifest_path}" + ) + return + report = await app.run_batch() typer.echo(f"Batch complete: {report.succeeded} succeeded, {report.dead} dead") if report.failed_sweeps: diff --git a/tests/ingester/test_cli.py b/tests/ingester/test_cli.py index 1e69cdad..2b5a0903 100644 --- a/tests/ingester/test_cli.py +++ b/tests/ingester/test_cli.py @@ -1,13 +1,22 @@ """haiku-ingester CLI: exercises every subcommand via CliRunner with IngesterApp / open_queue patched out so no real ingestion runs.""" +from datetime import UTC, datetime from unittest.mock import AsyncMock, MagicMock import pytest +import yaml from typer.testing import CliRunner from haiku.rag.ingester.app import BatchReport +from haiku.rag.ingester.batch import ( + BatchChange, + BatchDryRunReport, + BatchManifest, + BatchSourceSummary, +) from haiku.rag.ingester.cli import _cli as cli +from haiku.rag.ingester.queue.models import JobOp runner = CliRunner() @@ -22,6 +31,43 @@ def _fake_app(report: BatchReport, monkeypatch) -> AsyncMock: return fake +def _manifest() -> BatchManifest: + now = datetime(2026, 6, 22, 10, 30, tzinfo=UTC) + return BatchManifest( + generated_at=now, + sources=[ + BatchSourceSummary( + source_id="docs", + upsert_count=1, + delete_count=1, + unchanged_count=2, + ) + ], + changes=[ + BatchChange( + op=JobOp.UPSERT, + source_id="docs", + uri="file:///a.md", + revision="r1", + discovered_at=now, + ), + BatchChange( + op=JobOp.DELETE, + source_id="docs", + uri="file:///gone.md", + discovered_at=now, + ), + ], + ) + + +def _fake_dry_run_app(report: BatchDryRunReport, monkeypatch) -> AsyncMock: + fake = AsyncMock() + fake.run_batch_dry_run.return_value = report + monkeypatch.setattr("haiku.rag.ingester.cli.IngesterApp", lambda **_: fake) + return fake + + def test_run_batch_reports_and_exits_zero(monkeypatch): fake = _fake_app(BatchReport(succeeded=3, dead=0), monkeypatch) @@ -50,6 +96,62 @@ def test_run_batch_exits_nonzero_when_sweep_fails(monkeypatch): assert "failed to sweep: docs" in result.output +def test_run_batch_dry_run_writes_default_manifest(monkeypatch, tmp_path): + fake = _fake_dry_run_app( + BatchDryRunReport(manifest=_manifest()), + monkeypatch, + ) + + with runner.isolated_filesystem(temp_dir=tmp_path): + result = runner.invoke(cli, ["run-batch", "--dry-run", "--db", "x.lancedb"]) + + assert result.exit_code == 0 + assert "Dry run complete: 1 upsert, 1 delete, 2 unchanged -> manifest-" in ( + result.output + ) + written = list(tmp_path.glob("*/manifest-*.yaml")) + assert len(written) == 1 + data = yaml.safe_load(written[0].read_text()) + + assert data["version"] == 1 + assert data["sources"][0]["source_id"] == "docs" + assert [change["op"] for change in data["changes"]] == ["upsert", "delete"] + fake.run_batch_dry_run.assert_awaited_once() + fake.run_batch.assert_not_awaited() + + +def test_run_batch_dry_run_writes_explicit_output(monkeypatch, tmp_path): + output = tmp_path / "custom.yaml" + _fake_dry_run_app(BatchDryRunReport(manifest=_manifest()), monkeypatch) + + result = runner.invoke( + cli, + ["run-batch", "--dry-run", "--output", str(output), "--db", "x.lancedb"], + ) + + assert result.exit_code == 0 + assert f"-> {output}" in result.output + data = yaml.safe_load(output.read_text()) + assert data["changes"][0]["uri"] == "file:///a.md" + + +def test_run_batch_dry_run_exits_nonzero_when_sweep_fails(monkeypatch, tmp_path): + output = tmp_path / "failed.yaml" + _fake_dry_run_app( + BatchDryRunReport(manifest=_manifest(), failed_sweeps=["docs"]), + monkeypatch, + ) + + result = runner.invoke( + cli, + ["run-batch", "--dry-run", "--output", str(output), "--db", "x.lancedb"], + ) + + assert result.exit_code == 1 + assert "failed to sweep: docs" in result.output + assert not output.exists() + + # --- serve ---