diff --git a/crates/e2e/tests/aggregate.rs b/crates/e2e/tests/aggregate.rs
new file mode 100644
index 00000000..a56c1813
--- /dev/null
+++ b/crates/e2e/tests/aggregate.rs
@@ -0,0 +1,179 @@
+//! `QueryExpr::Aggregate` — cross-series aggregation tests.
+//!
+//! topk/bottomk are omitted — dispatch is deferred.
+//!
+//! Cross-series aggregates lower to a single `Aggregate` node with no
+//! `TimeRange` child (range functions use `TimeRange` — see `time_range.rs`).
+//! Group keys land on `Aggregate.by` as positional `ColumnId`s.
+//! Single-stat PromQL aggregates always get `output_names: [""]` (no alias)
+//! and `having: None`.
+
+use asap_control_core::intent_algebra::{AggIntent, QueryExpr, Source};
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+use asap_e2e::fixtures::metric_schema;
+
+fn lower(q: &str) -> QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+fn scan(metric: &str, labels: &[&str]) -> QueryExpr {
+    QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: metric.into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(labels),
+    }
+}
+
+fn agg(by: Vec<usize>, intent: AggIntent, child: QueryExpr) -> QueryExpr {
+    QueryExpr::Aggregate {
+        by,
+        aggs: vec![intent],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(child),
+    }
+}
+
+// #5 — sum with no group keys
+#[test]
+fn q05_sum_no_group() {
+    assert_eq!(
+        lower("sum(http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::Sum { col: None },
+            scan("http_requests_total", &[])
+        ),
+    );
+}
+
+// #6 — sum grouped by job; job is col 2 in [ts, value, job]
+#[test]
+fn q06_sum_by_job() {
+    assert_eq!(
+        lower("sum by (job) (http_requests_total)"),
+        agg(
+            vec![2],
+            AggIntent::Sum { col: None },
+            scan("http_requests_total", &["job"])
+        ),
+    );
+}
+
+// #7 — PromQL `count` is cross-series cardinality, not per-sample Count
+#[test]
+fn q07_count_is_cardinality() {
+    assert_eq!(
+        lower("count(http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::Cardinality {
+                accuracy: AccuracyTarget::Exact
+            },
+            scan("http_requests_total", &[]),
+        ),
+    );
+}
+
+// #8 — avg grouped by datacenter; datacenter is col 2
+#[test]
+fn q08_avg_by_datacenter() {
+    assert_eq!(
+        lower("avg by (datacenter) (http_requests_total)"),
+        agg(
+            vec![2],
+            AggIntent::Avg { col: None },
+            scan("http_requests_total", &["datacenter"])
+        ),
+    );
+}
+
+// multiple group keys — sum by (job, status); alphabetical: job(2), status(3)
+#[test]
+fn q_sum_by_job_and_status() {
+    assert_eq!(
+        lower("sum by (job, status) (http_requests_total)"),
+        agg(
+            vec![2, 3],
+            AggIntent::Sum { col: None },
+            scan("http_requests_total", &["job", "status"])
+        ),
+    );
+}
+
+// min — cross-series minimum
+#[test]
+fn q_min_no_group() {
+    assert_eq!(
+        lower("min(http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::Min { col: None },
+            scan("http_requests_total", &[])
+        ),
+    );
+}
+
+// max — cross-series maximum, grouped by job
+#[test]
+fn q_max_by_job() {
+    assert_eq!(
+        lower("max by (job) (http_requests_total)"),
+        agg(
+            vec![2],
+            AggIntent::Max { col: None },
+            scan("http_requests_total", &["job"])
+        ),
+    );
+}
+
+// stddev — cross-series standard deviation; PromQL stddev uses population=false in the lowering
+#[test]
+fn q_stddev_no_group() {
+    assert_eq!(
+        lower("stddev(http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::StdDev {
+                col: None,
+                population: false
+            },
+            scan("http_requests_total", &[])
+        ),
+    );
+}
+
+// stdvar — cross-series variance
+#[test]
+fn q_stdvar_no_group() {
+    assert_eq!(
+        lower("stdvar(http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::Variance {
+                col: None,
+                population: false
+            },
+            scan("http_requests_total", &[])
+        ),
+    );
+}
+
+// #10 — cross-series quantile; no TimeRange node (no range window)
+#[test]
+fn q10_quantile_cross_series() {
+    assert_eq!(
+        lower("quantile(0.99, http_requests_total)"),
+        agg(
+            vec![],
+            AggIntent::Quantile {
+                q: 0.99,
+                accuracy: AccuracyTarget::Exact
+            },
+            scan("http_requests_total", &[]),
+        ),
+    );
+}
diff --git a/crates/e2e/tests/binary_op.rs b/crates/e2e/tests/binary_op.rs
new file mode 100644
index 00000000..8149763d
--- /dev/null
+++ b/crates/e2e/tests/binary_op.rs
@@ -0,0 +1,228 @@
+//! `QueryExpr::BinaryOp` — arithmetic, comparison, and vector-match tests.
+//!
+//! Each side of a `BinaryOp` is bound independently by the Binder, so each
+//! gets its own scan schema derived from the labels it references.
+//! `VectorMatch` labels (e.g. `on(job)`) are carried as strings on the node
+//! and are NOT resolved to column ids — the Binder does not see them.
+
+use std::time::Duration;
+
+use asap_control_core::intent_algebra::{
+    AggIntent, ArithOp, BinaryOpKind, CompareOp, GroupSide, QueryExpr, Source, VectorGrouping,
+    VectorMatch, VectorMatchKind,
+};
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+use asap_e2e::fixtures::metric_schema;
+
+fn lower(q: &str) -> QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+fn scan(metric: &str, labels: &[&str]) -> QueryExpr {
+    QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: metric.into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(labels),
+    }
+}
+
+fn rate_agg(metric: &str) -> QueryExpr {
+    QueryExpr::Aggregate {
+        by: vec![],
+        aggs: vec![AggIntent::Rate],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(QueryExpr::TimeRange {
+            range: Duration::from_secs(300),
+            child: Box::new(scan(metric, &[])),
+        }),
+    }
+}
+
+fn sum_by_job(metric: &str) -> QueryExpr {
+    QueryExpr::Aggregate {
+        by: vec![2],
+        aggs: vec![AggIntent::Sum { col: None }],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(scan(metric, &["job"])),
+    }
+}
+
+// #18 — arithmetic binary op between two bare scans; no vector match
+#[test]
+fn q18_div_bare_scans() {
+    let expected = QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arith(ArithOp::Div),
+        lhs: Box::new(scan("http_requests_total", &[])),
+        rhs: Box::new(scan("http_requests_total", &[])),
+        vector_match: None,
+    };
+    assert_eq!(lower("http_requests_total / http_requests_total"), expected);
+}
+
+// #19 — add with on(job) vector match; match labels are strings, not column ids
+#[test]
+fn q19_add_with_on_match() {
+    let expected = QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arith(ArithOp::Add),
+        lhs: Box::new(scan("http_requests_total", &[])),
+        rhs: Box::new(scan("http_requests_total", &[])),
+        vector_match: Some(VectorMatch {
+            kind: VectorMatchKind::On,
+            labels: vec!["job".into()],
+            grouping: None,
+        }),
+    };
+    assert_eq!(
+        lower("http_requests_total + on(job) http_requests_total"),
+        expected
+    );
+}
+
+// #20 — divide two rate aggregates over different metrics
+#[test]
+fn q20_div_two_rates() {
+    let expected = QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arith(ArithOp::Div),
+        lhs: Box::new(rate_agg("http_requests_total")),
+        rhs: Box::new(rate_agg("http_errors_total")),
+        vector_match: None,
+    };
+    assert_eq!(
+        lower("rate(http_requests_total[5m]) / rate(http_errors_total[5m])"),
+        expected,
+    );
+}
+
+// comparison ops — filter semantics; each op between two instant vectors
+#[test]
+fn q_gt_comparison() {
+    assert_eq!(
+        lower("http_requests_total > http_errors_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Compare(CompareOp::Gt),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("http_errors_total", &[])),
+            vector_match: None,
+        }
+    );
+}
+
+#[test]
+fn q_lt_comparison() {
+    assert_eq!(
+        lower("http_requests_total < http_errors_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Compare(CompareOp::Lt),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("http_errors_total", &[])),
+            vector_match: None,
+        }
+    );
+}
+
+#[test]
+fn q_ge_comparison() {
+    assert_eq!(
+        lower("http_requests_total >= http_errors_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Compare(CompareOp::Ge),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("http_errors_total", &[])),
+            vector_match: None,
+        }
+    );
+}
+
+#[test]
+fn q_le_comparison() {
+    assert_eq!(
+        lower("http_requests_total <= http_errors_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Compare(CompareOp::Le),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("http_errors_total", &[])),
+            vector_match: None,
+        }
+    );
+}
+
+// ignoring(job) — match on all labels except job; labels are strings, not column ids
+#[test]
+fn q_add_with_ignoring() {
+    assert_eq!(
+        lower("http_requests_total + ignoring(job) http_errors_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Arith(ArithOp::Add),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("http_errors_total", &[])),
+            vector_match: Some(VectorMatch {
+                kind: VectorMatchKind::Ignoring,
+                labels: vec!["job".into()],
+                grouping: None,
+            }),
+        }
+    );
+}
+
+// group_left — many-to-one: left side has higher cardinality
+#[test]
+fn q_mul_group_left() {
+    assert_eq!(
+        lower("http_requests_total * on(job) group_left() node_info"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Arith(ArithOp::Mul),
+            lhs: Box::new(scan("http_requests_total", &[])),
+            rhs: Box::new(scan("node_info", &[])),
+            vector_match: Some(VectorMatch {
+                kind: VectorMatchKind::On,
+                labels: vec!["job".into()],
+                grouping: Some(VectorGrouping {
+                    side: GroupSide::Left,
+                    labels: vec![],
+                }),
+            }),
+        }
+    );
+}
+
+// group_right — one-to-many: right side has higher cardinality
+#[test]
+fn q_mul_group_right() {
+    assert_eq!(
+        lower("node_info * on(job) group_right() http_requests_total"),
+        QueryExpr::BinaryOp {
+            op: BinaryOpKind::Arith(ArithOp::Mul),
+            lhs: Box::new(scan("node_info", &[])),
+            rhs: Box::new(scan("http_requests_total", &[])),
+            vector_match: Some(VectorMatch {
+                kind: VectorMatchKind::On,
+                labels: vec!["job".into()],
+                grouping: Some(VectorGrouping {
+                    side: GroupSide::Right,
+                    labels: vec![],
+                }),
+            }),
+        }
+    );
+}
+
+// #21 — divide two sum-by-job aggregates over different metrics
+//   each side: Aggregate{Sum, by=[2]} over Scan([ts, value, job])
+#[test]
+fn q21_div_two_sum_by_job() {
+    let expected = QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arith(ArithOp::Div),
+        lhs: Box::new(sum_by_job("http_requests_total")),
+        rhs: Box::new(sum_by_job("http_errors_total")),
+        vector_match: None,
+    };
+    assert_eq!(
+        lower("sum by (job) (http_requests_total) / sum by (job) (http_errors_total)"),
+        expected,
+    );
+}
diff --git a/crates/e2e/tests/nested.rs b/crates/e2e/tests/nested.rs
new file mode 100644
index 00000000..72eee964
--- /dev/null
+++ b/crates/e2e/tests/nested.rs
@@ -0,0 +1,177 @@
+//! Multi-node pipeline tests — nested `Aggregate`, `TimeRange`, `BinaryOp`, and `Scan`.
+//!
+//! Key invariant: `rate`/`increase` are label-preserving (per-series), so an
+//! outer `Aggregate.by` resolves its group keys against the inner aggregate's
+//! output schema, which still carries all label columns.
+//!
+//! Label column ordering is always alphabetical, so in a query that references
+//! both `job` and `status`:
+//!   schema = [ts(0), value(1), job(2), status(3)]
+
+use std::time::Duration;
+
+use asap_control_core::intent_algebra::{
+    AggIntent, ArithOp, BinaryOpKind, CompareOp, L3Expr, L3Scalar, Predicate, QueryExpr, Source,
+};
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+use asap_e2e::fixtures::metric_schema;
+
+fn lower(q: &str) -> QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+fn agg(by: Vec<usize>, intent: AggIntent, child: QueryExpr) -> QueryExpr {
+    QueryExpr::Aggregate {
+        by,
+        aggs: vec![intent],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(child),
+    }
+}
+
+// #22 — sum by job over rate; outer by=[2] resolves against rate's
+//   label-preserving output schema [ts, value, job]
+#[test]
+fn q22_sum_by_job_over_rate() {
+    let scan = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(&["job"]),
+    };
+    let inner_rate = agg(
+        vec![],
+        AggIntent::Rate,
+        QueryExpr::TimeRange {
+            range: Duration::from_secs(300),
+            child: Box::new(scan),
+        },
+    );
+    let expected = agg(vec![2], AggIntent::Sum { col: None }, inner_rate);
+    assert_eq!(
+        lower("sum by (job) (rate(http_requests_total[5m]))"),
+        expected
+    );
+}
+
+// #23 — sum by job over a filtered scan; status="200" is a filter-only label
+//   labels sorted: job(2) < status(3)
+//   predicate on status (col 3); group key job (col 2)
+#[test]
+fn q23_sum_by_job_over_filtered_scan() {
+    let scan = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![Predicate(L3Expr::Compare {
+            left: Box::new(L3Expr::Column(3)),
+            op: CompareOp::Eq,
+            right: Box::new(L3Expr::Literal(L3Scalar::Utf8("200".into()))),
+        })],
+        schema: metric_schema(&["job", "status"]),
+    };
+    let expected = agg(vec![2], AggIntent::Sum { col: None }, scan);
+    assert_eq!(
+        lower(r#"sum by (job) (http_requests_total{status="200"})"#),
+        expected
+    );
+}
+
+// #25 — binary op over two complex subtrees
+//   LHS: sum by (job) over rate over filtered scan
+//     schema [ts, value, job, status]; outer by=[2] (job)
+//   RHS: sum by (job) over rate over bare scan
+//     schema [ts, value, job]; outer by=[2] (job)
+#[test]
+fn q25_div_over_complex_subtrees() {
+    let lhs_scan = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![Predicate(L3Expr::Compare {
+            left: Box::new(L3Expr::Column(3)),
+            op: CompareOp::Eq,
+            right: Box::new(L3Expr::Literal(L3Scalar::Utf8("200".into()))),
+        })],
+        schema: metric_schema(&["job", "status"]),
+    };
+    let lhs = agg(
+        vec![2],
+        AggIntent::Sum { col: None },
+        agg(
+            vec![],
+            AggIntent::Rate,
+            QueryExpr::TimeRange {
+                range: Duration::from_secs(300),
+                child: Box::new(lhs_scan),
+            },
+        ),
+    );
+
+    let rhs_scan = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_errors_total".into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(&["job"]),
+    };
+    let rhs = agg(
+        vec![2],
+        AggIntent::Sum { col: None },
+        agg(
+            vec![],
+            AggIntent::Rate,
+            QueryExpr::TimeRange {
+                range: Duration::from_secs(300),
+                child: Box::new(rhs_scan),
+            },
+        ),
+    );
+
+    let expected = QueryExpr::BinaryOp {
+        op: BinaryOpKind::Arith(ArithOp::Div),
+        lhs: Box::new(lhs),
+        rhs: Box::new(rhs),
+        vector_match: None,
+    };
+    assert_eq!(
+        lower(
+            r#"sum by (job) (rate(http_requests_total{status="200"}[5m])) / sum by (job) (rate(http_errors_total[5m]))"#
+        ),
+        expected
+    );
+}
+
+// #24 — sum by job over rate over a filtered scan
+//   same schema [ts, value, job, status]; rate is label-preserving,
+//   so outer sum by job still finds job at col 2
+#[test]
+fn q24_sum_by_job_over_rate_over_filtered_scan() {
+    let scan = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![Predicate(L3Expr::Compare {
+            left: Box::new(L3Expr::Column(3)),
+            op: CompareOp::Eq,
+            right: Box::new(L3Expr::Literal(L3Scalar::Utf8("200".into()))),
+        })],
+        schema: metric_schema(&["job", "status"]),
+    };
+    let inner_rate = agg(
+        vec![],
+        AggIntent::Rate,
+        QueryExpr::TimeRange {
+            range: Duration::from_secs(300),
+            child: Box::new(scan),
+        },
+    );
+    let expected = agg(vec![2], AggIntent::Sum { col: None }, inner_rate);
+    assert_eq!(
+        lower(r#"sum by (job) (rate(http_requests_total{status="200"}[5m]))"#),
+        expected,
+    );
+}
diff --git a/crates/e2e/tests/scan.rs b/crates/e2e/tests/scan.rs
new file mode 100644
index 00000000..94a8f79d
--- /dev/null
+++ b/crates/e2e/tests/scan.rs
@@ -0,0 +1,143 @@
+//! `QueryExpr::Scan` — label matcher / predicate tests.
+//!
+//! The Scan schema is always [ts(0), value(1), label_a(2), label_b(3), …]
+//! where labels are appended alphabetically after dedup by the Binder.
+//! Filter-only labels (not group keys) still land in the schema because the
+//! predicate expression references them positionally.
+//! Predicates are canonicalized alphabetically by label name at lowering time.
+
+use asap_control_core::intent_algebra::{
+    CompareOp, L3Expr, L3Scalar, Predicate, QueryExpr, Source,
+};
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+use asap_e2e::fixtures::metric_schema;
+
+fn lower(q: &str) -> QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+fn bare_scan(metric: &str, labels: &[&str]) -> QueryExpr {
+    QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: metric.into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(labels),
+    }
+}
+
+fn eq_pred(col_id: usize, value: &str) -> Predicate {
+    Predicate(L3Expr::Compare {
+        left: Box::new(L3Expr::Column(col_id)),
+        op: CompareOp::Eq,
+        right: Box::new(L3Expr::Literal(L3Scalar::Utf8(value.into()))),
+    })
+}
+
+fn ne_pred(col_id: usize, value: &str) -> Predicate {
+    Predicate(L3Expr::Compare {
+        left: Box::new(L3Expr::Column(col_id)),
+        op: CompareOp::Ne,
+        right: Box::new(L3Expr::Literal(L3Scalar::Utf8(value.into()))),
+    })
+}
+
+fn regex_pred(col_id: usize, pattern: &str) -> Predicate {
+    Predicate(L3Expr::Compare {
+        left: Box::new(L3Expr::Column(col_id)),
+        op: CompareOp::Regex,
+        right: Box::new(L3Expr::Literal(L3Scalar::Utf8(pattern.into()))),
+    })
+}
+
+fn notregex_pred(col_id: usize, pattern: &str) -> Predicate {
+    Predicate(L3Expr::Compare {
+        left: Box::new(L3Expr::Column(col_id)),
+        op: CompareOp::NotRegex,
+        right: Box::new(L3Expr::Literal(L3Scalar::Utf8(pattern.into()))),
+    })
+}
+
+// #1 — bare metric name, no matchers
+#[test]
+fn q01_bare_scan() {
+    assert_eq!(
+        lower("http_requests_total"),
+        bare_scan("http_requests_total", &[])
+    );
+}
+
+// #2 — single equality matcher
+//   schema: [ts(0), value(1), job(2)]
+#[test]
+fn q02_equality_predicate() {
+    let expected = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![eq_pred(2, "api-server")],
+        schema: metric_schema(&["job"]),
+    };
+    assert_eq!(lower(r#"http_requests_total{job="api-server"}"#), expected);
+}
+
+// #3 — single inequality matcher
+//   schema: [ts(0), value(1), status(2)]
+#[test]
+fn q03_inequality_predicate() {
+    let expected = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![ne_pred(2, "500")],
+        schema: metric_schema(&["status"]),
+    };
+    assert_eq!(lower(r#"http_requests_total{status!="500"}"#), expected);
+}
+
+// #4 — regex matcher; RHS is the pattern string, op is Regex
+//   schema: [ts(0), value(1), job(2)]
+#[test]
+fn q04_regex_predicate() {
+    let expected = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![regex_pred(2, "api.*")],
+        schema: metric_schema(&["job"]),
+    };
+    assert_eq!(lower(r#"http_requests_total{job=~"api.*"}"#), expected);
+}
+
+// negative regex matcher; RHS is the pattern string, op is NotRegex
+//   schema: [ts(0), value(1), job(2)]
+#[test]
+fn q_notregex_predicate() {
+    let expected = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![notregex_pred(2, "internal.*")],
+        schema: metric_schema(&["job"]),
+    };
+    assert_eq!(lower(r#"http_requests_total{job!~"internal.*"}"#), expected);
+}
+
+// multiple matchers — two predicates, canonicalized alphabetically by label name
+//   labels sorted: job(2) < status(3)  →  schema: [ts, value, job, status]
+//   predicates in same alphabetical order: job first, then status
+#[test]
+fn q_multi_two_predicates() {
+    let expected = QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: "http_requests_total".into(),
+        },
+        predicates: vec![eq_pred(2, "api-server"), ne_pred(3, "500")],
+        schema: metric_schema(&["job", "status"]),
+    };
+    assert_eq!(
+        lower(r#"http_requests_total{job="api-server",status!="500"}"#),
+        expected,
+    );
+}
diff --git a/crates/e2e/tests/schema.rs b/crates/e2e/tests/schema.rs
new file mode 100644
index 00000000..f088a569
--- /dev/null
+++ b/crates/e2e/tests/schema.rs
@@ -0,0 +1,104 @@
+//! `Schema::closed` propagation — open/closed invariant tests.
+//!
+//! Verifies that `QueryExpr::output_schema()` propagates the open/closed
+//! completeness flag correctly through a lowered query tree.
+//!
+//! Key invariant: a PromQL scan is always `closed: false` (open) because its
+//! label set is runtime-only.  The schema freezes to `closed: true` exactly at
+//! the first operator that fully enumerates its output columns: a cross-series
+//! `Aggregate` or a `Project`.  Per-series reductions (`rate`, `*_over_time`)
+//! are label-preserving and keep the schema open.
+
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+
+fn lower(q: &str) -> asap_control_core::intent_algebra::QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+// bare scan is open — the metric's full label set is unknown at plan time
+#[test]
+fn schema_bare_scan_is_open() {
+    let s = lower("http_requests_total").output_schema().unwrap();
+    assert!(!s.closed, "PromQL scan must be open");
+}
+
+// scan with label filter is also open — the listed labels are referenced, not exhaustive
+#[test]
+fn schema_filtered_scan_is_open() {
+    let s = lower(r#"http_requests_total{job="api-server"}"#)
+        .output_schema()
+        .unwrap();
+    assert!(!s.closed, "PromQL scan with predicates must remain open");
+    assert_eq!(s.columns.len(), 3, "[ts, value, job]");
+}
+
+// per-series rate is label-preserving → output stays open
+#[test]
+fn schema_rate_stays_open() {
+    let s = lower("rate(http_requests_total[5m])")
+        .output_schema()
+        .unwrap();
+    assert!(!s.closed, "per-series rate is label-preserving; stays open");
+}
+
+// per-series count_over_time is also label-preserving → stays open
+#[test]
+fn schema_count_over_time_stays_open() {
+    let s = lower("count_over_time(http_requests_total[5m])")
+        .output_schema()
+        .unwrap();
+    assert!(!s.closed, "per-series count_over_time stays open");
+}
+
+// cross-series sum with no group keys freezes to closed
+#[test]
+fn schema_sum_freezes_to_closed() {
+    let s = lower("sum(http_requests_total)").output_schema().unwrap();
+    assert!(s.closed, "cross-series aggregate must freeze to closed");
+}
+
+// cross-series sum grouped by job also freezes to closed
+#[test]
+fn schema_sum_by_job_freezes_to_closed() {
+    let s = lower("sum by (job) (http_requests_total)")
+        .output_schema()
+        .unwrap();
+    assert!(
+        s.closed,
+        "grouped cross-series aggregate must freeze to closed"
+    );
+}
+
+// open scan → per-series rate → cross-series sum: freezes at the outer aggregate
+#[test]
+fn schema_sum_over_rate_freezes_to_closed() {
+    let s = lower("sum by (job) (rate(http_requests_total[5m]))")
+        .output_schema()
+        .unwrap();
+    assert!(
+        s.closed,
+        "cross-series aggregate over rate must freeze to closed"
+    );
+}
+
+// binary op between two open scans: output stays open (open && open → open)
+#[test]
+fn schema_binary_op_two_open_stays_open() {
+    let s = lower("http_requests_total / http_errors_total")
+        .output_schema()
+        .unwrap();
+    assert!(!s.closed, "binary op over two open scans must stay open");
+}
+
+// binary op between two closed aggregates: output is closed (closed && closed → closed)
+#[test]
+fn schema_binary_op_two_closed_is_closed() {
+    let s = lower("sum by (job) (http_requests_total) / sum by (job) (http_errors_total)")
+        .output_schema()
+        .unwrap();
+    assert!(
+        s.closed,
+        "binary op over two closed aggregates must be closed"
+    );
+}
diff --git a/crates/e2e/tests/time_range.rs b/crates/e2e/tests/time_range.rs
new file mode 100644
index 00000000..d04f7fd9
--- /dev/null
+++ b/crates/e2e/tests/time_range.rs
@@ -0,0 +1,159 @@
+//! `QueryExpr::TimeRange` — range / streaming function tests.
+//!
+//! All range functions lower to `Aggregate { child: TimeRange { range, child: Scan } }`.
+//! The temporal range lives on the `TimeRange` node, not in the `AggIntent`.
+//! `rate` / `increase` use `AggIntent::Rate` / `AggIntent::Increase` (no window field).
+//! `*_over_time` functions reuse the corresponding cross-series intents
+//! (`Count`, `Sum`, `Quantile`, …) — the `TimeRange` child is what marks them
+//! as per-series reductions.
+
+use std::time::Duration;
+
+use asap_control_core::intent_algebra::{AggIntent, QueryExpr, Source};
+use asap_control_core::types::AccuracyTarget;
+use asap_control_lower::lower_promql;
+use asap_e2e::fixtures::metric_schema;
+
+fn lower(q: &str) -> QueryExpr {
+    lower_promql(q, AccuracyTarget::Exact).unwrap_or_else(|e| panic!("lower failed for {q:?}: {e}"))
+}
+
+fn scan(metric: &str) -> QueryExpr {
+    QueryExpr::Scan {
+        source: Source::TimeSeries {
+            metric: metric.into(),
+        },
+        predicates: vec![],
+        schema: metric_schema(&[]),
+    }
+}
+
+fn range_agg(range_secs: u64, intent: AggIntent, metric: &str) -> QueryExpr {
+    QueryExpr::Aggregate {
+        by: vec![],
+        aggs: vec![intent],
+        output_names: vec!["".into()],
+        having: None,
+        child: Box::new(QueryExpr::TimeRange {
+            range: Duration::from_secs(range_secs),
+            child: Box::new(scan(metric)),
+        }),
+    }
+}
+
+// #13 — rate: counter-reset-aware per-second rate; range on TimeRange node
+#[test]
+fn q13_rate() {
+    assert_eq!(
+        lower("rate(http_requests_total[5m])"),
+        range_agg(300, AggIntent::Rate, "http_requests_total"),
+    );
+}
+
+// #14 — increase: cumulative increase over the range window
+#[test]
+fn q14_increase() {
+    assert_eq!(
+        lower("increase(http_requests_total[1h])"),
+        range_agg(3600, AggIntent::Increase, "http_requests_total"),
+    );
+}
+
+// #15 — count_over_time: sample count per series over the window
+#[test]
+fn q15_count_over_time() {
+    assert_eq!(
+        lower("count_over_time(http_requests_total[5m])"),
+        range_agg(
+            300,
+            AggIntent::Count {
+                accuracy: AccuracyTarget::Exact
+            },
+            "http_requests_total"
+        ),
+    );
+}
+
+// #16 — sum_over_time: sum of samples per series over the window
+#[test]
+fn q16_sum_over_time() {
+    assert_eq!(
+        lower("sum_over_time(http_requests_total[5m])"),
+        range_agg(300, AggIntent::Sum { col: None }, "http_requests_total"),
+    );
+}
+
+// avg_over_time: per-series mean over the window
+#[test]
+fn q_avg_over_time() {
+    assert_eq!(
+        lower("avg_over_time(http_requests_total[5m])"),
+        range_agg(300, AggIntent::Avg { col: None }, "http_requests_total"),
+    );
+}
+
+// min_over_time: per-series minimum over the window
+#[test]
+fn q_min_over_time() {
+    assert_eq!(
+        lower("min_over_time(http_requests_total[5m])"),
+        range_agg(300, AggIntent::Min { col: None }, "http_requests_total"),
+    );
+}
+
+// max_over_time: per-series maximum over the window
+#[test]
+fn q_max_over_time() {
+    assert_eq!(
+        lower("max_over_time(http_requests_total[5m])"),
+        range_agg(300, AggIntent::Max { col: None }, "http_requests_total"),
+    );
+}
+
+// stddev_over_time: per-series standard deviation over the window; population=false per lowering
+#[test]
+fn q_stddev_over_time() {
+    assert_eq!(
+        lower("stddev_over_time(http_requests_total[5m])"),
+        range_agg(
+            300,
+            AggIntent::StdDev {
+                col: None,
+                population: false
+            },
+            "http_requests_total"
+        ),
+    );
+}
+
+// stdvar_over_time: per-series variance over the window
+#[test]
+fn q_stdvar_over_time() {
+    assert_eq!(
+        lower("stdvar_over_time(http_requests_total[5m])"),
+        range_agg(
+            300,
+            AggIntent::Variance {
+                col: None,
+                population: false
+            },
+            "http_requests_total"
+        ),
+    );
+}
+
+// #17 — quantile_over_time: per-series quantile over the window
+#[test]
+fn q17_quantile_over_time() {
+    assert_eq!(
+        lower("quantile_over_time(0.99, http_requests_total[5m])"),
+        range_agg(
+            300,
+            AggIntent::Quantile {
+                q: 0.99,
+                accuracy: AccuracyTarget::Exact
+            },
+            "http_requests_total",
+        ),
+    );
+}