diff --git a/CHANGELOG.md b/CHANGELOG.md index 649f9dd..b11275b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ Notable changes to Loafer are documented here. This project follows ## [Unreleased] +### Added + +- A framework-independent application service with strict, JSON-roundtrippable contracts for run + requests, execution plans, batch envelopes, events, snapshots, and results. +- Runtime ports and local adapters for cancellation, checkpoints, secret resolution, event + publication, and interactive transform review. + +### Changed + +- The CLI, scheduler, and legacy Python runner now share the same application boundary while core + execution orchestration remains independent of client frameworks. +- Durable application contracts now exclude credentials, connector instances, iterators, provider + clients, row payloads, and other ephemeral runtime objects. + ## [0.4.0] - 2026-07-29 ### Added diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3252023..be7127e 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -62,12 +62,19 @@ The core domain never imports from infrastructure. ## Git Hygiene +- Name branches for the work using a conventional work-type prefix: + `feat/`, `fix/`, `docs/`, `refactor/`, + `test/`, or `chore/` +- Never use an agent/tool prefix such as `agent/`, and never name branches after roadmap phases + such as `phase-1` or `phase-2` - Commits are incremental. One logical unit of work per commit - Commit messages are lowercase, imperative, and descriptive - Never start a commit message with `feat:`, `fix:`, `chore:` — just describe what it does - Never start a commit message with `Phase 0`, `Phase 1`, or any phase label - Stage only the files you actually created or modified - Never commit local prompt scratchpads, credentials, or generated test secrets +- Before every commit, update the `[Unreleased]` section of `CHANGELOG.md` with the change; never + commit implementation work first and backfill its changelog entry later ## Testing diff --git a/PRODUCTION_READINESS.md b/PRODUCTION_READINESS.md index 8c185c4..fa76634 100644 --- a/PRODUCTION_READINESS.md +++ b/PRODUCTION_READINESS.md @@ -336,6 +336,15 @@ Exit gate: - import-boundary tests prevent engine-to-client/API dependencies; - all durable contract types serialize and round-trip. +**Current status:** complete. `ExecutionPlan`, `BatchEnvelope`, `Checkpoint`, `RunEvent`, +`RunSnapshot`, and `RunResult` are strict JSON-round-trippable contracts; cancellation, +checkpoint, secret, event, and generated-code review behavior is expressed through ports. The +`RunPipeline` application use case now owns plan/run orchestration, while `loafer/engine.py` owns +the in-process ETL/ELT graph and `runner.py` is a compatibility facade. The CLI and local scheduler +call the same application service. A real CSV → custom transform → JSON pipeline passes through +that interface, import tests keep client frameworks out of the engine, and the repository suite is +green with 683 passed and 50 skipped. + ### Phase 2 — Build the bounded, correct data plane **Goal:** make memory and publication behavior a property of the execution contract rather than a @@ -537,16 +546,18 @@ Exit gate: ## What to implement next -With Phase 0 complete, start Phase 1: +With Phase 0 and Phase 1 complete, start Phase 2: -1. Define the serializable `ExecutionPlan`, `BatchEnvelope`, `RunEvent`, `RunResult`, - `CancellationPort`, `CheckpointPort`, and `SecretResolver` contracts. -2. Extract one `RunPipeline` application use case from the CLI/runner. -3. Migrate one vertical slice—CSV → row-local transform → JSON—through the new boundary while - preserving current CLI behavior. +1. Add a `transform_batch` execution path for declared row-local transforms. +2. Keep bounded `BatchEnvelope` units flowing through CSV extract → validate → transform → + staged JSON publication without populating full-run `raw_data` or `transformed_data`. +3. Generate and version AI transform artifacts once per run, then execute the validated artifact + per batch. +4. Reconcile batch/input/output/rejected counts and checksums, and test cancellation or target + failure without false success or final partial output. -This slice creates the seam needed for every later phase without prematurely introducing Better -Auth, PostgreSQL metadata, NATS, or a second execution path. +Do not add Better Auth, PostgreSQL run metadata, NATS, or distributed workers until the bounded +single-node data-plane contract is real. ## Definition of the 100M-row claim diff --git a/README.md b/README.md index 1309df1..aee2556 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,25 @@ loafer run pipeline.yaml Loafer infers connector and transform types from URLs, file extensions, and configuration fields. Use an explicit `type` when inference would be ambiguous. +## Python application interface + +The CLI and local scheduler use the same application service available to Python callers: + +```python +from loafer.application import RunRequest, get_local_application + +service = get_local_application() +result = service.run_pipeline.run( + RunRequest(config_path="pipeline.yaml", auto_confirm=True) +) + +print(result.status, result.snapshot.rows_loaded) +``` + +`RunResult` and streamed `RunEvent` values are JSON-round-trippable, sanitized contracts. They do +not contain source rows, credentials, connectors, iterators, or live LLM provider objects. The +legacy `loafer.runner.run_pipeline()` API remains available as a compatibility facade. + ## Transform options ### SQL diff --git a/loafer/adapters/runtime.py b/loafer/adapters/runtime.py new file mode 100644 index 0000000..22c59aa --- /dev/null +++ b/loafer/adapters/runtime.py @@ -0,0 +1,53 @@ +"""Local adapters for application runtime ports.""" + +from __future__ import annotations + +import os + +from loafer.contracts import Checkpoint, RunEvent + + +class NeverCancelled: + """Default cancellation adapter for synchronous local execution.""" + + def is_cancelled(self, run_id: str) -> bool: + del run_id + return False + + +class NullCheckpointStore: + """No-op checkpoint adapter until Phase 3 adds durable metadata.""" + + def load(self, run_id: str, partition_id: str) -> Checkpoint | None: + del run_id, partition_id + return None + + def save(self, checkpoint: Checkpoint) -> None: + del checkpoint + + +class EnvironmentSecretResolver: + """Resolve local secret references from environment variables.""" + + def resolve(self, reference: str) -> str | None: + return os.environ.get(reference) + + +class NullEventPublisher: + """Discard events for callers that consume the returned iterator.""" + + def publish(self, event: RunEvent) -> None: + del event + + +class InputReviewPort: + """Portable stdin reviewer used by the local Python API.""" + + def approve_transform(self, generated_code: str) -> bool: + print("\nAI-generated transform code:\n") + print(generated_code) + try: + answer = input("Execute this code? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt): + return False + return answer in {"y", "yes"} diff --git a/loafer/application/__init__.py b/loafer/application/__init__.py new file mode 100644 index 0000000..7cceeec --- /dev/null +++ b/loafer/application/__init__.py @@ -0,0 +1,38 @@ +"""Versioned application boundary for Loafer clients.""" + +from loafer.application.local import get_local_application +from loafer.application.service import LocalApplicationService, RunPipeline +from loafer.contracts import ( + BatchEnvelope, + Checkpoint, + ConnectorCatalog, + ExecutionPlan, + RunEvent, + RunRequest, + RunResult, + RunSnapshot, + RunStatus, + StageStatus, + ValidationResult, +) +from loafer.ports.runtime import CancellationPort, CheckpointPort, SecretResolver + +__all__ = [ + "BatchEnvelope", + "CancellationPort", + "Checkpoint", + "CheckpointPort", + "ConnectorCatalog", + "ExecutionPlan", + "LocalApplicationService", + "RunEvent", + "RunPipeline", + "RunRequest", + "RunResult", + "RunSnapshot", + "RunStatus", + "SecretResolver", + "StageStatus", + "ValidationResult", + "get_local_application", +] diff --git a/loafer/application/local.py b/loafer/application/local.py new file mode 100644 index 0000000..58a0445 --- /dev/null +++ b/loafer/application/local.py @@ -0,0 +1,32 @@ +"""Local composition root for the application service.""" + +from __future__ import annotations + +from loafer.adapters.runtime import ( + EnvironmentSecretResolver, + InputReviewPort, + NeverCancelled, + NullCheckpointStore, + NullEventPublisher, +) +from loafer.application.service import LocalApplicationService, RunPipeline +from loafer.engine import ProviderFactory +from loafer.ports.runtime import ReviewPort + + +def get_local_application( + *, + reviewer: ReviewPort | None = None, + provider_factory: ProviderFactory | None = None, +) -> LocalApplicationService: + """Build the synchronous local application service.""" + return LocalApplicationService( + RunPipeline( + cancellation=NeverCancelled(), + checkpoints=NullCheckpointStore(), + secrets=EnvironmentSecretResolver(), + events=NullEventPublisher(), + reviewer=reviewer or InputReviewPort(), + provider_factory=provider_factory, + ) + ) diff --git a/loafer/application/service.py b/loafer/application/service.py new file mode 100644 index 0000000..5e5efb4 --- /dev/null +++ b/loafer/application/service.py @@ -0,0 +1,314 @@ +"""Local application service and pipeline use cases.""" + +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Iterator +from dataclasses import asdict, is_dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from loafer.config import PipelineConfig, load_config +from loafer.contracts import ( + ConnectorCatalog, + ExecutionPlan, + RunEvent, + RunRequest, + RunResult, + RunSnapshot, + RunStatus, + StageStatus, + TransformStepResult, + ValidationResult, +) +from loafer.engine import ProviderFactory, stream_pipeline +from loafer.exceptions import PipelineError +from loafer.graph.state import PipelineState +from loafer.ports.runtime import ( + CancellationPort, + CheckpointPort, + EventPublisher, + ReviewPort, + SecretResolver, +) + +_URL_PASSWORD = re.compile(r"(?P[a-z][a-z0-9+.-]*://[^:/@\s]+):[^@\s]+@") + + +def _json_value(value: Any) -> Any: + """Convert config metadata to a JSON-compatible value.""" + return json.loads(json.dumps(value, default=str)) + + +def _redact(text: str | None, config: PipelineConfig) -> str | None: + """Remove configured keys and URL passwords from durable messages.""" + if text is None: + return None + redacted = _URL_PASSWORD.sub(r"\g:***@", text) + if config.llm.api_key: + redacted = redacted.replace(config.llm.api_key, "***") + return redacted + + +def _config_digest(config: PipelineConfig) -> str: + rendered = json.dumps( + config.model_dump(mode="json"), + sort_keys=True, + separators=(",", ":"), + ) + return hashlib.sha256(rendered.encode("utf-8")).hexdigest() + + +def _build_plan( + config: PipelineConfig, + request: RunRequest, +) -> ExecutionPlan: + digest = _config_digest(config) + resolved_path = str(Path(request.config_path).resolve()) + plan_id = hashlib.sha256(f"{resolved_path}:{digest}".encode()).hexdigest()[:16] + cursor_value: Any = None + + if config.incremental is not None: + cursor_value = config.incremental.initial + if not request.full_refresh: + from loafer.core.incremental import StateStore, state_path_for + + stored = StateStore(state_path_for(resolved_path)).get_cursor( + config.name or Path(resolved_path).stem + ) + if stored is not None: + cursor_value = stored + + return ExecutionPlan( + plan_id=plan_id, + config_digest=digest, + config_path=resolved_path, + pipeline_name=config.name or Path(resolved_path).stem, + mode=config.mode, + source_type=config.source.type, + target_type=config.target.type, + transform_type=config.transform.type, + chunk_size=config.chunk_size, + streaming_threshold=config.streaming_threshold, + validation_strict=config.validation.strict, + llm_provider=config.llm.provider, + llm_model=config.llm.model, + incremental_column=config.incremental.column if config.incremental else None, + cursor_value=_json_value(cursor_value), + dry_run=request.dry_run, + auto_confirm=request.auto_confirm, + full_refresh=request.full_refresh, + ) + + +def _step_result(value: Any, config: PipelineConfig) -> TransformStepResult: + if isinstance(value, dict): + data = dict(value) + elif is_dataclass(value): + data = asdict(value) + else: + data = { + name: getattr(value, name) + for name in ( + "index", + "name", + "type", + "rows_in", + "rows_out", + "duration_ms", + "success", + "error", + "token_usage", + ) + } + data["error"] = _redact(data.get("error"), config) + data["token_usage"] = data.get("token_usage") or {} + return TransformStepResult.model_validate(data) + + +def _snapshot( + state: PipelineState, + plan: ExecutionPlan, + config: PipelineConfig, +) -> RunSnapshot: + warnings = tuple(_redact(str(item), config) or "" for item in state.get("warnings", [])) + durations = { + str(name): max(0.0, float(value)) for name, value in state.get("duration_ms", {}).items() + } + token_usage = {str(name): int(value) for name, value in state.get("token_usage", {}).items()} + steps = tuple(_step_result(item, config) for item in state.get("step_results", [])) + + return RunSnapshot( + run_id=state.get("run_id", ""), + plan_id=plan.plan_id, + pipeline_name=plan.pipeline_name, + mode=plan.mode, + source_type=plan.source_type, + target_type=plan.target_type, + transform_type=plan.transform_type, + rows_extracted=max(0, int(state.get("rows_extracted", 0))), + rows_transformed=max(0, len(state.get("transformed_data", []))), + rows_loaded=max(0, int(state.get("rows_loaded", 0))), + validation_passed=bool(state.get("validation_passed", False)), + duration_ms=durations, + warnings=warnings, + token_usage=token_usage, + step_results=steps, + error=_redact(state.get("last_error"), config), + ) + + +class RunPipeline: + """Application use case for planning and executing one pipeline.""" + + def __init__( + self, + *, + cancellation: CancellationPort, + checkpoints: CheckpointPort, + secrets: SecretResolver, + events: EventPublisher, + reviewer: ReviewPort, + provider_factory: ProviderFactory | None = None, + ) -> None: + self._cancellation = cancellation + self._checkpoints = checkpoints + self._secrets = secrets + self._events = events + self._reviewer = reviewer + self._provider_factory = provider_factory + + def create_plan(self, request: RunRequest) -> ExecutionPlan: + """Validate a config and return its credential-free execution plan.""" + config = self._load_config(request.config_path) + return _build_plan(config, request) + + def validate_config_model(self, config_path: str | Path) -> PipelineConfig: + """Return the validated runtime model for legacy local callers.""" + return self._load_config(config_path) + + def run(self, request: RunRequest) -> RunResult: + """Execute a pipeline and return only its durable result.""" + started_at = datetime.now(UTC) + config, plan = self._prepare(request) + state, output_published = self._consume(config, plan, request) + return RunResult( + run_id=request.run_id, + plan_id=plan.plan_id, + status=RunStatus.SUCCEEDED, + started_at=started_at, + finished_at=datetime.now(UTC), + output_published=output_published, + snapshot=_snapshot(state, plan, config), + ) + + def run_state(self, request: RunRequest) -> PipelineState: + """Execute through the use case while returning legacy runtime state.""" + config, plan = self._prepare(request) + state, _output_published = self._consume(config, plan, request) + return state + + def stream(self, request: RunRequest) -> Iterator[RunEvent]: + """Yield sanitized, serializable application events.""" + for event, _state in self.stream_states(request): + yield event + + def stream_states( + self, + request: RunRequest, + ) -> Iterator[tuple[RunEvent, PipelineState]]: + """Yield events plus ephemeral state for the compatibility facade.""" + config, plan = self._prepare(request) + yield from self._stream_prepared(config, plan, request) + + def _consume( + self, + config: PipelineConfig, + plan: ExecutionPlan, + request: RunRequest, + ) -> tuple[PipelineState, bool]: + final_state: PipelineState | None = None + output_published = False + for event, state in self._stream_prepared(config, plan, request): + final_state = state + if event.status is StageStatus.DONE and event.stage in { + "load", + "transform_in_target", + }: + output_published = not request.dry_run + if final_state is None: + raise PipelineError(f"Pipeline produced no stages (run_id={request.run_id})") + return final_state, output_published + + def _stream_prepared( + self, + config: PipelineConfig, + plan: ExecutionPlan, + request: RunRequest, + ) -> Iterator[tuple[RunEvent, PipelineState]]: + updates = stream_pipeline( + config, + config_path=plan.config_path, + dry_run=request.dry_run, + auto_confirm=request.auto_confirm, + full_refresh=request.full_refresh, + run_id=request.run_id, + reviewer=self._reviewer, + secret_resolver=self._secrets, + provider_factory=self._provider_factory, + ) + sequence = 0 + + while True: + if self._cancellation.is_cancelled(request.run_id): + raise PipelineError(f"Pipeline cancelled (run_id={request.run_id})") + try: + stage, status, state = next(updates) + except StopIteration: + return + + sequence += 1 + event = RunEvent( + run_id=request.run_id, + plan_id=plan.plan_id, + sequence=sequence, + stage=stage, + status=StageStatus(status), + snapshot=_snapshot(state, plan, config), + ) + self._events.publish(event) + yield event, state + + def _prepare(self, request: RunRequest) -> tuple[PipelineConfig, ExecutionPlan]: + config = self._load_config(request.config_path) + return config, _build_plan(config, request) + + @staticmethod + def _load_config(config_path: str | Path) -> PipelineConfig: + try: + return load_config(config_path) + except Exception as exc: + raise PipelineError(f"Config validation failed: {exc}") from exc + + +class LocalApplicationService: + """Application boundary used by the local CLI, scheduler, and Python API.""" + + def __init__(self, run_pipeline: RunPipeline) -> None: + self.run_pipeline = run_pipeline + + def validate(self, config_path: str | Path) -> ValidationResult: + request = RunRequest(config_path=str(config_path)) + return ValidationResult(plan=self.run_pipeline.create_plan(request)) + + def validate_config_model(self, config_path: str | Path) -> PipelineConfig: + return self.run_pipeline.validate_config_model(config_path) + + def list_connectors(self) -> ConnectorCatalog: + from loafer.connectors.registry import list_registered_connectors + + sources, targets = list_registered_connectors() + return ConnectorCatalog(sources=tuple(sources), targets=tuple(targets)) diff --git a/loafer/cli.py b/loafer/cli.py index cbd05eb..952ef08 100644 --- a/loafer/cli.py +++ b/loafer/cli.py @@ -21,11 +21,12 @@ from rich.panel import Panel from rich.rule import Rule from rich.spinner import Spinner +from rich.syntax import Syntax from rich.table import Table +from loafer.application import RunRequest, get_local_application from loafer.exceptions import LLMError, PipelineError, SchedulerError from loafer.llm.models import DEFAULT_GEMINI_MODEL, default_model_for, provider_for_model -from loafer.runner import list_connectors, run_pipeline_streaming, validate_config app = typer.Typer( name="loafer", @@ -39,6 +40,27 @@ _config_arg = typer.Argument(..., help="Path to pipeline YAML config") +class RichReviewPort: + """Render generated code and collect approval in the CLI client.""" + + def approve_transform(self, generated_code: str) -> bool: + console.print() + console.print( + Panel( + "[yellow]AI-generated transform code is ready for review.[/yellow]\n" + "Review the code below. Type 'y' to execute or 'n' to skip it.", + title="[bold yellow]⚠ Human Review Required[/bold yellow]", + ) + ) + console.print(Syntax(generated_code, "python", theme="monokai", line_numbers=True)) + try: + answer = input("Execute this code? [y/N]: ").strip().lower() + except (EOFError, KeyboardInterrupt): + console.print("\n[dim]No input received. Skipping AI transform.[/dim]") + return False + return answer in {"y", "yes"} + + def _resolve_version() -> str: """Return the installed package version, or a dev fallback.""" from importlib.metadata import PackageNotFoundError, version @@ -382,17 +404,17 @@ def _get_stage_label(node_name: str, state: Any) -> str: base = _STAGE_LABELS.get(node_name, node_name.capitalize()) if node_name == "extract": - src = state.get("source_config") - if src and hasattr(src, "type"): - base = f"Extracting from {src.type.upper()}" + source_type = state.get("source_type") + if source_type: + base = f"Extracting from {str(source_type).upper()}" elif node_name == "transform": - tc = state.get("transform_config") - if tc and hasattr(tc, "type"): - base = f"Transforming data ({tc.type})" + transform_type = state.get("transform_type") + if transform_type: + base = f"Transforming data ({transform_type})" elif node_name in ("load", "load_raw"): - tgt = state.get("target_config") - if tgt and hasattr(tgt, "type"): - base = f"Loading to {tgt.type.upper()}" + target_type = state.get("target_type") + if target_type: + base = f"Loading to {str(target_type).upper()}" return base @@ -408,7 +430,7 @@ def _get_row_info(node_name: str, state: Any) -> str: return f"{n} passed" if passed else "failed" if node_name == "transform": src = state.get("rows_extracted", 0) - dst = len(state.get("transformed_data", [])) + dst = state.get("rows_transformed", 0) if src and dst and src != dst: return f"{src} → {dst}" return f"{dst} row{'s' if dst != 1 else ''}" if dst else "—" @@ -443,16 +465,23 @@ def _add_step_breakdown_rows(table: Any, state: dict[str, Any]) -> None: return for i, step in enumerate(step_results): + step_name = step["name"] if isinstance(step, dict) else step.name + step_type = step["type"] if isinstance(step, dict) else step.type + step_success = step["success"] if isinstance(step, dict) else step.success + rows_in = step["rows_in"] if isinstance(step, dict) else step.rows_in + rows_out = step["rows_out"] if isinstance(step, dict) else step.rows_out + duration_ms = step["duration_ms"] if isinstance(step, dict) else step.duration_ms + token_usage = step.get("token_usage") if isinstance(step, dict) else step.token_usage connector = "└─" if i == len(step_results) - 1 else "├─" - icon = _STATUS_ICONS.get("done" if step.success else "failed", " ") - detail = f"({step.type})" - if step.token_usage and step.token_usage.get("total_tokens"): - detail = f"({step.type}, {step.token_usage['total_tokens']:,} tok)" + icon = _STATUS_ICONS.get("done" if step_success else "failed", " ") + detail = f"({step_type})" + if token_usage and token_usage.get("total_tokens"): + detail = f"({step_type}, {token_usage['total_tokens']:,} tok)" table.add_row( - f" {connector} {step.name}", + f" {connector} {step_name}", icon, - f"{step.rows_in:,} → {step.rows_out:,}", - f"{step.duration_ms / 1000:.1f}s [dim]{detail}[/dim]", + f"{rows_in:,} → {rows_out:,}", + f"{duration_ms / 1000:.1f}s [dim]{detail}[/dim]", ) @@ -563,41 +592,30 @@ def run( err_console.print(f"[red]Config file not found: {actual_config}[/red]") raise typer.Exit(1) - from loafer.config import load_config as _load_config - + service = get_local_application(reviewer=RichReviewPort()) + request = RunRequest( + config_path=str(actual_config), + dry_run=dry_run, + auto_confirm=yes, + full_refresh=full_refresh, + ) try: - cfg = _load_config(actual_config) - pipeline_name = cfg.name or actual_config.stem - mode = cfg.mode - except Exception as exc: + plan = service.run_pipeline.create_plan(request) + pipeline_name = plan.pipeline_name + mode = plan.mode + except PipelineError as exc: user_msg = _format_user_error(exc) err_console.print(f"\n[red]{user_msg}[/red]") raise typer.Exit(1) from exc - # Validate LLM provider is available before starting - if cfg.transform.type == "ai" and not cfg.transform.bypass_ai: - from loafer.runner import _build_llm_provider - - try: - _build_llm_provider(cfg) - except LLMError as exc: - user_msg = _format_user_error(exc) - err_console.print(f"\n[red]{user_msg}[/red]") - raise typer.Exit(1) from exc - console.print(f"\n[bold]Running: {pipeline_name}[/bold] [{mode.upper()}]") - if cfg.incremental is not None: - from loafer.core.incremental import StateStore, state_path_for - + if plan.incremental_column is not None: if full_refresh: - console.print(f"[dim]Incremental on '{cfg.incremental.column}' — full refresh[/dim]") + console.print(f"[dim]Incremental on '{plan.incremental_column}' — full refresh[/dim]") else: - saved = StateStore(state_path_for(actual_config)).get_cursor( - cfg.name or actual_config.stem - ) - shown = saved if saved is not None else cfg.incremental.initial console.print( - f"[dim]Incremental on '{cfg.incremental.column}' — cursor: {shown!r}[/dim]" + f"[dim]Incremental on '{plan.incremental_column}' — " + f"cursor: {plan.cursor_value!r}[/dim]" ) console.print(Rule(style="dim")) @@ -606,12 +624,10 @@ def run( active_animator: StageAnimator | None = None try: - for node_name, status, state in run_pipeline_streaming( - config_path=actual_config, - dry_run=dry_run, - yes=yes, - full_refresh=full_refresh, - ): + for event in service.run_pipeline.stream(request): + node_name = event.stage + status = event.status.value + state = event.snapshot.model_dump(mode="python") final_state = state label = _get_stage_label(node_name, state) row_info = _get_row_info(node_name, state) @@ -684,9 +700,9 @@ def validate( raise typer.Exit(1) try: - config = validate_config(config_file) + result = get_local_application().validate(config_file) except PipelineError as exc: - # validate_config already prefixes "Config validation failed:" — don't + # The application service already prefixes "Config validation failed:" — don't # wrap it a second time (BUG: doubled error prefix). err_console.print(f"[red]{exc}[/red]") raise typer.Exit(1) from exc @@ -697,17 +713,18 @@ def validate( table.add_column("Setting", style="cyan") table.add_column("Value", style="green") - if config.name: - table.add_row("Name", config.name) - table.add_row("Mode", config.mode) - table.add_row("Source", config.source.type) - table.add_row("Target", config.target.type) - table.add_row("Transform", config.transform.type) - table.add_row("Chunk size", str(config.chunk_size)) - table.add_row("Streaming threshold", str(config.streaming_threshold)) - table.add_row("Validation strict", str(config.validation.strict)) - table.add_row("LLM provider", config.llm.provider) - table.add_row("LLM model", config.llm.model) + plan = result.plan + if plan.pipeline_name: + table.add_row("Name", plan.pipeline_name) + table.add_row("Mode", plan.mode) + table.add_row("Source", plan.source_type) + table.add_row("Target", plan.target_type) + table.add_row("Transform", plan.transform_type) + table.add_row("Chunk size", str(plan.chunk_size)) + table.add_row("Streaming threshold", str(plan.streaming_threshold)) + table.add_row("Validation strict", str(plan.validation_strict)) + table.add_row("LLM provider", plan.llm_provider) + table.add_row("LLM model", plan.llm_model) console.print(table) @@ -715,19 +732,19 @@ def validate( @app.command() def connectors() -> None: """List available source and target connectors.""" - result = list_connectors() + result = get_local_application().list_connectors() console.print("[bold]Available Connectors[/bold]\n") source_table = Table(title="Sources") source_table.add_column("Type", style="cyan") - for source_type in result["sources"]: + for source_type in result.sources: source_table.add_row(source_type) console.print(source_table) target_table = Table(title="Targets") target_table.add_column("Type", style="cyan") - for target_type in result["targets"]: + for target_type in result.targets: target_table.add_row(target_type) console.print(target_table) diff --git a/loafer/connectors/registry.py b/loafer/connectors/registry.py index e5ab437..442dc6b 100644 --- a/loafer/connectors/registry.py +++ b/loafer/connectors/registry.py @@ -69,6 +69,11 @@ def _register_target(type_name: str, cls: type[TargetConnector]) -> None: _register_target("mongo", _MongoTarget) +def list_registered_connectors() -> tuple[list[str], list[str]]: + """Return sorted source and target connector type names.""" + return sorted(_SOURCE_REGISTRY), sorted(_TARGET_REGISTRY) + + def get_source_connector( config: SourceConfig, *, diff --git a/loafer/contracts.py b/loafer/contracts.py new file mode 100644 index 0000000..f71ddb8 --- /dev/null +++ b/loafer/contracts.py @@ -0,0 +1,183 @@ +"""Serializable engine/application contracts shared by clients and workers.""" + +from __future__ import annotations + +import uuid +from datetime import UTC, datetime +from enum import StrEnum +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, JsonValue, field_validator + + +def _utc_now() -> datetime: + return datetime.now(UTC) + + +class ContractModel(BaseModel): + """Strict, immutable base for durable application contracts.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + +class StageStatus(StrEnum): + RUNNING = "running" + DONE = "done" + SKIPPED = "skipped" + FAILED = "failed" + CANCELLED = "cancelled" + + +class RunStatus(StrEnum): + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELLED = "cancelled" + + +class RunRequest(ContractModel): + """Command to execute one immutable local pipeline configuration.""" + + config_path: str + run_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:12]) + dry_run: bool = False + auto_confirm: bool = False + full_refresh: bool = False + + +class ExecutionPlan(ContractModel): + """Credential-free description of a validated pipeline execution.""" + + contract_version: Literal[1] = 1 + plan_id: str + config_digest: str + config_path: str + pipeline_name: str + mode: Literal["etl", "elt"] + source_type: str + target_type: str + transform_type: str + chunk_size: int = Field(gt=0) + streaming_threshold: int = Field(gt=0) + validation_strict: bool + llm_provider: str + llm_model: str + incremental_column: str | None = None + cursor_value: JsonValue = None + dry_run: bool = False + auto_confirm: bool = False + full_refresh: bool = False + + @field_validator("config_digest") + @classmethod + def digest_is_sha256(cls, value: str) -> str: + if len(value) != 64 or any(char not in "0123456789abcdef" for char in value): + raise ValueError("config_digest must be a lowercase SHA-256 hex digest") + return value + + +class BatchEnvelope(ContractModel): + """Durable metadata for one bounded data-plane batch.""" + + contract_version: Literal[1] = 1 + run_id: str + stage_id: str + partition_id: str + batch_id: str + attempt: int = Field(ge=0) + source_position_start: JsonValue = None + source_position_end: JsonValue = None + schema_version: str + transform_artifact_version: str | None = None + rows_in: int = Field(ge=0) + rows_out: int = Field(ge=0) + rows_rejected: int = Field(ge=0) + bytes_in: int = Field(ge=0) + bytes_out: int = Field(ge=0) + checksum: str | None = None + + +class Checkpoint(ContractModel): + """Last durable source position for a committed target effect.""" + + contract_version: Literal[1] = 1 + checkpoint_id: str + run_id: str + partition_id: str + batch_id: str + source_position: JsonValue + committed_at: datetime = Field(default_factory=_utc_now) + + +class TransformStepResult(ContractModel): + """Serializable outcome of one multi-step transform entry.""" + + index: int = Field(ge=0) + name: str + type: str + rows_in: int = Field(ge=0) + rows_out: int = Field(ge=0) + duration_ms: float = Field(ge=0) + success: bool + error: str | None = None + token_usage: dict[str, int] = Field(default_factory=dict) + + +class RunSnapshot(ContractModel): + """Sanitized durable projection of ephemeral graph state.""" + + run_id: str + plan_id: str + pipeline_name: str + mode: Literal["etl", "elt"] + source_type: str + target_type: str + transform_type: str + rows_extracted: int = Field(ge=0) + rows_transformed: int = Field(ge=0) + rows_loaded: int = Field(ge=0) + validation_passed: bool + duration_ms: dict[str, float] = Field(default_factory=dict) + warnings: tuple[str, ...] = () + token_usage: dict[str, int] = Field(default_factory=dict) + step_results: tuple[TransformStepResult, ...] = () + error: str | None = None + + +class RunEvent(ContractModel): + """Append-friendly stage event emitted by the application service.""" + + contract_version: Literal[1] = 1 + run_id: str + plan_id: str + sequence: int = Field(gt=0) + stage: str + status: StageStatus + occurred_at: datetime = Field(default_factory=_utc_now) + snapshot: RunSnapshot + + +class RunResult(ContractModel): + """Final sanitized result of an application-level pipeline run.""" + + contract_version: Literal[1] = 1 + run_id: str + plan_id: str + status: RunStatus + started_at: datetime + finished_at: datetime + output_published: bool + snapshot: RunSnapshot + + +class ValidationResult(ContractModel): + """Successful validation response for a client.""" + + valid: Literal[True] = True + plan: ExecutionPlan + + +class ConnectorCatalog(ContractModel): + """Serializable list of connector types available to clients.""" + + sources: tuple[str, ...] + targets: tuple[str, ...] diff --git a/loafer/engine.py b/loafer/engine.py new file mode 100644 index 0000000..42cac54 --- /dev/null +++ b/loafer/engine.py @@ -0,0 +1,592 @@ +"""In-process ETL/ELT engine. + +This module owns graph selection and execution. It has no CLI rendering or +client-framework dependencies; callers use the application service. +""" + +from __future__ import annotations + +import time +import uuid +from collections.abc import Iterator +from pathlib import Path +from typing import TYPE_CHECKING, Any, Protocol + +from loafer.config import PipelineConfig +from loafer.exceptions import LLMError, PipelineError +from loafer.graph.state import PipelineState + +if TYPE_CHECKING: + from collections.abc import Iterator + + from loafer.llm.base import LLMProvider + from loafer.ports.runtime import ReviewPort, SecretResolver + +_PROVIDER_ENV_VARS = { + "gemini": "GEMINI_API_KEY", + "claude": "ANTHROPIC_API_KEY", + "openai": "OPENAI_API_KEY", + "qwen": "DASHSCOPE_API_KEY", +} + +# Hard backstop on graph hops. The ELT graph caps its own retries via the +# transform_in_target counter, but a stuck conditional edge could still loop; +# this ensures LangGraph raises GraphRecursionError instead of spinning +# forever (the original BUG-3 symptom under graph.stream). +_GRAPH_CONFIG = {"recursion_limit": 25} + + +class ProviderFactory(Protocol): + """Construct an LLM provider using the caller's secret boundary.""" + + def __call__( + self, + config: PipelineConfig, + secret_resolver: SecretResolver | None, + ) -> LLMProvider: ... + + +def _build_llm_provider( + config: PipelineConfig, + secret_resolver: SecretResolver | None = None, +) -> LLMProvider: + """Instantiate the LLM provider from config.""" + llm_config = config.llm + provider = llm_config.provider + api_key = llm_config.api_key + + if not api_key: + env_var = _PROVIDER_ENV_VARS.get(provider) + if env_var and secret_resolver is not None: + api_key = secret_resolver.resolve(env_var) + if not api_key: + raise LLMError( + f"Missing API key for {provider}.\n" + f"Set 'llm.api_key' in your config file, or export the environment variable:\n" + f' export {_PROVIDER_ENV_VARS.get(provider, "API_KEY")}="your-key"' + ) + + match provider: + case "gemini": + from loafer.llm.gemini import GeminiProvider + + return GeminiProvider(api_key=api_key, model=llm_config.model) + case "claude": + from loafer.llm.claude import ClaudeProvider + + return ClaudeProvider(api_key=api_key, model=llm_config.model) + case "openai": + from loafer.llm.openai import OpenAIProvider + + return OpenAIProvider(api_key=api_key, model=llm_config.model) + case "qwen": + from loafer.llm.qwen import QwenProvider + + return QwenProvider(api_key=api_key, model=llm_config.model) + case _: + available = ", ".join(_PROVIDER_ENV_VARS.keys()) + raise LLMError(f"Unknown LLM provider: {provider!r}.\nSupported providers: {available}") + + +def _build_initial_state( + config: PipelineConfig, + config_path: str | Path | None = None, + full_refresh: bool = False, + run_id: str | None = None, + reviewer: ReviewPort | None = None, +) -> PipelineState: + """Build the initial PipelineState from a validated config. + + When ``config.incremental`` is set, the saved watermark is loaded from the + state file next to *config_path* (unless *full_refresh*), falling back to + ``incremental.initial``. + """ + state_key = config.name or (Path(config_path).stem if config_path else "pipeline") + state_store_path: str | None = None + cursor_value: Any = None + + if config.incremental is not None and config_path is not None: + from loafer.core.incremental import StateStore, state_path_for + + store_path = state_path_for(config_path) + state_store_path = str(store_path) + if not full_refresh: + cursor_value = StateStore(store_path).get_cursor(state_key) + if cursor_value is None: + cursor_value = config.incremental.initial + + return PipelineState( + source_config=config.source, + target_config=config.target, + transform_config=config.transform, + llm_config=config.llm, + transform_instruction=_get_transform_instruction(config), + mode=config.mode, + chunk_size=config.chunk_size, + streaming_threshold=config.streaming_threshold, + destructive_filter_threshold=config.destructive_filter_threshold, + raw_data=[], + transformed_data=[], + schema_sample={}, + validation_report={}, + validation_passed=False, + max_null_rate=config.validation.max_null_rate, + strict_validation=config.validation.strict, + generated_code="", + retry_count=0, + transform_retry_count=0, + last_error=None, + token_usage={}, + raw_table_name=None, + generated_sql=None, + run_id=run_id or uuid.uuid4().hex[:12], + rows_extracted=0, + rows_loaded=0, + duration_ms={}, + warnings=[], + is_streaming=False, + stream_iterator=None, + destructive_warnings=[], + auto_confirmed=False, + incremental_config=config.incremental, + cursor_value=cursor_value, + new_cursor=cursor_value, + state_key=state_key, + state_store_path=state_store_path, + sandbox_config=config.sandbox, + reviewer=reviewer, + ) + + +def _get_transform_instruction(config: PipelineConfig) -> str: + """Extract the transform instruction from the config.""" + transform = config.transform + if hasattr(transform, "instruction"): + return transform.instruction + if hasattr(transform, "path"): + return transform.path + if hasattr(transform, "query"): + return transform.query + return "" + + +def execute_pipeline( + config: PipelineConfig, + *, + config_path: str | Path | None = None, + dry_run: bool = False, + auto_confirm: bool = False, + full_refresh: bool = False, + run_id: str | None = None, + reviewer: ReviewPort | None = None, + secret_resolver: SecretResolver | None = None, + provider_factory: ProviderFactory | None = None, +) -> PipelineState: + """Execute a validated ETL or ELT pipeline. + + Args: + config: Validated pipeline configuration. + config_path: Original path, used for incremental cursor state. + dry_run: If True, stop after transform without loading to target. + auto_confirm: If True, skip destructive operation confirmations. + full_refresh: If True, ignore a saved incremental cursor. + run_id: Caller-assigned run identifier. + reviewer: Human-review port for generated code. + secret_resolver: Resolver for provider credential references. + + Returns: + The final PipelineState after pipeline execution. + + Raises: + PipelineError: If any stage of the pipeline fails. + """ + start = time.monotonic() + + state = _build_initial_state( + config, + config_path, + full_refresh, + run_id=run_id, + reviewer=reviewer, + ) + state["auto_confirmed"] = auto_confirm + + if _transform_requires_llm(config): + factory = provider_factory or _build_llm_provider + state["llm_provider"] = factory(config, secret_resolver) + + mode = config.mode + + if mode == "etl": + graph = _build_etl_graph() + elif mode == "elt": + graph = _build_elt_graph() + else: + raise PipelineError(f"Unknown pipeline mode: {mode}") + + try: + if dry_run: + state = _run_dry_run(graph, state, mode) + else: + state = graph.invoke(state, config=_GRAPH_CONFIG) + _raise_on_terminal_failure(state, mode) + except Exception as exc: + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + _cleanup_source_connector(state) + _cleanup_elt_staging(state, mode) + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + _cleanup_source_connector(state) + _cleanup_elt_staging(state, mode) + + if not dry_run: + _persist_cursor(state) + + return state + + +def stream_pipeline( + config: PipelineConfig, + *, + config_path: str | Path | None = None, + dry_run: bool = False, + auto_confirm: bool = False, + full_refresh: bool = False, + run_id: str | None = None, + reviewer: ReviewPort | None = None, + secret_resolver: SecretResolver | None = None, + provider_factory: ProviderFactory | None = None, +) -> Iterator[tuple[str, str, PipelineState]]: + """Execute a validated pipeline and yield per-stage runtime updates. + + Yields: + ("extract", "done"|"failed", state) — after extraction completes + ("validate", "done"|"failed"|"skipped", state) — after validation + ("transform", "done"|"failed"|"skipped", state) — after transform + ("load", "done"|"failed"|"skipped", state) — after load + + The final yield always has the complete state. Raises PipelineError on + any stage failure. + """ + start = time.monotonic() + + state = _build_initial_state( + config, + config_path, + full_refresh, + run_id=run_id, + reviewer=reviewer, + ) + state["auto_confirmed"] = auto_confirm + + if _transform_requires_llm(config): + factory = provider_factory or _build_llm_provider + state["llm_provider"] = factory(config, secret_resolver) + + mode = config.mode + + if mode == "etl": + graph = _build_etl_graph() + elif mode == "elt": + graph = _build_elt_graph() + else: + raise PipelineError(f"Unknown pipeline mode: {mode}") + + try: + if dry_run: + yield from _stream_dry_run(graph, state, mode, start) + else: + yield from _stream_graph(graph, state, mode, start) + except PipelineError: + _cleanup_source_connector(state) + _cleanup_elt_staging(state, mode) + raise + except Exception as exc: + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + _cleanup_source_connector(state) + _cleanup_elt_staging(state, mode) + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + else: + _cleanup_source_connector(state) + _cleanup_elt_staging(state, mode) + if not dry_run: + _persist_cursor(state) + + +def _stream_graph( + graph: Any, + state: PipelineState, + mode: str, + start: float, +) -> Iterator[tuple[str, str, PipelineState]]: + """Stream graph execution, yielding per-node updates.""" + nodes_executed: set[str] = set() + + stage_order = ( + ["extract", "validate", "transform", "load"] + if mode == "etl" + else ["extract", "load_raw", "transform_in_target"] + ) + + # Yield "running" before the first stage starts + if stage_order: + yield (stage_order[0], "running", state) + + try: + for event in graph.stream(state, stream_mode="updates", config=_GRAPH_CONFIG): + for node_name, delta in event.items(): + nodes_executed.add(node_name) + + # Merge delta into state + for key, value in delta.items(): + state[key] = value # type: ignore[literal-required] + + # Yield "done" for completed stages + if node_name in ( + "extract", + "validate", + "transform", + "load", + "load_raw", + "transform_in_target", + ): + if node_name == "transform_in_target" and state.get("last_error"): + continue + yield (node_name, "done", state) + + # Yield "running" for the next expected stage + next_stages = [s for s in stage_order if s not in nodes_executed] + if next_stages: + yield (next_stages[0], "running", state) + + # Mark skipped stages + expected = set(stage_order) + + for stage in expected - nodes_executed: + yield (stage, "skipped", state) + + if mode == "elt" and state.get("last_error"): + yield ("transform_in_target", "failed", state) + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + raise PipelineError( + f"Pipeline failed (run_id={state['run_id']}): {state['last_error']}" + ) + + except PipelineError: + raise + except Exception as exc: + # The stage that failed is the next expected stage that hasn't completed + failed_stage = next( + (s for s in stage_order if s not in nodes_executed), + _last_executed_node(nodes_executed, mode), + ) + if failed_stage: + yield (failed_stage, "failed", state) + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + else: + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + + +def _stream_dry_run( + graph: Any, + state: PipelineState, + mode: str, + start: float, +) -> Iterator[tuple[str, str, PipelineState]]: + """Stream dry-run graph execution.""" + from langgraph.graph import END, START, StateGraph + + from loafer.agents.extract import extract_agent + from loafer.agents.transform import transform_agent + from loafer.agents.validate import validate_agent + + dry_graph = StateGraph(state_schema=PipelineState) + dry_graph.add_node("extract", extract_agent) + dry_graph.add_node("validate", validate_agent) + dry_graph.add_node("transform", transform_agent) + dry_graph.add_edge(START, "extract") + dry_graph.add_edge("extract", "validate") + + def _check_validation_dry(state: PipelineState) -> str: + if state.get("validation_passed", False): + return "transform" + return "end" + + dry_graph.add_conditional_edges( + "validate", + _check_validation_dry, + {"transform": "transform", "end": END}, + ) + dry_graph.add_edge("transform", END) + + compiled = dry_graph.compile() + nodes_executed: set[str] = set() + + try: + for event in compiled.stream(state, stream_mode="updates"): # type: ignore[arg-type] + for node_name, delta in event.items(): + nodes_executed.add(node_name) + for key, value in delta.items(): + state[key] = value # type: ignore[literal-required] + yield (node_name, "done", state) + + for stage in {"extract", "validate", "transform"} - nodes_executed: + yield (stage, "skipped", state) + + except Exception as exc: + failed_stage = _last_executed_node(nodes_executed, "etl") + if failed_stage: + yield (failed_stage, "failed", state) + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc + else: + total_ms = (time.monotonic() - start) * 1000 + state["duration_ms"]["total"] = total_ms + + +def _last_executed_node(nodes_executed: set[str], mode: str) -> str | None: + """Return the last node that was executed, for error reporting.""" + if mode == "etl": + order = ["extract", "validate", "transform", "load"] + else: + order = ["extract", "load_raw", "transform_in_target"] + for node in reversed(order): + if node in nodes_executed: + return node + return None + + +def _run_dry_run(graph: Any, state: PipelineState, mode: str) -> PipelineState: + """Run pipeline without the final load step.""" + from langgraph.graph import END, START, StateGraph + + from loafer.agents.extract import extract_agent + from loafer.agents.transform import transform_agent + from loafer.agents.validate import validate_agent + + dry_graph = StateGraph(state_schema=PipelineState) + dry_graph.add_node("extract", extract_agent) + dry_graph.add_node("validate", validate_agent) + dry_graph.add_node("transform", transform_agent) + + dry_graph.add_edge(START, "extract") + dry_graph.add_edge("extract", "validate") + + def _check_validation_dry(state: PipelineState) -> str: + if state.get("validation_passed", False): + return "transform" + return "end" + + dry_graph.add_conditional_edges( + "validate", + _check_validation_dry, + {"transform": "transform", "end": END}, + ) + dry_graph.add_edge("transform", END) + + compiled = dry_graph.compile() + return compiled.invoke(state) # type: ignore[arg-type, return-value] + + +def _build_etl_graph() -> Any: + """Build the ETL graph.""" + from loafer.graph.etl import build_etl_graph + + return build_etl_graph() + + +def _build_elt_graph() -> Any: + """Build the ELT graph.""" + from loafer.graph.elt import build_elt_graph + + return build_elt_graph() + + +def _persist_cursor(state: PipelineState) -> None: + """Advance the saved watermark after a successful incremental run.""" + if state.get("incremental_config") is None: + return + store_path = state.get("state_store_path") + new_cursor = state.get("new_cursor") + if not store_path or new_cursor is None: + return + + from loafer.core.incremental import StateStore + + StateStore(store_path).set_cursor(state["state_key"], new_cursor) + + +def _transform_requires_llm(config: PipelineConfig) -> bool: + """Return whether the top-level transform or any pipeline step uses AI.""" + transform = config.transform + if transform.type == "ai": + return not transform.bypass_ai + if transform.type == "pipeline": + return any(step.type == "ai" and not step.bypass_ai for step in transform.steps) + return False + + +def _cleanup_source_connector(state: PipelineState) -> None: + """Disconnect the source connector if it was kept alive for streaming.""" + connector = state.get("_source_connector") + if connector is not None: + try: + connector.disconnect() + except Exception: + pass + state["_source_connector"] = None + + +def _raise_on_terminal_failure(state: PipelineState, mode: str) -> None: + """Raise when a graph exhausted retries but returned an error state.""" + if mode == "elt" and state.get("last_error"): + raise PipelineError(str(state["last_error"])) + + +def _cleanup_elt_staging(state: PipelineState, mode: str) -> None: + """Best-effort removal of the per-run PostgreSQL staging table.""" + if mode != "elt": + return + + raw_table = state.get("raw_table_name") + target_config = state.get("target_config") + if not raw_table or not hasattr(target_config, "url"): + return + + conn: Any | None = None + cursor: Any | None = None + try: + import psycopg2 + from psycopg2 import sql as pg_sql + + from loafer.adapters.postgres_sql import qualified_identifier + + conn = psycopg2.connect(target_config.url) + cursor = conn.cursor() + cursor.execute( + pg_sql.SQL("DROP TABLE IF EXISTS {}").format(qualified_identifier(raw_table)) + ) + conn.commit() + except Exception as exc: + if conn is not None: + try: + conn.rollback() + except Exception: + pass + state.setdefault("warnings", []).append( + f"Could not remove ELT staging table '{raw_table}': {exc}" + ) + finally: + if cursor is not None: + cursor.close() + if conn is not None: + conn.close() diff --git a/loafer/graph/state.py b/loafer/graph/state.py index 51d8937..cc713a3 100644 --- a/loafer/graph/state.py +++ b/loafer/graph/state.py @@ -1,7 +1,11 @@ -"""PipelineState — the single source of truth for all data flowing through the system. +"""Ephemeral in-process state for the current LangGraph execution. Every agent receives this state, operates on it, and returns an updated copy. LangGraph nodes must return updated state, never mutate in place. + +This is deliberately not a persistence contract: it may contain live +connectors, iterators, providers, and review callbacks. Durable clients use +the sanitized contracts in :mod:`loafer.application.contracts`. """ from __future__ import annotations @@ -55,6 +59,7 @@ class PipelineState(TypedDict, total=False): # LLM llm_provider: Any + reviewer: Any generated_code: str retry_count: int last_error: str | None diff --git a/loafer/ports/runtime.py b/loafer/ports/runtime.py new file mode 100644 index 0000000..86c7158 --- /dev/null +++ b/loafer/ports/runtime.py @@ -0,0 +1,45 @@ +"""Runtime ports shared by the engine and application service.""" + +from __future__ import annotations + +from typing import Protocol + +from loafer.contracts import Checkpoint, RunEvent + + +class CancellationPort(Protocol): + """Report whether a run should stop at the next safe boundary.""" + + def is_cancelled(self, run_id: str) -> bool: + """Return true when cancellation has been requested.""" + + +class CheckpointPort(Protocol): + """Load and commit durable batch checkpoints.""" + + def load(self, run_id: str, partition_id: str) -> Checkpoint | None: + """Return the latest committed checkpoint for a partition.""" + + def save(self, checkpoint: Checkpoint) -> None: + """Persist a checkpoint after its target effect is durable.""" + + +class SecretResolver(Protocol): + """Resolve a server-side secret reference without exposing it to clients.""" + + def resolve(self, reference: str) -> str | None: + """Return a secret value, or ``None`` when the reference is absent.""" + + +class EventPublisher(Protocol): + """Publish sanitized application events.""" + + def publish(self, event: RunEvent) -> None: + """Publish one monotonically sequenced run event.""" + + +class ReviewPort(Protocol): + """Approve or reject generated transform code before execution.""" + + def approve_transform(self, generated_code: str) -> bool: + """Return true only when the candidate may execute.""" diff --git a/loafer/runner.py b/loafer/runner.py index 5a2f098..3278052 100644 --- a/loafer/runner.py +++ b/loafer/runner.py @@ -1,157 +1,31 @@ -"""Pipeline runner — composition root. +"""Backward-compatible local runner facade. -Parses config, builds state, instantiates LLM provider, selects the -correct graph (ETL or ELT), and invokes it. +New clients should use :mod:`loafer.application`. These functions preserve +the pre-Phase 1 Python API while delegating orchestration to the same +application use cases used by the CLI and scheduler. """ from __future__ import annotations -import os -import time -import uuid from collections.abc import Iterator from pathlib import Path -from typing import TYPE_CHECKING, Any -from loafer.config import PipelineConfig, load_config -from loafer.exceptions import LLMError, PipelineError +from loafer.adapters.runtime import EnvironmentSecretResolver +from loafer.application import RunRequest, get_local_application +from loafer.config import PipelineConfig +from loafer.engine import ( + _build_initial_state, + _raise_on_terminal_failure, + _transform_requires_llm, +) +from loafer.engine import ( + _build_llm_provider as _engine_build_llm_provider, +) from loafer.graph.state import PipelineState +from loafer.ports.llm import LLMProvider +from loafer.ports.runtime import SecretResolver -if TYPE_CHECKING: - from collections.abc import Iterator - - from loafer.llm.base import LLMProvider - -_PROVIDER_ENV_VARS = { - "gemini": "GEMINI_API_KEY", - "claude": "ANTHROPIC_API_KEY", - "openai": "OPENAI_API_KEY", - "qwen": "DASHSCOPE_API_KEY", -} - -# Hard backstop on graph hops. The ELT graph caps its own retries via the -# transform_in_target counter, but a stuck conditional edge could still loop; -# this ensures LangGraph raises GraphRecursionError instead of spinning -# forever (the original BUG-3 symptom under graph.stream). -_GRAPH_CONFIG = {"recursion_limit": 25} - - -def _build_llm_provider(config: PipelineConfig) -> LLMProvider: - """Instantiate the LLM provider from config.""" - llm_config = config.llm - provider = llm_config.provider - api_key = llm_config.api_key - - if not api_key: - env_var = _PROVIDER_ENV_VARS.get(provider) - if env_var: - api_key = os.environ.get(env_var, "") - if not api_key: - raise LLMError( - f"Missing API key for {provider}.\n" - f"Set 'llm.api_key' in your config file, or export the environment variable:\n" - f' export {_PROVIDER_ENV_VARS.get(provider, "API_KEY")}="your-key"' - ) - - match provider: - case "gemini": - from loafer.llm.gemini import GeminiProvider - - return GeminiProvider(api_key=api_key, model=llm_config.model) - case "claude": - from loafer.llm.claude import ClaudeProvider - - return ClaudeProvider(api_key=api_key, model=llm_config.model) - case "openai": - from loafer.llm.openai import OpenAIProvider - - return OpenAIProvider(api_key=api_key, model=llm_config.model) - case "qwen": - from loafer.llm.qwen import QwenProvider - - return QwenProvider(api_key=api_key, model=llm_config.model) - case _: - available = ", ".join(_PROVIDER_ENV_VARS.keys()) - raise LLMError(f"Unknown LLM provider: {provider!r}.\nSupported providers: {available}") - - -def _build_initial_state( - config: PipelineConfig, - config_path: str | Path | None = None, - full_refresh: bool = False, -) -> PipelineState: - """Build the initial PipelineState from a validated config. - - When ``config.incremental`` is set, the saved watermark is loaded from the - state file next to *config_path* (unless *full_refresh*), falling back to - ``incremental.initial``. - """ - state_key = config.name or (Path(config_path).stem if config_path else "pipeline") - state_store_path: str | None = None - cursor_value: Any = None - - if config.incremental is not None and config_path is not None: - from loafer.core.incremental import StateStore, state_path_for - - store_path = state_path_for(config_path) - state_store_path = str(store_path) - if not full_refresh: - cursor_value = StateStore(store_path).get_cursor(state_key) - if cursor_value is None: - cursor_value = config.incremental.initial - - return PipelineState( - source_config=config.source, - target_config=config.target, - transform_config=config.transform, - llm_config=config.llm, - transform_instruction=_get_transform_instruction(config), - mode=config.mode, - chunk_size=config.chunk_size, - streaming_threshold=config.streaming_threshold, - destructive_filter_threshold=config.destructive_filter_threshold, - raw_data=[], - transformed_data=[], - schema_sample={}, - validation_report={}, - validation_passed=False, - max_null_rate=config.validation.max_null_rate, - strict_validation=config.validation.strict, - generated_code="", - retry_count=0, - transform_retry_count=0, - last_error=None, - token_usage={}, - raw_table_name=None, - generated_sql=None, - run_id=uuid.uuid4().hex[:12], - rows_extracted=0, - rows_loaded=0, - duration_ms={}, - warnings=[], - is_streaming=False, - stream_iterator=None, - destructive_warnings=[], - auto_confirmed=False, - incremental_config=config.incremental, - cursor_value=cursor_value, - new_cursor=cursor_value, - state_key=state_key, - state_store_path=state_store_path, - sandbox_config=config.sandbox, - ) - - -def _get_transform_instruction(config: PipelineConfig) -> str: - """Extract the transform instruction from the config.""" - transform = config.transform - if hasattr(transform, "instruction"): - return transform.instruction - if hasattr(transform, "path"): - return transform.path - if hasattr(transform, "query"): - return transform.query - return "" +_build_llm_provider = _engine_build_llm_provider def run_pipeline( @@ -161,62 +35,18 @@ def run_pipeline( yes: bool = False, full_refresh: bool = False, ) -> PipelineState: - """Run a full ETL or ELT pipeline from a YAML config file. - - Args: - config_path: Path to the pipeline YAML config. - dry_run: If True, stop after transform without loading to target. - verbose: If True, print detailed agent output. - yes: If True, skip destructive operation confirmations. - - Returns: - The final PipelineState after pipeline execution. - - Raises: - PipelineError: If any stage of the pipeline fails. - """ - start = time.monotonic() - - config = load_config(config_path) - state = _build_initial_state(config, config_path, full_refresh) - state["auto_confirmed"] = yes - - if _transform_requires_llm(config): - state["llm_provider"] = _build_llm_provider(config) - - mode = config.mode - - if mode == "etl": - graph = _build_etl_graph() - elif mode == "elt": - graph = _build_elt_graph() - else: - raise PipelineError(f"Unknown pipeline mode: {mode}") - - try: - if dry_run: - state = _run_dry_run(graph, state, mode) - else: - state = graph.invoke(state, config=_GRAPH_CONFIG) - _raise_on_terminal_failure(state, mode) - except Exception as exc: - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - _cleanup_source_connector(state) - _cleanup_elt_staging(state, mode) - raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc - - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - _cleanup_source_connector(state) - _cleanup_elt_staging(state, mode) - - if not dry_run: - _persist_cursor(state) - + """Run a pipeline and return its legacy in-process state.""" + request = RunRequest( + config_path=str(config_path), + dry_run=dry_run, + auto_confirm=yes, + full_refresh=full_refresh, + ) + state = get_local_application(provider_factory=_compat_provider_factory).run_pipeline.run_state( + request + ) if verbose: - _print_summary(state) - + _print_legacy_summary(state) return state @@ -226,239 +56,47 @@ def run_pipeline_streaming( yes: bool = False, full_refresh: bool = False, ) -> Iterator[tuple[str, str, PipelineState]]: - """Run pipeline and yield (stage_name, status, state) per completed node. - - Yields: - ("extract", "done"|"failed", state) — after extraction completes - ("validate", "done"|"failed"|"skipped", state) — after validation - ("transform", "done"|"failed"|"skipped", state) — after transform - ("load", "done"|"failed"|"skipped", state) — after load - - The final yield always has the complete state. Raises PipelineError on - any stage failure. - """ - start = time.monotonic() - - config = load_config(config_path) - state = _build_initial_state(config, config_path, full_refresh) - state["auto_confirmed"] = yes - - if _transform_requires_llm(config): - state["llm_provider"] = _build_llm_provider(config) - - mode = config.mode - - if mode == "etl": - graph = _build_etl_graph() - elif mode == "elt": - graph = _build_elt_graph() - else: - raise PipelineError(f"Unknown pipeline mode: {mode}") - - try: - if dry_run: - yield from _stream_dry_run(graph, state, mode, start) - else: - yield from _stream_graph(graph, state, mode, start) - except PipelineError: - _cleanup_source_connector(state) - _cleanup_elt_staging(state, mode) - raise - except Exception as exc: - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - _cleanup_source_connector(state) - _cleanup_elt_staging(state, mode) - raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc - else: - _cleanup_source_connector(state) - _cleanup_elt_staging(state, mode) - if not dry_run: - _persist_cursor(state) - - -def _stream_graph( - graph: Any, - state: PipelineState, - mode: str, - start: float, -) -> Iterator[tuple[str, str, PipelineState]]: - """Stream graph execution, yielding per-node updates.""" - nodes_executed: set[str] = set() - - stage_order = ( - ["extract", "validate", "transform", "load"] - if mode == "etl" - else ["extract", "load_raw", "transform_in_target"] + """Yield legacy stage tuples through the application use case.""" + request = RunRequest( + config_path=str(config_path), + dry_run=dry_run, + auto_confirm=yes, + full_refresh=full_refresh, ) + service = get_local_application(provider_factory=_compat_provider_factory) + for event, state in service.run_pipeline.stream_states(request): + yield event.stage, event.status, state - # Yield "running" before the first stage starts - if stage_order: - yield (stage_order[0], "running", state) - - try: - for event in graph.stream(state, stream_mode="updates", config=_GRAPH_CONFIG): - for node_name, delta in event.items(): - nodes_executed.add(node_name) - # Merge delta into state - for key, value in delta.items(): - state[key] = value # type: ignore[literal-required] - - # Yield "done" for completed stages - if node_name in ( - "extract", - "validate", - "transform", - "load", - "load_raw", - "transform_in_target", - ): - if node_name == "transform_in_target" and state.get("last_error"): - continue - yield (node_name, "done", state) - - # Yield "running" for the next expected stage - next_stages = [s for s in stage_order if s not in nodes_executed] - if next_stages: - yield (next_stages[0], "running", state) +def validate_config(config_path: str | Path) -> PipelineConfig: + """Validate a config and return the legacy Pydantic model.""" + return get_local_application().validate_config_model(config_path) - # Mark skipped stages - expected = set(stage_order) - for stage in expected - nodes_executed: - yield (stage, "skipped", state) +def list_connectors() -> dict[str, list[str]]: + """Return the legacy connector catalog mapping.""" + catalog = get_local_application().list_connectors() + return { + "sources": list(catalog.sources), + "targets": list(catalog.targets), + } - if mode == "elt" and state.get("last_error"): - yield ("transform_in_target", "failed", state) - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - raise PipelineError( - f"Pipeline failed (run_id={state['run_id']}): {state['last_error']}" - ) - except PipelineError: - raise - except Exception as exc: - # The stage that failed is the next expected stage that hasn't completed - failed_stage = next( - (s for s in stage_order if s not in nodes_executed), - _last_executed_node(nodes_executed, mode), +def _compat_provider_factory( + config: PipelineConfig, + secret_resolver: SecretResolver | None, +) -> LLMProvider: + """Keep monkeypatching the legacy provider factory effective.""" + if _build_llm_provider is _engine_build_llm_provider: + return _build_llm_provider( + config, + secret_resolver or EnvironmentSecretResolver(), ) - if failed_stage: - yield (failed_stage, "failed", state) - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc - else: - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - - -def _stream_dry_run( - graph: Any, - state: PipelineState, - mode: str, - start: float, -) -> Iterator[tuple[str, str, PipelineState]]: - """Stream dry-run graph execution.""" - from langgraph.graph import END, START, StateGraph + return _build_llm_provider(config) - from loafer.agents.extract import extract_agent - from loafer.agents.transform import transform_agent - from loafer.agents.validate import validate_agent - dry_graph = StateGraph(state_schema=PipelineState) - dry_graph.add_node("extract", extract_agent) - dry_graph.add_node("validate", validate_agent) - dry_graph.add_node("transform", transform_agent) - dry_graph.add_edge(START, "extract") - dry_graph.add_edge("extract", "validate") - - def _check_validation_dry(state: PipelineState) -> str: - if state.get("validation_passed", False): - return "transform" - return "end" - - dry_graph.add_conditional_edges( - "validate", - _check_validation_dry, - {"transform": "transform", "end": END}, - ) - dry_graph.add_edge("transform", END) - - compiled = dry_graph.compile() - nodes_executed: set[str] = set() - - try: - for event in compiled.stream(state, stream_mode="updates"): # type: ignore[arg-type] - for node_name, delta in event.items(): - nodes_executed.add(node_name) - for key, value in delta.items(): - state[key] = value # type: ignore[literal-required] - yield (node_name, "done", state) - - for stage in {"extract", "validate", "transform"} - nodes_executed: - yield (stage, "skipped", state) - - except Exception as exc: - failed_stage = _last_executed_node(nodes_executed, "etl") - if failed_stage: - yield (failed_stage, "failed", state) - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - raise PipelineError(f"Pipeline failed (run_id={state['run_id']}): {exc}") from exc - else: - total_ms = (time.monotonic() - start) * 1000 - state["duration_ms"]["total"] = total_ms - - -def _last_executed_node(nodes_executed: set[str], mode: str) -> str | None: - """Return the last node that was executed, for error reporting.""" - if mode == "etl": - order = ["extract", "validate", "transform", "load"] - else: - order = ["extract", "load_raw", "transform_in_target"] - for node in reversed(order): - if node in nodes_executed: - return node - return None - - -def _run_dry_run(graph: Any, state: PipelineState, mode: str) -> PipelineState: - """Run pipeline without the final load step.""" - from langgraph.graph import END, START, StateGraph - - from loafer.agents.extract import extract_agent - from loafer.agents.transform import transform_agent - from loafer.agents.validate import validate_agent - - dry_graph = StateGraph(state_schema=PipelineState) - dry_graph.add_node("extract", extract_agent) - dry_graph.add_node("validate", validate_agent) - dry_graph.add_node("transform", transform_agent) - - dry_graph.add_edge(START, "extract") - dry_graph.add_edge("extract", "validate") - - def _check_validation_dry(state: PipelineState) -> str: - if state.get("validation_passed", False): - return "transform" - return "end" - - dry_graph.add_conditional_edges( - "validate", - _check_validation_dry, - {"transform": "transform", "end": END}, - ) - dry_graph.add_edge("transform", END) - - compiled = dry_graph.compile() - return compiled.invoke(state) # type: ignore[arg-type, return-value] - - -def _print_summary(state: PipelineState) -> None: - """Print a summary of the pipeline run.""" +def _print_legacy_summary(state: PipelineState) -> None: + """Preserve ``run_pipeline(verbose=True)`` as a client-side concern.""" from rich.console import Console console = Console() @@ -466,140 +104,18 @@ def _print_summary(state: PipelineState) -> None: console.print(f" Rows extracted: {state.get('rows_extracted', 0)}") console.print(f" Rows loaded: {state.get('rows_loaded', 0)}") console.print(f" Warnings: {len(state.get('warnings', []))}") - if state.get("token_usage"): console.print(f" Token usage: {state.get('token_usage', {})}") - console.print(f" Duration: {state.get('duration_ms', {}).get('total', 0):.0f}ms") - if state.get("warnings"): - console.print("\n[yellow]Warnings:[/yellow]") - for w in state["warnings"]: - console.print(f" - {w}") - - -def _build_etl_graph() -> Any: - """Build the ETL graph.""" - from loafer.graph.etl import build_etl_graph - - return build_etl_graph() - - -def _build_elt_graph() -> Any: - """Build the ELT graph.""" - from loafer.graph.elt import build_elt_graph - - return build_elt_graph() - -def validate_config(config_path: str | Path) -> PipelineConfig: - """Validate a pipeline config file without running it. - - Args: - config_path: Path to the pipeline YAML config. - - Returns: - The validated PipelineConfig. - - Raises: - PipelineError: If the config is invalid. - """ - try: - return load_config(config_path) - except Exception as exc: - raise PipelineError(f"Config validation failed: {exc}") from exc - - -def list_connectors() -> dict[str, list[str]]: - """List all registered source and target connectors. - - Returns: - Dict with 'sources' and 'targets' keys listing connector types. - """ - from loafer.connectors.registry import _SOURCE_REGISTRY, _TARGET_REGISTRY - - return { - "sources": sorted(_SOURCE_REGISTRY.keys()), - "targets": sorted(_TARGET_REGISTRY.keys()), - } - - -def _persist_cursor(state: PipelineState) -> None: - """Advance the saved watermark after a successful incremental run.""" - if state.get("incremental_config") is None: - return - store_path = state.get("state_store_path") - new_cursor = state.get("new_cursor") - if not store_path or new_cursor is None: - return - - from loafer.core.incremental import StateStore - - StateStore(store_path).set_cursor(state["state_key"], new_cursor) - - -def _transform_requires_llm(config: PipelineConfig) -> bool: - """Return whether the top-level transform or any pipeline step uses AI.""" - transform = config.transform - if transform.type == "ai": - return not transform.bypass_ai - if transform.type == "pipeline": - return any(step.type == "ai" and not step.bypass_ai for step in transform.steps) - return False - - -def _cleanup_source_connector(state: PipelineState) -> None: - """Disconnect the source connector if it was kept alive for streaming.""" - connector = state.get("_source_connector") - if connector is not None: - try: - connector.disconnect() - except Exception: - pass - state["_source_connector"] = None - - -def _raise_on_terminal_failure(state: PipelineState, mode: str) -> None: - """Raise when a graph exhausted retries but returned an error state.""" - if mode == "elt" and state.get("last_error"): - raise PipelineError(str(state["last_error"])) - - -def _cleanup_elt_staging(state: PipelineState, mode: str) -> None: - """Best-effort removal of the per-run PostgreSQL staging table.""" - if mode != "elt": - return - - raw_table = state.get("raw_table_name") - target_config = state.get("target_config") - if not raw_table or not hasattr(target_config, "url"): - return - - conn: Any | None = None - cursor: Any | None = None - try: - import psycopg2 - from psycopg2 import sql as pg_sql - - from loafer.adapters.postgres_sql import qualified_identifier - - conn = psycopg2.connect(target_config.url) - cursor = conn.cursor() - cursor.execute( - pg_sql.SQL("DROP TABLE IF EXISTS {}").format(qualified_identifier(raw_table)) - ) - conn.commit() - except Exception as exc: - if conn is not None: - try: - conn.rollback() - except Exception: - pass - state.setdefault("warnings", []).append( - f"Could not remove ELT staging table '{raw_table}': {exc}" - ) - finally: - if cursor is not None: - cursor.close() - if conn is not None: - conn.close() +__all__ = [ + "_build_initial_state", + "_build_llm_provider", + "_raise_on_terminal_failure", + "_transform_requires_llm", + "list_connectors", + "run_pipeline", + "run_pipeline_streaming", + "validate_config", +] diff --git a/loafer/scheduler.py b/loafer/scheduler.py index fb3a13a..93a0ac2 100644 --- a/loafer/scheduler.py +++ b/loafer/scheduler.py @@ -17,8 +17,8 @@ from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore from apscheduler.schedulers.background import BackgroundScheduler +from loafer.application import RunRequest, get_local_application from loafer.exceptions import SchedulerError -from loafer.runner import run_pipeline logger = logging.getLogger("loafer.scheduler") @@ -102,7 +102,7 @@ def _run_pipeline_job(config_path: str, name: str = "") -> None: display = f"{name} ({config_path})" if name else config_path logger.info("Starting scheduled run %s for %s", run_id, display) try: - run_pipeline(config_path=config_path, verbose=False) + get_local_application().run_pipeline.run(RunRequest(config_path=config_path, run_id=run_id)) logger.info("Completed scheduled run %s", run_id) except Exception as exc: logger.error("Scheduled run %s failed: %s", run_id, exc) diff --git a/loafer/transform/ai_runner.py b/loafer/transform/ai_runner.py index 4336443..3a2a6e8 100644 --- a/loafer/transform/ai_runner.py +++ b/loafer/transform/ai_runner.py @@ -145,33 +145,13 @@ def _execute_code( def _ask_user_confirmation(generated_code: str) -> bool: - """Show generated code and ask user to confirm execution.""" - from rich.console import Console - from rich.panel import Panel - from rich.syntax import Syntax - - console = Console() - - console.print() - console.print( - Panel( - "[yellow]AI-generated transform code is ready for review.[/yellow]\n" - "Review the code below. If it looks correct, type 'y' to execute.\n" - "Type 'n' to skip AI transform (custom transform will still run if configured).", - title="[bold yellow]⚠ Human Review Required[/bold yellow]", - ) - ) - console.print() - - # Show the code with syntax highlighting - syntax = Syntax(generated_code, "python", theme="monokai", line_numbers=True) - console.print(syntax) - console.print() + """Portable fallback for direct engine use without a review adapter.""" + print("\nAI-generated transform code:\n") + print(generated_code) try: answer = input("Execute this code? [y/N]: ").strip().lower() except (EOFError, KeyboardInterrupt): - console.print("\n[dim]No input received. Skipping AI transform.[/dim]") return False return answer in ("y", "yes") @@ -232,11 +212,18 @@ def run(self, state: PipelineState) -> PipelineState: ) # Human review if requested - if transform_config.review and not _ask_user_confirmation(ai_code): - # User rejected — skip AI, keep custom result if any - state["transformed_data"] = data - state["duration_ms"]["transform"] = (time.monotonic() - start) * 1000 - return state + if transform_config.review: + reviewer = state.get("reviewer") + approved = ( + reviewer.approve_transform(ai_code) + if reviewer is not None + else _ask_user_confirmation(ai_code) + ) + if not approved: + # User rejected — skip AI, keep custom result if any + state["transformed_data"] = data + state["duration_ms"]["transform"] = (time.monotonic() - start) * 1000 + return state try: data = self._run_ai_code(ai_code, data, state) diff --git a/skills/loafer-engineering/SKILL.md b/skills/loafer-engineering/SKILL.md index 4279c9e..61e2b0d 100644 --- a/skills/loafer-engineering/SKILL.md +++ b/skills/loafer-engineering/SKILL.md @@ -38,8 +38,13 @@ boundaries. 4. Read [references/connectors.md](references/connectors.md) before adding or changing a connector. 5. Read [references/product-ui.md](references/product-ui.md) for web or operator workflow work. 6. Inspect `git status --short`. Preserve unrelated and pre-existing changes. -7. Trace the complete path affected by the request: config → composition root → graph/agent → +7. When creating a branch, use a purpose-based work-type prefix: `feat/`, `fix/`, `docs/`, + `refactor/`, `test/`, or `chore/`. Never use an agent/tool prefix such as `agent/`, and never + use roadmap phase names such as `phase-1` or `phase-2`. +8. Trace the complete path affected by the request: config → composition root → graph/agent → port → adapter → state/error/reporting → tests. +9. Before every commit, update the `[Unreleased]` section of `CHANGELOG.md` with the implementation + or documentation change. Never commit first and backfill the changelog entry afterward. Treat every roadmap statement in the references as intent, not proof. Search for its implementation and tests first. diff --git a/skills/loafer-engineering/references/architecture.md b/skills/loafer-engineering/references/architecture.md index d4784ad..2476bfb 100644 --- a/skills/loafer-engineering/references/architecture.md +++ b/skills/loafer-engineering/references/architecture.md @@ -38,33 +38,39 @@ a clearly labeled product preview; it has no authenticated runtime API. ```text YAML → config.load_config - → runner._build_initial_state - → runner selects ETL or ELT LangGraph + → application.RunPipeline creates a credential-free ExecutionPlan + → engine._build_initial_state + → engine selects ETL or ELT LangGraph → extract agent resolves source adapter and samples schema → validate agent applies sample-based checks → ETL: transform runner → load target adapter → ELT: load_raw target adapter → in-target SQL transform - → runner persists incremental cursor after graph completion + → engine persists the local incremental cursor after graph completion + → application emits sanitized RunEvent / RunResult contracts ``` `PipelineState` mixes configuration, data, execution metadata, live iterators, provider objects, and -connectors. It is an in-process coordination object, not a durable workflow record. +connectors. It is explicitly an ephemeral in-process coordination object, not a durable workflow +record. Persistence and client surfaces use the credential-free contracts in `loafer/contracts.py`. ## Module ownership | Area | Ownership | |---|---| | `loafer/config.py` | Pydantic schema, environment substitution, auto-detection | +| `loafer/contracts.py` | Serializable execution plans, batch metadata, events, snapshots, and results | | `loafer/core/` | Destructive-change policy, sandbox process, incremental state | -| `loafer/ports/` | Connector and LLM interfaces | +| `loafer/ports/` | Connector, LLM, cancellation, checkpoint, secret, event, and review interfaces | | `loafer/adapters/` | Database/file/API source and target implementations | | `loafer/connectors/registry.py` | Connector registration and construction | | `loafer/agents/` | LangGraph stage functions | | `loafer/transform/` | AI, Python, SQL, and multi-step execution | | `loafer/graph/` | Separate ETL and ELT topology | -| `loafer/runner.py` | Composition root and invocation | +| `loafer/engine.py` | In-process graph composition and execution | +| `loafer/application/` | Plan, run, validate, and connector-listing use cases | +| `loafer/runner.py` | Backward-compatible Python facade over the application service | | `loafer/cli.py` | Typer/Rich user experience | -| `loafer/scheduler.py`, `daemon.py` | Local APScheduler service lifecycle | +| `loafer/scheduler.py`, `daemon.py` | Local APScheduler lifecycle; runs call the application service | | `web/` | Next.js web control-plane shell, marketing site, and MDX documentation | ## Implemented surface diff --git a/tests/e2e/test_application_pipeline.py b/tests/e2e/test_application_pipeline.py new file mode 100644 index 0000000..4cad15a --- /dev/null +++ b/tests/e2e/test_application_pipeline.py @@ -0,0 +1,59 @@ +"""End-to-end execution through the Phase 1 application interface.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from loafer.application import RunRequest, get_local_application +from loafer.contracts import RunStatus + + +def test_csv_transform_json_runs_through_application_service(tmp_path: Path) -> None: + source = tmp_path / "input.csv" + source.write_text("id,name\n1,Alice\n2,Bob\n", encoding="utf-8") + transform = tmp_path / "transform.py" + transform.write_text( + "def transform(data):\n return [{**row, 'name': row['name'].upper()} for row in data]\n", + encoding="utf-8", + ) + output = tmp_path / "output.json" + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "name: phase-one-vertical-slice", + "source:", + " type: csv", + f" path: {source}", + "target:", + " type: json", + f" path: {output}", + "transform:", + " type: custom", + f" path: {transform}", + "mode: etl", + "chunk_size: 1", + "", + ] + ), + encoding="utf-8", + ) + + result = get_local_application().run_pipeline.run( + RunRequest( + config_path=str(config), + run_id="phase-one-e2e", + auto_confirm=True, + ) + ) + + assert result.status is RunStatus.SUCCEEDED + assert result.snapshot.rows_extracted == 2 + assert result.snapshot.rows_loaded == 2 + assert result.snapshot.source_type == "csv" + assert result.snapshot.target_type == "json" + assert json.loads(output.read_text(encoding="utf-8")) == [ + {"id": "1", "name": "ALICE"}, + {"id": "2", "name": "BOB"}, + ] diff --git a/tests/unit/test_application_contracts.py b/tests/unit/test_application_contracts.py new file mode 100644 index 0000000..021539b --- /dev/null +++ b/tests/unit/test_application_contracts.py @@ -0,0 +1,159 @@ +"""Round-trip tests for durable Phase 1 contracts.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import TypeVar + +from loafer.contracts import ( + BatchEnvelope, + Checkpoint, + ConnectorCatalog, + ExecutionPlan, + RunEvent, + RunRequest, + RunResult, + RunSnapshot, + RunStatus, + StageStatus, + TransformStepResult, + ValidationResult, +) + +ContractT = TypeVar("ContractT") + + +def _round_trip(value: ContractT) -> ContractT: + model_type = type(value) + restored = model_type.model_validate_json(value.model_dump_json()) # type: ignore[attr-defined] + assert restored == value + return restored + + +def _plan() -> ExecutionPlan: + return ExecutionPlan( + plan_id="plan-1", + config_digest="a" * 64, + config_path="/pipelines/orders.yaml", + pipeline_name="orders", + mode="etl", + source_type="csv", + target_type="json", + transform_type="custom", + chunk_size=500, + streaming_threshold=10_000, + validation_strict=False, + llm_provider="openai", + llm_model="gpt-test", + incremental_column="updated_at", + cursor_value="2026-07-30T00:00:00Z", + ) + + +def _snapshot() -> RunSnapshot: + return RunSnapshot( + run_id="run-1", + plan_id="plan-1", + pipeline_name="orders", + mode="etl", + source_type="csv", + target_type="json", + transform_type="pipeline", + rows_extracted=10, + rows_transformed=9, + rows_loaded=9, + validation_passed=True, + duration_ms={"extract": 12.5, "total": 30.0}, + warnings=("one warning",), + token_usage={"total_tokens": 42}, + step_results=( + TransformStepResult( + index=0, + name="normalize", + type="custom", + rows_in=10, + rows_out=9, + duration_ms=4.2, + success=True, + ), + ), + ) + + +def test_all_durable_contracts_round_trip_through_json() -> None: + now = datetime(2026, 7, 30, tzinfo=UTC) + plan = _round_trip(_plan()) + snapshot = _round_trip(_snapshot()) + request = _round_trip( + RunRequest( + config_path="/pipelines/orders.yaml", + run_id="run-1", + dry_run=True, + ) + ) + batch = _round_trip( + BatchEnvelope( + run_id="run-1", + stage_id="transform", + partition_id="partition-1", + batch_id="batch-1", + attempt=0, + source_position_start={"offset": 0}, + source_position_end={"offset": 499}, + schema_version="schema-1", + transform_artifact_version="transform-1", + rows_in=500, + rows_out=490, + rows_rejected=10, + bytes_in=4096, + bytes_out=3900, + checksum="sha256:example", + ) + ) + checkpoint = _round_trip( + Checkpoint( + checkpoint_id="checkpoint-1", + run_id="run-1", + partition_id="partition-1", + batch_id="batch-1", + source_position={"offset": 499}, + committed_at=now, + ) + ) + event = _round_trip( + RunEvent( + run_id="run-1", + plan_id="plan-1", + sequence=1, + stage="extract", + status=StageStatus.DONE, + occurred_at=now, + snapshot=snapshot, + ) + ) + result = _round_trip( + RunResult( + run_id="run-1", + plan_id="plan-1", + status=RunStatus.SUCCEEDED, + started_at=now, + finished_at=now, + output_published=True, + snapshot=snapshot, + ) + ) + validation = _round_trip(ValidationResult(plan=plan)) + catalog = _round_trip(ConnectorCatalog(sources=("csv", "postgres"), targets=("json",))) + + assert request.run_id == batch.run_id == checkpoint.run_id == event.run_id == result.run_id + assert validation.valid is True + assert catalog.sources == ("csv", "postgres") + + +def test_execution_plan_contains_no_credentials_or_runtime_objects() -> None: + rendered = _plan().model_dump_json() + assert "api_key" not in rendered + assert "password" not in rendered + assert "connector" not in rendered + assert "iterator" not in rendered + assert "provider" in rendered # Provider name is metadata, not a live provider object. diff --git a/tests/unit/test_application_service.py b/tests/unit/test_application_service.py new file mode 100644 index 0000000..eb2c117 --- /dev/null +++ b/tests/unit/test_application_service.py @@ -0,0 +1,155 @@ +"""Tests for the Phase 1 local application boundary.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from loafer.adapters.runtime import ( + EnvironmentSecretResolver, + InputReviewPort, + NullCheckpointStore, + NullEventPublisher, +) +from loafer.application import RunRequest, get_local_application +from loafer.application.service import RunPipeline +from loafer.contracts import RunStatus, StageStatus +from loafer.exceptions import PipelineError + + +def _pipeline_fixture(tmp_path: Path) -> tuple[Path, Path]: + source = tmp_path / "input.csv" + source.write_text("id,name\n1,Alice\n2,Bob\n", encoding="utf-8") + transform = tmp_path / "transform.py" + transform.write_text( + "def transform(data):\n return [{**row, 'name': row['name'].lower()} for row in data]\n", + encoding="utf-8", + ) + output = tmp_path / "output.json" + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "name: application-boundary", + "source:", + " type: csv", + f" path: {source}", + "target:", + " type: json", + f" path: {output}", + "transform:", + " type: custom", + f" path: {transform}", + "mode: etl", + "chunk_size: 10", + "", + ] + ), + encoding="utf-8", + ) + return config, output + + +def test_csv_to_json_executes_through_application_interface(tmp_path: Path) -> None: + config, output = _pipeline_fixture(tmp_path) + request = RunRequest(config_path=str(config), run_id="application-e2e", auto_confirm=True) + + result = get_local_application().run_pipeline.run(request) + + assert result.status is RunStatus.SUCCEEDED + assert result.run_id == "application-e2e" + assert result.snapshot.rows_extracted == 2 + assert result.snapshot.rows_transformed == 2 + assert result.snapshot.rows_loaded == 2 + assert result.output_published is True + assert json.loads(output.read_text(encoding="utf-8"))[0]["name"] == "alice" + + +def test_stream_emits_monotonic_serializable_events(tmp_path: Path) -> None: + config, _output = _pipeline_fixture(tmp_path) + request = RunRequest(config_path=str(config), run_id="stream-e2e", auto_confirm=True) + + events = list(get_local_application().run_pipeline.stream(request)) + + assert [event.sequence for event in events] == list(range(1, len(events) + 1)) + assert {event.stage for event in events} >= {"extract", "validate", "transform", "load"} + assert events[-1].status in {StageStatus.DONE, StageStatus.SKIPPED} + assert all( + type(event).model_validate_json(event.model_dump_json()) == event for event in events + ) + rendered = "".join(event.model_dump_json() for event in events) + assert "raw_data" not in rendered + assert "stream_iterator" not in rendered + assert "llm_provider" not in rendered + + +def test_plan_is_stable_and_does_not_expose_inline_api_key(tmp_path: Path) -> None: + source = tmp_path / "input.csv" + source.write_text("id\n1\n", encoding="utf-8") + config = tmp_path / "pipeline.yaml" + config.write_text( + "\n".join( + [ + "source:", + " type: csv", + f" path: {source}", + "target:", + " type: json", + f" path: {tmp_path / 'output.json'}", + "transform:", + " type: ai", + " instruction: return rows", + "llm:", + " provider: openai", + " model: test-model", + " api_key: super-secret-key", + "", + ] + ), + encoding="utf-8", + ) + request = RunRequest(config_path=str(config)) + use_case = get_local_application().run_pipeline + + first = use_case.create_plan(request) + second = use_case.create_plan(request) + + assert first.plan_id == second.plan_id + assert first.config_digest == second.config_digest + assert "super-secret-key" not in first.model_dump_json() + + +class _AlwaysCancelled: + def is_cancelled(self, run_id: str) -> bool: + return run_id == "cancel-me" + + +def test_cancellation_is_checked_before_engine_execution(tmp_path: Path) -> None: + config, output = _pipeline_fixture(tmp_path) + use_case = RunPipeline( + cancellation=_AlwaysCancelled(), + checkpoints=NullCheckpointStore(), + secrets=EnvironmentSecretResolver(), + events=NullEventPublisher(), + reviewer=InputReviewPort(), + ) + + with pytest.raises(PipelineError, match="cancelled"): + list(use_case.stream(RunRequest(config_path=str(config), run_id="cancel-me"))) + + assert not output.exists() + + +def test_validation_and_connector_listing_use_application_service(tmp_path: Path) -> None: + config, _output = _pipeline_fixture(tmp_path) + service = get_local_application() + + validation = service.validate(config) + catalog = service.list_connectors() + + assert validation.valid is True + assert validation.plan.source_type == "csv" + assert "csv" in catalog.sources + assert "json" in catalog.targets diff --git a/tests/unit/test_import_boundaries.py b/tests/unit/test_import_boundaries.py new file mode 100644 index 0000000..0a17d72 --- /dev/null +++ b/tests/unit/test_import_boundaries.py @@ -0,0 +1,70 @@ +"""Architecture tests for the Phase 1 engine/client boundary.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +_REPOSITORY = Path(__file__).resolve().parents[2] +_LOAFER = _REPOSITORY / "loafer" +_ENGINE_PATHS = ( + _LOAFER / "engine.py", + _LOAFER / "config.py", + _LOAFER / "contracts.py", + _LOAFER / "core", + _LOAFER / "graph", + _LOAFER / "agents", + _LOAFER / "transform", + _LOAFER / "ports", +) +_BANNED_EXTERNAL = {"typer", "rich", "fastapi", "starlette", "flask", "django"} +_BANNED_INTERNAL = { + "loafer.cli", + "loafer.scheduler", + "loafer.daemon", + "loafer.application.service", + "loafer.application.local", +} + + +def _python_files(path: Path) -> list[Path]: + return [path] if path.is_file() else sorted(path.rglob("*.py")) + + +def _imports(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + modules: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules.update(alias.name for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + modules.add(node.module) + return modules + + +def test_engine_does_not_import_client_or_web_frameworks() -> None: + violations: list[str] = [] + for root in _ENGINE_PATHS: + for path in _python_files(root): + for module in _imports(path): + top_level = module.split(".", 1)[0] + if top_level in _BANNED_EXTERNAL or any( + module == banned or module.startswith(f"{banned}.") + for banned in _BANNED_INTERNAL + ): + violations.append(f"{path.relative_to(_REPOSITORY)} imports {module}") + + assert violations == [] + + +def test_application_service_has_no_cli_rendering_dependency() -> None: + modules = set() + for path in (_LOAFER / "application").rglob("*.py"): + modules.update(_imports(path)) + + assert not ({module.split(".", 1)[0] for module in modules} & _BANNED_EXTERNAL) + + for client in (_LOAFER / "cli.py", _LOAFER / "scheduler.py"): + client_imports = _imports(client) + assert "loafer.application" in client_imports + assert "loafer.runner" not in client_imports