Add bounded batch data plane

[lupppig]

Summary

• add an opt-in bounded row-local ETL path with one batch in flight
• validate every batch with explicit schema drift, quarantine, count, byte, and checksum contracts
• prepare custom and AI transform artifacts once per run and reuse them across batches
• atomically publish CSV/JSON outputs and add run-scoped PostgreSQL staging with transactional replace, append, and keyed upsert
• harden PDF extraction with provenance, file/page/time limits, and explicit failure policies
• expose target delivery guarantees and update the README, changelog, readiness roadmap, and execution skills

Why

The materialized compatibility path retains full-run input and output rows, so memory scales with dataset size and failed database runs can expose partial effects. This change gives declared row-local transforms bounded memory and makes publication behavior part of the execution contract.

During the 30M-row gate, the benchmark also exposed two harness issues: an unbounded CSV encoding scan and normal short-lived process races in /proc. Both are fixed, and Git provenance is now optional inside minimal production images.

Impact

Existing pipelines keep the materialized default. Users opt into bounded execution with execution.transform_class: row_local. SQL remains classified as global relational work. MongoDB row-local targets remain blocked until they have an equivalent staging protocol.

PostgreSQL delivery is explicit: replace is transactional replacement, append is at-least-once across an ambiguous target-commit/checkpoint gap, and deterministic keyed upsert is replay-safe.

Validation

• 731 passed, 54 skipped
• live PostgreSQL 16 integration: 24 passed
• Ruff check, Ruff format check, and diff integrity passed
• clean production image built from b2d474b and smoke-tested as 0.4.1.dev0+b2d474b
• 30,000,000-row production-image benchmark:
  • exact 30M output row count
  • exact 1,140,588,914-byte input/output SHA-256 match
  • 118.227 MiB peak process-tree RSS under a 512 MiB limit
  • 2,028.26 seconds, 14,791 rows/second
  • atomic publication with no temporary output remaining

The benchmark artifact pins source revision, image ID, Python runtime, container limits, and disk-backed storage in benchmarks/results/30m-row-local.json.

Summary by CodeRabbit

• New Features
  • Added opt-in bounded row-local ETL with batch validation, schema-drift policies, quarantine handling, cancellation, checkpoints, and reconciliation metrics.
  • Added atomic staged publication for PostgreSQL, including replace, append, error, and upsert modes.
  • Added PDF extraction limits, per-page timeouts, failure policies, diagnostics, and provenance metadata.
• Bug Fixes
  • CSV encoding detection now uses bounded reads.
  • Improved handling of interrupted runs and partial output.
• Documentation
  • Expanded execution guarantees, configuration guidance, limitations, and benchmark reporting.

[coderabbitai]

Caution

Review failed

The pull request is closed.

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: c40410a5-de6e-48dc-ab2a-2cf3fdd603d0

📥 Commits

Reviewing files that changed from the base of the PR and between 1c4f73f and 1a2a0bc.

📒 Files selected for processing (31)

• CHANGELOG.md
• PRODUCTION_READINESS.md
• README.md
• benchmarks/full_pipeline.py
• benchmarks/results/30m-row-local.json
• benchmarks/results/README.md
• loafer/adapters/sources/csv_source.py
• loafer/adapters/sources/pdf.py
• loafer/adapters/targets/postgres_staging.py
• loafer/agents/extract.py
• loafer/application/service.py
• loafer/cli.py
• loafer/config.py
• loafer/connectors/registry.py
• loafer/contracts.py
• loafer/core/batches.py
• loafer/data_plane.py
• loafer/engine.py
• loafer/graph/state.py
• loafer/ports/connector.py
• loafer/transform/batch_runner.py
• skills/loafer-engine/references/execution-contract.md
• skills/loafer-engineering/references/architecture.md
• tests/e2e/test_bounded_data_plane.py
• tests/integration/test_postgres_staging.py
• tests/unit/connectors/test_csv_source.py
• tests/unit/connectors/test_pdf_source.py
• tests/unit/connectors/test_postgres_staging_target.py
• tests/unit/test_batch_policies.py
• tests/unit/test_bounded_config.py
• tests/unit/test_full_pipeline_benchmark.py

---

📝 Walkthrough

Walkthrough

Adds an opt-in bounded row-local ETL path with per-batch validation, schema handling, reusable transforms, cancellation, reconciliation, quarantine, and atomic publication. It also adds PostgreSQL staging, PDF limits and diagnostics, benchmark updates, documentation, and extensive unit, integration, and end-to-end coverage.

Changes

Bounded row-local execution

Layer / File(s)	Summary
Execution contracts and batch policies loafer/config.py, loafer/contracts.py, loafer/core/batches.py, loafer/application/service.py, tests/unit/test_batch_policies.py, tests/unit/test_bounded_config.py	Adds execution modes, schema-drift policies, validation rules, batch envelopes, rolling digests, rejection metadata, and explicit delivery guarantees.
Batch transformation and orchestration loafer/transform/batch_runner.py, loafer/data_plane.py, tests/e2e/test_bounded_data_plane.py	Adds reusable custom/AI transform artifacts, bounded batch processing, quarantine handling, cancellation, checkpoints, checksums, and failure cleanup.
Engine routing and run reporting loafer/engine.py, loafer/graph/state.py, loafer/ports/connector.py, loafer/agents/extract.py, loafer/cli.py	Routes declared row-local pipelines through the bounded path, carries runtime ports and diagnostics, and reports batch metrics and events.
PostgreSQL staged publication loafer/adapters/targets/postgres_staging.py, loafer/connectors/registry.py, tests/unit/connectors/test_postgres_staging_target.py, tests/integration/test_postgres_staging.py	Stages batches in run-scoped PostgreSQL tables and publishes them using transactional replace, error, append, and upsert modes.
Source limits and diagnostics loafer/adapters/sources/csv_source.py, loafer/adapters/sources/pdf.py, tests/unit/connectors/test_csv_source.py, tests/unit/connectors/test_pdf_source.py	Bounds CSV encoding scans and adds PDF size, page, timeout, failure-policy, provenance, and diagnostic behavior.
Benchmarks and documentation benchmarks/*, benchmarks/results/*, README.md, PRODUCTION_READINESS.md, CHANGELOG.md, skills/loafer-*/references/*	Documents bounded execution semantics, records the 30-million-row benchmark, and updates production-readiness and known-limitations details.

Estimated code review effort: 5 (Critical) | ~120 minutes

Possibly related PRs

• lupppig/loafer#17: Adds application-boundary plumbing and shared contracts extended by this execution-path work.

Sequence Diagram(s)

sequenceDiagram
  participant Client
  participant RunPipeline
  participant stream_bounded_pipeline
  participant TargetConnector
  Client->>RunPipeline: run declared row-local pipeline
  RunPipeline->>stream_bounded_pipeline: stream batches with cancellation and checkpoints
  stream_bounded_pipeline->>TargetConnector: write and finalize published output
  TargetConnector-->>stream_bounded_pipeline: publication status
  stream_bounded_pipeline-->>RunPipeline: batch events and final state
  RunPipeline-->>Client: run snapshot and progress

Loading

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/bounded-batch-data-plane

---

_{Comment @coderabbitai help to get the list of available commands.}