feat: python-csv node kind (contract-only CSV loader + validation header check)

.github/workflows/ci.yml

@@ -24,9 +24,15 @@ jobs:
         run: uv run --package continuo-python-runtime-trino mypy adapters/trino/continuo_python_runtime_trino
       # `-m "not image"` deselects tests/test_image_smoke_validation.py, which
       # needs a built engine image and the env naming it. Those tests run in
-      # images.yml's smoke jobs, where an image actually exists.
+      # images.yml's smoke jobs, where an image actually exists. `and not
+      # integration` deselects the csv-reader/validation-runner tests that
+      # need a real minio backend (started via `docker run` by the
+      # `minio_container` fixture, not docker-compose) -- those run in the
+      # dedicated step below, on the same runner, where docker is available.
       - name: Tests (runtime)
-        run: uv run pytest --cov=continuo_python_runtime -m "not image" -v
+        run: uv run pytest --cov=continuo_python_runtime -m "not image and not integration" -v
+      - name: Tests (runtime, integration)
+        run: uv run pytest tests/test_csv_readers_integration.py tests/test_validation_runner.py -m integration -v
       - name: Tests (contract)
         run: uv run pytest contract/tests -v
       - name: Tests (adapter units)

CONTRIBUTING.md

@@ -39,7 +39,8 @@ repository; please do not add headers to new files.
 ## Development setup
 
 Prerequisites: Python 3.14+, [uv](https://docs.astral.sh/uv/), and Docker (only needed
-for the Postgres/Trino integration tests).
+for the Postgres/Trino integration tests and the csv-reader/validation-runner
+integration tests, which start a real minio backend via `docker run`).
 
 ```bash
 uv sync --all-packages --all-groups
@@ -58,14 +59,16 @@ uv run mypy continuo_python_runtime
 uv run mypy contract/continuo_engine_contract
 uv run --package continuo-python-runtime-postgres mypy adapters/postgres/continuo_python_runtime_postgres
 uv run --package continuo-python-runtime-trino mypy adapters/trino/continuo_python_runtime_trino
-uv run pytest --cov=continuo_python_runtime -m "not image" -v
+uv run pytest --cov=continuo_python_runtime -m "not image and not integration" -v
+uv run pytest tests/test_csv_readers_integration.py tests/test_validation_runner.py -m integration -v
 uv run pytest contract/tests -v
 uv run pytest adapters/postgres/tests adapters/trino/tests -m "not integration" -v
 ```
 
 These are exactly what `.github/workflows/ci.yml` runs. Integration tests against a real
-Postgres/Trino stack need Docker and are not required for most changes — see
-`.github/workflows/ci.yml` for how CI stands them up if you want to run them locally.
+Postgres/Trino stack, or against the csv-reader/validation-runner minio backend, need
+Docker and are not required for most changes — see `.github/workflows/ci.yml` for how CI
+stands them up if you want to run them locally.
 
 Also run the security scan before opening a pull request that touches dependencies or
 anything that could carry a credential:

README.md

@@ -94,7 +94,9 @@ the Go parser has not been taught is a production outage, not a refactor.
    login` step to `release.yml`.
 5. Write a contract file under `contracts/` (see
    `template/contracts/example.yml`) and a script under `scripts/` that
-   implements `run(ctx)` (see `template/scripts/example.py`).
+   implements `run(ctx)` (see `template/scripts/example.py`). A node that
+   only needs to land a csv file needs no script at all — see
+   `template/contracts/example_csv.yml` and "Node kinds" below.
 6. Push to `main`. The `release.yml` workflow lints the scripts, validates
    and merges the contracts, builds and pushes the image, uploads the merged
    contract to S3, and POSTs the release.
@@ -121,6 +123,30 @@ The runtime image does not re-run this gate, so a read that passes here is
 not re-judged under a different grammar in production. See
 `docs/boundary-contract.md` §13.1.
 
+## Node kinds
+
+A contract node's `kind:` field selects how the node produces its rows.
+Every rule below (`extra_columns`, `output_columns`, "Conform rules") applies
+to both kinds identically — `kind` only changes how the pre-conform table is
+produced, never how it is checked or written.
+
+- **`python-model`** (the default; the field may be omitted) — a script node.
+  It requires `script:` and a `reads:` map of one or more named SQL queries,
+  as described in "The script API" below.
+- **`python-csv`** — a contract-only node: it has no script and its `reads:`
+  map must be exactly `{csv: <uri>}`, where the uri is `s3://bucket/key` or
+  an `https://` url (`http://` is rejected at validate time, not run time).
+  The harness fetches the file, parses it with RFC 4180 defaults, and feeds
+  the result straight into `conform()` — declared `output_columns` types
+  decide the warehouse schema, not whatever pyarrow infers from the csv.
+  Because there is no script, `script:` is a forbidden key for this kind;
+  `continuo-runtime validate`/`merge`/`lint` reject one that sets it. The
+  csv's header row must contain every declared output column (checked again,
+  independently, at release time before promotion); columns present in the
+  header but not declared are governed by the same `extra_columns` policy as
+  a script node's output — `raise` (default) fails the run, `warn` drops
+  them and logs a warning. See `template/contracts/example_csv.yml`.
+
 ## The script API
 
 A node script is a Python file with exactly one required entry point:

