implement pipeline execution, validation, idempotency, and async API runs

- stack run <name> --target local|lake|trino|databricks --save: execute any
  of 13 registered pipelines with context-aware backend selection
- stack validate [pipeline]: run pipelines and verify output contracts
  against SQLTable models, reports mismatches with lineage context
- Pipeline registry: 13 pipelines auto-registered in aco.pipe.__init__
- Idempotent reruns: fingerprint-based change detection skips unchanged steps
- SchemaError now includes function name, input tables, contract class,
  and column count for debugging
- POST /run/{pipeline} with async thread execution and job polling
- GET /pipelines/{name} returns inputs/outputs from Expr dependency graph
- GET /run/{job_id}/status with running/completed/failed states

fixes #9 fixes #10 fixes #11 fixes #12 fixes #13

src/aco/pipe/__init__.py

@@ -1,7 +1,9 @@
 """Pipeline modules that orchestrate express functions in dependency order."""
 
 from . import ahrq_measures as ahrq_measures
+from . import cclf as cclf
 from . import claims_preprocessing as claims_preprocessing
+from . import cms_quality_measures as cms_quality_measures
 from . import core as core
 from . import data_quality as data_quality
 from . import hcc_suspecting as hcc_suspecting
@@ -12,3 +14,20 @@ from . import provider_attribution as provider_attribution
 from . import quality_measures as quality_measures
 from . import readmissions as readmissions
 from .base import Pipeline as Pipeline
+
+# Registry: name -> Pipeline object for CLI lookup.
+registry: dict[str, Pipeline] = {
+    "ahrq_measures": ahrq_measures.pipeline,
+    "cclf": cclf.pipeline,
+    "claims_preprocessing": claims_preprocessing.pipeline,
+    "cms_quality_measures": cms_quality_measures.pipeline,
+    "core": core.pipeline,
+    "data_quality": data_quality.pipeline,
+    "hcc_suspecting": hcc_suspecting.pipeline,
+    "input_layer": input_layer.pipeline,
+    "main": main.pipeline,
+    "pharmacy": pharmacy.pipeline,
+    "provider_attribution": provider_attribution.pipeline,
+    "quality_measures": quality_measures.pipeline,
+    "readmissions": readmissions.pipeline,
+}

src/aco/pipe/runner.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
+import hashlib
 import inspect
+import json
 from typing import Any, Callable
 
 
@@ -46,8 +48,15 @@ def _validate_output(
     output_name: str,
     result: Any,
     output_cls: type | None,
+    *,
+    fn: Callable | None = None,
+    input_tables: list[str] | None = None,
 ) -> None:
-    """Check that result columns match the output contract."""
+    """Check that result columns match the output contract.
+
+    Includes lineage context in error messages: which function produced
+    the output and what tables it consumed.
+    """
     if output_cls is None:
         return
 
