From 5615d48731e656c0a65d2c84a457a67840d52b6e Mon Sep 17 00:00:00 2001
From: zzylol <50204836+zzylol@users.noreply.github.com>
Date: Thu, 1 Oct 2026 23:07:00 +0000
Subject: [PATCH] fix: plan each root's lifecycle against its own entries

MajorPass bound every root's lifecycle assembly to all workload entries,
so a root's expected reads, accuracy targets and prepared-state
eligibility were the whole workload's. Selection already costs each
target against the entries whose roots reach it; assembly now uses the
same binding (`workload_entries_by_target`), so roots shared by several
entries still see all of them and unrelated queries no longer amortize
each other's state.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/asap-aware-mapping/src/pass/major.rs | 11 ++++-
 crates/planner/tests/e2e_plan.rs            | 45 +++++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

diff --git a/crates/asap-aware-mapping/src/pass/major.rs b/crates/asap-aware-mapping/src/pass/major.rs
index 6d148020f..d7aa0c7d7 100644
--- a/crates/asap-aware-mapping/src/pass/major.rs
+++ b/crates/asap-aware-mapping/src/pass/major.rs
@@ -72,13 +72,22 @@ impl OptimizationPass for MajorPass {
             models.cost,
         )
         .map_err(OptimizeError::LifecycleSelection)?;
+        // Each root's lifecycle is planned against the entries that consume
+        // it — the same binding selection costed it with — not the whole
+        // workload, so one query's reads never amortize another's state.
+        let bindings = space
+            .workload_entries_by_target(demand.workload, &entry_indices)
+            .map_err(|error| OptimizeError::LifecycleSelection(error.into()))?;
 
         let mut plans = Vec::with_capacity(space.roots.len());
         for (entry_index, root) in &space.roots {
             let plan = assemble_selected_dag_with_summary_maintenance_lifecycles(
                 &selection,
                 root,
-                demand,
+                WorkloadDemand {
+                    entry_indices: &bindings[&Rc::as_ptr(root)],
+                    ..demand
+                },
                 lifecycle.now_ms,
                 lifecycle.horizon,
                 lifecycle.capabilities,
diff --git a/crates/planner/tests/e2e_plan.rs b/crates/planner/tests/e2e_plan.rs
index c4d5120d7..684de5f79 100644
--- a/crates/planner/tests/e2e_plan.rs
+++ b/crates/planner/tests/e2e_plan.rs
@@ -392,3 +392,48 @@ async fn lifecycle_decisions_ride_inside_each_plan() {
     let _: &Rc<_> = &output.plans[0].plan.root;
     assert_eq!(output.dags().len(), 1);
 }
+
+/// Each root's lifecycle is planned against the entries that read it: a
+/// query polled every minute and an unrelated one polled every ten minutes
+/// each see only their own reads over the hour, not the workload's 66.
+#[tokio::test]
+async fn each_plan_counts_only_its_own_entries_reads() {
+    let repeating = |query: &str, interval_ms: u32| RepeatingEntry {
+        query: Query(query.into()),
+        demand: RepeatedDemand::FixedInterval(RepetitionInterval(interval_ms)),
+        requirements: approximate(),
+        predictability: Predictability::Unknown,
+        time_selection: TimeSelection::default(),
+    };
+    let workload = PlanningWorkload {
+        query_workload: QueryWorkload {
+            language: QueryLanguage::PromQL,
+            query_batch: None,
+            repeating_queries: Some(vec![
+                repeating("count_over_time(up[5m])", 60_000),
+                repeating("sum_over_time(latency[5m])", 600_000),
+            ]),
+        },
+        data_workload: Some(DataWorkload {
+            arrival: DataArrival::ContinuouslyIngesting,
+            data_ingestion_interval: Evidence {
+                value: Some(DurationMs(15_000)),
+                ..Default::default()
+            },
+            ..Default::default()
+        }),
+    };
+    let input = UserInput::new(
+        &workload,
+        FrontendInput::Promql {
+            now_ms: NOW_MS,
+            histograms: None,
+        },
+        PlanningModels::builtin(),
+        lifecycle().with_horizon(Horizon(3_600.0)),
+    );
+
+    let output = e2e_plan(input).await.expect("workload plans");
+    let reads: Vec<_> = output.plans.iter().map(|p| p.plan.expected_reads).collect();
+    assert_eq!(reads, vec![Some(60.0), Some(6.0)]);
+}