adapters/postgres/tests/conftest.py

@@ -0,0 +1,114 @@
+"""Shared fixtures for adapters/postgres/tests.
+
+``adapters/postgres/tests`` is its own top-level pytest package (it carries
+``__init__.py``, per the root pyproject's import-mode note), so it cannot see
+fixtures declared in the root ``tests/conftest.py`` — pytest only walks
+conftest files up a test's own directory ancestry, and this directory is not
+an ancestor of the root ``tests/`` package. ``minio_container`` is duplicated
+here rather than imported, matching the root fixture byte-for-byte in
+behavior (real minio via a plain ``docker run``, dynamic host port, health
+wait, ``docker rm -f`` teardown) so the postgres-adapter integration suite
+gets the same "real backends, no stubs, no silent skip" guarantee without a
+cross-package import.
+"""
+
+import os
+import subprocess
+import time
+import urllib.error
+import urllib.request
+import uuid
+
+import pytest
+
+
+@pytest.fixture(scope="session")
+def minio_container():
+    """Session-scoped real minio backend, started via a plain `docker run`.
+
+    No testcontainers/pytest-docker dependency in this repo, so this drives
+    docker directly. Publishes minio's 9000 to an EPHEMERAL host port (colima
+    may already hold 9000 for another stack) and waits for minio's own
+    /minio/health/live endpoint to return 200 before yielding. Any failure —
+    docker missing, `docker run` erroring, the health check never turning
+    green — raises so the dependent tests error loudly instead of silently
+    skipping, per this suite's "real backends, no stubs, no silent skip"
+    integration-testing policy.
+    """
+    name = f"postgres-adapter-minio-{uuid.uuid4().hex[:12]}"
+    try:
+        subprocess.run(
+            [
+                "docker", "run", "-d", "--name", name,
+                "-p", "0:9000",
+                "-e", "MINIO_ROOT_USER=minioadmin",
+                "-e", "MINIO_ROOT_PASSWORD=minioadmin",
+                "minio/minio:latest",
+                "server", "/data", "--address", ":9000",
+            ],
+            check=True, capture_output=True, text=True, timeout=60,
+        )
+    except FileNotFoundError as exc:
+        raise RuntimeError(
+            "docker is not available; the postgres adapter's csv integration "
+            "test requires a live docker daemon"
+        ) from exc
+    except subprocess.CalledProcessError as exc:
+        raise RuntimeError(f"docker run for minio failed: {exc.stderr}") from exc
+
+    try:
+        port_out = subprocess.run(
+            ["docker", "port", name, "9000/tcp"],
+            check=True, capture_output=True, text=True, timeout=10,
+        ).stdout.strip()
+        # e.g. "0.0.0.0:54321\n[::]:54321" -- take the first (ipv4) mapping.
+        host_port = port_out.splitlines()[0].rsplit(":", 1)[1]
+        endpoint = f"http://127.0.0.1:{host_port}"
+
+        deadline = time.monotonic() + 30
+        last_error: Exception | None = None
+        healthy = False
+        while time.monotonic() < deadline:
+            try:
+                with urllib.request.urlopen(
+                    f"{endpoint}/minio/health/live", timeout=2
+                ) as resp:
+                    if resp.status == 200:
+                        healthy = True
+                        break
+            except (urllib.error.URLError, ConnectionError, TimeoutError) as exc:
+                last_error = exc
+            time.sleep(0.5)
+        if not healthy:
+            raise RuntimeError(
+                f"minio container {name} never became healthy at "
+                f"{endpoint}/minio/health/live: {last_error}"
+            )
+
+        # make_s3_client() forwards only S3_ENDPOINT_URL and otherwise leaves
+        # credentials to boto3's own chain, so the chain needs something to
+        # find. This is session-scoped (not monkeypatch, which is
+        # function-scoped) -- set directly and restore on teardown.
+        # WARNING: any other test that runs while this fixture is active and
+        # makes a REAL (non-mocked, non-S3_ENDPOINT_URL-redirected) AWS call
+        # would authenticate with these fake minioadmin credentials, not the
+        # caller's real ones.
+        prior_key = os.environ.get("AWS_ACCESS_KEY_ID")
+        prior_secret = os.environ.get("AWS_SECRET_ACCESS_KEY")
+        os.environ["AWS_ACCESS_KEY_ID"] = "minioadmin"
+        os.environ["AWS_SECRET_ACCESS_KEY"] = "minioadmin"
+        try:
+            yield (endpoint, "minioadmin", "minioadmin")
+        finally:
+            if prior_key is None:
+                os.environ.pop("AWS_ACCESS_KEY_ID", None)
+            else:
+                os.environ["AWS_ACCESS_KEY_ID"] = prior_key
+            if prior_secret is None:
+                os.environ.pop("AWS_SECRET_ACCESS_KEY", None)
+            else:
+                os.environ["AWS_SECRET_ACCESS_KEY"] = prior_secret
+    finally:
+        subprocess.run(
+            ["docker", "rm", "-f", name], capture_output=True, text=True, timeout=30
+        )

