Skip to content

Add blueprint workflow CLI commands #385

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
txmed82 merged 1 commit into from
Jun 4, 2026
Merged

Add blueprint workflow CLI commands #385

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 64 additions & 2 deletions src/casecrawler/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2076,7 +2076,7 @@ def export_blueprints(
blueprints = store.list_blueprints(
dataset_id=dataset_id,
cohort_plan_id=cohort_plan_id,
limit=1_000_000,
limit=None,
)
Comment on lines 2076 to 2080

@coderabbitai coderabbitai Bot Jun 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Unbounded scans still eagerly load every blueprint into memory.

DatasetStore.list_blueprints() returns a full Python list, so limit=None turns these CLI paths into all-at-once loads. On large datasets, export-blueprints, validate-blueprints, and materialize-blueprints can become OOM-prone or stall before doing useful work. If the intent is truly unbounded processing, this needs a paged/streaming store API rather than removing the cap at the call site.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/casecrawler/cli.py` around lines 2076 - 2080, The CLI is calling
DatasetStore.list_blueprints(...) with limit=None which forces an eager load of
every blueprint into memory (blueprints) and makes export-blueprints /
validate-blueprints / materialize-blueprints OOM-prone; instead either (A)
change the DatasetStore API to provide a streaming/paginated iterator (e.g.,
stream_blueprints or make list_blueprints yield an iterator) and consume that
here, or (B) replace the single call with a paginated loop that repeatedly calls
list_blueprints with a fixed batch size and a cursor/offset until no more
results, processing each batch as you go; update the CLI paths
(export-blueprints, validate-blueprints, materialize-blueprints) to use the
streaming/paginated approach rather than limit=None so blueprints are processed
in bounded batches.

if dataset_id and not blueprints:
raise click.ClickException(f"No blueprints found for dataset {dataset_id}.")
Expand Down Expand Up @@ -2113,7 +2113,7 @@ def validate_blueprints(
blueprints = store.list_blueprints(
dataset_id=dataset_id,
cohort_plan_id=cohort_plan_id,
limit=1_000_000,
limit=None,
)
if dataset_id and not blueprints:
raise click.ClickException(f"No blueprints found for dataset {dataset_id}.")
Expand All @@ -2133,6 +2133,68 @@ def validate_blueprints(
click.echo(f"Clinically plausible: {clinically_plausible}")


@cli.command("materialize-blueprints")
@click.option("--dataset-id", required=True, help="Dataset id filter")
@click.option("--cohort-plan-id", default=None, help="Cohort plan id filter")
@click.option(
"--require-release-ready",
is_flag=True,
help="Only materialize blueprints with research-release-ready validation reports.",
)
def materialize_blueprints(
dataset_id: str,
cohort_plan_id: str | None,
require_release_ready: bool,
) -> None:
"""Materialize persisted clinical blueprints into synthetic record scaffolds."""
from casecrawler.generation.blueprint_materializer import BlueprintMaterializer
from casecrawler.storage.dataset_store import DatasetStore

store = DatasetStore()
blueprints = store.list_blueprints(
dataset_id=dataset_id,
cohort_plan_id=cohort_plan_id,
limit=None,
)
if not blueprints:
raise click.ClickException(f"No blueprints found for dataset {dataset_id}.")

materializer = BlueprintMaterializer()
materialized_count = 0
for blueprint in blueprints:
validation_report = store.get_blueprint_validation_report(
blueprint.blueprint_id
)
try:
materializer.materialize(
blueprint,
validation_report=validation_report,
store=store,
require_release_ready=require_release_ready,
)
except ValueError as exc:
raise click.ClickException(str(exc)) from exc
Comment on lines +2164 to +2176

@coderabbitai coderabbitai Bot Jun 4, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Avoid partial materialization on a later release-readiness failure.

Because records are persisted inside the same loop that can still raise on a later blueprint, this command can exit non-zero after already saving a partial dataset. Preflight the selected blueprints first, then materialize once the whole batch is known to be eligible.

💡 One way to make the command fail before any writes 
     materializer = BlueprintMaterializer()
+    validation_reports = {
+        blueprint.blueprint_id: store.get_blueprint_validation_report(blueprint.blueprint_id)
+        for blueprint in blueprints
+    }
+
+    if require_release_ready:
+        for blueprint in blueprints:
+            report = validation_reports[blueprint.blueprint_id]
+            if report is None or not report.research_release_ready:
+                raise click.ClickException(
+                    "Blueprint must be research release ready before materialization."
+                )
+
     materialized_count = 0
     for blueprint in blueprints:
-        validation_report = store.get_blueprint_validation_report(
-            blueprint.blueprint_id
-        )
         try:
             materializer.materialize(
                 blueprint,
-                validation_report=validation_report,
+                validation_report=validation_reports[blueprint.blueprint_id],
                 store=store,
                 require_release_ready=require_release_ready,
             )
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/casecrawler/cli.py` around lines 2164 - 2176, The current loop calls
materializer.materialize per blueprint and may persist partial results before a
later blueprint fails release-readiness; first preflight all selected blueprints
by calling store.get_blueprint_validation_report(...) and invoking
materializer.materialize(...) in a dry-run/preflight mode (or call a
validation-only API) for each blueprint to detect
ValueError/require_release_ready failures, raising click.ClickException
immediately if any fail; only after all blueprints pass the preflight, run a
second loop that calls materializer.materialize(...) for real (the existing
call) to perform writes. Ensure you reference the existing symbols blueprint,
store.get_blueprint_validation_report, materializer.materialize, and the
require_release_ready flag when implementing the two-phase
preflight-then-materialize change.

materialized_count += 1

click.echo(f"Materialized {materialized_count} blueprint record(s)")


@cli.command("export-blueprint-release-summary")
@click.option("--dataset-id", required=True, help="Dataset id to summarize")
@click.option("--output", required=True, help="Output JSON path")
def export_blueprint_release_summary(dataset_id: str, output: str) -> None:
"""Export blueprint dataset release readiness and audit summary JSON."""
from casecrawler.export.blueprint_release import write_blueprint_release_summary
from casecrawler.storage.dataset_store import DatasetStore

store = DatasetStore()
summary = write_blueprint_release_summary(store, dataset_id, output)
click.echo(
"Wrote blueprint release summary "
f"for {summary['blueprint_count']} blueprint artifact(s) to {output}"
)


@cli.command("verify-fhir-export")
@click.argument("path", type=click.Path(exists=True, dir_okay=False))
def verify_fhir_export(path: str) -> None:
Expand Down
64 changes: 64 additions & 0 deletions tests/test_cli_synthetic.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
CohortArchetype,
CohortPlan,
GenerationRole,
JudgeReport,
)
from casecrawler.models.synthetic import Modality
from casecrawler.storage.dataset_store import DatasetStore
Expand Down Expand Up @@ -46,6 +47,7 @@ def _blueprint_cli_result(dataset_id: str) -> BlueprintPipelineResult:
"supporting_findings": ["ECG confirms AF"],
}
],
clinical_reasoning_targets=["Review renal dosing and bleeding risk."],
evidence=BlueprintEvidence(
supported_claims=["AF anticoagulation requires renal-dose review."],
citations=[{"source": "dailymed", "claim": "renal-dose review"}],
Expand Down Expand Up @@ -177,3 +179,65 @@ def test_validate_blueprints_command_persists_validation_reports(tmp_path, monke
assert report is not None
assert report.blueprint_id == "bp-1"
assert report.clinically_plausible is True


def test_blueprint_cli_materializes_and_exports_release_summary(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
store = DatasetStore()
result = _blueprint_cli_result("ds-blueprint")
store.save_cohort_plan(result.plan)
for blueprint in result.blueprints:
store.save_blueprint(blueprint)
store.save_judge_report(
JudgeReport(
report_id="judge-1",
dataset_id="ds-blueprint",
artifact_id=blueprint.blueprint_id,
role=GenerationRole.JUDGE,
score=0.93,
passed=True,
rubric="blueprint_plausibility",
)
)
runner = CliRunner()

validated = runner.invoke(
cli,
[
"validate-blueprints",
"--dataset-id",
"ds-blueprint",
],
)
materialized = runner.invoke(
cli,
[
"materialize-blueprints",
"--dataset-id",
"ds-blueprint",
"--require-release-ready",
],
)
exported = runner.invoke(
cli,
[
"export-blueprint-release-summary",
"--dataset-id",
"ds-blueprint",
"--output",
"blueprint-release-summary.json",
],
)

summary = json.loads((tmp_path / "blueprint-release-summary.json").read_text())
records = DatasetStore().list_records(dataset_id="ds-blueprint")
assert validated.exit_code == 0
assert materialized.exit_code == 0
assert "Materialized 1 blueprint record(s)" in materialized.output
assert exported.exit_code == 0
assert "Wrote blueprint release summary" in exported.output
assert len(records) == 1
assert records[0].metadata["blueprint_id"] == "bp-1"
assert summary["blueprint_count"] == 1
assert summary["research_release_ready_count"] == 1
assert summary["materialized_record_count"] == 1
Loading