# Phase 7.7b — User-Defined Functions (UDFs) Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Expose `FunctionRegistry` as a public extension point — letting embedders register custom aggregate functions and filter predicates that plug into the same grouping, window, and `:where` evaluation paths as built-ins.

**Architecture:** Extend `AggregateDesc` with an `AggImpl` discriminator (`Builtin(WindowOps)` vs `Udf(UdfOps)`) keeping the built-in path untouched. Add `UdfOps` (Arc-boxed closures with type-erased `Box<dyn Any + Send>` accumulator) and `PredicateDesc` to `FunctionRegistry`. Surface `register_aggregate` and `register_predicate` on `Minigraf`. Update parser to emit `WindowFunc::Udf(String)` and `UnaryOp::Udf(String)` for unknown names; executor resolves at runtime.

**Tech Stack:** Rust, `std::any::Any`, `std::sync::{Arc, RwLock}`, existing `anyhow::Result`, `minigraf` crate internals.

---

## File Map

| File | Action | What changes |
|------|--------|-------------|
| `src/query/datalog/types.rs` | Modify | Add `WindowFunc::Udf(String)`, `UnaryOp::Udf(String)` |
| `src/query/datalog/functions.rs` | Modify | Add `UdfOps`, `AggImpl`, `PredicateDesc`; update `AggregateDesc`, `FunctionRegistry` |
| `src/query/datalog/parser.rs` | Modify | `parse_window_expr`: unknown → `Udf`; `parse_expr`: unknown 1-arg → `Udf` |
| `src/query/datalog/executor.rs` | Modify | `eval_expr` + `apply_expr_clauses` registry param; `compute_aggregation` UDF arm; `apply_window_functions` `Udf` arm; `WindowSpec::func_name` updated |
| `src/db.rs` | Modify | Add `register_aggregate`, `register_predicate` to `Minigraf` |
| `tests/udf_test.rs` | Create | 9 integration tests |

---

## Task 1: Extend `types.rs` — add `Udf` variants

**Files:**
- Modify: `src/query/datalog/types.rs:99-109` (`WindowFunc` enum)
- Modify: `src/query/datalog/types.rs:170-177` (`UnaryOp` enum)
- Modify: `src/query/datalog/types.rs:131-143` (`WindowSpec::func_name`)

- [ ] **Step 1: Add `WindowFunc::Udf` variant**

In `src/query/datalog/types.rs`, change `WindowFunc` from:

```rust
pub enum WindowFunc {
    Sum,
    Count,
    Min,
    Max,
    Avg,
    Rank,
    RowNumber,
}
```

to:

```rust
pub enum WindowFunc {
    Sum,
    Count,
    Min,
    Max,
    Avg,
    Rank,
    RowNumber,
    /// Any function name not recognised at parse time — resolved against
    /// `FunctionRegistry` at query execution time.
    Udf(String),
}
```

- [ ] **Step 2: Add `UnaryOp::Udf` variant**

In `src/query/datalog/types.rs`, change `UnaryOp` from:

```rust
pub enum UnaryOp {
    StringQ,
    IntegerQ,
    FloatQ,
    BooleanQ,
    NilQ,
}
```

to:

```rust
pub enum UnaryOp {
    StringQ,
    IntegerQ,
    FloatQ,
    BooleanQ,
    NilQ,
    /// A UDF predicate name not in the built-in whitelist, resolved at execution time.
    Udf(String),
}
```

- [ ] **Step 3: Update `WindowSpec::func_name` to handle `Udf` variant**

In `src/query/datalog/types.rs`, `WindowSpec::func_name()` currently returns `&'static str` but `Udf(String)` requires returning an owned `String`. Change return type and callers:

```rust
impl WindowSpec {
    /// Returns the FunctionRegistry key name for this function.
    pub fn func_name(&self) -> String {
        match &self.func {
            WindowFunc::Sum => "sum".to_string(),
            WindowFunc::Count => "count".to_string(),
            WindowFunc::Min => "min".to_string(),
            WindowFunc::Max => "max".to_string(),
            WindowFunc::Avg => "avg".to_string(),
            WindowFunc::Rank => "rank".to_string(),
            WindowFunc::RowNumber => "row-number".to_string(),
            WindowFunc::Udf(name) => name.clone(),
        }
    }
}
```

Also update `FindSpec::display_name()` — it calls `ws.func_name()`. Since `func_name()` now returns `String`, the existing call `ws.func_name()` already works (no change needed there — Rust will use the owned value).

- [ ] **Step 4: Build to confirm no regressions**

```bash
cargo build 2>&1 | head -40
```

Expected: compiler errors only about `func_name()` callers in `executor.rs` that stored the result as `&'static str` (if any). Fix by using `let name = ws.func_name();` with owned `String` where needed. Or no errors if callers already used the value by reference.

- [ ] **Step 5: Run existing tests**

```bash
cargo test 2>&1 | tail -20
```

Expected: all existing tests pass (new `Udf` variants are unreachable from existing code paths).

- [ ] **Step 6: Commit**

```bash
git add src/query/datalog/types.rs
git commit -m "feat(types): add WindowFunc::Udf and UnaryOp::Udf variants for 7.7b"
```

---

## Task 2: Extend `functions.rs` — `UdfOps`, `AggImpl`, `PredicateDesc`

**Files:**
- Modify: `src/query/datalog/functions.rs`

- [ ] **Step 1: Write the failing registry unit tests**

At the bottom of `src/query/datalog/functions.rs`, inside the `#[cfg(test)] mod tests` block, add:

```rust
    #[test]
    fn register_udf_aggregate_is_known() {
        let mut reg = FunctionRegistry::with_builtins();
        reg.register_aggregate_desc(
            "myfn".to_string(),
            AggregateDesc {
                impl_: AggImpl::Udf(UdfOps {
                    init: Arc::new(|| Box::new(0i64) as Box<dyn Any + Send>),
                    step: Arc::new(|acc, v| {
                        if let (Some(n), Value::Integer(i)) = (acc.downcast_mut::<i64>(), v) {
                            *n += i;
                        }
                    }),
                    finalise: Arc::new(|acc, _n| {
                        acc.downcast_ref::<i64>()
                            .map(|n| Value::Integer(*n))
                            .unwrap_or(Value::Null)
                    }),
                }),
                is_builtin: false,
            },
        ).expect("register should succeed");
        assert!(reg.is_known("myfn"));
        assert!(reg.is_window_compatible("myfn"));
    }

    #[test]
    fn register_udf_duplicate_rejected() {
        let mut reg = FunctionRegistry::with_builtins();
        let make_desc = || AggregateDesc {
            impl_: AggImpl::Udf(UdfOps {
                init: Arc::new(|| Box::new(0i64) as Box<dyn Any + Send>),
                step: Arc::new(|_acc, _v| {}),
                finalise: Arc::new(|_acc, _n| Value::Null),
            }),
            is_builtin: false,
        };
        reg.register_aggregate_desc("myfn".to_string(), make_desc()).expect("first ok");
        assert!(reg.register_aggregate_desc("myfn".to_string(), make_desc()).is_err());
    }

    #[test]
    fn register_builtin_name_rejected() {
        let mut reg = FunctionRegistry::with_builtins();
        let result = reg.register_aggregate_desc(
            "sum".to_string(),
            AggregateDesc {
                impl_: AggImpl::Udf(UdfOps {
                    init: Arc::new(|| Box::new(0i64) as Box<dyn Any + Send>),
                    step: Arc::new(|_acc, _v| {}),
                    finalise: Arc::new(|_acc, _n| Value::Null),
                }),
                is_builtin: false,
            },
        );
        assert!(result.is_err());
    }

    #[test]
    fn register_predicate_works_and_rejects_duplicate() {
        let mut reg = FunctionRegistry::with_builtins();
        reg.register_predicate_desc(
            "email?".to_string(),
            PredicateDesc {
                f: Arc::new(|v| matches!(v, Value::String(s) if s.contains('@'))),
                is_builtin: false,
            },
        ).expect("first registration ok");
        assert!(reg.get_predicate("email?").is_some());
        let second = reg.register_predicate_desc(
            "email?".to_string(),
            PredicateDesc {
                f: Arc::new(|_v| false),
                is_builtin: false,
            },
        );
        assert!(second.is_err());
    }

    #[test]
    fn register_builtin_predicate_name_rejected() {
        let mut reg = FunctionRegistry::with_builtins();
        let result = reg.register_predicate_desc(
            "string?".to_string(),
            PredicateDesc {
                f: Arc::new(|_v| false),
                is_builtin: false,
            },
        );
        assert!(result.is_err());
    }

    #[test]
    fn udf_accumulator_runs() {
        let reg = FunctionRegistry::with_builtins();
        // Just confirm the existing window_ops path still works (regression guard).
        let desc = reg.get("sum").unwrap();
        if let AggImpl::Builtin(ops) = &desc.impl_ {
            let mut acc = (ops.init)();
            (ops.step)(&mut acc, &Value::Integer(5));
            assert_eq!((ops.finalise)(&acc), Value::Integer(5));
        } else {
            panic!("sum should be Builtin");
        }
    }
```

- [ ] **Step 2: Run tests to confirm they fail (types don't exist yet)**

```bash
cargo test --lib query::datalog::functions 2>&1 | head -40
```

Expected: compile errors — `UdfOps`, `AggImpl`, `PredicateDesc`, `AggImpl::Builtin`, `AggImpl::Udf` not found.

- [ ] **Step 3: Add `UdfOps`, `AggImpl`, `PredicateDesc` and update `AggregateDesc`**

At the top of `src/query/datalog/functions.rs`, add imports:

```rust
use std::any::Any;
use std::sync::Arc;
```

After the existing `WindowOps` struct, add:

```rust
/// Closure-based aggregate ops for UDFs.
/// The accumulator is type-erased as `Box<dyn Any + Send>`.
pub struct UdfOps {
    pub init:     Arc<dyn Fn() -> Box<dyn Any + Send> + Send + Sync>,
    pub step:     Arc<dyn Fn(&mut Box<dyn Any + Send>, &Value) + Send + Sync>,
    pub finalise: Arc<dyn Fn(&Box<dyn Any + Send>, usize) -> Value + Send + Sync>,
}

/// Implementation discriminator for aggregate functions.
pub enum AggImpl {
    /// Built-in: uses `fn()` function pointers and `AggState`.
    Builtin(WindowOps),
    /// User-defined: uses `Arc<dyn Fn>` closures and `Box<dyn Any + Send>`.
    Udf(UdfOps),
}

/// Descriptor for a registered predicate function.
pub struct PredicateDesc {
    pub f:          Arc<dyn Fn(&Value) -> bool + Send + Sync>,
    pub is_builtin: bool,
}
```

Replace `AggregateDesc`:

```rust
/// Descriptor for one registered aggregate function.
pub struct AggregateDesc {
    pub impl_:      AggImpl,
    /// True for built-in functions handled by `apply_builtin_aggregate`.
    pub is_builtin: bool,
}
```

- [ ] **Step 4: Add `predicates` field to `FunctionRegistry` and update `with_builtins`**

Update the `FunctionRegistry` struct:

```rust
pub struct FunctionRegistry {
    aggregates: HashMap<String, AggregateDesc>,
    /// Registered filter predicates, including built-in names as sentinels.
    predicates: HashMap<String, PredicateDesc>,
}
```

In `with_builtins()`, update all `AggregateDesc` constructions to use `AggImpl::Builtin(WindowOps { ... })` instead of `window_compatible` + `window_ops` fields. Also register built-in predicate names as sentinels so collision detection works. The full updated `with_builtins` body:

```rust
pub fn with_builtins() -> Self {
    let mut reg = Self {
        aggregates: HashMap::new(),
        predicates: HashMap::new(),
    };

    // ── Aggregate built-ins ──────────────────────────────────────────────
    reg.aggregates.insert("count".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::Count(0),
            step: |state, v| {
                if !matches!(v, Value::Null) {
                    let AggState::Count(n) = state else { return };
                    *n += 1;
                }
            },
            finalise: |state| {
                if let AggState::Count(n) = state { Value::Integer(*n) } else { Value::Null }
            },
        }),
    });
    reg.aggregates.insert("sum".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::Sum { total: 0.0, is_float: false },
            step: |state, v| {
                if let AggState::Sum { total, is_float } = state {
                    match v {
                        Value::Integer(i) => *total += *i as f64,
                        Value::Float(f) => { *total += f; *is_float = true; }
                        _ => {}
                    }
                }
            },
            finalise: |state| {
                if let AggState::Sum { total, is_float } = state {
                    if *is_float { Value::Float(*total) } else { Value::Integer(*total as i64) }
                } else { Value::Null }
            },
        }),
    });
    reg.aggregates.insert("min".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::MinMax { current: None },
            step: |state, v| {
                if matches!(v, Value::Null) { return; }
                if let AggState::MinMax { current } = state {
                    match current {
                        None => *current = Some(v.clone()),
                        Some(cur) => { if value_lt(v, cur) { *current = Some(v.clone()); } }
                    }
                }
            },
            finalise: |state| {
                if let AggState::MinMax { current } = state {
                    current.clone().unwrap_or(Value::Null)
                } else { Value::Null }
            },
        }),
    });
    reg.aggregates.insert("max".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::MinMax { current: None },
            step: |state, v| {
                if matches!(v, Value::Null) { return; }
                if let AggState::MinMax { current } = state {
                    match current {
                        None => *current = Some(v.clone()),
                        Some(cur) => { if value_lt(cur, v) { *current = Some(v.clone()); } }
                    }
                }
            },
            finalise: |state| {
                if let AggState::MinMax { current } = state {
                    current.clone().unwrap_or(Value::Null)
                } else { Value::Null }
            },
        }),
    });
    reg.aggregates.insert("avg".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::Avg { sum: 0.0, count: 0 },
            step: |state, v| {
                if let AggState::Avg { sum, count } = state {
                    match v {
                        Value::Integer(i) => { *sum += *i as f64; *count += 1; }
                        Value::Float(f) => { *sum += f; *count += 1; }
                        _ => {}
                    }
                }
            },
            finalise: |state| {
                if let AggState::Avg { sum, count } = state {
                    if *count == 0 { Value::Null } else { Value::Float(*sum / *count as f64) }
                } else { Value::Null }
            },
        }),
    });
    // count-distinct / sum-distinct: not window-compatible, no WindowOps.
    // Represented as Builtin with a sentinel WindowOps that panics — never called
    // for window computation. The executor uses is_builtin=true to route to apply_builtin_aggregate.
    // Use a minimal WindowOps that is never invoked:
    reg.aggregates.insert("count-distinct".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::Count(0),
            step: |_, _| {},
            finalise: |_| Value::Null,
        }),
    });
    reg.aggregates.insert("sum-distinct".into(), AggregateDesc {
        is_builtin: true,
        impl_: AggImpl::Builtin(WindowOps {
            init: || AggState::Count(0),
            step: |_, _| {},
            finalise: |_| Value::Null,
        }),
    });

    // ── Predicate built-in sentinels ─────────────────────────────────────
    // Registered so that `register_predicate_desc("string?", ...)` returns Err.
    for name in ["string?", "integer?", "float?", "boolean?", "nil?",
                 "starts-with?", "ends-with?", "contains?", "matches?"] {
        reg.predicates.insert(name.to_string(), PredicateDesc {
            f: Arc::new(|_| false), // sentinel; never called via registry
            is_builtin: true,
        });
    }

    reg
}
```

- [ ] **Step 5: Add new public methods to `FunctionRegistry`**

After `with_builtins`, add:

```rust
pub fn get(&self, name: &str) -> Option<&AggregateDesc> {
    self.aggregates.get(name)
}

pub fn is_known(&self, name: &str) -> bool {
    self.aggregates.contains_key(name)
}

/// All aggregates (built-in and UDF) are window-compatible, except
/// count-distinct and sum-distinct which must go through the builtin batch path.
pub fn is_window_compatible(&self, name: &str) -> bool {
    match name {
        "count-distinct" | "sum-distinct" => false,
        _ => self.aggregates.contains_key(name),
    }
}

/// Register a UDF aggregate descriptor. Returns Err if the name is already taken.
pub fn register_aggregate_desc(&mut self, name: String, desc: AggregateDesc) -> anyhow::Result<()> {
    if self.aggregates.contains_key(&name) {
        anyhow::bail!("aggregate function '{}' is already registered", name);
    }
    self.aggregates.insert(name, desc);
    Ok(())
}

/// Register a predicate descriptor. Returns Err if the name is already taken.
pub fn register_predicate_desc(&mut self, name: String, desc: PredicateDesc) -> anyhow::Result<()> {
    if self.predicates.contains_key(&name) {
        anyhow::bail!("predicate '{}' is already registered", name);
    }
    self.predicates.insert(name, desc);
    Ok(())
}

/// Look up a registered predicate by name.
pub fn get_predicate(&self, name: &str) -> Option<&PredicateDesc> {
    // Don't expose built-in sentinels as callable predicates.
    self.predicates.get(name).filter(|d| !d.is_builtin)
}
```

- [ ] **Step 6: Fix the existing tests that used `window_compatible` / `window_ops` fields**

The existing tests in `functions.rs` used `desc.window_compatible` and `desc.window_ops`. Update them to use the new `AggImpl`:

Test `window_compatible_flags` — update to check `is_window_compatible`:
```rust
#[test]
fn window_compatible_flags() {
    let reg = FunctionRegistry::with_builtins();
    for name in ["count", "sum", "min", "max", "avg"] {
        assert!(reg.is_window_compatible(name), "'{}' should be window-compatible", name);
    }
    for name in ["count-distinct", "sum-distinct"] {
        assert!(!reg.is_window_compatible(name), "'{}' should NOT be window-compatible", name);
    }
}
```

Tests `window_ops_sum_accumulator`, `window_ops_count_accumulator`, `window_ops_avg_accumulator` — update to destructure `AggImpl::Builtin`:
```rust
#[test]
fn window_ops_sum_accumulator() {
    let reg = FunctionRegistry::with_builtins();
    let desc = reg.get("sum").unwrap();
    let AggImpl::Builtin(ops) = &desc.impl_ else { panic!("sum should be Builtin") };
    let mut state = (ops.init)();
    (ops.step)(&mut state, &Value::Integer(10));
    assert_eq!((ops.finalise)(&state), Value::Integer(10));
    (ops.step)(&mut state, &Value::Integer(20));
    assert_eq!((ops.finalise)(&state), Value::Integer(30));
}

#[test]
fn window_ops_count_accumulator() {
    let reg = FunctionRegistry::with_builtins();
    let desc = reg.get("count").unwrap();
    let AggImpl::Builtin(ops) = &desc.impl_ else { panic!("count should be Builtin") };
    let mut state = (ops.init)();
    (ops.step)(&mut state, &Value::Integer(1));
    (ops.step)(&mut state, &Value::Integer(2));
    assert_eq!((ops.finalise)(&state), Value::Integer(2));
}

#[test]
fn window_ops_avg_accumulator() {
    let reg = FunctionRegistry::with_builtins();
    let desc = reg.get("avg").unwrap();
    let AggImpl::Builtin(ops) = &desc.impl_ else { panic!("avg should be Builtin") };
    let mut state = (ops.init)();
    (ops.step)(&mut state, &Value::Integer(10));
    (ops.step)(&mut state, &Value::Integer(20));
    assert_eq!((ops.finalise)(&state), Value::Float(15.0));
}
```

- [ ] **Step 7: Run library tests**

```bash
cargo test --lib query::datalog::functions 2>&1 | tail -30
```

Expected: all tests in the functions module pass, including the 6 new UDF registry tests.

- [ ] **Step 8: Commit**

```bash
git add src/query/datalog/functions.rs
git commit -m "feat(functions): add UdfOps, AggImpl, PredicateDesc; update FunctionRegistry for 7.7b"
```

---

## Task 3: Update `parser.rs`

**Files:**
- Modify: `src/query/datalog/parser.rs` — `parse_window_expr` (line ~497–511) and `parse_expr` (line ~975–988)

- [ ] **Step 1: Write the failing parser tests**

At the end of the `#[cfg(test)] mod tests` block in `parser.rs`, add:

```rust
    #[test]
    fn parse_udf_window_function() {
        // A name not in the built-in list should produce WindowFunc::Udf
        let result = parse_datalog_command(
            r#"(query [:find (geomean ?score :over (:order-by ?score))
                     :where [?e :item/score ?score]])"#,
        );
        assert!(result.is_ok(), "UDF window function should parse");
        if let Ok(DatalogCommand::Query(q)) = result {
            assert_eq!(q.find_specs.len(), 1);
            if let FindSpec::Window(ws) = &q.find_specs[0] {
                assert_eq!(ws.func, WindowFunc::Udf("geomean".to_string()));
            } else {
                panic!("expected Window find spec");
            }
        }
    }

    #[test]
    fn parse_udf_predicate_clause() {
        // Unknown 1-arg predicate should produce UnaryOp::Udf
        let result = parse_datalog_command(
            r#"(query [:find ?e :where [?e :person/email ?addr] [(email? ?addr)]])"#,
        );
        assert!(result.is_ok(), "UDF predicate should parse");
        if let Ok(DatalogCommand::Query(q)) = result {
            let has_udf = q.where_clauses.iter().any(|c| {
                matches!(c, WhereClause::Expr { expr: Expr::UnaryOp(UnaryOp::Udf(name), _), .. }
                    if name == "email?")
            });
            assert!(has_udf, "should have UnaryOp::Udf(\"email?\") clause");
        }
    }

    #[test]
    fn parse_unknown_two_arg_predicate_rejected() {
        // Unknown 2-arg form should still be a parse error
        let result = parse_datalog_command(
            r#"(query [:find ?e :where [?e :x ?v] [(myfn? ?v ?v)]])"#,
        );
        assert!(result.is_err(), "unknown 2-arg form should be rejected");
    }

    #[test]
    fn lag_still_rejected() {
        let result = parse_datalog_command(
            r#"(query [:find (lag ?v :over (:order-by ?v)) :where [?e :x ?v]])"#,
        );
        assert!(result.is_err());
    }

    #[test]
    fn lead_still_rejected() {
        let result = parse_datalog_command(
            r#"(query [:find (lead ?v :over (:order-by ?v)) :where [?e :x ?v]])"#,
        );
        assert!(result.is_err());
    }
```

- [ ] **Step 2: Run tests to confirm failures**

```bash
cargo test --lib query::datalog::parser::tests::parse_udf_window_function 2>&1 | tail -20
```

Expected: test fails because `other => return Err(format!("unknown window function: '{}'", other))` in `parse_window_expr`.

- [ ] **Step 3: Update `parse_window_expr` to emit `WindowFunc::Udf` for unknown names**

In `src/query/datalog/parser.rs`, find the match block starting at line ~497:

```rust
    let func = match func_name {
        "sum" => WindowFunc::Sum,
        "count" => WindowFunc::Count,
        "min" => WindowFunc::Min,
        "max" => WindowFunc::Max,
        "avg" => WindowFunc::Avg,
        "rank" => WindowFunc::Rank,
        "row-number" => WindowFunc::RowNumber,
        "count-distinct" | "sum-distinct" => {
            return Err(format!(
                "'{}' is not window-compatible and cannot be used with ':over'",
                func_name
            ));
        }
        other => return Err(format!("unknown window function: '{}'", other)),
    };
```

Replace the last arm:

```rust
        other => WindowFunc::Udf(other.to_string()),
```

- [ ] **Step 4: Update `parse_window_expr` variable parsing for `Udf` variant**

The `match func` block immediately after (around line ~516) that checks for `Rank | RowNumber` to decide whether a `?var` argument is required. `Udf` functions always require a `?var` (consistent with all accumulator-based functions). Find:

```rust
    let (var, over_keyword_idx) = match func {
        WindowFunc::Rank | WindowFunc::RowNumber => {
```

This is already a catch-all `_` pattern for the rest. No change needed — `Udf` falls into the `_` arm which requires a `?var` argument. ✓

- [ ] **Step 5: Update `parse_expr` to emit `UnaryOp::Udf` for unknown 1-arg predicates**

In `src/query/datalog/parser.rs`, find `parse_expr` around line ~973. The match is:

```rust
    match head {
        // Unary type predicates
        "string?" | "integer?" | "float?" | "boolean?" | "nil?" => {
            ...
        }
        // Binary operators
        "<" | ">" | ... => {
            ...
        }
        other => Err(format!("unknown expression operator: {}", other)),
    }
```

Replace the final `other =>` arm:

```rust
        other => {
            // Unknown name in 1-arg position: treat as UDF predicate.
            // 2-arg unknown forms are still rejected.
            if list.len() == 2 {
                let arg = parse_expr_arg(&list[1])?;
                Ok(Expr::UnaryOp(UnaryOp::Udf(other.to_string()), Box::new(arg)))
            } else {
                Err(format!("unknown expression operator: {}", other))
            }
        }
```

- [ ] **Step 6: Run the new parser tests**

```bash
cargo test --lib query::datalog::parser 2>&1 | tail -30
```

Expected: all parser tests pass, including the 5 new tests. `lag`/`lead` still rejected; `geomean` emits `Udf`; `email?` emits `UnaryOp::Udf`; unknown 2-arg still errors.

- [ ] **Step 7: Commit**

```bash
git add src/query/datalog/parser.rs
git commit -m "feat(parser): emit Udf variants for unknown window functions and 1-arg predicates"
```

---

## Task 4: Update `executor.rs` — UDF evaluation paths

**Files:**
- Modify: `src/query/datalog/executor.rs`

This is the largest task. Changes:
1. `eval_expr`: add `registry: Option<&FunctionRegistry>` param, handle `UnaryOp::Udf`
2. `apply_expr_clauses`: add `registry: &FunctionRegistry`, return `Result<Vec<Binding>>`; update all 5 call sites
3. `compute_aggregation`: handle `AggImpl::Udf` in the aggregate loop
4. `apply_window_functions`: handle `WindowFunc::Udf` arm
5. Fix `func_name()` call sites (now returns `String` instead of `&'static str`)

- [ ] **Step 1: Fix `func_name()` call sites**

In `apply_window_functions`, the current code calls `ws.func_name()` and stores it as `let func_name = ws.func_name()`. Since `func_name()` now returns `String`, this still compiles. But `registry.get(func_name)` now expects `&str` — use `registry.get(&func_name)` or `registry.get(func_name.as_str())`. Search for all uses of `ws.func_name()` and confirm they compile.

```bash
cargo build 2>&1 | grep "func_name" | head -20
```

Fix any type errors (e.g., `let func_name = ws.func_name();` then `registry.get(&func_name)` or `.get(&*func_name)`).

- [ ] **Step 2: Update `eval_expr` signature and add `UnaryOp::Udf` arm**

Find `pub(crate) fn eval_expr` (line ~1347). Change its signature to accept an optional registry:

```rust
pub(crate) fn eval_expr(
    expr: &Expr,
    binding: &std::collections::HashMap<String, Value>,
    registry: Option<&FunctionRegistry>,
) -> Result<Value, ()> {
```

In the `Expr::UnaryOp` match arm, add `UnaryOp::Udf(name)` after the five built-in variants:

```rust
        Expr::UnaryOp(op, arg) => {
            let v = eval_expr(arg, binding, registry)?;
            match op {
                UnaryOp::StringQ  => Ok(Value::Boolean(matches!(v, Value::String(_)))),
                UnaryOp::IntegerQ => Ok(Value::Boolean(matches!(v, Value::Integer(_)))),
                UnaryOp::FloatQ   => Ok(Value::Boolean(matches!(v, Value::Float(_)))),
                UnaryOp::BooleanQ => Ok(Value::Boolean(matches!(v, Value::Boolean(_)))),
                UnaryOp::NilQ     => Ok(Value::Boolean(matches!(v, Value::Null))),
                UnaryOp::Udf(name) => {
                    let desc = registry
                        .and_then(|r| r.get_predicate(name))
                        .ok_or(())?;
                    Ok(Value::Boolean((desc.f)(&v)))
                }
            }
        }
```

Also update the recursive calls inside `eval_expr` itself (the `BinOp` arm calls `eval_expr(lhs, binding)` — add the registry arg):

```rust
        Expr::BinOp(op, lhs, rhs) => {
            let l = eval_expr(lhs, binding, registry)?;
            let r = eval_expr(rhs, binding, registry)?;
            eval_binop(op, l, r)
        }
```

- [ ] **Step 3: Update `apply_expr_clauses` signature and return type**

Find `pub(crate) fn apply_expr_clauses` (line ~1377). Change it to:

```rust
pub(crate) fn apply_expr_clauses(
    mut bindings: Vec<Binding>,
    where_clauses: &[WhereClause],
    registry: &FunctionRegistry,
) -> anyhow::Result<Vec<Binding>> {
    // Pre-validate: check for unknown UDF predicates before filtering rows.
    for clause in where_clauses {
        if let WhereClause::Expr { expr, .. } = clause {
            if let Expr::UnaryOp(UnaryOp::Udf(name), _) = expr {
                if registry.get_predicate(name).is_none() {
                    anyhow::bail!("unknown predicate: '{}'", name);
                }
            }
        }
    }

    for clause in where_clauses {
        if let WhereClause::Expr { expr, binding: out } = clause {
            bindings = bindings
                .into_iter()
                .filter_map(|mut b| match eval_expr(expr, &b, Some(registry)) {
                    Ok(v) => {
                        if let Some(var) = out {
                            b.insert(var.clone(), v);
                            Some(b)
                        } else if is_truthy(&v) {
                            Some(b)
                        } else {
                            None
                        }
                    }
                    Err(()) => None,
                })
                .collect();
        }
    }
    Ok(bindings)
}
```

- [ ] **Step 4: Update all 5 call sites of `apply_expr_clauses`**

Search for all calls to `apply_expr_clauses` in `executor.rs` (lines 332, 557, 574, 670, 1128). Each needs:
- The `registry` argument added
- The result unwrapped with `?`

The executor functions that call `apply_expr_clauses` already have access to `registry` (it's a parameter of the enclosing functions). In each location, change:

```rust
// BEFORE:
let filtered_bindings = apply_expr_clauses(not_filtered, &query.where_clauses);
// AFTER:
let filtered_bindings = apply_expr_clauses(not_filtered, &query.where_clauses, registry)?;
```

Similarly for the other 4 call sites. To locate exact lines:

```bash
grep -n "apply_expr_clauses" src/query/datalog/executor.rs
```

Each call site is inside a function that already has `registry: &FunctionRegistry` in scope (passed down from `execute_query` / `execute_query_with_rules`). If any call site doesn't have a `registry` in scope, thread it through from the caller.

- [ ] **Step 5: Fix `eval_expr` call sites in tests and elsewhere**

The `eval_expr` function is `pub(crate)` and called in tests at ~20 locations (lines 2909–3061). Each test call passes two arguments; now it needs three. Search:

```bash
grep -n "eval_expr(" src/query/datalog/executor.rs | grep -v "fn eval_expr"
```

In tests, pass `None` as the third argument (no registry needed for built-in predicates):

```rust
// BEFORE:
assert_eq!(eval_expr(&e, &HashMap::new()), Ok(Value::Integer(42)));
// AFTER:
assert_eq!(eval_expr(&e, &HashMap::new(), None), Ok(Value::Integer(42)));
```

Make this replacement across all test call sites.

- [ ] **Step 6: Update `compute_aggregation` to handle `AggImpl::Udf`**

Find the aggregate loop in `compute_aggregation` (around line ~835):

```rust
                let agg_val: anyhow::Result<Value> = if let Some(desc) = registry.get(func) {
                    if desc.is_builtin {
                        apply_builtin_aggregate(func, &non_null)
                    } else if let Some(ops) = &desc.window_ops {
                        ...
                    } else {
                        apply_builtin_aggregate(func, &non_null)
                    }
                } else {
                    apply_builtin_aggregate(func, &non_null)
                };
```

Replace with a clean match on `AggImpl`:

```rust
                let agg_val: anyhow::Result<Value> = match registry.get(func.as_str()) {
                    Some(desc) if desc.is_builtin => {
                        apply_builtin_aggregate(func, &non_null)
                    }
                    Some(desc) => {
                        // UDF aggregate: init → step each non-null value → finalise.
                        if let AggImpl::Udf(ops) = &desc.impl_ {
                            if non_null.is_empty() {
                                Ok(Value::Null)
                            } else {
                                let mut acc = (ops.init)();
                                for v in &non_null {
                                    (ops.step)(&mut acc, v);
                                }
                                Ok((ops.finalise)(&acc, non_null.len()))
                            }
                        } else {
                            // AggImpl::Builtin for a non-is_builtin desc: shouldn't happen.
                            apply_builtin_aggregate(func, &non_null)
                        }
                    }
                    None => {
                        Err(anyhow::anyhow!("unknown aggregate function: '{}'", func))
                    }
                };
```

- [ ] **Step 7: Update `apply_window_functions` to handle `WindowFunc::Udf`**

Find the `match ws.func` in `apply_window_functions` (around line ~928). The current fallthrough `_` arm handles all accumulator-based built-ins via `ws.func_name()`. Extend it to handle `Udf`:

The existing `_` arm already calls `registry.get(func_name)`. Update that arm to also handle the UDF case:

```rust
                _ => {
                    // Accumulator-based: built-ins (sum, count, min, max, avg) and UDFs.
                    let func_name = ws.func_name(); // now returns String
                    let desc = registry.get(&func_name).ok_or_else(|| {
                        anyhow::anyhow!("unknown window function '{}' — register it with register_aggregate() before querying", func_name)
                    })?;

                    let mut values = Vec::with_capacity(row_indices.len());
                    match &desc.impl_ {
                        AggImpl::Builtin(ops) => {
                            let mut acc = (ops.init)();
                            for &row_idx in row_indices.iter() {
                                let val = ws.var.as_ref()
                                    .and_then(|v| bindings[row_idx].get(v))
                                    .unwrap_or(&Value::Null);
                                (ops.step)(&mut acc, val);
                                values.push((ops.finalise)(&acc));
                            }
                        }
                        AggImpl::Udf(ops) => {
                            let mut acc = (ops.init)();
                            let mut row_count = 0usize;
                            for &row_idx in row_indices.iter() {
                                let val = ws.var.as_ref()
                                    .and_then(|v| bindings[row_idx].get(v))
                                    .unwrap_or(&Value::Null);
                                (ops.step)(&mut acc, val);
                                row_count += 1;
                                values.push((ops.finalise)(&acc, row_count));
                            }
                        }
                    }
                    values
                }
```

- [ ] **Step 8: Build and run all tests**

```bash
cargo build 2>&1 | head -40
cargo test 2>&1 | tail -30
```

Expected: all existing tests pass. Fix any remaining compile errors (most likely `eval_expr` call sites missing the third argument or `apply_expr_clauses` missing the `registry` argument at some call site).

- [ ] **Step 9: Commit**

```bash
git add src/query/datalog/executor.rs
git commit -m "feat(executor): add UDF eval paths for aggregation, window functions, and predicates"
```

---

## Task 5: Add `register_aggregate` / `register_predicate` to `db.rs`

**Files:**
- Modify: `src/db.rs`

- [ ] **Step 1: Add necessary imports to `db.rs`**

At the top of `src/db.rs`, add:

```rust
use crate::query::datalog::functions::{AggImpl, AggregateDesc, PredicateDesc, UdfOps};
use std::any::Any;
```

- [ ] **Step 2: Add `register_aggregate` to `Minigraf`**

Inside `impl Minigraf`, add after the existing public methods:

```rust
/// Register a custom aggregate function.
///
/// `Acc` is any `Send + 'static` type that serves as the accumulator.
/// It is type-erased internally. The function is usable in both `:find`
/// grouping position and `:over` (window) position.
///
/// Returns `Err` if `name` is already registered (built-in or UDF).
///
/// # Example
/// ```
/// # use minigraf::db::Minigraf;
/// # use minigraf::graph::types::Value;
/// let db = Minigraf::in_memory().unwrap();
/// db.register_aggregate(
///     "mysum",
///     || 0i64,
///     |acc: &mut i64, v: &Value| { if let Value::Integer(i) = v { *acc += i; } },
///     |acc: &i64, _n: usize| Value::Integer(*acc),
/// ).unwrap();
/// ```
pub fn register_aggregate<Acc>(
    &self,
    name: &str,
    init:     impl Fn() -> Acc              + Send + Sync + 'static,
    step:     impl Fn(&mut Acc, &Value)     + Send + Sync + 'static,
    finalise: impl Fn(&Acc, usize) -> Value + Send + Sync + 'static,
) -> Result<()>
where
    Acc: Any + Send + 'static,
{
    use std::sync::Arc;
    let init_boxed: Arc<dyn Fn() -> Box<dyn Any + Send> + Send + Sync> =
        Arc::new(move || Box::new(init()) as Box<dyn Any + Send>);
    let step_boxed: Arc<dyn Fn(&mut Box<dyn Any + Send>, &Value) + Send + Sync> =
        Arc::new(move |acc, v| {
            // SAFETY: `init_boxed` always creates `Box<Acc>`, so downcast is infallible.
            step(acc.downcast_mut::<Acc>().expect("UDF accumulator type mismatch"), v);
        });
    let finalise_boxed: Arc<dyn Fn(&Box<dyn Any + Send>, usize) -> Value + Send + Sync> =
        Arc::new(move |acc, n| {
            finalise(acc.downcast_ref::<Acc>().expect("UDF accumulator type mismatch"), n)
        });

    let desc = AggregateDesc {
        impl_: AggImpl::Udf(UdfOps {
            init: init_boxed,
            step: step_boxed,
            finalise: finalise_boxed,
        }),
        is_builtin: false,
    };
    self.inner
        .functions
        .write()
        .map_err(|e| anyhow::anyhow!("function registry lock poisoned: {}", e))?
        .register_aggregate_desc(name.to_string(), desc)
}

/// Register a custom single-argument filter predicate.
///
/// The predicate is usable in `[(name? ?var)]` `:where` clauses.
/// Returns `Err` if `name` is already registered (built-in or UDF).
///
/// # Example
/// ```
/// # use minigraf::db::Minigraf;
/// # use minigraf::graph::types::Value;
/// let db = Minigraf::in_memory().unwrap();
/// db.register_predicate(
///     "email?",
///     |v: &Value| matches!(v, Value::String(s) if s.contains('@')),
/// ).unwrap();
/// ```
pub fn register_predicate(
    &self,
    name: &str,
    f: impl Fn(&Value) -> bool + Send + Sync + 'static,
) -> Result<()> {
    use std::sync::Arc;
    let desc = PredicateDesc {
        f: Arc::new(f),
        is_builtin: false,
    };
    self.inner
        .functions
        .write()
        .map_err(|e| anyhow::anyhow!("function registry lock poisoned: {}", e))?
        .register_predicate_desc(name.to_string(), desc)
}
```

- [ ] **Step 3: Build and run all tests**

```bash
cargo build 2>&1 | head -20
cargo test 2>&1 | tail -20
```

Expected: clean build; all tests pass.

- [ ] **Step 4: Commit**

```bash
git add src/db.rs
git commit -m "feat(db): add register_aggregate and register_predicate to Minigraf"
```

---

## Task 6: Integration tests in `tests/udf_test.rs`

**Files:**
- Create: `tests/udf_test.rs`

- [ ] **Step 1: Write all 9 integration tests**

Create `tests/udf_test.rs`:

```rust
//! Integration tests for Phase 7.7b — User-Defined Functions (UDFs).
//!
//! Testing conventions: no `{:?}` of Result/Value/Fact in assert messages.

use minigraf::db::Minigraf;
use minigraf::graph::types::Value;

// ─── helpers ─────────────────────────────────────────────────────────────────

fn db() -> Minigraf {
    Minigraf::in_memory().expect("in_memory failed")
}

fn seed(db: &Minigraf, cmds: &[&str]) {
    for cmd in cmds {
        db.execute(cmd).expect("seed failed");
    }
}

// ─── Test 1: custom aggregate (geometric mean) ───────────────────────────────

#[test]
fn custom_aggregate_geometric_mean() {
    let db = db();
    // geometric mean = exp(mean of ln(values))
    db.register_aggregate(
        "geomean",
        || (0.0_f64, 0usize), // (sum_of_ln, count)
        |acc: &mut (f64, usize), v: &Value| {
            if let Value::Float(f) = v {
                if *f > 0.0 {
                    acc.0 += f.ln();
                    acc.1 += 1;
                }
            } else if let Value::Integer(i) = v {
                if *i > 0 {
                    acc.0 += (*i as f64).ln();
                    acc.1 += 1;
                }
            }
        },
        |acc: &(f64, usize), _n: usize| {
            if acc.1 == 0 { Value::Null } else { Value::Float((acc.0 / acc.1 as f64).exp()) }
        },
    ).expect("register geomean");

    seed(&db, &[
        r#"(transact [[:a :item/score 2] [:b :item/score 8]])"#,
    ]);

    let result = db.execute(r#"(query [:find (geomean ?score) :where [?e :item/score ?score]])"#)
        .expect("query failed");

    if let minigraf::query::QueryResult::QueryResults { results, .. } = result {
        assert_eq!(results.len(), 1);
        // geomean(2, 8) = sqrt(16) = 4.0
        if let Value::Float(f) = &results[0][0] {
            assert!((*f - 4.0).abs() < 1e-9, "expected ~4.0");
        } else {
            panic!("expected Float result");
        }
    } else {
        panic!("expected QueryResults");
    }
}

// ─── Test 2: custom aggregate — empty result returns Null ────────────────────

#[test]
fn custom_aggregate_empty_result() {
    let db = db();
    db.register_aggregate(
        "geomean2",
        || (0.0_f64, 0usize),
        |acc: &mut (f64, usize), v: &Value| {
            if let Value::Float(f) = v { if *f > 0.0 { acc.0 += f.ln(); acc.1 += 1; } }
        },
        |acc: &(f64, usize), _n: usize| {
            if acc.1 == 0 { Value::Null } else { Value::Float((acc.0 / acc.1 as f64).exp()) }
        },
    ).expect("register geomean2");

    // No facts — empty result
    let result = db.execute(r#"(query [:find (geomean2 ?score) :where [?e :item/score ?score]])"#)
        .expect("query failed");

    if let minigraf::query::QueryResult::QueryResults { results, .. } = result {
        // Empty input → no groups → no output rows (consistent with built-in behaviour
        // for non-count aggregates when no facts match)
        assert!(results.is_empty() || results[0][0] == Value::Null);
    } else {
        panic!("expected QueryResults");
    }
}

// ─── Test 3: custom predicate filter ────────────────────────────────────────

#[test]
fn custom_predicate_filter() {
    let db = db();
    db.register_predicate(
        "email?",
        |v: &Value| matches!(v, Value::String(s) if s.contains('@')),
    ).expect("register email?");

    seed(&db, &[r#"(transact [
        [:alice :person/email "alice@example.com"]
        [:bob   :person/email "notanemail"]
    ])"#]);

    let result = db.execute(
        r#"(query [:find ?e :where [?e :person/email ?addr] [(email? ?addr)]])"#,
    ).expect("query failed");

    if let minigraf::query::QueryResult::QueryResults { results, .. } = result {
        assert_eq!(results.len(), 1, "only alice has a valid email");
        assert_eq!(results[0][0], Value::Keyword("alice".to_string()));
    } else {
        panic!("expected QueryResults");
    }
}

// ─── Test 4: UDF as window function ─────────────────────────────────────────

#[test]
fn udf_as_window_function() {
    let db = db();
    db.register_aggregate(
        "winsum",
        || 0i64,
        |acc: &mut i64, v: &Value| {
            if let Value::Integer(i) = v { *acc += i; }
        },
        |acc: &i64, _n: usize| Value::Integer(*acc),
    ).expect("register winsum");

    seed(&db, &[r#"(transact [
        [:a :item/score 1]
        [:b :item/score 2]
        [:c :item/score 3]
    ])"#]);

    let result = db.execute(
        r#"(query [:find ?e (winsum ?score :over (:order-by ?score))
                  :where [?e :item/score ?score]])"#,
    ).expect("query failed");

    if let minigraf::query::QueryResult::QueryResults { results, .. } = result {
        assert_eq!(results.len(), 3, "three rows");
        // After ordering by score: rows are 1,2,3; cumulative sums are 1,3,6
        let mut sums: Vec<i64> = results.iter()
            .map(|r| if let Value::Integer(n) = r[1] { n } else { -1 })
            .collect();
        sums.sort();
        assert_eq!(sums, vec![1, 3, 6]);
    } else {
        panic!("expected QueryResults");
    }
}

// ─── Test 5: name collision — shadowing a built-in aggregate ─────────────────

#[test]
fn name_collision_builtin_aggregate() {
    let db = db();
    let result = db.register_aggregate(
        "sum",
        || 0i64,
        |_acc: &mut i64, _v: &Value| {},
        |acc: &i64, _n: usize| Value::Integer(*acc),
    );
    assert!(result.is_err(), "shadowing built-in 'sum' must return Err");
}

// ─── Test 6: name collision — UDF-on-UDF ─────────────────────────────────────

#[test]
fn name_collision_udf_on_udf() {
    let db = db();
    db.register_aggregate(
        "myfn",
        || 0i64,
        |_acc: &mut i64, _v: &Value| {},
        |acc: &i64, _n: usize| Value::Integer(*acc),
    ).expect("first registration");

    let result = db.register_aggregate(
        "myfn",
        || 0i64,
        |_acc: &mut i64, _v: &Value| {},
        |acc: &i64, _n: usize| Value::Integer(*acc),
    );
    assert!(result.is_err(), "duplicate UDF name must return Err");
}

// ─── Test 7: unknown aggregate function → runtime error ──────────────────────

#[test]
fn unknown_function_runtime_error() {
    let db = db();
    seed(&db, &[r#"(transact [[:a :x 1]])"#]);
    let result = db.execute(r#"(query [:find (nosuchfn ?x) :where [?e :x ?x]])"#);
    assert!(result.is_err(), "unknown aggregate should return Err, not panic");
}

// ─── Test 8: unknown predicate → runtime error ───────────────────────────────

#[test]
fn unknown_predicate_runtime_error() {
    let db = db();
    seed(&db, &[r#"(transact [[:a :x "hello"]])"#]);
    let result = db.execute(r#"(query [:find ?e :where [?e :x ?v] [(nosuchpred? ?v)]])"#);
    assert!(result.is_err(), "unknown predicate should return Err, not panic");
}

// ─── Test 9: thread safety ───────────────────────────────────────────────────

#[test]
fn thread_safety() {
    use std::sync::Arc;

    let db = Arc::new(db());

    // Seed some facts.
    db.execute(r#"(transact [[:a :x 1] [:b :x 2]])"#).expect("seed");

    // Spawn reader threads.
    let mut handles = Vec::new();
    for _ in 0..4 {
        let db2 = Arc::clone(&db);
        handles.push(std::thread::spawn(move || {
            for _ in 0..10 {
                db2.execute(r#"(query [:find ?e :where [?e :x _]])"#)
                    .expect("concurrent read");
            }
        }));
    }

    // Register a UDF from the main thread while readers run.
    db.register_aggregate(
        "threadfn",
        || 0i64,
        |acc: &mut i64, v: &Value| { if let Value::Integer(i) = v { *acc += i; } },
        |acc: &i64, _n: usize| Value::Integer(*acc),
    ).expect("register threadfn");

    for h in handles {
        h.join().expect("reader thread panicked");
    }

    // Verify the UDF works after concurrent registration.
    let result = db.execute(r#"(query [:find (threadfn ?x) :where [?e :x ?x]])"#)
        .expect("post-registration query");
    if let minigraf::query::QueryResult::QueryResults { results, .. } = result {
        assert_eq!(results.len(), 1);
        assert_eq!(results[0][0], Value::Integer(3)); // 1 + 2
    } else {
        panic!("expected QueryResults");
    }
}
```

- [ ] **Step 2: Run the new tests**

```bash
cargo test --test udf_test 2>&1 | tail -40
```

Expected: all 9 tests pass.

- [ ] **Step 3: Run the full test suite**

```bash
cargo test 2>&1 | tail -20
```

Expected: all tests pass (707 existing + new tests).

- [ ] **Step 4: Commit**

```bash
git add tests/udf_test.rs
git commit -m "test: add 9 integration tests for Phase 7.7b UDFs"
```

---

## Task 7: Docs, version bump, and final checks

**Files:**
- Modify: `Cargo.toml` (version bump to 0.17.0)
- Modify: `CHANGELOG.md`
- Modify: `CLAUDE.md` (test count)
- Modify: `ROADMAP.md` (mark 7.7b complete)

- [ ] **Step 1: Bump version to 0.17.0**

In `Cargo.toml`, change `version = "0.16.0"` to `version = "0.17.0"`.

- [ ] **Step 2: Run clippy**

```bash
cargo clippy -- -D warnings 2>&1 | head -40
```

Fix any warnings before proceeding.

- [ ] **Step 3: Run the full test suite one final time**

```bash
cargo test 2>&1 | tail -10
```

Note the total test count (it will be 707 + number of new tests). Record it.

- [ ] **Step 4: Update `CHANGELOG.md`**

Add an entry for v0.17.0 at the top of `CHANGELOG.md`:

```markdown
## v0.17.0 — Phase 7.7b: User-Defined Functions (2026-04-02)

### Added
- `Minigraf::register_aggregate(name, init, step, finalise)` — register a custom aggregate
  function usable in both `:find` grouping and `:over` (window) clauses
- `Minigraf::register_predicate(name, f)` — register a single-argument filter predicate
  usable in `[(name? ?var)]` `:where` clauses
- `FunctionRegistry::register_aggregate_desc` / `register_predicate_desc` (internal API)
- `WindowFunc::Udf(String)` and `UnaryOp::Udf(String)` AST variants for runtime-resolved functions
- `UdfOps`, `AggImpl`, `PredicateDesc` types in `functions.rs`

### Changed
- `AggregateDesc` now uses `AggImpl` discriminator instead of `window_compatible`+`window_ops`
- `apply_expr_clauses` now returns `Result<Vec<Binding>>` and accepts `&FunctionRegistry`
- `eval_expr` accepts `Option<&FunctionRegistry>` for UDF predicate resolution
- `WindowSpec::func_name()` now returns `String` instead of `&'static str`
- Parser emits `Udf` variants for unknown names instead of erroring (runtime validation)

### Test count: [N] tests
```

- [ ] **Step 5: Update `ROADMAP.md`**

Find the `7.7b` sub-section and mark it complete:

```markdown
#### 7.7b User-Defined Functions (UDFs) ✅ COMPLETE

**Status**: ✅ Complete (v0.17.0, 2026-04-02)

**Summary**: `register_aggregate` and `register_predicate` public API added to `Minigraf`.
UDFs plug into grouping aggregation, window computation, and `:where` predicate filtering
via type-erased `Box<dyn Any + Send>` accumulators and `Arc<dyn Fn>` closures.
`FunctionRegistry` extended with `AggImpl` discriminator and `PredicateDesc`.
[N] tests.
```

- [ ] **Step 6: Update `CLAUDE.md` test count**

Find the line `**707 tests passing**` and update it to the new count.

- [ ] **Step 7: Commit docs**

```bash
git add Cargo.toml CHANGELOG.md ROADMAP.md CLAUDE.md
git commit -m "docs: update CHANGELOG, ROADMAP, CLAUDE.md for v0.17.0 (Phase 7.7b complete)"
```

- [ ] **Step 8: Tag the release**

```bash
git tag -a v0.17.0 -m "Phase 7.7b complete — User-Defined Functions (UDFs)"
```

---

## Self-Review

**Spec coverage:**

| Spec requirement | Task |
|---|---|
| `UdfOps` with `Arc<dyn Fn>` closures | Task 2 |
| `AggImpl::Builtin` / `AggImpl::Udf` discriminator | Task 2 |
| `PredicateDesc` and `predicates` map | Task 2 |
| Built-in predicate name sentinels for collision detection | Task 2 |
| `WindowFunc::Udf(String)` | Task 1 |
| `UnaryOp::Udf(String)` | Task 1 |
| Parser emits `Udf` for unknown window names | Task 3 |
| Parser emits `Udf` for unknown 1-arg predicates | Task 3 |
| Unknown 2-arg still errors at parse time | Task 3 |
| `lag`/`lead` still rejected | Task 3 |
| `eval_expr` handles `UnaryOp::Udf` | Task 4 |
| `apply_expr_clauses` returns named error for unknown predicate | Task 4 |
| `compute_aggregation` handles `AggImpl::Udf` | Task 4 |
| `apply_window_functions` handles `WindowFunc::Udf` | Task 4 |
| `Minigraf::register_aggregate` | Task 5 |
| `Minigraf::register_predicate` | Task 5 |
| All 9 integration tests | Task 6 |
| Version bump + docs + tag | Task 7 |

**Placeholder scan:** No TBDs, all steps have complete code. ✓

**Type consistency:**
- `UdfOps` defined in Task 2 used in Tasks 4, 5 — names match ✓
- `AggImpl::Udf(UdfOps)` defined in Task 2 used in Tasks 4, 5 — matches ✓
- `PredicateDesc` defined in Task 2 used in Tasks 4, 5 — matches ✓
- `WindowFunc::Udf(String)` defined in Task 1 matched in Tasks 3, 4 — matches ✓
- `UnaryOp::Udf(String)` defined in Task 1 matched in Tasks 3, 4 — matches ✓
- `register_aggregate_desc` / `register_predicate_desc` defined in Task 2, called in Task 5 — match ✓
- `get_predicate` defined in Task 2, used in Task 4 — match ✓
- `func_name()` return type changed to `String` in Task 1, callers updated in Task 4 — ✓
