refactor: finish native compiler and planner ownership

CONTRIBUTING.md

@@ -111,6 +111,28 @@ uv run fensu dupes --since origin/main
 - Use project-local config files for local overrides, and do not commit credentials or personal settings.
 - Follow existing code style and module organization before introducing new patterns.
 
+### Native ownership and Python interfaces
+
+Compiler and planner algorithms belong in the Rust crates. Keep parsed header metadata, config
+nodes, dependency graphs and selections native between their producers and consumers. Constructing
+a Python container or public record only to extract it again is not an interface requirement.
+
+The supported Python interfaces have specific owners:
+
+| Interface | Why Python remains | Ownership rule |
+| --- | --- | --- |
+| Adapters and warehouse execution | Projects subclass adapters; the executor runs warehouse operations | Call through the adapter/executor boundary, retaining their public inputs and results |
+| User functions, factories, hooks, rules and custom constructors | These execute project-authored Python | Preserve invocation timing, errors and opaque return values; do not execute them speculatively |
+| Public records and caller-authored protocol objects | Python callers access, replace, compare and serialize documented records | Project native values on public access; import caller-authored values at entry, not repeatedly inside native algorithms |
+| Config and sample values | Python can supply values without a lossless native representation | Retain opaque values and their Python semantics; keep ordinary builtin containers native |
+| Streams, progress, logging and interpreter shutdown | These integrate with Python's runtime and caller-owned resources | Keep resource lifetime and error propagation at the owning boundary |
+
+`python-boundary.toml` lists the Python source that intentionally stays. The native crossing,
+storage and layer rules in `fensu.toml` prevent new accidental dependencies; pending budgets are
+an inventory to reduce, not permission to add callbacks. `native_host` owns genuine host edges,
+not a destination for movable compiler or planner algorithms. Remove an obsolete private
+interface with its last caller, while retaining supported public facades and stubs.
+
 ## Commits And Pull Requests
 
 - Use concise conventional commit messages, such as `fix: ...`, `feat: ...`, `test: ...`, or `docs: ...`.

benchmarks/duckshop/project/models/marts/edge_cases/rpt_edge_header_contracts.sql

@@ -0,0 +1,23 @@
+MODEL (
+  description "Order contract metadata retained beside a templated destination alias",
+  tags [edge_case],
+  alias "rpt_edge_header_contracts_${reporting_currency}",
+  contract enforced,
+  columns (
+    order_id (type INTEGER, nullable false, audits [not_null, unique]),
+    status (type order_status, nullable false, audits [not_null]),
+    amount (type DECIMAL(10, 2)),
+  ),
+);
+
+WITH final AS (
+  SELECT
+    7::INTEGER AS order_id,
+    @enum("order_status").PENDING::VARCHAR AS status,
+    12.50::DECIMAL(10, 2) AS amount
+  FROM __seed("currencies")
+  WHERE currency_code = 'USD'
+)
+
+SELECT order_id, status, amount
+FROM final

benchmarks/duckshop/project/models/marts/edge_cases/rpt_edge_planner_fan_in.sql

@@ -0,0 +1,46 @@
+MODEL (
+  description "Ledger report whose branches share an upstream model during migration planning",
+  tags [edge_case],
+  columns (
+    entry_id (type VARCHAR, nullable false, audits [not_null, unique]),
+    channel (type VARCHAR),
+    unit_amount (type NUMBER(38, 10)),
+  ),
+);
+
+WITH ledger AS (
+  SELECT entry_id
+  FROM __ref("stg_edge_ledger_entries")
+),
+
+amounts AS (
+  SELECT
+    entry_id,
+    unit_amount
+  FROM __ref("int_edge_line_amounts")
+),
+
+channels AS (
+  SELECT
+    entry_id,
+    channel
+  FROM __ref("int_edge_orders_all_channels")
+),
+
+final AS (
+  SELECT
+    l.entry_id,
+    c.channel,
+    a.unit_amount
+  FROM ledger AS l
+  INNER JOIN amounts AS a
+    ON l.entry_id = a.entry_id
+  INNER JOIN channels AS c
+    ON l.entry_id = c.entry_id
+)
+
+SELECT
+  entry_id,
+  channel,
+  unit_amount
+FROM final

benchmarks/duckshop/project/tests/unit/marts/edge_cases/test_rpt_edge_header_contracts.sql

@@ -0,0 +1,15 @@
+TEST (name "rpt_edge_header_contracts__keeps_enum_contract_with_templated_alias");
+
+WITH __seed__currencies AS (
+  SELECT 'USD' AS currency_code
+  UNION ALL
+  SELECT 'EUR' AS currency_code
+),
+__expected__rpt_edge_header_contracts AS (
+  SELECT
+    7::INTEGER AS order_id,
+    'PENDING'::VARCHAR AS status,
+    12.50::DECIMAL(10, 2) AS amount
+)
+
+SELECT 1

benchmarks/duckshop/project/tests/unit/marts/edge_cases/test_rpt_edge_planner_fan_in.sql

@@ -0,0 +1,29 @@
+TEST (name "rpt_edge_planner_fan_in__keeps_only_entries_in_every_branch");
+
+WITH __ref__stg_edge_ledger_entries AS (
+  SELECT v.entry_id
+  FROM VALUES ('LE-01'), ('LE-02'), ('LE-03') AS v(entry_id)
+),
+
+__ref__int_edge_line_amounts AS (
+  SELECT
+    v.entry_id,
+    v.unit_amount::DECIMAL(38, 10) AS unit_amount
+  FROM VALUES ('LE-01', 2.50), ('LE-02', 7.25), ('LE-04', 9.50) AS v(entry_id, unit_amount)
+),
+
+__ref__int_edge_orders_all_channels AS (
+  SELECT
+    v.entry_id,
+    v.channel
+  FROM VALUES ('LE-01', 'web'), ('LE-03', 'app'), ('LE-04', 'partner') AS v(entry_id, channel)
+),
+
+__expected__rpt_edge_planner_fan_in AS (
+  SELECT
+    'LE-01' AS entry_id,
+    'web' AS channel,
+    2.50::DECIMAL(38, 10) AS unit_amount
+)
+
+SELECT 1

crates/sqlbuild-analysis/src/graph/_helpers/indexes.rs

@@ -4,20 +4,14 @@ use std::collections::{HashMap, HashSet};
 
 use crate::assembly::project::types::ObjectKey;
 use crate::graph::constants::{MODEL_ROOT, MODEL_ROOT_PREFIX, SQL_TEST_RESOURCE};
-use crate::graph::models::{GraphIndexes, GraphResource, ProjectGraph};
+use crate::graph::main::downstream_edges::downstream_edges;
+use crate::graph::models::{GraphIndexes, GraphResource, OrderedEdges, ProjectGraph};
 
 type Edges = Vec<(ObjectKey, Vec<ObjectKey>)>;
 
