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 e03033c..f914840 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:
 
@@ -388,31 +385,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),
@@ -423,6 +417,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 d31cb3d..7d842f3 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 count of each event type. This Trouve reads the 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 86f3fab..8028814 100644
--- a/site_docs/docs/concepts/trouve.md
+++ b/site_docs/docs/concepts/trouve.md
@@ -35,7 +35,6 @@ trouve = Trouve(
 | `tests` | `list[AnyTest]` | `[]` | Data quality tests. See [Tests](../guides/data-quality-tests.md). |
 | `docs` | `str` | `""` | Documentation string. `clair docs` shows it. |
 | `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`)
 
-If SQL is not the correct tool, give the Trouve a `df_fn` in place of `sql`. You supply a Python function. clair reads the upstream tables from Snowflake as DataFrames. Then it calls your function on the machine that runs clair, and writes the result to Snowflake.
+If SQL is not the correct tool, use a `PandasTrouve` in place of a `Trouve`. You supply a Python function. clair reads the upstream tables from Snowflake as DataFrames. Then it calls your function on the machine that runs clair, and writes the result 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 with the upstream Trouve as its default value.
+clair binds each item of `inputs` to a transform parameter by position. The transform takes plain DataFrames, thus you can call it directly in a test or in a notebook.
 
-The two run types have these differences:
+The two backends have these differences:
 
-| | `sql` | `df_fn` |
+| | `Trouve` | `PandasTrouve` |
 |---|---|---|
 | Runs | In Snowflake | On the clair machine |
 | Output type | TABLE or VIEW | TABLE only |
 | Incremental | Yes | Full refresh only |
-| Dependencies | f-string references | Parameter default values |
+| Dependencies | f-string references in `sql` | The `inputs` list |
 
-All the other behaviour is the same: the DAG, the `--select` flag, the data quality tests, and the `clair dag` output. You cannot use the two fields together. A Trouve with `sql` and `df_fn` causes an error.
+All the other behaviour is the same: the DAG, the `--select` flag, the data quality tests, and the `clair dag` output. The two classes have the same base class, `TrouveAbc`. That base class holds `columns`, `tests`, `docs`, and `run_config`.
 
 See the [Pandas-native guide](../guides/pandas-native.md) for a full example.
 
diff --git a/site_docs/docs/guides/pandas-native.md b/site_docs/docs/guides/pandas-native.md
index b422aa1..06c98c8 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 reads your upstream tables from Snowflake as DataFrames. Then it calls your function on the machine that runs clair, and writes the result to Snowflake. The DAG, the lineage, the selectors, and the data quality tests all apply.
+A `PandasTrouve` lets you write a pipeline step as a plain Python function. clair reads your upstream tables from Snowflake as DataFrames. Then it calls your function on the machine that runs clair, and writes the result to Snowflake. The DAG, the lineage, the selectors, and the data quality tests all apply.
 
-## When to use `df_fn`
+## When to use `PandasTrouve`
 
 Use it if SQL is the incorrect tool for the task:
 
@@ -11,11 +11,11 @@ Use it if SQL is the incorrect tool for the task:
 - Aggregations of many steps that depend on Python state between the steps
 - Logic that you already have as pandas code
 
-For all other work, give the `Trouve` a `sql` string. The SQL runs in Snowflake, and the data does not move on the network.
+For all other work, use a `Trouve` with a `sql` string. The SQL runs in Snowflake, and the data does not move on 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. **Read** — 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 DataFrame from your function to Snowflake. clair creates or replaces the table.
 4. **Test** — run the attached tests against the output table in Snowflake.
 
@@ -85,9 +90,16 @@ If your function returns a different type than `DataFrame`, the run fails with a
 !!! note
     clair reads the data into the memory of 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.
 - **A full table read.** clair reads all the upstream rows into memory. Chunked reads are 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 each one to a transform parameter by position. |
 | `columns` | `list[Column]` | `[]` | Column definitions. Optional — clair uses them for the documentation. |
 | `tests` | `list[AnyTest]` | `[]` | Data quality tests. They run after clair writes the output. |
 | `docs` | `str` | `""` | Documentation string. `clair docs` shows it. |
 
 ## 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 2c7d002..e912d54 100644
--- a/site_docs/docs/index.md
+++ b/site_docs/docs/index.md
@@ -29,7 +29,7 @@ Import the upstream Trouve and use it in the f-string. clair does the rest.
 - **Compile first, run second.** `clair compile` resolves the full DAG and writes SQL to `_clairtifacts/` before it connects to Snowflake.
 - **clair includes incremental strategies.** Use 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 reads the upstream tables as DataFrames, runs your code on your machine, then writes the result to Snowflake.
