separate application execution boundaries

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

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/<scope>`, `fix/<scope>`, `docs/<scope>`, `refactor/<scope>`,
+  `test/<scope>`, or `chore/<scope>`
+- 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
 

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
 

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

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"}

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",
+]

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,
+        )
+    )