feat: udfs= surface — pyo3 trampoline, wide list boundary, .pyi contract

DuckDBInferFn(..., udfs=[...]) accepts duck-typed protocol objects (name/
takes/returns, scalar __call__, an instances attribute marking the implicit
leading i64 id), builds ExternSpecs for prepare and boxed GIL trampolines
for both backends. Width-k bare items emit one list|None field at every
boundary (marshaller, generic, infer_arrow as pa.list_). Constant fallback
is off under declared udfs. Contract parameterized, not weakened: bit-exact
vs DuckDB with the same udfs registered, or refuse.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: ahrzb

docs/superpowers/specs/2026-07-29-confit-udf-externs-design.md

@@ -0,0 +1,62 @@
+# Confit: params-join wiring + UDF externs (DRAFT-22, steps 2+3)
+
+Full design: `backlog/drafts/draft-22 - UDF-protocol-and-Confit-externs-serving-transformers-without-special-support.md`
+(+ its addendum and DRAFT-23). Scope per AmirHossein: **PythonTransform tier
+only** — the opaque-callable fallback/oracle; native `confit.udfs.*` families
+are the next loop (DRAFT-23).
+
+## What Confit gains
+
+**Step 3 — the marginalizer's join shape builds.**
+
+- `IS NOT DISTINCT FROM` join keys against static build sides: a NULL key is
+  an ordinary key value (one bucket) — encoded as a (validity, payload) pair
+  in the probe map, on both the build and probe sides.
+- The keyless always-true LEFT JOIN (`ON ((1 = 1))`) against a one-row
+  static.
+- Both under "map"/"filter" shapes: these joins are 0-or-1 by the params
+  contract, and the LEFT-join rule of the exactly-one proof covers them.
+
+**Step 2 — declared UDF externs.**
+
+- New program header `extern @N: name(tys) -> (tys)` + `ecall` instruction.
+  Operand layout: one (validity, payload) pair per declared param in; a
+  whole-call validity (the NULL list, distinct from a list of NULLs) plus a
+  (validity, payload) pair per declared return out.
+- Both backends route through one shared `call_extern` that enforces the
+  declared return shape — wrong length/type or a raised exception is a
+  named trap. The cranelift path is an `extern "C"` helper over the same
+  function (the no-drift rule).
+- `DuckDBInferFn(..., udfs=[...])`: duck-typed protocol objects (`name`,
+  `takes`, `returns`, scalar `__call__`; an `instances` attribute marks the
+  implicit leading nullable-i64 instance id). The trampoline attaches to
+  Python per call, converts NULL-masked scalars both ways under the
+  declared types.
+- Width-1 calls are ordinary scalar expressions (mid-expression works).
+  Width-k calls are bare SELECT items emitting ONE `list | None` field —
+  assembled at every boundary (row marshaller, generic path, `infer_arrow`
+  as `pa.list_`), NULL list when the id/whole-call is NULL. Mid-expression
+  width-k refuses by name.
+- The contract is parameterized, not weakened: bit-exact vs DuckDB *with
+  the same udfs registered* (`create_function`), or refuse. Unknown
+  functions without a matching declaration refuse exactly as before; the
+  static-only constant fallback is off when udfs are declared.
+
+## Gate
+
+- Rust: 189 tests including the DRAFT-22 serving-shape end-to-end, IR
+  round-trip/verify for `extern`/`ecall`, and the cranelift-vs-interpreter
+  random-IR differential.
+- Python: `tests/test_params_joins.py` (INDF/keyless differentials vs
+  DuckDB) and `tests/test_udfs.py` (marginalizer shape, wide list fields,
+  arrow boundary, both backends agree, trap/refusal surface) — all
+  differential against DuckDB with the same objects registered.
+
+## Deliberately out (next loops)
+
+- Native typed entries (`confit.udfs.PCA/StandardScaler/TreeEnsemble`) —
+  DRAFT-23, behind the same extern slots.
+- `SQLProjection.infer/infer_batch` wiring through this surface (the
+  integration loop; needs this PR + sql-transform's step 1, both merged).
+- Vectorized `apply_batch` binding for `infer_arrow` (today: per-row calls
+  through the scalar protocol — correct, boundary-amortization later).

packages/confit/confit/_engine.pyi

@@ -10,8 +10,9 @@ class DuckDBInferFn:
     """SQL specialized against frozen static tables, served bit-exact with DuckDB.
 
     Construction either succeeds with a function that matches DuckDB
-    bit-for-bit, or raises `ValueError` naming the unsupported construct. There
-    is no third mode.
+    bit-for-bit — *with the declared `udfs` registered*, when there are any —
+    or raises `ValueError` naming the unsupported construct. There is no
+    third mode.
     """
 
     output_model: type[BaseModel] | None
@@ -21,11 +22,27 @@ class DuckDBInferFn:
         sql: str,
         row_tables: dict[str, type[BaseModel]],
         static_tables: dict[str, pa.Table],
+        udfs: list[Any] | None = None,
         output_model: type[BaseModel] | None = None,
         output: str | None = None,
         shape: str | None = None,
     ) -> None:
-        """`output`: "model" (typed, default) or "dict".
+        """`udfs`: declared opaque scalar functions the SQL may call
+        (DRAFT-22). Each object carries `name: str`, `takes: tuple[str, ...]`
+        and `returns: tuple[str, ...]` in the engine type vocabulary
+        ("i1" | "i64" | "f64" | "str"), and a scalar `__call__(*args)`
+        receiving one plain Python value per argument (None for NULL) and
+        returning a tuple matching `returns`, or None for an all-NULL
+        result. An object with an `instances` attribute is a fitted
+        transformer: its implicit leading argument is a nullable BIGINT
+        instance id (never written in `takes`). Width-1 calls are scalar
+        expressions; width-k calls are bare SELECT items emitting one
+        `list | None` field. The oracle statement is parameterized, not
+        weakened: the engine matches DuckDB running the same SQL with these
+        same objects registered via `create_function`. UDFs must be
+        deterministic.
+
+        `output`: "model" (typed, default) or "dict".
 
         `shape`: the output-multiplicity contract, proven at build time --
         "map" (exactly one row out per row in; rejects WHERE and inner joins),

packages/confit/src/duckdb/arrow.rs

@@ -230,15 +230,50 @@ fn bitmap(oks: impl Iterator<Item = bool>, n: usize) -> (Vec<u8>, usize) {
     (bits, nulls)
 }
 
-/// Engine output -> pa.Table, one `Array.from_buffers` per column.
-pub fn emit(py: Python<'_>, out_cols: &[Col], st: &RunState) -> PyResult<Py<PyAny>> {
+/// Engine output -> pa.Table, one `Array.from_buffers` per scalar field; a
+/// wide UDF field assembles as `pa.array` of `None | [components]` (the
+/// python-list path — the wide boundary is transformer output, not the
+/// scalar hot lane).
+pub fn emit(
+    py: Python<'_>,
+    out_cols: &[Col],
+    plan: &[super::EmitField],
+    st: &RunState,
+) -> PyResult<Py<PyAny>> {
     let pa = PyModule::import(py, "pyarrow")?;
     let from_buffers = pa.getattr("Array")?.getattr("from_buffers")?;
     let py_buffer = pa.getattr("py_buffer")?;
     let n = st.emitted;
-    let mut arrays = Vec::with_capacity(out_cols.len());
-    let mut names = Vec::with_capacity(out_cols.len());
-    for (c, oc) in out_cols.iter().zip(&st.out) {
+    let mut arrays = Vec::with_capacity(plan.len());
+    let mut names = Vec::with_capacity(plan.len());
+    for field in plan {
+        let (c, oc) = match field {
+            super::EmitField::Scalar(i) => (&out_cols[*i], &st.out[*i]),
+            super::EmitField::Wide {
+                name,
+                valid,
+                first,
+                width,
+            } => {
+                names.push(name.clone());
+                let elem = match out_cols[*first].ty.ty {
+                    crate::specializer::ir::Ty::I1 => pa.call_method0("bool_")?,
+                    crate::specializer::ir::Ty::I64 => pa.call_method0("int64")?,
+                    crate::specializer::ir::Ty::F64 => pa.call_method0("float64")?,
+                    crate::specializer::ir::Ty::Str => pa.call_method0("large_string")?,
+                };
+                let list_ty = pa.call_method1("list_", (elem,))?;
+                let mut values = Vec::with_capacity(n);
+                for r in 0..n {
+                    values.push(super::wide_py(py, st, *valid, *first, *width, r)?);
+                }
+                let vals = PyList::new(py, values)?;
+                let kw = pyo3::types::PyDict::new(py);
+                kw.set_item("type", list_ty)?;
+                arrays.push(pa.call_method("array", (vals,), Some(&kw))?);
+                continue;
+            }
+        };
         names.push(c.name.clone());
         let (dtype, validity, bufs): (Bound<'_, PyAny>, (Vec<u8>, usize), Vec<Py<PyAny>>) =
             match oc {