+- **Pandas-native transformations.** Use a [`PandasTrouve`](guides/pandas-native.md) to write any step as a Python function. clair reads the upstream tables as DataFrames, runs your code on your machine, then writes the result to Snowflake.
 
 ## Install
 
diff --git a/site_docs/docs/reference/index.md b/site_docs/docs/reference/index.md
index 34cebc9..c7076de 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 bd70e43..28f4505 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` | Tells you if this is a SOURCE, a TABLE, or a 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. `clair docs` shows it. |
@@ -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 @@ Discovery sets these attributes on `Trouve.compiled`. They are available after `
 |-----------|------|-------------|
 | `full_name` | `str` | The routed Snowflake name. clair uses it in the SQL and the DDL. |
 | `logical_name` | `str` | The name from the file system. clair uses it for the DAG edges and the selectors. |
-| `resolved_sql` | `str` | The SQL. clair replaced each placeholder token with a real full_name. |
+| `resolved_sql` | `str` | The SQL. clair replaced each placeholder token with a real full_name. It is empty for a `PandasTrouve`. |
+| `resolved_transform` | `str` | The source text of the transform function. It is empty for a SQL `Trouve`. |
 | `file_path` | `Path` | Absolute path to the Trouve file |
 | `imports` | `list[str]` | The logical names of the 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` with a `Trouve` as its default value becomes an upstream dependency. clair passes the DataFrame as that keyword argument. |
+| Dependencies | Each Trouve in `inputs` becomes an upstream dependency. clair binds each one to a transform parameter 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 970fe84..1c38b08 100644
--- a/src/clair/__init__.py
+++ b/src/clair/__init__.py
@@ -46,6 +46,7 @@
 from clair.environments.routing import RoutingEntry, RoutingTable, TrouveAddress
 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,
@@ -64,7 +65,7 @@
     TestUnique,
     TestUniqueColumns,
 )
-from clair.trouves.trouve import Trouve, TrouveType
+from clair.trouves.trouve import Trouve, TrouveAbc, TrouveType
 
 __version__ = "0.1.0"
 
@@ -85,6 +86,7 @@
     "ColumnType",
     "DatabaseDefaults",
     "IncrementalMode",
+    "PandasTrouve",
     "RoutingEntry",
     "RoutingTable",
     "RunConfig",
@@ -97,6 +99,7 @@
     "TestUnique",
     "TestUniqueColumns",
     "Trouve",
+    "TrouveAbc",
     "TrouveAddress",
     "TrouveType",
     "UpsertConfig",
diff --git a/src/clair/core/compiler.py b/src/clair/core/compiler.py
index 853529c..2e2273e 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"Clair did not compile {name}"
         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 a207b5b..2b56e33 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 2978840..acca423 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: RoutingEntry | 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 5955b2e..f28f6ad 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"Clair cannot read the input '{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"Clair cannot read the input '{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/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.
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 56bc88f..4230ef9 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 a callable object")
-            if self.sql.strip():
-                raise ValueError("a Trouve must have sql or df_fn, but not both")
-            if self.type != TrouveType.TABLE:
-                raise ValueError(f"a df_fn Trouve must have the TABLE type, but this Trouve has the type '{self.type.value}'")
-            if self.run_config.run_mode == RunMode.INCREMENTAL:
-                raise ValueError("a df_fn Trouve cannot use the 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"a Trouve with the type '{self.type.value}' must have sql"
-            )
-        if self.type == TrouveType.SOURCE and self.sql.strip():
-            raise ValueError("a SOURCE Trouve must not have sql")
-        if self.run_config.run_mode == RunMode.INCREMENTAL and self.type != TrouveType.TABLE:
-            raise ValueError("only a TABLE Trouve can use the 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() needs 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"a Trouve with the type '{self.type.value}' must have sql"
+            )
+        if self.type == TrouveType.SOURCE and self.sql.strip():
+            raise ValueError("a SOURCE Trouve must not have sql")
+        if self.run_config.run_mode == RunMode.INCREMENTAL and self.type != TrouveType.TABLE:
+            raise ValueError("only a TABLE Trouve can use the 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 0fc0371..f5561a9 100644
--- a/tests/unit/test_discovery.py
+++ b/tests/unit/test_discovery.py
@@ -15,7 +15,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
 from tests.helpers import DatabaseOverrideRouting, SchemaIsolationRouting
 
 
@@ -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 464e716..18ce57e 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 divide the 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 "The 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 "cannot read the input" 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 "cannot write the 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 7f47eee..78a5c43 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="must have sql or df_fn, but not both"):
-            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="a df_fn Trouve must have the 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="a df_fn Trouve must have the 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 a callable object"):
-            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="a df_fn Trouve cannot use the 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