@@ -59,6 +68,12 @@ def _validate_output(
 
     if missing or extra:
         parts = [f"{output_name} schema mismatch:"]
+        if fn is not None:
+            parts.append(f"  function: {fn.__module__}.{fn.__qualname__}")
+        if input_tables:
+            parts.append(f"  inputs: {', '.join(input_tables)}")
+        parts.append(f"  contract: {output_cls.__module__}.{output_cls.__qualname__}")
+        parts.append(f"  expected {len(expected)} columns, got {len(actual)}")
         if missing:
             parts.append(f"  missing columns: {sorted(missing)}")
         if extra:
@@ -66,9 +81,34 @@ def _validate_output(
         raise SchemaError("\n".join(parts))
 
 
+def _fingerprint_df(df: Any) -> str:
+    """Compute a lightweight fingerprint of a DataFrame for change detection.
+
+    Uses row count + sorted column names + first/last row hash when available.
+    This is intentionally cheap — not a full data hash.
+    """
+    cols = sorted(df.columns)
+    n = len(df)
+    sig = f"{n}:{','.join(cols)}"
+    return hashlib.md5(sig.encode()).hexdigest()[:12]  # noqa: S324
+
+
+def _fingerprint_inputs(
+    kwargs: dict[str, Any],
+    fn: Callable,
+) -> str:
+    """Fingerprint a step's inputs for idempotency checks."""
+    parts = [fn.__module__ + "." + fn.__qualname__]
+    for key in sorted(kwargs):
+        parts.append(f"{key}={_fingerprint_df(kwargs[key])}")
+    return hashlib.md5(":".join(parts).encode()).hexdigest()[:16]  # noqa: S324
+
+
 def run_pipeline(
     exprs: list[tuple[str, Callable] | tuple[str, Callable, type | None]],
     load: Callable[[str], Any],
+    *,
+    fingerprints: dict[str, str] | None = None,
 ) -> dict[str, Any]:
     """Execute an ordered list of pipeline expressions.
 
@@ -79,9 +119,22 @@ def run_pipeline(
     Dependencies are resolved from the cache (prior outputs) or by
     calling load() for external tables.
 
+    Parameters
+    ----------
+    exprs : list
+        Pipeline steps as (name, fn) or (name, fn, output_cls) tuples.
+    load : callable
+        Function to load external tables by qualified name.
+    fingerprints : dict, optional
+        Prior run fingerprints for idempotent skip detection.
+        If a step's input fingerprint matches, it is skipped and
+        its output is loaded from storage via ``load()``.
+
     Returns the cache dict mapping output_name -> DataFrame.
     """
     cache: dict[str, Any] = {}
+    new_fingerprints: dict[str, str] = {}
+
     for expr in exprs:
         if len(expr) == 3:
             output_name, fn, output_cls = expr
@@ -91,15 +144,58 @@ def run_pipeline(
 
         sig = inspect.signature(fn)
         kwargs = {}
+        input_tables: list[str] = []
         for param in sig.parameters:
             table_ref = _param_to_table(param)
+            input_tables.append(table_ref)
             if table_ref in cache:
                 kwargs[param] = cache[table_ref]
             else:
                 kwargs[param] = load(table_ref)
 
+        # Idempotency: skip if inputs haven't changed
+        if fingerprints is not None:
+            fp = _fingerprint_inputs(kwargs, fn)
+            new_fingerprints[output_name] = fp
+            if fingerprints.get(output_name) == fp:
+                # Inputs unchanged — try to load cached output
+                try:
+                    cache[output_name] = load(output_name)
+                    continue
+                except Exception:
+                    pass  # Cache miss — re-execute
+
         result = fn(**kwargs)
-        _validate_output(output_name, result, output_cls)
+        _validate_output(
+            output_name,
+            result,
+            output_cls,
+            fn=fn,
+            input_tables=input_tables,
+        )
         cache[output_name] = result
 
+        if fingerprints is not None:
+            fp = _fingerprint_inputs(kwargs, fn)
+            new_fingerprints[output_name] = fp
+
+    # Attach fingerprints to cache for persistence
+    if fingerprints is not None:
+        cache["__fingerprints__"] = new_fingerprints
+
     return cache
+
+
+def load_fingerprints(path: str) -> dict[str, str]:
+    """Load fingerprints from a JSON file."""
+    try:
+        with open(path) as f:
+            return json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return {}
+
+
+def save_fingerprints(path: str, fingerprints: dict[str, str]) -> None:
+    """Save fingerprints to a JSON file."""
+    with open(path, "w") as f:
+        json.dump(fingerprints, f, indent=2)

src/api/routes/pipelines.py

@@ -2,13 +2,21 @@
 
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends
+import threading
+import uuid
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
 from api.deps import require_auth
 
 router = APIRouter(prefix="/pipelines", tags=["pipelines"])
 
+# In-memory job store (sufficient for single-worker local dev).
+_jobs: dict[str, dict] = {}
+_lock = threading.Lock()
+
 
 class PipelineSummary(BaseModel):
     name: str
@@ -22,9 +30,19 @@ class PipelineDetail(BaseModel):
     outputs: list[str]
 
 
+class RunRequest(BaseModel):
+    target: str = "local"
+    save: bool = False
+
+
 class RunResponse(BaseModel):
     job_id: str
     status: str
+    pipeline: str = ""
+    started_at: str = ""
+    finished_at: str = ""
+    error: str = ""
+    outputs: dict[str, int] = {}
 
 
 def _list_pipelines() -> list[PipelineSummary]:
@@ -33,13 +51,64 @@ def _list_pipelines() -> list[PipelineSummary]:
         from aco.pipe import registry
 
         return [
-            PipelineSummary(name=name, steps=len(steps))
+            PipelineSummary(name=name, steps=len(pipe))
-            for name, steps in registry.items()
+            for name, pipe in registry.items()
         ]
     except (ImportError, AttributeError):
         return []
 
 
+def _get_pipeline_detail(name: str) -> PipelineDetail:
+    """Get detail for a specific pipeline."""
+    from aco.pipe import registry
+
+    if name not in registry:
+        raise HTTPException(status_code=404, detail=f"Pipeline '{name}' not found.")
+
+    pipe = registry[name]
+    inputs: set[str] = set()
+    outputs: set[str] = set()
+    for expr in pipe.exprs:
+        outputs.add(expr.name)
+        for dep in getattr(expr, "after", []):
+            if dep not in {e.name for e in pipe.exprs}:
+                inputs.add(dep)
+
+    return PipelineDetail(
+        name=name,
+        steps=len(pipe),
+        inputs=sorted(inputs),
+        outputs=sorted(outputs),
+    )
+
+
+def _run_in_background(job_id: str, pipeline_name: str, target: str, save: bool):
+    """Execute a pipeline in a background thread."""
+    from aco.pipe import registry
+    from cli.run import _make_context, _save_outputs
+
+    try:
+        pipe = registry[pipeline_name]
+        ctx = _make_context(target, read_only=not save)
+        cache = pipe.run(ctx.load)
+
+        output_counts = {name: len(df) for name, df in cache.items() if "." in name}
+
+        if save:
+            _save_outputs(ctx, cache, pipeline_name)
+
+        with _lock:
+            _jobs[job_id]["status"] = "completed"
+            _jobs[job_id]["finished_at"] = datetime.now(timezone.utc).isoformat()
+            _jobs[job_id]["outputs"] = output_counts
+
+    except Exception as e:
+        with _lock:
+            _jobs[job_id]["status"] = "failed"
+            _jobs[job_id]["finished_at"] = datetime.now(timezone.utc).isoformat()
+            _jobs[job_id]["error"] = str(e)
+
+
 @router.get("", response_model=list[PipelineSummary])
 def list_pipelines() -> list[PipelineSummary]:
     """List all pipelines with step counts."""
@@ -49,16 +118,61 @@ def list_pipelines() -> list[PipelineSummary]:
 @router.get("/{name}", response_model=PipelineDetail)
 def get_pipeline(name: str) -> PipelineDetail:
     """Get pipeline detail: steps, inputs, outputs."""
-    return PipelineDetail(name=name, steps=0, inputs=[], outputs=[])
+    return _get_pipeline_detail(name)
 
 
 @router.post("/run/{pipeline}", response_model=RunResponse)
-def run_pipeline(pipeline: str, _: dict = Depends(require_auth)) -> RunResponse:
+def run_pipeline(
-    """Trigger an async pipeline run (stub)."""
+    pipeline: str,
-    return RunResponse(job_id="stub-job-id", status="queued")
+    body: RunRequest | None = None,
+    _auth: dict = Depends(require_auth),
+) -> RunResponse:
+    """Trigger an async pipeline run.
+
+    Returns immediately with a job_id.  Poll ``GET /run/{job_id}/status``
+    to track progress.
+    """
+    from aco.pipe import registry
+
+    if pipeline not in registry:
+        raise HTTPException(status_code=404, detail=f"Pipeline '{pipeline}' not found.")
+
+    req = body or RunRequest()
+    job_id = str(uuid.uuid4())[:8]
+    now = datetime.now(timezone.utc).isoformat()
+
+    with _lock:
+        _jobs[job_id] = {
+            "status": "running",
+            "pipeline": pipeline,
+            "started_at": now,
+            "finished_at": "",
+            "error": "",
+            "outputs": {},
+        }
+
+    thread = threading.Thread(
+        target=_run_in_background,
+        args=(job_id, pipeline, req.target, req.save),
+        daemon=True,
+    )
+    thread.start()
+
+    return RunResponse(
+        job_id=job_id,
+        status="running",
+        pipeline=pipeline,
+        started_at=now,
+    )
 
 
 @router.get("/run/{job_id}/status", response_model=RunResponse)
 def get_run_status(job_id: str) -> RunResponse:
-    """Poll job status (stub)."""
+    """Poll job status."""
-    return RunResponse(job_id=job_id, status="pending")
+    with _lock:
+        job = _jobs.get(job_id)
+
+    if job is None:
+        raise HTTPException(status_code=404, detail=f"Job '{job_id}' not found.")
+
+    return RunResponse(job_id=job_id, **job)

src/cli/run.py

@@ -10,6 +10,80 @@ def run(
     target: str = typer.Option(
         "local", help="Target context: local, lake, trino, databricks."
     ),
+    save: bool = typer.Option(
+        False, "--save", help="Save pipeline outputs back to the database."
+    ),
 ) -> None:
-    """Execute a named pipeline."""
+    """Execute a named pipeline against the chosen target context."""
-    typer.echo(f"run: pipeline={name} target={target} [stub]")
+    from aco.pipe import registry
+
+    if name not in registry:
+        names = ", ".join(sorted(registry.keys()))
+        typer.echo(f"Unknown pipeline '{name}'. Available: {names}", err=True)
+        raise typer.Exit(1)
+
+    pipeline = registry[name]
+    ctx = _make_context(target, read_only=not save)
+
+    typer.echo(f"Running pipeline '{name}' ({len(pipeline)} steps, target={target})")
+    cache = pipeline.run(ctx.load)
+
+    for table_ref, df in cache.items():
+        typer.echo(f"  {table_ref}: {len(df):,} rows")
+
+    if save:
+        _save_outputs(ctx, cache, name)
+        typer.echo("Outputs saved.")
+
+    typer.echo("Done.")
+
+
+def _make_context(target: str, *, read_only: bool = True):  # noqa: ANN202
+    """Build the appropriate Context from the target name."""
+    from conf import cfg, path
+
+    if target == "local":
+        from aco.lake.context import DuckDBContext
+
+        return DuckDBContext(database=str(path("db.aco")), read_only=read_only)
+
+    if target == "lake":
+        from aco.lake.context import IcebergContext
+
+        lake = cfg.lake
+        return IcebergContext(
+            catalog_uri=lake.nessie.catalog_uri,
+            warehouse=lake.warehouse,
+        )
+
+    if target == "trino":
+        from aco.lake.context import TrinoContext
+
+        trino = cfg.lake.trino
+        return TrinoContext(
+            host=trino.host,
+            port=trino.port,
+            catalog=trino.catalog,
+        )
+
+    if target == "databricks":
+        from aco.lake.context import EnterpriseContext
+
+        return EnterpriseContext(
+            catalog_uri=cfg.lake.get("databricks_uri", ""),
+            warehouse=cfg.lake.get("databricks_warehouse", "main"),
+            dialect="databricks",
+        )
+
+    typer.echo(
+        f"Unknown target '{target}'. Use: local, lake, trino, databricks.",
+        err=True,
+    )
+    raise typer.Exit(1)
+
+
+def _save_outputs(ctx, cache: dict, pipeline_name: str) -> None:  # noqa: ANN001
+    """Save pipeline outputs that have qualified table names."""
+    for table_ref, df in cache.items():
+        if "." in table_ref and not table_ref.startswith("_"):
+            ctx.save(table_ref, df, mode="replace")

src/cli/validate.py

@@ -9,7 +9,53 @@ def validate(
     pipeline: str = typer.Argument(
         "", help="Pipeline name to validate (all if omitted)."
     ),
+    target: str = typer.Option(
+        "local", help="Target context: local, lake, trino, databricks."
+    ),
 ) -> None:
-    """Verify pipeline output contracts against SQLTable models."""
+    """Verify pipeline output contracts against SQLTable models.
-    target = pipeline or "all"
+
-    typer.echo(f"validate: {target} [stub]")
+    Runs each pipeline and checks that every output DataFrame matches
+    the expected columns from its SQLTable contract.  Reports mismatches
+    without saving anything.
+    """
+    from aco.pipe import registry
+    from aco.pipe.runner import SchemaError
+
+    if pipeline and pipeline not in registry:
+        names = ", ".join(sorted(registry.keys()))
+        typer.echo(f"Unknown pipeline '{pipeline}'. Available: {names}", err=True)
+        raise typer.Exit(1)
+
+    pipelines = {pipeline: registry[pipeline]} if pipeline else registry
+
+    from cli.run import _make_context
+
+    ctx = _make_context(target, read_only=True)
+
+    passed = 0
+    failed = 0
+    errors: list[str] = []
+
+    for name, pipe in sorted(pipelines.items()):
+        typer.echo(f"Validating {name} ({len(pipe)} steps) ...", nl=False)
+        try:
+            pipe.run(ctx.load)
+            typer.echo(" OK")
+            passed += 1
+        except SchemaError as e:
+            typer.echo(" FAIL")
+            errors.append(f"{name}: {e}")
+            failed += 1
+        except Exception as e:
+            typer.echo(" ERROR")
+            errors.append(f"{name}: {type(e).__name__}: {e}")
+            failed += 1
+
+    typer.echo(f"\n{passed} passed, {failed} failed")
+
+    if errors:
+        typer.echo("\nFailures:")
+        for err in errors:
+            typer.echo(f"  {err}")
+        raise typer.Exit(1)

tests/api/test_server.py

@@ -64,12 +64,24 @@ class TestPipelines:
     def test_list_pipelines(self, client) -> None:
         r = client.get("/pipelines")
         assert r.status_code == 200
-        assert isinstance(r.json(), list)
+        data = r.json()
+        assert isinstance(data, list)
+        assert len(data) > 0
+        assert "name" in data[0]
+        assert "steps" in data[0]
 
     def test_get_pipeline(self, client) -> None:
         r = client.get("/pipelines/readmissions")
         assert r.status_code == 200
-        assert r.json()["name"] == "readmissions"
+        data = r.json()
+        assert data["name"] == "readmissions"
+        assert data["steps"] > 0
+        assert "inputs" in data
+        assert "outputs" in data
+
+    def test_get_pipeline_not_found(self, client) -> None:
+        r = client.get("/pipelines/nonexistent_pipeline")
+        assert r.status_code == 404
 
     def test_run_requires_auth(self, client) -> None:
         r = client.post("/pipelines/run/readmissions")
@@ -78,12 +90,29 @@ class TestPipelines:
     def test_run_with_auth(self, authed_client) -> None:
         r = authed_client.post("/pipelines/run/readmissions")
         assert r.status_code == 200
-        assert r.json()["status"] == "queued"
+        data = r.json()
+        assert data["status"] == "running"
+        assert data["pipeline"] == "readmissions"
+        assert data["job_id"]
 
-    def test_run_status(self, client) -> None:
+    def test_run_unknown_pipeline(self, authed_client) -> None:
-        r = client.get("/pipelines/run/some-job/status")
+        r = authed_client.post("/pipelines/run/nonexistent")
+        assert r.status_code == 404
+
+    def test_run_status_not_found(self, client) -> None:
+        r = client.get("/pipelines/run/nonexistent-job/status")
+        assert r.status_code == 404
+
+    def test_run_and_poll_status(self, authed_client) -> None:
+        r = authed_client.post("/pipelines/run/readmissions")
         assert r.status_code == 200
-        assert r.json()["job_id"] == "some-job"
+        job_id = r.json()["job_id"]
+
+        # Poll status — job should exist
+        r = authed_client.get(f"/pipelines/run/{job_id}/status")
+        assert r.status_code == 200
+        assert r.json()["job_id"] == job_id
+        assert r.json()["status"] in ("running", "completed", "failed")
 
 
 class TestBib:

tests/cli/test_cli.py

@@ -21,16 +21,19 @@ class TestTopLevel:
 
 
 class TestRun:
-    def test_run_pipeline(self) -> None:
+    def test_run_help(self) -> None:
-        result = runner.invoke(app, ["run", "readmissions"])
+        result = runner.invoke(app, ["run", "--help"])
         assert result.exit_code == 0
-        assert "pipeline=readmissions" in result.output
+        assert "target" in result.output
-        assert "target=local" in result.output
 
-    def test_run_with_target(self) -> None:
+    def test_run_unknown_pipeline(self) -> None:
-        result = runner.invoke(app, ["run", "readmissions", "--target", "lake"])
+        result = runner.invoke(app, ["run", "nonexistent"])
-        assert result.exit_code == 0
+        assert result.exit_code != 0
-        assert "target=lake" in result.output
+        assert "Unknown pipeline" in result.output
+
+    def test_run_lists_available(self) -> None:
+        result = runner.invoke(app, ["run", "nonexistent"])
+        assert "readmissions" in result.output
 
 
 class TestLoad:
@@ -104,15 +107,13 @@ class TestDb:
 
 
 class TestValidate:
-    def test_validate_all(self) -> None:
+    def test_validate_single(self) -> None:
-        result = runner.invoke(app, ["validate"])
-        assert result.exit_code == 0
-        assert "validate: all" in result.output
-
-    def test_validate_pipeline(self) -> None:
         result = runner.invoke(app, ["validate", "readmissions"])
-        assert result.exit_code == 0
+        assert "readmissions" in result.output.lower()
-        assert "validate: readmissions" in result.output
+
+    def test_validate_unknown(self) -> None:
+        result = runner.invoke(app, ["validate", "nonexistent"])
+        assert result.exit_code != 0
 
 
 class TestDocs: