From cb9253d2dececde451da9d0d347e7ec1ea456c02 Mon Sep 17 00:00:00 2001
From: OmerBaddour <omer.m.baddour@gmail.com>
Date: Sat, 1 Aug 2026 19:26:47 -0400
Subject: [PATCH 1/2] feat: split Trouve into TrouveAbc + Trouve + PandasTrouve

Replace the df_fn field with a PandasTrouve class. A pandas transform is now
a plain function that takes DataFrames, and the upstream Trouves come from an
inputs list. Clair binds the inputs to the parameters by position.

Before, a transform declared each dependency as a parameter default value.
That made the pd.DataFrame annotation false, thus each parameter needed a
`# type: ignore`, and a user could not call the function directly.

The split introduces TrouveAbc, the abstract base that holds the attributes
of every backend. Trouve keeps the SQL backend and its validation. This makes
space for a Databricks backend later.

- PandasTrouve validates the arity of the transform, and rejects *args/**kwargs
- ClairDag, discovery, the compiler and the runner accept a TrouveAbc
- The example project and every documentation page use the new API

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 .../project_docs_are_the_source_of_truth.md   |   9 +-
 README.md                                     |  42 +++--
 .../derived/daily_event_counts.py             |  20 +--
 site_docs/docs/concepts/trouve.md             |  33 ++--
 site_docs/docs/guides/pandas-native.md        |  81 +++++----
 site_docs/docs/index.md                       |   2 +-
 site_docs/docs/reference/index.md             |   2 +-
 site_docs/docs/reference/trouve-api.md        | 100 ++++++++---
 src/clair/__init__.py                         |   5 +-
 src/clair/core/compiler.py                    |  19 ++-
 src/clair/core/dag.py                         |  16 +-
 src/clair/core/discovery.py                   |  43 ++---
 src/clair/core/runner.py                      |  50 +++---
 src/clair/trouves/__init__.py                 |   5 +-
 src/clair/trouves/pandas_trouve.py            |  84 +++++++++
 src/clair/trouves/trouve.py                   | 103 +++++++----
 tests/unit/test_compiler_pandas.py            |  39 ++---
 tests/unit/test_discovery.py                  |   3 +-
 tests/unit/test_discovery_pandas.py           |  74 ++++----
 tests/unit/test_runner_pandas.py              | 105 ++++++------
 tests/unit/test_trouves.py                    | 161 ++++++++++++++----
 21 files changed, 638 insertions(+), 358 deletions(-)
 create mode 100644 src/clair/trouves/pandas_trouve.py

diff --git a/.claude/memory/project_docs_are_the_source_of_truth.md b/.claude/memory/project_docs_are_the_source_of_truth.md
index be2d989..d645b67 100644
--- a/.claude/memory/project_docs_are_the_source_of_truth.md
+++ b/.claude/memory/project_docs_are_the_source_of_truth.md
@@ -12,11 +12,12 @@ metadata:
 `src/clair/auth/environments.py`, a path that does not exist. The documentation is the
 better source because users read it, so an error there gets a report.
 
-**But the documentation also rots.** The pandas guide, the landing page and the README
+**But the documentation also rots.** The pandas guide, the landing page and the README once
 documented a `PandasTrouve` class that was never built, while the API reference correctly
 documented the `df_fn` field that was. The pages came from a design spec, and nobody
 changed them when the implementation took a different shape. mkdocs does not execute the
-examples, so CI did not catch it.
+examples, so CI did not catch it. (The 2026-08-01 backend split built the real
+`PandasTrouve` and deleted `df_fn`, thus the two agree again — but the lesson holds.)
 
 **Therefore: the code is the final authority.** Read the documentation first for
 orientation, then confirm any API detail against `src/` or `example_projects/` before you
@@ -26,8 +27,8 @@ depend on it. When the two disagree, the code wins and the page is a bug.
 
 - To learn what a feature does, `grep` `site_docs/docs/` first. Then confirm the exact API
   against `src/` or a project in `example_projects/`.
-- Search for the field name, not only the class name. The pandas feature was invisible to a
-  search for `PandasTrouve`, because the real name is `df_fn`.
+- Search for the field name, not only the class name. The pandas feature was once invisible
+  to a search for `PandasTrouve`, because the name in the code was `df_fn`.
 - Map for orientation: `concepts/` (Trouve, DAG, project layout, environments),
   `guides/` (routing, incrementality, tests, selectors, pandas, per-database config),
   `cli/` (one page for each subcommand), `reference/` (API for Trouve, Column, RunConfig,
diff --git a/README.md b/README.md
index 59aaa15..4be3b7c 100644
--- a/README.md
+++ b/README.md
@@ -42,18 +42,15 @@ import pandas as pd
 from refined.products.catalog import trouve as catalog_trouve
 from refined.products.reviews import trouve as reviews_trouve
 
-from clair import Trouve
+from clair import PandasTrouve
 
 
-def summarize(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
-    df = catalog.merge(reviews, on="product_id")
-    return df.groupby("name", as_index=False)["rating"].mean()
+def summarize(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
+    merged = catalog.merge(reviews, on="product_id")
+    return merged.groupby("name", as_index=False).agg(rating=("rating", "mean"))
 
 
-trouve = Trouve(df_fn=summarize)
+trouve = PandasTrouve(transform=summarize, inputs=[catalog_trouve, reviews_trouve])
 ```
 
 clair fetches the upstream tables from Snowflake, runs your function locally, then writes the result back. The DAG, lineage, `--select` filters, and data quality tests all work unchanged.
@@ -73,10 +70,10 @@ my_project/
         └── reviews.py   →   source.products.reviews
 ```
 
-A Trouve runs in one of two ways:
+Clair has one Trouve class for each backend. Both derive from `TrouveAbc`:
 
-- **`sql`** — compiles to a Snowflake `TABLE` or `VIEW`. Runs inside Snowflake.
-- **`df_fn`** — runs a Python function on the machine executing clair, then writes the result back to Snowflake.
+- **`Trouve`** — compiles `sql` to a Snowflake `TABLE` or `VIEW`. Runs inside Snowflake.
+- **`PandasTrouve`** — runs a Python function on the machine executing clair, then writes the result back to Snowflake.
 
 `Trouve` has three types:
 
@@ -385,31 +382,28 @@ Pass `--run-mode full_refresh` on the CLI to force a full rebuild of everything
 
 ## Pandas-native transformations
 
-When SQL isn't the right tool — complex reshaping, ML feature engineering, multi-step aggregations — give the Trouve a `df_fn` in place of `sql`. Your function receives upstream tables as DataFrames, runs locally on the machine executing clair, and the result is written back to Snowflake automatically.
+When SQL isn't the right tool — complex reshaping, ML feature engineering, multi-step aggregations — use a `PandasTrouve` in place of a `Trouve`. Your function receives upstream tables as DataFrames, runs locally on the machine executing clair, and clair writes the result back to Snowflake.
 
 ```python
 import pandas as pd
 from refined.products.catalog import trouve as catalog_trouve
 from refined.products.reviews import trouve as reviews_trouve
 
-from clair import Column, ColumnType, TestNotNull, Trouve
+from clair import Column, ColumnType, PandasTrouve, TestNotNull
 
 
-def top_rated(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
-    df = catalog.merge(reviews, on="product_id")
+def top_rated(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
+    merged = catalog.merge(reviews, on="product_id")
     return (
-        df.groupby(["product_id", "name"], as_index=False)["rating"]
-        .mean()
-        .rename(columns={"rating": "avg_rating"})
+        merged.groupby(["product_id", "name"], as_index=False)
+        .agg(avg_rating=("rating", "mean"))
         .query("avg_rating >= 4")
     )
 
 
-trouve = Trouve(
-    df_fn=top_rated,
+trouve = PandasTrouve(
+    transform=top_rated,
+    inputs=[catalog_trouve, reviews_trouve],
     columns=[
         Column(name="product_id", type=ColumnType.STRING),
         Column(name="name",       type=ColumnType.STRING),
@@ -420,6 +414,8 @@ trouve = Trouve(
 )
 ```
 
+Clair binds `inputs` to the transform parameters by position. Because the transform takes plain DataFrames, you can call it directly in a unit test or in a notebook.
+
 > **Note:** pandas transformations run on the machine executing clair, not inside Snowflake. Keep this in mind for large tables.
 
 See the [Pandas-Native Transformations guide](https://rivage-sh.github.io/clair/guides/pandas-native/) for the full field reference, DAG integration details, and limitations.
diff --git a/example_projects/example_4/example_4_database/derived/daily_event_counts.py b/example_projects/example_4/example_4_database/derived/daily_event_counts.py
index 3f27cb9..b2289b9 100644
--- a/example_projects/example_4/example_4_database/derived/daily_event_counts.py
+++ b/example_projects/example_4/example_4_database/derived/daily_event_counts.py
@@ -1,22 +1,18 @@
 import pandas as pd
 from example_4_database.refined.events import trouve as example_4_database_refined_events
 
-from clair import Column, ColumnType, Trouve
+from clair import Column, ColumnType, PandasTrouve
 
 
-def daily_event_counts(
-    refined_events: pd.DataFrame = example_4_database_refined_events,  # type: ignore
-) -> pd.DataFrame:
-    return (
-        refined_events
-        .groupby(["event_date", "event_type"], as_index=False)
-        .size()
-        .rename(columns={"size": "event_count"})  # type: ignore
-    )
+def daily_event_counts(refined_events: pd.DataFrame) -> pd.DataFrame:
+    return refined_events.groupby(
+        ["event_date", "event_type"], as_index=False
+    ).agg(event_count=("event_type", "size"))
 
 
-trouve = Trouve(
-    df_fn=daily_event_counts,
+trouve = PandasTrouve(
+    transform=daily_event_counts,
+    inputs=[example_4_database_refined_events],
     docs="Daily counts of each event type, aggregated from refined events.",
     columns=[
         Column(name="event_date", type=ColumnType.DATE),
diff --git a/site_docs/docs/concepts/trouve.md b/site_docs/docs/concepts/trouve.md
index fe831fc..92c1eca 100644
--- a/site_docs/docs/concepts/trouve.md
+++ b/site_docs/docs/concepts/trouve.md
@@ -35,7 +35,6 @@ When `{source_catalog}` is evaluated, it calls `Trouve.__format__`, which regist
 | `tests` | `list[AnyTest]` | `[]` | Data quality tests. See [Tests](../guides/data-quality-tests.md). |
 | `docs` | `str` | `""` | Documentation string shown in `clair docs`. |
 | `run_config` | `RunConfig` | full refresh | Materialization strategy. See [Incrementality](../guides/incrementality.md). |
-| `df_fn` | `Callable \| None` | `None` | Pandas execution mode (alternative to `sql`). TABLE-only, full-refresh-only. |
 
 ## Examples
 
@@ -109,9 +108,9 @@ trouve = Trouve(
 )
 ```
 
-## Pandas execution (`df_fn`)
+## Pandas execution (`PandasTrouve`)
 
-When SQL is not the right tool, give the Trouve a `df_fn` in place of `sql`. You supply a Python function. Clair fetches the upstream tables from Snowflake as DataFrames, calls your function on the machine that runs clair, then writes the result back to Snowflake.
+When SQL is not the right tool, use a `PandasTrouve` in place of a `Trouve`. You supply a Python function. Clair fetches the upstream tables from Snowflake as DataFrames, calls your function on the machine that runs clair, then writes the result back to Snowflake.
 
 ```python
 # derived/products/top_rated.py
@@ -119,36 +118,36 @@ import pandas as pd
 from refined.products.catalog import trouve as catalog_trouve
 from refined.products.reviews import trouve as reviews_trouve
 
-from clair import Trouve
+from clair import PandasTrouve
 
 
-def top_rated(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
-    df = catalog.merge(reviews, on="product_id")
+def top_rated(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
+    merged = catalog.merge(reviews, on="product_id")
     return (
-        df.groupby(["product_id", "name"], as_index=False)["rating"]
-        .mean()
+        merged.groupby(["product_id", "name"], as_index=False)
+        .agg(rating=("rating", "mean"))
         .query("rating >= 4")
     )
 
 
-trouve = Trouve(df_fn=top_rated)
+trouve = PandasTrouve(
+    transform=top_rated,
+    inputs=[catalog_trouve, reviews_trouve],
+)
 ```
 
-Each dependency is a parameter whose default value is the upstream Trouve.
+Clair binds `inputs` to the transform parameters by position. The transform takes plain DataFrames, thus you can call it directly in a test or in a notebook.
 
-Differences between the two execution types:
+Differences between the two backends:
 
-| | `sql` | `df_fn` |
+| | `Trouve` | `PandasTrouve` |
 |---|---|---|
 | Execution | Inside Snowflake | Locally on the clair machine |
 | Output type | TABLE or VIEW | TABLE only |
 | Incremental | Supported | Full-refresh only |
-| Dependencies | f-string references | Parameter default values |
+| Dependencies | f-string references in `sql` | the `inputs` list |
 
-Everything else — DAG integration, `--select` filtering, data quality tests, `clair dag` output — operates the same way. The two fields are mutually exclusive: a Trouve with both `sql` and `df_fn` raises an error.
+Everything else — DAG integration, `--select` filtering, data quality tests, `clair dag` output — operates the same way. Both classes derive from `TrouveAbc`, the abstract base that holds `columns`, `tests`, `docs`, and `run_config`.
 
 See the [Pandas-native guide](../guides/pandas-native.md) for a full walkthrough.
 
diff --git a/site_docs/docs/guides/pandas-native.md b/site_docs/docs/guides/pandas-native.md
index ff14161..099a552 100644
--- a/site_docs/docs/guides/pandas-native.md
+++ b/site_docs/docs/guides/pandas-native.md
@@ -1,8 +1,8 @@
 # Pandas-Native Transformations
 
-A `Trouve` with a `df_fn` lets you write a pipeline step as a plain Python function. Clair fetches your upstream tables from Snowflake as DataFrames, runs your function on the machine executing clair, and writes the result back to Snowflake — with full DAG integration, lineage, selectors, and data quality tests.
+A `PandasTrouve` lets you write a pipeline step as a plain Python function. Clair fetches your upstream tables from Snowflake as DataFrames, runs your function on the machine executing clair, and writes the result back to Snowflake — with full DAG integration, lineage, selectors, and data quality tests.
 
-## When to use `df_fn`
+## When to use `PandasTrouve`
 
 Use it when SQL is the wrong tool for the job:
 
@@ -11,11 +11,11 @@ Use it when SQL is the wrong tool for the job:
 - Multi-step aggregations that depend on intermediate Python state
 - Logic you already have as pandas code
 
-For everything else, give the `Trouve` a `sql` string — it runs entirely inside Snowflake and does not move data over the network.
+For everything else, use a `Trouve` with a `sql` string — it runs entirely inside Snowflake and does not move data over the network.
 
 ## Installation
 
-`df_fn` needs no extra installation. pandas is a dependency of clair.
+`PandasTrouve` needs no extra installation. pandas is a dependency of clair.
 
 ## Basic example
 
@@ -25,39 +25,44 @@ import pandas as pd
 from refined.products.catalog import trouve as catalog_trouve
 from refined.products.reviews import trouve as reviews_trouve
 
-from clair import Trouve
+from clair import PandasTrouve
 
 
-def top_rated(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
-    df = catalog.merge(reviews, on="product_id")
+def top_rated(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
+    merged = catalog.merge(reviews, on="product_id")
     return (
-        df.groupby(["product_id", "name"], as_index=False)["rating"]
-        .mean()
-        .rename(columns={"rating": "avg_rating"})
+        merged.groupby(["product_id", "name"], as_index=False)
+        .agg(avg_rating=("rating", "mean"))
         .query("avg_rating >= 4")
     )
 
 
-trouve = Trouve(df_fn=top_rated)
+trouve = PandasTrouve(
+    transform=top_rated,
+    inputs=[catalog_trouve, reviews_trouve],
+)
 ```
 
-You declare each dependency as a **parameter default value**: annotate the parameter as `pd.DataFrame` and give it the upstream Trouve object as its default. At run time clair replaces each default with the fetched DataFrame and calls your function. The parameter name is the name you use in the function body — it does not need to match the import.
+Clair binds `inputs` to the parameters of `transform` **by position**. The first Trouve in `inputs` becomes the first parameter, the second becomes the second parameter, and so on. The parameter names are yours to choose — they do not need to match the import names.
 
-!!! note
-    The `# type: ignore` comment is necessary. A type checker sees a `Trouve` object assigned to a `pd.DataFrame` parameter and reports a mismatch. clair substitutes the DataFrame before it calls the function, so the annotation is correct at run time.
+Because `transform` is an ordinary function that takes DataFrames, you can call it directly in a unit test or in a notebook:
+
+```python
+from derived.products.top_rated import top_rated
+
+result = top_rated(my_catalog_dataframe, my_reviews_dataframe)
+```
 
 ## With columns and tests
 
 `columns` and `tests` work the same as they do for SQL Trouves:
 
 ```python
-from clair import Column, ColumnType, TestNotNull, TestRowCount, Trouve
+from clair import Column, ColumnType, PandasTrouve, TestNotNull, TestRowCount
 
-trouve = Trouve(
-    df_fn=top_rated,
+trouve = PandasTrouve(
+    transform=top_rated,
+    inputs=[catalog_trouve, reviews_trouve],
     columns=[
         Column(name="product_id", type=ColumnType.STRING),
         Column(name="name",       type=ColumnType.STRING),
@@ -73,10 +78,10 @@ trouve = Trouve(
 
 ## How it runs
 
-`clair run` gives a `df_fn` Trouve these four steps:
+`clair run` gives a `PandasTrouve` these four steps:
 
-1. **Fetch** — for each parameter whose default is a Trouve, run `SELECT * FROM <full_name>` and load the result into a DataFrame. Column names become lowercase.
-2. **Transform** — call your function locally on the clair machine, with one keyword argument for each parameter.
+1. **Fetch** — for each Trouve in `inputs`, run `SELECT * FROM <full_name>` and load the result into a DataFrame. Column names become lowercase.
+2. **Transform** — call your function locally on the clair machine, with one DataFrame for each parameter, in the order of `inputs`.
 3. **Write** — write the returned DataFrame back to Snowflake. The table is created or replaced.
 4. **Test** — run the attached tests against the output table in Snowflake.
 
@@ -85,9 +90,16 @@ If your function returns something other than a `DataFrame`, the run fails with
 !!! note
     Clair reads the data into memory on the machine that runs clair. For large upstream tables this is slow and it uses much memory. Chunked reads are not available.
 
+## Validation
+
+Clair examines the `transform` signature when Python loads your file. Thus a mistake stops the run immediately, and it names the fault:
+
+- The count of `inputs` must equal the count of parameters. An error tells you both counts and lists the parameter names.
+- The transform must not use `*args` or `**kwargs`. Clair binds each input to a named parameter.
+
 ## DAG integration
 
-Dependencies come from the parameter defaults. No extra configuration is necessary. `clair dag` marks these nodes with a `[PANDAS]` tag, in place of the `[TABLE]` or `[VIEW]` tag:
+Dependencies come from `inputs`. No extra configuration is necessary. `clair dag` marks these nodes with a `[PANDAS]` tag, in place of the `[TABLE]` or `[VIEW]` tag:
 
 ```
 === Clair DAG: 3 models, 1 source ===
@@ -98,7 +110,7 @@ example_4_database.source.events  [SOURCE]
         └── example_4_database.derived.top_event_types  [TABLE]
 ```
 
-SQL Trouves can depend on the output of a `df_fn` Trouve, and a `df_fn` Trouve can depend on other `df_fn` Trouves. The tree above shows both: a `df_fn` node reads a SQL table, and a SQL table reads the `df_fn` output.
+A SQL `Trouve` can depend on the output of a `PandasTrouve`, and a `PandasTrouve` can depend on other `PandasTrouve` nodes. The tree above shows both: a pandas node reads a SQL table, and a SQL table reads the pandas output.
 
 ## Selectors
 
@@ -110,7 +122,7 @@ clair run --project=. --env=dev --select='derived.products.top_rated'
 
 ## Compile output
 
-`clair compile` writes a `.py` artifact for a `df_fn` Trouve, in place of the `.sql` file it writes for a SQL Trouve. The artifact holds a header, the imports of the source module, and the source of your function:
+`clair compile` writes a `.py` artifact for a `PandasTrouve`, in place of the `.sql` file it writes for a SQL Trouve. The artifact holds a header, the imports of the source module, and the source of your function. The header shows which upstream Trouve clair binds to each parameter:
 
 ```python
 # clair compiled: derived.products.top_rated
@@ -121,31 +133,28 @@ clair run --project=. --env=dev --select='derived.products.top_rated'
 
 import pandas as pd
 
-def top_rated(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
+def top_rated(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
     ...
 ```
 
 ## Limitations
 
-- **Full-refresh only.** Incremental strategies are not available. A `df_fn` Trouve always replaces the table. A `RunConfig` with an incremental mode raises an error.
-- **TABLE output only.** Views are not available.
+- **Full-refresh only.** Incremental strategies are not available. A `PandasTrouve` always replaces the table. A `RunConfig` with an incremental mode raises an error.
+- **TABLE output only.** Views and sources are not available.
 - **Full table fetch.** Clair reads all upstream rows into memory. Chunking is not available.
-- **`sql` and `df_fn` are mutually exclusive.** A Trouve with both raises an error.
 
 ## Field reference
 
-These are the `Trouve` fields that apply to pandas execution. See the [Trouve API reference](../reference/trouve-api.md) for the full list.
+These are the `PandasTrouve` fields. See the [Trouve API reference](../reference/trouve-api.md) for the full list.
 
 | Field | Type | Default | Description |
 |-------|------|---------|-------------|
-| `df_fn` | `Callable \| None` | `None` | Python function that returns the output DataFrame. Its parameter defaults declare the upstream Trouves. Mutually exclusive with `sql`. |
+| `transform` | `Callable[..., pd.DataFrame]` | required | Python function that gives the output DataFrame. |
+| `inputs` | `list[TrouveAbc]` | `[]` | The upstream Trouves. Clair binds them to the transform parameters by position. |
 | `columns` | `list[Column]` | `[]` | Column definitions. Optional — used for documentation. |
 | `tests` | `list[AnyTest]` | `[]` | Data quality tests, run after clair writes the output. |
 | `docs` | `str` | `""` | Documentation string shown in `clair docs`. |
 
 ## Complete example
 
-`example_projects/example_4/` in the repository is a runnable project that uses `df_fn`. See `example_4_database/derived/daily_event_counts.py`.
+`example_projects/example_4/` in the repository is a runnable project that uses `PandasTrouve`. See `example_4_database/derived/daily_event_counts.py`.
diff --git a/site_docs/docs/index.md b/site_docs/docs/index.md
index bfd4356..523d77d 100644
--- a/site_docs/docs/index.md
+++ b/site_docs/docs/index.md
@@ -29,7 +29,7 @@ Import the upstream, use it in the f-string — clair figures out the rest.
 - **Compile first, run second.** `clair compile` resolves the full DAG and writes SQL to `_clairtifacts/` before touching Snowflake.
 - **Incremental strategies built in.** APPEND and UPSERT modes with no boilerplate — attach a [`RunConfig`](reference/run-config-api.md) to any [`Trouve`](concepts/trouve.md).
 - **Data quality as code.** Tests are Pydantic objects on the Trouve itself, not a separate test file.
-- **Pandas-native transformations.** Give a Trouve a [`df_fn`](guides/pandas-native.md) to write any step as a Python function — clair fetches upstream tables as DataFrames, runs your code locally, and writes the result back to Snowflake.
+- **Pandas-native transformations.** Use a [`PandasTrouve`](guides/pandas-native.md) to write any step as a Python function — clair fetches upstream tables as DataFrames, runs your code locally, and writes the result back to Snowflake.
 
 ## Install
 
diff --git a/site_docs/docs/reference/index.md b/site_docs/docs/reference/index.md
index ebaca07..1cf3b1f 100644
--- a/site_docs/docs/reference/index.md
+++ b/site_docs/docs/reference/index.md
@@ -14,7 +14,7 @@ from clair import (
 )
 ```
 
-- **[Trouve](trouve-api.md)** — the core class, for both SQL and pandas (`df_fn`) transformations
+- **[Trouve](trouve-api.md)** — the core classes: `TrouveAbc`, `Trouve` for SQL, and `PandasTrouve` for pandas
 - **[Column](column-api.md)** — column definitions
 - **[RunConfig](run-config-api.md)** — incremental materialization config
 - **[Tests](tests-api.md)** — data quality test classes
diff --git a/site_docs/docs/reference/trouve-api.md b/site_docs/docs/reference/trouve-api.md
index 245bae8..1567955 100644
--- a/site_docs/docs/reference/trouve-api.md
+++ b/site_docs/docs/reference/trouve-api.md
@@ -1,9 +1,19 @@
 # Trouve API
 
 ```python
-from clair import Trouve, TrouveType
+from clair import PandasTrouve, Trouve, TrouveAbc, TrouveType
 ```
 
+Clair has one Trouve class for each backend. `TrouveAbc` is the abstract base
+that they share. `Trouve` runs SQL in Snowflake. `PandasTrouve` runs a Python
+function on the machine executing clair.
+
+| Class | Backend | Declares dependencies with |
+|-------|---------|----------------------------|
+| `TrouveAbc` | none — abstract base | — |
+| `Trouve` | Snowflake SQL | f-string references in `sql` |
+| `PandasTrouve` | pandas | the `inputs` list |
+
 ## `TrouveType`
 
 ```python
@@ -21,18 +31,35 @@ class ExecutionType(StrEnum):
     PANDAS    = "pandas"
 ```
 
-## `Trouve`
+## `TrouveAbc`
+
+The abstract base of every backend. It holds the fields that each backend
+shares. You do not instantiate it directly — a subclass supplies the backend.
 
 ```python
-class Trouve(BaseModel):
+class TrouveAbc(BaseModel, ABC):
     type:       TrouveType = TrouveType.TABLE
-    sql:        str = ""
-    df_fn:      Callable | None = None
     columns:    list[Column] = []
     tests:      list[AnyTest] = []
     docs:       str = ""
     run_config: RunConfig = RunConfig()
     compiled:   CompiledAttributes | None = None
+
+    @property
+    def execution_type(self) -> ExecutionType: ...   # each subclass supplies it
+
+    def upstream_trouves(self) -> list[TrouveAbc]: ...  # each subclass supplies it
+```
+
+Write `isinstance(obj, TrouveAbc)` to accept a Trouve of any backend.
+
+## `Trouve`
+
+The SQL backend. Snowflake materializes it.
+
+```python
+class Trouve(TrouveAbc):
+    sql: str = ""
 ```
 
 ### Fields
@@ -40,8 +67,7 @@ class Trouve(BaseModel):
 | Field | Type | Default | Description |
 |-------|------|---------|-------------|
 | `type` | `TrouveType` | `TABLE` | Whether this is a SOURCE, TABLE, or VIEW |
-| `sql` | `str` | `""` | SQL query. Required for TABLE/VIEW. Must be empty for SOURCE. Use f-strings to reference other Trouves. Mutually exclusive with `df_fn`. |
-| `df_fn` | `Callable \| None` | `None` | Pandas execution mode (alternative to `sql`). TABLE-only, full-refresh-only. |
+| `sql` | `str` | `""` | SQL query. Required for TABLE/VIEW. Must be empty for SOURCE. Use f-strings to reference other Trouves. |
 | `columns` | `list[Column]` | `[]` | Column definitions. Optional for TABLE/VIEW. Required for UPSERT. |
 | `tests` | `list[AnyTest]` | `[]` | Data quality tests. See [Tests](tests-api.md). |
 | `docs` | `str` | `""` | Documentation string shown in `clair docs`. |
@@ -60,7 +86,6 @@ class Trouve(BaseModel):
 - TABLE and VIEW: `sql` must be non-empty
 - SOURCE: `sql` must be empty
 - INCREMENTAL mode: only TABLE supports it (not VIEW)
-- `df_fn`: TABLE-only; full-refresh-only; mutually exclusive with `sql`
 
 ## `CompiledAttributes`
 
@@ -70,24 +95,47 @@ Set by discovery on each `Trouve.compiled`. Available after `clair compile` or `
 |-----------|------|-------------|
 | `full_name` | `str` | Routed Snowflake name (used in SQL and DDL) |
 | `logical_name` | `str` | Filesystem-derived name (used in DAG edges and selectors) |
-| `resolved_sql` | `str` | SQL with all placeholder tokens replaced by real full_names |
+| `resolved_sql` | `str` | SQL with all placeholder tokens replaced by real full_names. Empty for a `PandasTrouve`. |
+| `resolved_transform` | `str` | The source text of the transform function. Empty for a SQL `Trouve`. |
 | `file_path` | `Path` | Absolute path to the Trouve file |
 | `imports` | `list[str]` | Logical names of upstream Trouves |
 | `execution_type` | `ExecutionType` | SNOWFLAKE or PANDAS |
 
-## Pandas execution (`df_fn`)
+## `PandasTrouve`
+
+The pandas backend. A Python function materializes it.
 
-A Trouve runs in pandas when you set `df_fn` in place of `sql`. There is no separate class.
+```python
+class PandasTrouve(TrouveAbc):
+    transform: Callable[..., pd.DataFrame]
+    inputs:    list[TrouveAbc] = []
+```
 
 ```python
-from clair import Trouve
+from clair import PandasTrouve
 ```
 
+### Fields
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `transform` | `Callable[..., pd.DataFrame]` | required | The function that gives the output DataFrame. |
+| `inputs` | `list[TrouveAbc]` | `[]` | The upstream Trouves. Clair binds them to the transform parameters by position. |
+
+`TrouveAbc` holds the other fields: `columns`, `tests`, `docs`, `run_config`.
+
+### Methods
+
+| Method | Returns | Description |
+|--------|---------|-------------|
+| `upstream_trouves()` | `list[TrouveAbc]` | The inputs, in the parameter order of the transform. |
+| `parameter_names()` | `list[str]` | The parameter names of the transform, in order. |
+
 ### Behaviour
 
 | Aspect | Detail |
 |--------|--------|
-| Dependencies | Each parameter of `df_fn` whose default value is a `Trouve` becomes an upstream dependency. Clair passes the fetched DataFrame as that keyword argument. |
+| Dependencies | Each Trouve in `inputs` becomes an upstream dependency. Clair binds them to the transform parameters by position. |
 | Materialization | Always `TABLE`. Clair creates or replaces the table. |
 | Incremental | Not available. Full-refresh only. |
 | Return value | The function must return a `pd.DataFrame`. Any other type fails the run. |
@@ -95,10 +143,10 @@ from clair import Trouve
 
 ### Constraints
 
-- `sql` and `df_fn` are mutually exclusive. A Trouve with both raises `ValueError`.
-- `df_fn` must be callable.
-- A `df_fn` Trouve must be `TrouveType.TABLE`. A VIEW or SOURCE raises `ValueError`.
-- A `df_fn` Trouve does not support incremental run modes.
+- The count of `inputs` must equal the count of transform parameters.
+- The transform must not use `*args` or `**kwargs`.
+- A `PandasTrouve` must be `TrouveType.TABLE`. A VIEW or a SOURCE raises `ValueError`.
+- A `PandasTrouve` does not support incremental run modes.
 
 ### Example
 
@@ -107,23 +155,21 @@ import pandas as pd
 from refined.products.catalog import trouve as catalog_trouve
 from refined.products.reviews import trouve as reviews_trouve
 
-from clair import Column, ColumnType, TestNotNull, Trouve
+from clair import Column, ColumnType, PandasTrouve, TestNotNull
 
 
-def top_rated(
-    catalog: pd.DataFrame = catalog_trouve,  # type: ignore
-    reviews: pd.DataFrame = reviews_trouve,  # type: ignore
-) -> pd.DataFrame:
-    df = catalog.merge(reviews, on="product_id")
+def top_rated(catalog: pd.DataFrame, reviews: pd.DataFrame) -> pd.DataFrame:
+    merged = catalog.merge(reviews, on="product_id")
     return (
-        df.groupby(["product_id", "name"], as_index=False)["rating"]
-        .mean()
+        merged.groupby(["product_id", "name"], as_index=False)
+        .agg(rating=("rating", "mean"))
         .query("rating >= 4")
     )
 
 
-trouve = Trouve(
-    df_fn=top_rated,
+trouve = PandasTrouve(
+    transform=top_rated,
+    inputs=[catalog_trouve, reviews_trouve],
     columns=[
         Column(name="product_id", type=ColumnType.STRING),
         Column(name="name",       type=ColumnType.STRING),
diff --git a/src/clair/__init__.py b/src/clair/__init__.py
index 96fb145..5bc06b0 100644
--- a/src/clair/__init__.py
+++ b/src/clair/__init__.py
@@ -45,6 +45,7 @@
 
 from clair.trouves.column import Column, ColumnType
 from clair.trouves.config import DatabaseDefaults, SchemaDefaults
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.run_config import (
     SOURCE,
     TARGET,
@@ -63,7 +64,7 @@
     TestUnique,
     TestUniqueColumns,
 )
-from clair.trouves.trouve import Trouve, TrouveType
+from clair.trouves.trouve import Trouve, TrouveAbc, TrouveType
 
 __version__ = "0.1.0"
 
@@ -84,6 +85,7 @@
     "ColumnType",
     "DatabaseDefaults",
     "IncrementalMode",
+    "PandasTrouve",
     "RunConfig",
     "RunMode",
     "SchemaDefaults",
@@ -94,6 +96,7 @@
     "TestUnique",
     "TestUniqueColumns",
     "Trouve",
+    "TrouveAbc",
     "TrouveType",
     "UpsertConfig",
 ]
diff --git a/src/clair/core/compiler.py b/src/clair/core/compiler.py
index eb7dbd0..8ad33fa 100644
--- a/src/clair/core/compiler.py
+++ b/src/clair/core/compiler.py
@@ -14,6 +14,7 @@
 from clair.core.discovery import ARTIFACTS_DIR_NAME
 from clair.core.runner import resolve_effective_mode
 from clair.exceptions import CompileError
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.run_config import RunMode
 from clair.trouves.trouve import ExecutionType, Trouve, TrouveType
 
@@ -130,15 +131,16 @@ def write_compile_output(
         assert trouve.compiled is not None, f"{name} has not been compiled"
         node_info = None
         if trouve.compiled.execution_type == ExecutionType.PANDAS:
+            assert isinstance(trouve, PandasTrouve)
             try:
-                fn_source = inspect.getsource(trouve.df_fn)
+                fn_source = inspect.getsource(trouve.transform)
             except (OSError, TypeError):
                 # A lambda, a built-in, or a compiled extension has no source text.
-                fn_source = repr(trouve.df_fn)
+                fn_source = repr(trouve.transform)
 
             imports_section = ""
             try:
-                source_file = inspect.getfile(trouve.df_fn)
+                source_file = inspect.getfile(trouve.transform)
                 source_text = Path(source_file).read_text()
                 tree = ast.parse(source_text)
                 import_lines = [
@@ -152,10 +154,12 @@ def write_compile_output(
             except (OSError, SyntaxError):
                 pass
 
-            input_lines = []
-            for param in inspect.signature(trouve.df_fn).parameters.values():
-                if isinstance(param.default, Trouve):
-                    input_lines.append(f"#   {param.name}  ->  {param.default.full_name}")
+            input_lines = [
+                f"#   {parameter_name}  ->  {upstream.full_name}"
+                for parameter_name, upstream in zip(
+                    trouve.parameter_names(), trouve.upstream_trouves()
+                )
+            ]
 
             header = f"# clair compiled: {trouve.full_name}\n# execution_type: pandas\n"
             if input_lines:
@@ -177,6 +181,7 @@ def write_compile_output(
             artifact_path.parent.mkdir(parents=True, exist_ok=True)
             artifact_path.write_text(artifact_content)
         elif trouve.compiled.execution_type == ExecutionType.SNOWFLAKE:
+            assert isinstance(trouve, Trouve)
             effective_mode = resolve_effective_mode(trouve, run_mode)
             statements = trouve.build_sql(effective_mode, run_id=run_id)
 
diff --git a/src/clair/core/dag.py b/src/clair/core/dag.py
index 95c4f67..b69549a 100644
--- a/src/clair/core/dag.py
+++ b/src/clair/core/dag.py
@@ -2,10 +2,12 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
+
 import networkx as nx
 
 from clair.exceptions import CyclicDependencyError
-from clair.trouves.trouve import Trouve, TrouveType
+from clair.trouves.trouve import TrouveAbc, TrouveType
 
 
 class ClairDag(nx.DiGraph):
@@ -16,7 +18,7 @@ class ClairDag(nx.DiGraph):
     (dependency, dependent) pair. Clair reads the pairs from the Trouve imports.
     """
 
-    def add_trouve(self, trouve: Trouve) -> None:
+    def add_trouve(self, trouve: TrouveAbc) -> None:
         """Add a compiled Trouve as a node. The key of the node is its full_name."""
         self.add_node(trouve.full_name, trouve=trouve)
 
@@ -38,7 +40,7 @@ def add_dependency(self, dependency: str, dependent: str) -> None:
             )
         self.add_edge(dependency, dependent)
 
-    def get_trouve(self, full_name: str) -> Trouve:
+    def get_trouve(self, full_name: str) -> TrouveAbc:
         """Give the Trouve of a node.
 
         Raises:
@@ -63,9 +65,9 @@ def validate(self) -> None:
             assert trouve is not None, (
                 f"Node '{node}' is missing the 'trouve' attribute"
             )
-            assert isinstance(trouve, Trouve), (
+            assert isinstance(trouve, TrouveAbc), (
                 f"Node '{node}' has a 'trouve' attribute of type "
-                f"{type(trouve).__name__}, expected Trouve"
+                f"{type(trouve).__name__}, expected TrouveAbc"
             )
 
         if not nx.is_directed_acyclic_graph(self):
@@ -81,12 +83,12 @@ def validate(self) -> None:
             )
 
     @property
-    def trouves(self) -> list[Trouve]:
+    def trouves(self) -> list[TrouveAbc]:
         """Give each compiled Trouve object in the graph."""
         return [self.nodes[node]["trouve"] for node in self.nodes]
 
 
-def build_dag(trouves: list[Trouve]) -> ClairDag:
+def build_dag(trouves: Sequence[TrouveAbc]) -> ClairDag:
     """Make a directed acyclic graph from the compiled Trouves.
 
     Raises:
diff --git a/src/clair/core/discovery.py b/src/clair/core/discovery.py
index e39f517..adcca63 100644
--- a/src/clair/core/discovery.py
+++ b/src/clair/core/discovery.py
@@ -7,6 +7,7 @@
 import os
 import re
 import sys
+from collections.abc import Sequence
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -21,9 +22,10 @@
 from clair.trouves._refs import THIS_PLACEHOLDER, TROUVE_PLACEHOLDER_PREFIX
 from clair.trouves._refs import clear as clear_refs
 from clair.trouves.config import DatabaseDefaults, ResolvedConfig, SchemaDefaults
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.run_config import RunMode
 from clair.trouves.test import TestSql
-from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveType
+from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveAbc, TrouveType
 
 ARTIFACTS_DIR_NAME = "_clairtifacts"
 _SKIP_DIRS = {"clair", "tests", ARTIFACTS_DIR_NAME, "__pycache__", ".git", ".venv", "node_modules"}
@@ -151,7 +153,7 @@ def discover_project(
     routing: RoutingConfig | None = None,
     environment: Environment | None = None,
     run_mode: RunMode | None = None,
-) -> list[Trouve]:
+) -> list[TrouveAbc]:
     """Find each Trouve in a project.
 
     The function reads the project root and loads each Trouve file. It replaces
@@ -209,7 +211,7 @@ def discover_project(
 
     # Load each candidate. A file can be in sys.modules already, because an
     # earlier candidate imported it as a dependency.
-    collected: list[tuple[Trouve, str, Path, str]] = []
+    collected: list[tuple[TrouveAbc, str, Path, str]] = []
     errors: list[str] = []
 
     for file_path in candidates:
@@ -234,7 +236,7 @@ def discover_project(
                 continue
 
         trouve_obj = getattr(module, "trouve", None)
-        if not isinstance(trouve_obj, Trouve):
+        if not isinstance(trouve_obj, TrouveAbc):
             continue
 
         collected.append((trouve_obj, full_name, file_path, module_name))
@@ -254,9 +256,9 @@ def discover_project(
         if trouve_obj.type != TrouveType.SOURCE:
             collision_check[full_name.upper()] = routed
 
-    # Make a map from an id to a logical name, for the df_fn dependencies. With
-    # this map, clair finds the logical name of each Trouve that a df_fn holds as
-    # a parameter default.
+    # Make a map from an id to a logical name, for the pandas dependencies. With
+    # this map, clair finds the logical name of each Trouve that a PandasTrouve
+    # names in its inputs.
     id_to_logical_name: dict[int, str] = {
         id(trouve_obj): logical_names[id(trouve_obj)]
         for trouve_obj, _, _, _ in collected
@@ -271,27 +273,27 @@ def discover_project(
         logical = logical_names[id(trouve_obj)]
         routed = routed_names[id(trouve_obj)]
 
-        if trouve_obj.df_fn is not None:
-            df_imports = []
-            for param in inspect.signature(trouve_obj.df_fn).parameters.values():
-                if isinstance(param.default, Trouve):
-                    dep_logical = id_to_logical_name.get(id(param.default))
-                    if dep_logical and dep_logical != logical and dep_logical not in df_imports:
-                        df_imports.append(dep_logical)
+        if trouve_obj.execution_type == ExecutionType.PANDAS:
+            assert isinstance(trouve_obj, PandasTrouve)
+            transform_imports = []
+            for upstream in trouve_obj.upstream_trouves():
+                dep_logical = id_to_logical_name.get(id(upstream))
+                if dep_logical and dep_logical != logical and dep_logical not in transform_imports:
+                    transform_imports.append(dep_logical)
 
             try:
-                resolved_df_fn = inspect.getsource(trouve_obj.df_fn)
+                resolved_transform = inspect.getsource(trouve_obj.transform)
             except OSError:
-                resolved_df_fn = repr(trouve_obj.df_fn)
+                resolved_transform = repr(trouve_obj.transform)
 
             trouve_obj.compiled = CompiledAttributes(
                 full_name=routed,
                 logical_name=logical,
                 resolved_sql="",
-                resolved_df_fn=resolved_df_fn,
+                resolved_transform=resolved_transform,
                 file_path=file_path.relative_to(project_root),
                 module_name=module_name,
-                imports=df_imports,
+                imports=transform_imports,
                 config=_resolve_config(file_path, project_root, profile_defaults),
                 execution_type=ExecutionType.PANDAS,
             )
@@ -299,6 +301,7 @@ def discover_project(
                 if isinstance(test, TestSql):
                     test.sql = _resolve_sql(test.sql, logical_names, this_name=logical)
         else:
+            assert isinstance(trouve_obj, Trouve)
             trouve_obj.compiled = CompiledAttributes(
                 full_name=routed,
                 logical_name=logical,
@@ -319,7 +322,7 @@ def discover_project(
     return [trouve for trouve, _, _, _ in collected]
 
 
-def find_routing_collisions(trouves: list[Trouve]) -> list[tuple[str, list[str]]]:
+def find_routing_collisions(trouves: Sequence[TrouveAbc]) -> list[tuple[str, list[str]]]:
     """Give a (routed_target, [logical_sources]) pair for each routing collision.
 
     A collision occurs when two Trouves that are not SOURCE Trouves route to one
@@ -337,7 +340,7 @@ def find_routing_collisions(trouves: list[Trouve]) -> list[tuple[str, list[str]]
     return detect_routing_collisions(logical_to_routed)
 
 
-def recompile_for_selection(trouves: list[Trouve], selected_names: set[str]) -> None:
+def recompile_for_selection(trouves: Sequence[TrouveAbc], selected_names: set[str]) -> None:
     """Change each selected upstream name in the SQL from logical to routed.
 
     After discover_project(), the resolved_sql of each Trouve holds the logical
diff --git a/src/clair/core/runner.py b/src/clair/core/runner.py
index 6153248..9d308c1 100644
--- a/src/clair/core/runner.py
+++ b/src/clair/core/runner.py
@@ -2,11 +2,9 @@
 
 from __future__ import annotations
 
-import inspect
 import time
 from collections.abc import Callable, Iterator
 from enum import StrEnum
-from typing import Any
 
 import networkx as nx
 import pandas as pd
@@ -15,8 +13,9 @@
 
 from clair.adapters.base import WarehouseAdapter
 from clair.core.dag import ClairDag, get_executable_nodes
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.run_config import RunMode
-from clair.trouves.trouve import Trouve, TrouveType
+from clair.trouves.trouve import ExecutionType, Trouve, TrouveType
 
 
 class RunStatus(StrEnum):
@@ -174,34 +173,33 @@ def resolve_effective_mode(trouve: Trouve, cli_run_mode: RunMode) -> RunMode:
     return RunMode.INCREMENTAL
 
 
-def _run_df_fn_trouve(
-    trouve: Trouve,
+def _run_pandas_trouve(
+    trouve: PandasTrouve,
     adapter: WarehouseAdapter,
 ) -> RunResult:
-    """Execute a df_fn Trouve. Read the inputs, transform them, write the output.
+    """Execute a PandasTrouve. Read the inputs, transform them, write the output.
 
     Returns a RunResult with the SUCCESS status or the FAILURE status.
     """
     start = time.monotonic()
 
-    # 1. Read each input DataFrame. inspect.signature gives the parameters.
-    dataframe_kwargs: dict[str, Any] = {}
-    for param_name, param in inspect.signature(trouve.df_fn).parameters.items():
-        if isinstance(param.default, Trouve):
-            try:
-                dataframe_kwargs[param_name] = adapter.fetch_dataframe(param.default.full_name)
-            except Exception as fetch_error:  # noqa: BLE001 — each adapter fault becomes a RunResult with the FAILURE status
-                duration = time.monotonic() - start
-                return RunResult(
-                    full_name=trouve.full_name,
-                    status=RunStatus.FAILURE,
-                    error=f"Failed to fetch '{param_name}' ({param.default.full_name}): {fetch_error}",
-                    duration_seconds=duration,
-                )
+    # 1. Read each input DataFrame. Clair keeps the order of trouve.inputs.
+    input_dataframes: list[pd.DataFrame] = []
+    for parameter_name, upstream in zip(trouve.parameter_names(), trouve.upstream_trouves()):
+        try:
+            input_dataframes.append(adapter.fetch_dataframe(upstream.full_name))
+        except Exception as fetch_error:  # noqa: BLE001 — each adapter fault becomes a RunResult with the FAILURE status
+            duration = time.monotonic() - start
+            return RunResult(
+                full_name=trouve.full_name,
+                status=RunStatus.FAILURE,
+                error=f"Failed to fetch '{parameter_name}' ({upstream.full_name}): {fetch_error}",
+                duration_seconds=duration,
+            )
 
-    # 2. Call the df_fn function.
+    # 2. Call the transform function. Clair binds each input by position.
     try:
-        result_dataframe = trouve.df_fn(**dataframe_kwargs)
+        result_dataframe = trouve.transform(*input_dataframes)
     except Exception as transform_error:  # noqa: BLE001 — the user transform code is unknown
         duration = time.monotonic() - start
         return RunResult(
@@ -333,11 +331,12 @@ def run_project(
                 adapter.execute(f"CREATE DATABASE IF NOT EXISTS {routed_parts[0]}")
                 adapter.execute(f"CREATE SCHEMA IF NOT EXISTS {routed_parts[0]}.{routed_parts[1]}")
 
-        # A df_fn Trouve is different. Clair reads the data, transforms it and
+        # A PandasTrouve is different. Clair reads the data, transforms it and
         # writes it. Clair does not execute SQL.
-        if trouve.df_fn is not None:
+        if trouve.execution_type == ExecutionType.PANDAS:
+            assert isinstance(trouve, PandasTrouve)
             logger.info("run.node.start", trouve=name, effective_mode="full_refresh")
-            result = _run_df_fn_trouve(trouve, adapter)
+            result = _run_pandas_trouve(trouve, adapter)
             yield result
 
             if result.status == RunStatus.SUCCESS:
@@ -351,6 +350,7 @@ def run_project(
                     skip_reasons.setdefault(desc, name)
             continue
 
+        assert isinstance(trouve, Trouve)
         effective_mode = resolve_effective_mode(trouve, run_mode)
         # If the target table does not exist yet, change to the full refresh mode.
         if effective_mode == RunMode.INCREMENTAL:
diff --git a/src/clair/trouves/__init__.py b/src/clair/trouves/__init__.py
index 9e5d65f..1d03cde 100644
--- a/src/clair/trouves/__init__.py
+++ b/src/clair/trouves/__init__.py
@@ -1,5 +1,6 @@
 from clair.trouves.column import Column, ColumnType
 from clair.trouves.config import DatabaseDefaults, SchemaDefaults
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.test import (
     AnyTest,
     Test,
@@ -8,13 +9,14 @@
     TestUnique,
     TestUniqueColumns,
 )
-from clair.trouves.trouve import Trouve, TrouveType
+from clair.trouves.trouve import Trouve, TrouveAbc, TrouveType
 
 __all__ = [
     "AnyTest",
     "Column",
     "ColumnType",
     "DatabaseDefaults",
+    "PandasTrouve",
     "SchemaDefaults",
     "Test",
     "TestNotNull",
@@ -22,5 +24,6 @@
     "TestUnique",
     "TestUniqueColumns",
     "Trouve",
+    "TrouveAbc",
     "TrouveType",
 ]
diff --git a/src/clair/trouves/pandas_trouve.py b/src/clair/trouves/pandas_trouve.py
new file mode 100644
index 0000000..528c66f
--- /dev/null
+++ b/src/clair/trouves/pandas_trouve.py
@@ -0,0 +1,84 @@
+"""PandasTrouve -- a Trouve that a Python function materializes.
+
+Clair reads each upstream Trouve from Snowflake into a DataFrame, gives the
+DataFrames to your function, and writes the result back to Snowflake.
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Callable
+
+import pandas as pd
+from pydantic import Field, model_validator
+
+from clair.trouves.run_config import RunMode
+from clair.trouves.trouve import ExecutionType, TrouveAbc, TrouveType
+
+_VARIADIC_KINDS = (inspect.Parameter.VAR_POSITIONAL, inspect.Parameter.VAR_KEYWORD)
+
+
+class PandasTrouve(TrouveAbc):
+    """A Trouve that a pandas function materializes.
+
+    Clair binds ``inputs`` to the parameters of ``transform`` by position. Thus
+    the first Trouve in ``inputs`` becomes the first parameter, the second
+    becomes the second parameter, and so on. Your function receives plain
+    DataFrames, so you can call it directly in a test or in a notebook.
+
+    Example:
+        >>> def daily_counts(events: pd.DataFrame) -> pd.DataFrame:
+        ...     return events.groupby("day", as_index=False).agg(n=("day", "size"))
+        >>> trouve = PandasTrouve(transform=daily_counts, inputs=[events_trouve])
+
+    Attributes:
+        transform: The function that gives the output DataFrame.
+        inputs: The upstream Trouves, in the parameter order of ``transform``.
+
+    ``TrouveAbc`` holds the attributes that every backend shares.
+    """
+
+    transform: Callable[..., pd.DataFrame] = Field(exclude=True)
+    inputs: list[TrouveAbc] = Field(default_factory=list, exclude=True)
+
+    @property
+    def execution_type(self) -> ExecutionType:
+        return ExecutionType.PANDAS
+
+    def upstream_trouves(self) -> list[TrouveAbc]:
+        """Give the upstream Trouves, in the parameter order of the transform."""
+        return list(self.inputs)
+
+    def parameter_names(self) -> list[str]:
+        """Give the parameter names of the transform, in order.
+
+        The runner and the compiler use these names in their messages. Clair
+        binds the inputs by position, thus a name has no effect on the DAG.
+        """
+        return list(inspect.signature(self.transform).parameters)
+
+    @model_validator(mode="after")
+    def _validate_transform(self) -> PandasTrouve:
+        if self.type != TrouveType.TABLE:
+            raise ValueError(
+                f"PandasTrouve must be TABLE type, got '{self.type.value}'"
+            )
+        if self.run_config.run_mode == RunMode.INCREMENTAL:
+            raise ValueError("PandasTrouve does not support incremental mode")
+
+        parameters = list(inspect.signature(self.transform).parameters.values())
+        variadic = [p.name for p in parameters if p.kind in _VARIADIC_KINDS]
+        if variadic:
+            raise ValueError(
+                f"transform must not have *args or **kwargs, found: {', '.join(variadic)}. "
+                "Clair binds each input to a named parameter, by position."
+            )
+        if len(parameters) != len(self.inputs):
+            parameter_names = ", ".join(p.name for p in parameters) or "(none)"
+            raise ValueError(
+                f"transform takes {len(parameters)} parameter(s) but inputs has "
+                f"{len(self.inputs)} Trouve(s). Clair binds them by position. "
+                f"Parameters: {parameter_names}"
+            )
+        return self
+
diff --git a/src/clair/trouves/trouve.py b/src/clair/trouves/trouve.py
index 223abc9..faad0c7 100644
--- a/src/clair/trouves/trouve.py
+++ b/src/clair/trouves/trouve.py
@@ -3,10 +3,19 @@
 One Trouve maps to one object in Snowflake that you can query: a source table,
 a transformed table, or a view. Each Trouve stays in its own .py file. The
 framework finds each Trouve automatically.
+
+This module holds two classes:
+
+* ``TrouveAbc`` -- the abstract base. It holds the attributes that every backend
+  shares: the columns, the tests, the docs, and the compiled attributes.
+* ``Trouve`` -- the SQL backend. Snowflake materializes it from SQL.
+
+The pandas backend, ``PandasTrouve``, stays in ``pandas_trouve.py``.
 """
 
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
 from enum import StrEnum
 from pathlib import Path
 from typing import Any
@@ -33,13 +42,13 @@ class ExecutionType(StrEnum):
 class CompiledAttributes(BaseModel):
     """The attributes that discovery sets after it loads a Trouve.
 
-    These attributes exist only when ``Trouve.is_compiled`` is True.
+    These attributes exist only when ``TrouveAbc.is_compiled`` is True.
     """
 
     full_name: str       # The routed name. Clair puts it in the SQL and the DDL.
     logical_name: str    # The name from the file path. DAG edges and selectors use it.
     resolved_sql: str
-    resolved_df_fn: str = ""
+    resolved_transform: str = ""
     file_path: Path
     module_name: str
     imports: list[str]
@@ -47,25 +56,24 @@ class CompiledAttributes(BaseModel):
     execution_type: ExecutionType
 
 
-class Trouve(BaseModel):
-    """The primary class in Clair.
+class TrouveAbc(BaseModel, ABC):
+    """The base class of every Trouve backend.
+
+    A subclass supplies the backend. ``Trouve`` runs SQL in Snowflake.
+    ``PandasTrouve`` runs a Python function on the clair machine. Each subclass
+    gives its own ``execution_type`` and its own ``upstream_trouves``.
 
     Attributes:
         type: SOURCE, TABLE, or VIEW.
-        sql: The SQL query. A TABLE or a VIEW needs it. A SOURCE must leave it
-             empty. To point to a different Trouve, write
-             ``f"SELECT * FROM {other_trouve}"``. Discovery replaces the
-             f-string placeholder with the true full_name.
         columns: The column definitions, for the docs and for future checks.
         tests: The data quality tests.
         docs: The documentation text for this Trouve.
+        run_config: The materialization strategy.
         compiled: Discovery sets this. It stays None until discovery reads the
              project.
     """
 
     type: TrouveType = Field(default=TrouveType.TABLE)
-    sql: str = Field(default="", exclude=True)
-    df_fn: Any = Field(default=None, exclude=True)
     columns: list[Column] = []
     tests: list[AnyTest] = []
     docs: str = ""
@@ -74,28 +82,18 @@ class Trouve(BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    @model_validator(mode="after")
-    def _validate_sql(self) -> Trouve:
-        if self.df_fn is not None:
-            if not callable(self.df_fn):
-                raise ValueError("df_fn must be callable")
-            if self.sql.strip():
-                raise ValueError("Trouve cannot have both sql and df_fn")
-            if self.type != TrouveType.TABLE:
-                raise ValueError(f"df_fn Trouves must be TABLE type, got '{self.type.value}'")
-            if self.run_config.run_mode == RunMode.INCREMENTAL:
-                raise ValueError("df_fn Trouves do not support incremental mode")
-            return self
+    @property
+    @abstractmethod
+    def execution_type(self) -> ExecutionType:
+        """The backend that materializes this Trouve."""
 
-        if self.type in (TrouveType.TABLE, TrouveType.VIEW) and not self.sql.strip():
-            raise ValueError(
-                f"Trouve of type '{self.type.value}' requires non-empty sql"
-            )
-        if self.type == TrouveType.SOURCE and self.sql.strip():
-            raise ValueError("SOURCE Trouve must not have sql")
-        if self.run_config.run_mode == RunMode.INCREMENTAL and self.type != TrouveType.TABLE:
-            raise ValueError("only TABLE Trouves support incremental mode")
-        return self
+    @abstractmethod
+    def upstream_trouves(self) -> list[TrouveAbc]:
+        """Give the Trouves that this Trouve reads, in a stable order.
+
+        A SQL Trouve gives an empty list. Its dependencies come from the
+        placeholder tokens in its SQL, and discovery reads those tokens.
+        """
 
     def __format__(self, _spec: str) -> str:
         """Give a placeholder token for f-string SQL.
@@ -135,6 +133,45 @@ def sample(self) -> str:
         assert self.compiled is not None, "sample() requires a compiled Trouve"
         return f"(SELECT TOP 1000 * FROM {self.compiled.full_name})"
 
+    def get_full_table_name(self) -> str:
+        """An alias for .full_name. Use it in f-string SQL."""
+        return self.full_name
+
+
+class Trouve(TrouveAbc):
+    """A Trouve that Snowflake materializes from SQL.
+
+    Attributes:
+        sql: The SQL query. A TABLE or a VIEW needs it. A SOURCE must leave it
+             empty. To point to a different Trouve, write
+             ``f"SELECT * FROM {other_trouve}"``. Discovery replaces the
+             f-string placeholder with the true full_name.
+
+    ``TrouveAbc`` holds the attributes that every backend shares.
+    """
+
+    sql: str = Field(default="", exclude=True)
+
+    @property
+    def execution_type(self) -> ExecutionType:
+        return ExecutionType.SNOWFLAKE
+
+    def upstream_trouves(self) -> list[TrouveAbc]:
+        """Give an empty list. Discovery reads the SQL placeholder tokens."""
+        return []
+
+    @model_validator(mode="after")
+    def _validate_sql(self) -> Trouve:
+        if self.type in (TrouveType.TABLE, TrouveType.VIEW) and not self.sql.strip():
+            raise ValueError(
+                f"Trouve of type '{self.type.value}' requires non-empty sql"
+            )
+        if self.type == TrouveType.SOURCE and self.sql.strip():
+            raise ValueError("SOURCE Trouve must not have sql")
+        if self.run_config.run_mode == RunMode.INCREMENTAL and self.type != TrouveType.TABLE:
+            raise ValueError("only TABLE Trouves support incremental mode")
+        return self
+
     def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]:
         """Make the SQL statements that materialize this Trouve.
 
@@ -215,7 +252,3 @@ def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]:
         )
 
         return [stmt_1, stmt_2, stmt_3]
-
-    def get_full_table_name(self) -> str:
-        """An alias for .full_name. Use it in f-string SQL."""
-        return self.full_name
diff --git a/tests/unit/test_compiler_pandas.py b/tests/unit/test_compiler_pandas.py
index 53c1928..bf0f16f 100644
--- a/tests/unit/test_compiler_pandas.py
+++ b/tests/unit/test_compiler_pandas.py
@@ -1,4 +1,4 @@
-"""The tests of the compiler output for a df_fn Trouve node."""
+"""The tests of the compiler output for a PandasTrouve node."""
 
 from __future__ import annotations
 
@@ -14,7 +14,7 @@
 
 
 def _make_pandas_project(tmp_path: Path) -> Path:
-    """Make a project with a SOURCE and a df_fn Trouve that has columns."""
+    """Make a project with a SOURCE and a PandasTrouve that has columns."""
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "derived").mkdir(parents=True)
 
@@ -25,14 +25,15 @@ def _make_pandas_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "summary.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve, Column, ColumnType
+        from clair import PandasTrouve, Column, ColumnType
         from mydb.source.events import trouve as source_events
 
-        def summarize(events: pd.DataFrame = source_events) -> pd.DataFrame:
+        def summarize(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(
-            df_fn=summarize,
+        trouve = PandasTrouve(
+            transform=summarize,
+            inputs=[source_events],
             columns=[
                 Column(name="event_type", type=ColumnType.STRING),
                 Column(name="event_count", type=ColumnType.NUMBER),
@@ -44,7 +45,7 @@ def summarize(events: pd.DataFrame = source_events) -> pd.DataFrame:
 
 
 def _make_mixed_project(tmp_path: Path) -> Path:
-    """Make a project with a SQL Trouve and a df_fn Trouve."""
+    """Make a project with a SQL Trouve and a PandasTrouve."""
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "refined").mkdir(parents=True)
     (tmp_path / "mydb" / "derived").mkdir(parents=True)
@@ -62,20 +63,20 @@ def _make_mixed_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "summary.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve
+        from clair import PandasTrouve
         from mydb.refined.events import trouve as refined_events
 
-        def summarize(events: pd.DataFrame = refined_events) -> pd.DataFrame:
+        def summarize(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(df_fn=summarize)
+        trouve = PandasTrouve(transform=summarize, inputs=[refined_events])
     """))
 
     return tmp_path
 
 
 def _make_no_columns_project(tmp_path: Path) -> Path:
-    """Make a project with a df_fn Trouve that has no columns."""
+    """Make a project with a PandasTrouve that has no columns."""
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "derived").mkdir(parents=True)
 
@@ -86,19 +87,19 @@ def _make_no_columns_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "summary.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve
+        from clair import PandasTrouve
         from mydb.source.events import trouve as source_events
 
-        def summarize(events: pd.DataFrame = source_events) -> pd.DataFrame:
+        def summarize(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(df_fn=summarize)
+        trouve = PandasTrouve(transform=summarize, inputs=[refined_events])
     """))
 
     return tmp_path
 
 
-class TestDfFnTrouveArtifactFile:
+class TestPandasTrouveArtifactFile:
     def test_artifact_file_ends_in_py(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         dag = build_dag(discover_project(project))
@@ -108,7 +109,7 @@ def test_artifact_file_ends_in_py(self, tmp_path: Path):
         py_file = tmp_path / "_clairtifacts" / FAKE_RUN_ID / "mydb" / "derived" / "summary.py"
         assert py_file.exists()
 
-    def test_no_sql_file_for_df_fn_trouve(self, tmp_path: Path):
+    def test_no_sql_file_for_pandas_trouve(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         dag = build_dag(discover_project(project))
         selected = get_executable_nodes(dag)
@@ -117,7 +118,7 @@ def test_no_sql_file_for_df_fn_trouve(self, tmp_path: Path):
         sql_file = tmp_path / "_clairtifacts" / FAKE_RUN_ID / "mydb" / "derived" / "summary.sql"
         assert not sql_file.exists()
 
-    def test_no_json_file_for_df_fn_trouve(self, tmp_path: Path):
+    def test_no_json_file_for_pandas_trouve(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         dag = build_dag(discover_project(project))
         selected = get_executable_nodes(dag)
@@ -127,7 +128,7 @@ def test_no_json_file_for_df_fn_trouve(self, tmp_path: Path):
         assert not json_file.exists()
 
 
-class TestDfFnTrouveArtifactContent:
+class TestPandasTrouveArtifactContent:
     def _get_artifact_content(self, tmp_path: Path) -> str:
         project = _make_pandas_project(tmp_path)
         dag = build_dag(discover_project(project))
@@ -155,7 +156,7 @@ def test_contains_function_source(self, tmp_path: Path):
         assert "def summarize" in content
 
 
-class TestDfFnTrouveCompiledNodeInfo:
+class TestPandasTrouveCompiledNodeInfo:
     def test_compiled_node_type_is_pandas(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         dag = build_dag(discover_project(project))
diff --git a/tests/unit/test_discovery.py b/tests/unit/test_discovery.py
index 97abc11..6bc72e9 100644
--- a/tests/unit/test_discovery.py
+++ b/tests/unit/test_discovery.py
@@ -16,7 +16,7 @@
 from clair.trouves._refs import TROUVE_PLACEHOLDER_PREFIX
 from clair.trouves.run_config import RunMode
 from clair.trouves.test import TestSql
-from clair.trouves.trouve import TrouveType
+from clair.trouves.trouve import Trouve, TrouveType
 
 
 class TestComputeFullName:
@@ -68,6 +68,7 @@ def test_resolved_sql_contains_full_name(self, simple_project: Path):
     def test_raw_sql_contains_placeholder(self, simple_project: Path):
         trouves = discover_project(simple_project)
         table = next(t for t in trouves if t.full_name == "analytics.revenue.daily_orders")
+        assert isinstance(table, Trouve)
         assert TROUVE_PLACEHOLDER_PREFIX in table.sql
 
     def test_config_resolution(self, simple_project: Path):
diff --git a/tests/unit/test_discovery_pandas.py b/tests/unit/test_discovery_pandas.py
index 73ec752..018343e 100644
--- a/tests/unit/test_discovery_pandas.py
+++ b/tests/unit/test_discovery_pandas.py
@@ -1,4 +1,4 @@
-"""The tests of the discovery of a df_fn Trouve node."""
+"""The tests of the discovery of a PandasTrouve node."""
 
 from __future__ import annotations
 
@@ -6,15 +6,16 @@
 from pathlib import Path
 
 from clair.core.discovery import discover_project
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.trouve import Trouve
 
 
 def _make_pandas_project(tmp_path: Path) -> Path:
-    """Make a small project with a SOURCE and a df_fn Trouve.
+    """Make a small project with a SOURCE and a PandasTrouve.
 
     The structure is:
         mydb/source/events.py       [SOURCE]
-        mydb/derived/summary.py     [a df_fn Trouve] reads source.events
+        mydb/derived/summary.py     [a PandasTrouve] reads source.events
     """
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "derived").mkdir(parents=True)
@@ -26,14 +27,15 @@ def _make_pandas_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "summary.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve, Column, ColumnType
+        from clair import PandasTrouve, Column, ColumnType
         from mydb.source.events import trouve as source_events
 
-        def summarize(events: pd.DataFrame = source_events) -> pd.DataFrame:
+        def summarize(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(
-            df_fn=summarize,
+        trouve = PandasTrouve(
+            transform=summarize,
+            inputs=[source_events],
             columns=[
                 Column(name="event_type", type=ColumnType.STRING),
                 Column(name="event_count", type=ColumnType.NUMBER),
@@ -46,12 +48,12 @@ def summarize(events: pd.DataFrame = source_events) -> pd.DataFrame:
 
 
 def _make_mixed_project(tmp_path: Path) -> Path:
-    """Make a project with a SQL Trouve and a df_fn Trouve.
+    """Make a project with a SQL Trouve and a PandasTrouve.
 
     The structure is:
         mydb/source/events.py       [SOURCE]
         mydb/refined/events.py      [a TABLE with SQL] reads source.events
-        mydb/derived/summary.py     [a df_fn Trouve] reads refined.events
+        mydb/derived/summary.py     [a PandasTrouve] reads refined.events
     """
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "refined").mkdir(parents=True)
@@ -70,25 +72,25 @@ def _make_mixed_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "summary.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve
+        from clair import PandasTrouve
         from mydb.refined.events import trouve as refined_events
 
-        def summarize(events: pd.DataFrame = refined_events) -> pd.DataFrame:
+        def summarize(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(df_fn=summarize)
+        trouve = PandasTrouve(transform=summarize, inputs=[refined_events])
     """))
 
     return tmp_path
 
 
 def _make_chained_pandas_project(tmp_path: Path) -> Path:
-    """Make a project where one df_fn Trouve depends on a different df_fn Trouve.
+    """Make a project where one PandasTrouve depends on a different PandasTrouve.
 
     The structure is:
         mydb/source/events.py       [SOURCE]
-        mydb/derived/step_one.py    [a df_fn Trouve] reads source.events
-        mydb/derived/step_two.py    [a df_fn Trouve] reads derived.step_one
+        mydb/derived/step_one.py    [a PandasTrouve] reads source.events
+        mydb/derived/step_two.py    [a PandasTrouve] reads derived.step_one
     """
     (tmp_path / "mydb" / "source").mkdir(parents=True)
     (tmp_path / "mydb" / "derived").mkdir(parents=True)
@@ -100,51 +102,51 @@ def _make_chained_pandas_project(tmp_path: Path) -> Path:
 
     (tmp_path / "mydb" / "derived" / "step_one.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve
+        from clair import PandasTrouve
         from mydb.source.events import trouve as source_events
 
-        def transform_one(events: pd.DataFrame = source_events) -> pd.DataFrame:
+        def transform_one(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(df_fn=transform_one)
+        trouve = PandasTrouve(transform=transform_one, inputs=[source_events])
     """))
 
     (tmp_path / "mydb" / "derived" / "step_two.py").write_text(textwrap.dedent("""\
         import pandas as pd
-        from clair import Trouve
+        from clair import PandasTrouve
         from mydb.derived.step_one import trouve as step_one
 
-        def transform_two(step_one_data: pd.DataFrame = step_one) -> pd.DataFrame:
+        def transform_two(step_one_data: pd.DataFrame) -> pd.DataFrame:
             return step_one_data
 
-        trouve = Trouve(df_fn=transform_two)
+        trouve = PandasTrouve(transform=transform_two, inputs=[step_one])
     """))
 
     return tmp_path
 
 
-class TestDfFnTrouveDetection:
-    def test_df_fn_trouve_is_discovered(self, tmp_path: Path):
+class TestPandasTrouveDetection:
+    def test_pandas_trouve_is_discovered(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
         names = {t.full_name for t in trouves}
         assert "mydb.derived.summary" in names
 
-    def test_df_fn_trouve_is_trouve_instance(self, tmp_path: Path):
+    def test_pandas_trouve_is_a_pandas_trouve_instance(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
         summary = next(t for t in trouves if t.full_name == "mydb.derived.summary")
-        assert isinstance(summary, Trouve)
-        assert summary.df_fn is not None
+        assert isinstance(summary, PandasTrouve)
+        assert summary.transform is not None
 
-    def test_df_fn_trouve_is_compiled(self, tmp_path: Path):
+    def test_pandas_trouve_is_compiled(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
         summary = next(t for t in trouves if t.full_name == "mydb.derived.summary")
         assert summary.is_compiled
 
 
-class TestDfFnTrouveDependencyExtraction:
+class TestPandasTrouveDependencyExtraction:
     def test_imports_contain_upstream(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
@@ -152,7 +154,7 @@ def test_imports_contain_upstream(self, tmp_path: Path):
         assert summary.compiled is not None
         assert "mydb.source.events" in summary.compiled.imports
 
-    def test_chained_df_fn_trouve_has_correct_imports(self, tmp_path: Path):
+    def test_chained_pandas_trouve_has_correct_imports(self, tmp_path: Path):
         project = _make_chained_pandas_project(tmp_path)
         trouves = discover_project(project)
         step_two = next(t for t in trouves if t.full_name == "mydb.derived.step_two")
@@ -160,7 +162,7 @@ def test_chained_df_fn_trouve_has_correct_imports(self, tmp_path: Path):
         assert "mydb.derived.step_one" in step_two.compiled.imports
 
 
-class TestDfFnTrouveCompiledAttributes:
+class TestPandasTrouveCompiledAttributes:
     def test_logical_name_set_correctly(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
@@ -182,7 +184,7 @@ def test_file_path_set_correctly(self, tmp_path: Path):
         assert summary.compiled is not None
         assert summary.compiled.file_path == Path("mydb/derived/summary.py")
 
-    def test_resolved_sql_is_empty_for_df_fn_trouve(self, tmp_path: Path):
+    def test_resolved_sql_is_empty_for_pandas_trouve(self, tmp_path: Path):
         project = _make_pandas_project(tmp_path)
         trouves = discover_project(project)
         summary = next(t for t in trouves if t.full_name == "mydb.derived.summary")
@@ -205,16 +207,16 @@ def test_sql_trouve_is_trouve_instance(self, tmp_path: Path):
         trouves = discover_project(project)
         refined = next(t for t in trouves if t.full_name == "mydb.refined.events")
         assert isinstance(refined, Trouve)
-        assert refined.df_fn is None
+        assert not isinstance(refined, PandasTrouve)
 
-    def test_df_fn_trouve_has_df_fn_set(self, tmp_path: Path):
+    def test_pandas_trouve_has_a_transform(self, tmp_path: Path):
         project = _make_mixed_project(tmp_path)
         trouves = discover_project(project)
         summary = next(t for t in trouves if t.full_name == "mydb.derived.summary")
-        assert isinstance(summary, Trouve)
-        assert summary.df_fn is not None
+        assert isinstance(summary, PandasTrouve)
+        assert summary.transform is not None
 
-    def test_df_fn_trouve_depends_on_sql_trouve(self, tmp_path: Path):
+    def test_pandas_trouve_depends_on_sql_trouve(self, tmp_path: Path):
         project = _make_mixed_project(tmp_path)
         trouves = discover_project(project)
         summary = next(t for t in trouves if t.full_name == "mydb.derived.summary")
diff --git a/tests/unit/test_runner_pandas.py b/tests/unit/test_runner_pandas.py
index f994b09..36fa1e3 100644
--- a/tests/unit/test_runner_pandas.py
+++ b/tests/unit/test_runner_pandas.py
@@ -1,4 +1,4 @@
-"""The tests of _run_df_fn_trouve, the pandas path of the runner."""
+"""The tests of _run_pandas_trouve, the pandas path of the runner."""
 
 from __future__ import annotations
 
@@ -8,8 +8,9 @@
 import pandas as pd
 
 from clair.adapters.base import QueryResult, WarehouseAdapter
-from clair.core.runner import RunStatus, _run_df_fn_trouve
+from clair.core.runner import RunStatus, _run_pandas_trouve
 from clair.trouves.config import ResolvedConfig
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveType
 
 
@@ -66,21 +67,21 @@ def _fetch(full_name: str) -> pd.DataFrame:
     return adapter
 
 
-class TestRunDfFnTrouveHappyPath:
-    def test_df_fn_called_and_result_written(self):
+class TestRunPandasTrouveHappyPath:
+    def test_transform_called_and_result_written(self):
         source = _make_source("db.schema.events")
         input_df = pd.DataFrame({"event_type": ["a", "b"], "count": [1, 2]})
         result_df = pd.DataFrame({"summary": [3]})
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return result_df
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(fetch_dataframes={"db.schema.events": input_df})
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.SUCCESS
         assert result.full_name == "db.schema.summary"
@@ -94,55 +95,55 @@ def test_fetch_called_for_each_input(self):
         df_a = pd.DataFrame({"x": [1]})
         df_b = pd.DataFrame({"y": [2]})
 
-        def my_fn(a: pd.DataFrame = source_a, b: pd.DataFrame = source_b) -> pd.DataFrame:  # type: ignore
+        def my_fn(a: pd.DataFrame, b: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"z": [3]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source_a, source_b])
         trouve.compiled = _make_compiled("db.schema.output")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.a": df_a, "db.schema.b": df_b}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.SUCCESS
         assert adapter.fetch_dataframe.call_count == 2
 
-    def test_df_fn_receives_correct_keyword_args(self):
+    def test_transform_receives_the_fetched_dataframe(self):
         source = _make_source("db.schema.events")
         input_df = pd.DataFrame({"col": [1]})
         received_kwargs = {}
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             received_kwargs["events"] = events
             return pd.DataFrame({"out": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(fetch_dataframes={"db.schema.events": input_df})
-        _run_df_fn_trouve(trouve, adapter)
+        _run_pandas_trouve(trouve, adapter)
 
         assert "events" in received_kwargs
         pd.testing.assert_frame_equal(received_kwargs["events"], input_df)
 
 
-class TestRunDfFnTrouveFullNameParsing:
+class TestRunPandasTrouveFullNameParsing:
     def test_database_schema_table_parsed_correctly(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"x": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("mydb.myschema.mytable")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.SUCCESS
         call_kwargs = adapter.write_dataframe.call_args.kwargs
@@ -154,109 +155,109 @@ def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
     def test_full_name_with_wrong_part_count_fails(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"x": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("only_two_parts.table")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "Cannot parse full_name" in result.error
 
 
-class TestRunDfFnTrouveTransformErrors:
-    def test_df_fn_raises_results_in_failure(self):
+class TestRunPandasTrouveTransformErrors:
+    def test_transform_raises_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def bad_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def bad_fn(events: pd.DataFrame) -> pd.DataFrame:
             raise ValueError("something went wrong")
 
-        trouve = Trouve(df_fn=bad_fn)
+        trouve = PandasTrouve(transform=bad_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "Transform function failed" in result.error
         assert "something went wrong" in result.error
 
-    def test_df_fn_returns_non_dataframe_results_in_failure(self):
+    def test_transform_returns_non_dataframe_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def bad_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def bad_fn(events: pd.DataFrame) -> pd.DataFrame:
             return {"not": "a dataframe"}  # type: ignore
 
-        trouve = Trouve(df_fn=bad_fn)
+        trouve = PandasTrouve(transform=bad_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "must return a pandas DataFrame" in result.error
         assert "dict" in result.error
 
-    def test_df_fn_returns_none_results_in_failure(self):
+    def test_transform_returns_none_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def bad_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def bad_fn(events: pd.DataFrame) -> pd.DataFrame:
             return None  # type: ignore
 
-        trouve = Trouve(df_fn=bad_fn)
+        trouve = PandasTrouve(transform=bad_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "must return a pandas DataFrame" in result.error
 
 
-class TestRunDfFnTrouveFetchErrors:
+class TestRunPandasTrouveFetchErrors:
     def test_fetch_failure_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return events
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_side_effect=RuntimeError("Snowflake connection lost")
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "Failed to fetch" in result.error
         assert "events" in result.error
 
 
-class TestRunDfFnTrouveWriteErrors:
+class TestRunPandasTrouveWriteErrors:
     def test_write_exception_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"x": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
@@ -264,7 +265,7 @@ def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
             write_side_effect=RuntimeError("Write failed"),
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert "Failed to write DataFrame" in result.error
@@ -272,10 +273,10 @@ def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
     def test_write_returns_success_false_results_in_failure(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"x": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
@@ -283,28 +284,28 @@ def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
             write_success=False,
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert result.error
 
 
 
-class TestRunDfFnTrouveResultFields:
+class TestRunPandasTrouveResultFields:
     def test_success_result_has_duration(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             return pd.DataFrame({"x": [1]})
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_dataframes={"db.schema.events": pd.DataFrame({"col": [1]})}
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.SUCCESS
         assert result.duration_seconds >= 0.0
@@ -312,17 +313,17 @@ def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
     def test_failure_result_has_duration(self):
         source = _make_source("db.schema.events")
 
-        def my_fn(events: pd.DataFrame = source) -> pd.DataFrame:  # type: ignore
+        def my_fn(events: pd.DataFrame) -> pd.DataFrame:
             raise ValueError("oops")
 
-        trouve = Trouve(df_fn=my_fn)
+        trouve = PandasTrouve(transform=my_fn, inputs=[source])
         trouve.compiled = _make_compiled("db.schema.summary")
 
         adapter = _make_df_adapter(
             fetch_side_effect=RuntimeError("fetch failed")
         )
 
-        result = _run_df_fn_trouve(trouve, adapter)
+        result = _run_pandas_trouve(trouve, adapter)
 
         assert result.status == RunStatus.FAILURE
         assert result.duration_seconds >= 0.0
diff --git a/tests/unit/test_trouves.py b/tests/unit/test_trouves.py
index f0902e7..e6a0854 100644
--- a/tests/unit/test_trouves.py
+++ b/tests/unit/test_trouves.py
@@ -2,11 +2,13 @@
 
 from pathlib import Path
 
+import pandas as pd
 import pytest
 
 from clair.trouves._refs import TROUVE_PLACEHOLDER_PREFIX, _registry, clear
 from clair.trouves.column import Column, ColumnType
 from clair.trouves.config import DatabaseDefaults, ResolvedConfig, SchemaDefaults
+from clair.trouves.pandas_trouve import PandasTrouve
 from clair.trouves.run_config import (
     IncrementalMode,
     RunConfig,
@@ -19,7 +21,7 @@
     TestUnique,
     TestUniqueColumns,
 )
-from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveType
+from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveAbc, TrouveType
 
 # ---------------------------------------------------------------------------
 # The helper functions.
@@ -1019,12 +1021,12 @@ def test_schema_defaults(self):
 
 
 # ---------------------------------------------------------------------------
-# A Trouve with a df_fn
+# PandasTrouve and the abstract base
 # ---------------------------------------------------------------------------
 
 
-class TestTrouveWithDfFn:
-    """The tests of the df_fn field of Trouve."""
+class TestPandasTrouve:
+    """The tests of PandasTrouve, the pandas backend."""
 
     def _make_upstream(self) -> Trouve:
         """Make a compiled SOURCE Trouve. A different Trouve can depend on it."""
@@ -1032,60 +1034,153 @@ def _make_upstream(self) -> Trouve:
         source.compiled = _compiled_attrs(full_name="db.schema.upstream", resolved_sql="")
         return source
 
-    def test_valid_construction_with_df_fn(self):
+    def test_valid_construction(self):
         upstream = self._make_upstream()
 
-        def my_fn(events=upstream):
+        def my_fn(events):
             return events
 
-        trouve = Trouve(df_fn=my_fn)
-        assert trouve.df_fn is my_fn
+        trouve = PandasTrouve(transform=my_fn, inputs=[upstream])
+        assert trouve.transform is my_fn
         assert trouve.type == TrouveType.TABLE
-        assert trouve.sql == ""
+        assert trouve.execution_type == ExecutionType.PANDAS
 
-    def test_df_fn_with_sql_raises_value_error(self):
+    def test_inputs_keep_object_identity(self):
+        """Pydantic must not copy an input. Discovery matches each input by id()."""
         upstream = self._make_upstream()
 
-        def my_fn(events=upstream):
+        def my_fn(events):
             return events
 
-        with pytest.raises(ValueError, match="cannot have both sql and df_fn"):
-            Trouve(df_fn=my_fn, sql="SELECT 1")
+        trouve = PandasTrouve(transform=my_fn, inputs=[upstream])
+        assert trouve.inputs[0] is upstream
+        assert trouve.upstream_trouves()[0] is upstream
 
-    def test_df_fn_with_view_type_raises_value_error(self):
-        upstream = self._make_upstream()
+    def test_upstream_trouves_keeps_the_input_order(self):
+        first = self._make_upstream()
+        second = self._make_upstream()
 
-        def my_fn(events=upstream):
-            return events
+        def my_fn(a, b):
+            return a
 
-        with pytest.raises(ValueError, match="df_fn Trouves must be TABLE type"):
-            Trouve(df_fn=my_fn, type=TrouveType.VIEW)
+        trouve = PandasTrouve(transform=my_fn, inputs=[first, second])
+        assert trouve.upstream_trouves() == [first, second]
 
-    def test_df_fn_with_source_type_raises_value_error(self):
+    def test_parameter_names_gives_the_signature_order(self):
+        first = self._make_upstream()
+        second = self._make_upstream()
+
+        def my_fn(catalog, reviews):
+            return catalog
+
+        trouve = PandasTrouve(transform=my_fn, inputs=[first, second])
+        assert trouve.parameter_names() == ["catalog", "reviews"]
+
+    def test_transform_stays_directly_callable(self):
+        """A user must be able to call the transform in a test or a notebook."""
         upstream = self._make_upstream()
 
-        def my_fn(events=upstream):
+        def my_fn(events):
             return events
 
-        with pytest.raises(ValueError, match="df_fn Trouves must be TABLE type"):
-            Trouve(df_fn=my_fn, type=TrouveType.SOURCE)
+        trouve = PandasTrouve(transform=my_fn, inputs=[upstream])
+        frame = pd.DataFrame({"a": [1, 2]})
+        assert trouve.transform(frame) is frame
 
-    def test_df_fn_not_callable_raises_value_error(self):
-        with pytest.raises(ValueError, match="df_fn must be callable"):
-            Trouve(df_fn="not_a_function")
+    def test_no_inputs_and_no_parameters_is_valid(self):
+        def my_fn():
+            return pd.DataFrame()
 
-    def test_df_fn_with_incremental_raises_value_error(self):
-        upstream = self._make_upstream()
+        trouve = PandasTrouve(transform=my_fn)
+        assert trouve.inputs == []
 
-        def my_fn(events=upstream):
-            return events
+    def test_too_few_inputs_raises_value_error(self):
+        def my_fn(a, b):
+            return a
 
-        with pytest.raises(ValueError, match="df_fn Trouves do not support incremental mode"):
-            Trouve(
-                df_fn=my_fn,
+        with pytest.raises(ValueError, match="takes 2 parameter\\(s\\) but inputs has 1"):
+            PandasTrouve(transform=my_fn, inputs=[self._make_upstream()])
+
+    def test_too_many_inputs_raises_value_error(self):
+        def my_fn(a):
+            return a
+
+        with pytest.raises(ValueError, match="takes 1 parameter\\(s\\) but inputs has 2"):
+            PandasTrouve(
+                transform=my_fn,
+                inputs=[self._make_upstream(), self._make_upstream()],
+            )
+
+    def test_arity_error_names_the_parameters(self):
+        def my_fn(catalog, reviews):
+            return catalog
+
+        with pytest.raises(ValueError, match="Parameters: catalog, reviews"):
+            PandasTrouve(transform=my_fn, inputs=[])
+
+    def test_var_positional_raises_value_error(self):
+        def my_fn(*frames):
+            return frames
+
+        with pytest.raises(ValueError, match="must not have \\*args or \\*\\*kwargs"):
+            PandasTrouve(transform=my_fn, inputs=[self._make_upstream()])
+
+    def test_var_keyword_raises_value_error(self):
+        def my_fn(**frames):
+            return frames
+
+        with pytest.raises(ValueError, match="must not have \\*args or \\*\\*kwargs"):
+            PandasTrouve(transform=my_fn, inputs=[])
+
+    def test_view_type_raises_value_error(self):
+        def my_fn():
+            return pd.DataFrame()
+
+        with pytest.raises(ValueError, match="PandasTrouve must be TABLE type"):
+            PandasTrouve(transform=my_fn, type=TrouveType.VIEW)
+
+    def test_source_type_raises_value_error(self):
+        def my_fn():
+            return pd.DataFrame()
+
+        with pytest.raises(ValueError, match="PandasTrouve must be TABLE type"):
+            PandasTrouve(transform=my_fn, type=TrouveType.SOURCE)
+
+    def test_transform_not_callable_raises_value_error(self):
+        with pytest.raises(ValueError, match="transform"):
+            PandasTrouve(transform="not_a_function")
+
+    def test_incremental_raises_value_error(self):
+        def my_fn():
+            return pd.DataFrame()
+
+        with pytest.raises(ValueError, match="PandasTrouve does not support incremental mode"):
+            PandasTrouve(
+                transform=my_fn,
                 run_config=RunConfig(
                     run_mode=RunMode.INCREMENTAL,
                     incremental_mode=IncrementalMode.APPEND,
                 ),
             )
 
+    def test_pandas_trouve_is_a_trouve_abc(self):
+        """Discovery finds a node with isinstance(obj, TrouveAbc)."""
+        def my_fn():
+            return pd.DataFrame()
+
+        assert isinstance(PandasTrouve(transform=my_fn), TrouveAbc)
+
+
+class TestTrouveAbc:
+    """The tests of the abstract base of every backend."""
+
+    def test_trouve_abc_is_not_instantiable(self):
+        with pytest.raises(TypeError, match="abstract"):
+            TrouveAbc()  # type: ignore[abstract]
+
+    def test_sql_trouve_gives_no_upstream_trouves(self):
+        """A SQL Trouve declares its dependencies with placeholder tokens."""
+        assert Trouve(sql="SELECT 1").upstream_trouves() == []
+
+    def test_sql_trouve_execution_type(self):
+        assert Trouve(sql="SELECT 1").execution_type == ExecutionType.SNOWFLAKE

From 5d2931addc98444901621e8d95735a37a0df045a Mon Sep 17 00:00:00 2001
From: OmerBaddour <omer.m.baddour@gmail.com>
Date: Sun, 2 Aug 2026 19:44:07 -0400
Subject: [PATCH 2/2] fix: widen collect_routing_problems to accept TrouveAbc

Discovery gives a list[TrouveAbc] after the PandasTrouve split. The
function reads only .compiled and .type, which the base class holds.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 src/clair/environments/routing.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/clair/environments/routing.py b/src/clair/environments/routing.py
index 550bd73..c2dfaad 100644
--- a/src/clair/environments/routing.py
+++ b/src/clair/environments/routing.py
@@ -32,7 +32,7 @@
 from clair.trouves.trouve import TrouveType
 
 if TYPE_CHECKING:
-    from clair.trouves.trouve import Trouve
+    from clair.trouves.trouve import TrouveAbc
 
 
 # Snowflake accepts these characters in an unquoted identifier.
@@ -278,7 +278,7 @@ def route(
 
 
 def collect_routing_problems(
-    trouves: list[Trouve],
+    trouves: list[TrouveAbc],
     routing: RoutingEntry | None,
 ) -> list[tuple[str, str]]:
     """Apply a routing entry to every Trouve and collect all the failures.