-/// Insertion-ordered key lists, as Python's `dict.setdefault(key, []).append(...)` builds them.
-#[derive(Default)]
-struct OrderedEdges {
-    entries: Edges,
-    index: HashMap<ObjectKey, usize>,
-}
-
 impl OrderedEdges {
     /// The list stored for `key`, appended empty in insertion order when the key is new.
-    fn list(&mut self, key: &ObjectKey) -> &mut Vec<ObjectKey> {
+    pub(crate) fn list(&mut self, key: &ObjectKey) -> &mut Vec<ObjectKey> {
         let position: usize = match self.index.get(key) {
             Some(&position) => position,
             None => {
@@ -28,6 +22,13 @@ impl OrderedEdges {
         };
         &mut self.entries[position].1
     }
+
+    /// The dependencies already stored for `key`, without inserting a missing target.
+    pub(crate) fn get(&self, key: &ObjectKey) -> Option<&Vec<ObjectKey>> {
+        self.index
+            .get(key)
+            .map(|&position| &self.entries[position].1)
+    }
 }
 
 /// Tag to keys in first-tagged order, each key once, as Python's tag sets hold them.
@@ -70,7 +71,7 @@ pub(crate) fn build(resources: &[GraphResource]) -> ProjectGraph {
     {
         *upstream.list(&resource.key) = lineage_deps(&resource.deps);
     }
-    let downstream: Edges = downstream(&upstream.entries);
+    let downstream: Edges = downstream_edges(&upstream.entries);
     let mut tags: OrderedTags = OrderedTags::default();
     let mut graph = ProjectGraph {
         downstream_index: positions(&downstream),
@@ -125,24 +126,6 @@ fn lineage_deps(deps: &[ObjectKey]) -> Vec<ObjectKey> {
         .collect()
 }
 
-/// Inverted edges keyed in upstream order, each list sorted by `(resource type, name)`.
-fn downstream(upstream: &[(ObjectKey, Vec<ObjectKey>)]) -> Edges {
-    let mut inverted: OrderedEdges = OrderedEdges::default();
-    for (key, _) in upstream {
-        let _ = inverted.list(key);
-    }
-    for (key, deps) in upstream {
-        for dep in deps {
-            inverted.list(dep).push(key.clone());
-        }
-    }
-    let mut entries: Edges = inverted.entries;
-    for (_, keys) in &mut entries {
-        keys.sort();
-    }
-    entries
-}
-
 fn positions(entries: &[(ObjectKey, Vec<ObjectKey>)]) -> HashMap<ObjectKey, usize> {
     entries
         .iter()

crates/sqlbuild-analysis/src/graph/main/downstream_edges.rs

@@ -0,0 +1,24 @@
+//! Reverse ordered execution or lineage edges without projecting Python graph maps.
+
+use crate::assembly::project::types::ObjectKey;
+use crate::graph::models::OrderedEdges;
+
+/// Inverted edges keyed in upstream order, each list sorted by `(resource type, name)`.
+pub fn downstream_edges(
+    upstream: &[(ObjectKey, Vec<ObjectKey>)],
+) -> Vec<(ObjectKey, Vec<ObjectKey>)> {
+    let mut inverted = OrderedEdges::default();
+    for (key, _) in upstream {
+        let _ = inverted.list(key);
+    }
+    for (key, deps) in upstream {
+        for dep in deps {
+            inverted.list(dep).push(key.clone());
+        }
+    }
+    let mut entries = inverted.entries;
+    for (_, keys) in &mut entries {
+        keys.sort();
+    }
+    entries
+}

crates/sqlbuild-analysis/src/graph/main/mod.rs

@@ -1,10 +1,13 @@
 pub mod build_project_graph;
 pub mod build_resources;
+pub mod downstream_edges;
 pub mod edge_keys;
 pub mod graph_from_indexes;
 pub mod match_selector;
 pub mod model_layer_count;
 pub mod parse_selector;
+pub mod reachable_graph_keys;
+pub mod reachable_keys;
 pub mod resolve_selector_tokens;
 pub mod resolve_selectors;
 pub mod selector_name_help;

crates/sqlbuild-analysis/src/graph/main/reachable_graph_keys.rs

@@ -0,0 +1,23 @@
+//! Reachable keys over the project's existing edge indexes.
+
+use std::collections::HashSet;
+use std::convert::Infallible;
+
+use sqlbuild_core::graph_walk::main::breadth_first_reach::breadth_first_reach;
+
+use crate::assembly::project::types::ObjectKey;
+use crate::graph::main::edge_keys::edge_keys;
+use crate::graph::models::{EdgeDirection, ProjectGraph};
+
+/// The starts and every key reachable in the given direction.
+pub fn reachable_graph_keys(
+    starts: &[ObjectKey],
+    graph: &ProjectGraph,
+    direction: EdgeDirection,
+) -> HashSet<ObjectKey> {
+    let mut visited: HashSet<&ObjectKey> = starts.iter().collect();
+    let Ok(()) = breadth_first_reach(starts, None, &mut visited, |current| {
+        Ok::<_, Infallible>(edge_keys(graph, current, direction))
+    });
+    visited.into_iter().cloned().collect()
+}

crates/sqlbuild-analysis/src/graph/main/reachable_keys.rs

@@ -0,0 +1,21 @@
+//! Reachable compiled resource keys without projecting an intermediate host graph.
+
+use std::collections::{HashMap, HashSet};
+use std::convert::Infallible;
+
+use sqlbuild_core::graph_walk::main::breadth_first_reach::breadth_first_reach;
+
+use crate::assembly::project::types::ObjectKey;
+
+/// The starts and every key reachable through the ordered dependency lists.
+pub fn reachable_keys(
+    starts: &[ObjectKey],
+    edges: &[(ObjectKey, Vec<ObjectKey>)],
+) -> HashSet<ObjectKey> {
+    let index: HashMap<_, _> = edges.iter().map(|(key, deps)| (key, deps)).collect();
+    let mut visited: HashSet<&ObjectKey> = starts.iter().collect();
+    let Ok(()) = breadth_first_reach(starts, None, &mut visited, |current| {
+        Ok::<_, Infallible>(index.get(current).map_or(&[][..], |deps| deps.as_slice()))
+    });
+    visited.into_iter().cloned().collect()
+}

crates/sqlbuild-analysis/src/graph/models.rs

@@ -4,6 +4,13 @@ use std::collections::HashMap;
 
 use crate::assembly::project::types::ObjectKey;
 
+/// Dependency lists indexed without losing project insertion order.
+#[derive(Debug, Default)]
+pub(crate) struct OrderedEdges {
+    pub(crate) entries: Vec<(ObjectKey, Vec<ObjectKey>)>,
+    pub(crate) index: HashMap<ObjectKey, usize>,
+}
+
 /// One graph resource in project order: its key, upstream keys, tags and model folder.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct GraphResource {