adapters/postgres/tests/test_integration_runtime_postgres.py

@@ -7,10 +7,13 @@
 import os
 import uuid
 
+import boto3
 import psycopg2
 import pyarrow as pa
 import pytest
+import yaml
 
+from continuo_python_runtime.harness import run_node
 from continuo_python_runtime_postgres.adapter import PostgresAdapter
 
 PG = dict(
@@ -303,3 +306,79 @@ def test_ensure_schema_generic_failure_rolls_back_and_releases_advisory_lock():
     second.commit()
     second.close()
     assert acquired is True
+
+
+# --- python-csv node end-to-end: real minio -> run_node -> real postgres ---
+#
+# minio_container is declared in adapters/postgres/tests/conftest.py (this
+# package cannot see the root tests/conftest.py fixture of the same name --
+# see that conftest's module docstring).
+
+CSV_BODY = b"order_id,amount\n1,10.5\n2,20.0\n3,5.25\n"
+
+
+@pytest.fixture(scope="session")
+def csv_minio(minio_container):
+    """Seed a real minio bucket with the csv this test's node reads."""
+    endpoint, access, secret = minio_container
+    client = boto3.client(
+        "s3", endpoint_url=endpoint,
+        aws_access_key_id=access, aws_secret_access_key=secret,
+    )
+    client.create_bucket(Bucket="pg-drops")
+    client.put_object(Bucket="pg-drops", Key="orders.csv", Body=CSV_BODY)
+    return endpoint
+
+
+def _csv_contract_dir(tmp_path, schema):
+    """A contract dir with a single python-csv node reading s3://pg-drops/orders.csv."""
+    (tmp_path / "contracts").mkdir()
+    (tmp_path / "contracts" / "t.yml").write_text(yaml.safe_dump({"nodes": [{
+        "schema": schema, "table": "orders_csv", "owner": "m", "schedule": "daily",
+        "criticality": "SECONDARY", "kind": "python-csv",
+        "reads": {"csv": "s3://pg-drops/orders.csv"},
+        "output_columns": [
+            {"name": "order_id", "type": "INTEGER", "nullable": False},
+            {"name": "amount", "type": "DOUBLE PRECISION"},
+        ],
+    }]}))
+    return tmp_path
+
+
+@pytest.mark.integration
+def test_run_node_csv_kind_loads_minio_csv_into_postgres(clean_schema, csv_minio, monkeypatch, tmp_path):
+    """run_node on a python-csv node fetches from real minio and writes to real postgres.
+
+    Exercises the full production path added in this task: harness.run_node
+    dispatches on node.kind to csv_loader.produce_csv (no reader/adapter test
+    doubles here -- the S3CsvSourceReader from csv_readers.reader_for talks
+    to the real minio container, and PostgresAdapter writes to the real
+    postgres stack), then conform()/ensure_table()/load() proceed exactly as
+    for a python-model node.
+    """
+    monkeypatch.setenv("S3_ENDPOINT_URL", csv_minio)
+    repo = _csv_contract_dir(tmp_path, clean_schema)
+    env = {
+        "NODE_ID": f"python-csv.svc.{clean_schema}.orders_csv",
+        "TABLE_NAME": "orders_csv",
+        "TARGET_SCHEMA": clean_schema,
+        "CONTRACT_DIR": str(repo / "contracts"),
+        "APP_ROOT": str(repo),
+    }
+    a = _adapter()
+
+    assert run_node(env, adapter=a) == 0
+
+    assert _columns(clean_schema, "orders_csv") == [
+        ("order_id", "integer", "NO"),
+        ("amount", "double precision", "YES"),
+    ]
+    assert _count(clean_schema, "orders_csv") == 3
+    conn = _conn()
+    with conn.cursor() as cur:
+        cur.execute(
+            f'SELECT order_id, amount FROM "{clean_schema}"."orders_csv" ORDER BY order_id'
+        )
+        rows = cur.fetchall()
+    conn.close()
+    assert rows == [(1, 10.5), (2, 20.0), (3, 5.25)]