diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index e5057ca5..0fbebf62 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -372,6 +372,8 @@ pub struct ProducerContract {
 
 #[derive(Debug, Error, PartialEq, Eq)]
 pub enum PrecomputePlanError {
+    #[error("raw counter state {0} cannot preserve independent series resets/timestamps")]
+    UnsupportedRawCounter(u64),
     #[error("PrecomputePlan envelope does not match BackendPlan identity/lifecycle")]
     PlanIdentityMismatch,
     #[error("unsupported precompute ingest protocol/endpoint/identity contract")]
@@ -520,6 +522,20 @@ impl PrecomputePlan {
         }
         let mut materializations = BTreeSet::new();
         for materialization in &self.materializations {
+            if self.ingest.protocol == IngestProtocol::PrometheusRemoteWriteV1
+                && (matches!(
+                    materialization.aggregation_type,
+                    asap_types::AggregationType::Increase
+                ) || (materialization.aggregation_type
+                    == asap_types::AggregationType::SingleSubpopulation
+                    && materialization
+                        .aggregation_sub_type
+                        .eq_ignore_ascii_case("increase")))
+            {
+                return Err(PrecomputePlanError::UnsupportedRawCounter(
+                    materialization.policy_fp_u64(),
+                ));
+            }
             if !materializations.insert(materialization.policy_fingerprint()) {
                 return Err(PrecomputePlanError::DuplicateMaterialization(
                     materialization.policy_fp_u64(),
@@ -1666,6 +1682,7 @@ impl BackendLocalPlanningSnapshot {
             });
         }
         select_workload_roots(&mut queries, canonical_roots, &topk_evidence_by_id)?;
+        preserve_native_raw_counters(&mut queries)?;
         Ok((
             PlanningRequest {
                 queries,
@@ -1677,6 +1694,45 @@ impl BackendLocalPlanningSnapshot {
     }
 }
 
+/// Preserve native execution for counters until raw producers retain series identity.
+fn preserve_native_raw_counters(queries: &mut [PlanningQuery]) -> Result<(), CompileError> {
+    for query in queries {
+        let selected = collect_selected_materializations(&query.post_asap).map_err(|reason| {
+            CompileError::Query {
+                query_id: query.query_id.clone(),
+                reason,
+            }
+        })?;
+        if selected.iter().any(|state| {
+            matches!(
+                state.family,
+                SummaryFamilyType::ExactAggregate(
+                    planner_types::post_asap::ExactKind::Increase
+                        | planner_types::post_asap::ExactKind::Rate,
+                    _
+                )
+            )
+        }) {
+            let parsed = crate::query_parser::parse_query_expr_canonical(
+                &query.query_string,
+                query.accuracy.clone(),
+            )
+            .map_err(|error| CompileError::Query {
+                query_id: query.query_id.clone(),
+                reason: error.to_string(),
+            })?;
+            query.post_asap =
+                crate::planner_selection::keep_pre_asap(&parsed).map_err(|error| {
+                    CompileError::Query {
+                        query_id: query.query_id.clone(),
+                        reason: error.to_string(),
+                    }
+                })?;
+        }
+    }
+    Ok(())
+}
+
 impl PhysicalCompiler {
     pub fn compile(
         &self,
@@ -1698,6 +1754,10 @@ impl PhysicalCompiler {
             });
         }
 
+        if environment.target == PhysicalDeploymentTarget::BackendLocalRemoteWrite {
+            preserve_native_raw_counters(&mut request.queries)?;
+        }
+
         let roots = request
             .queries
             .iter()
@@ -2738,6 +2798,92 @@ fn stable_workload_plan_id(
 mod tests {
     use super::*;
 
+    // A pooled raw counter cannot distinguish same-timestamp series or independent resets.
+    #[test]
+    fn backend_local_rejects_pooled_counter_materialization() {
+        let request = request("counter", "sum(rate(m[1m]))");
+        let mut environment = environment(10_000);
+        environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        environment.collector_ids.clear();
+        let plan = PhysicalCompiler.compile(request, environment).unwrap();
+        assert!(
+            plan.precompute_plan.materializations.is_empty(),
+            "unsafe pooled raw counter was installed"
+        );
+        assert!(plan.query_plan.entries.values().all(|entry| matches!(
+            entry.nodes[&entry.root],
+            crate::query_plan::QueryPlanNode::ExactFallback { .. }
+        )));
+    }
+
+    // Old precompiled raw counter artifacts are rejected before installation too.
+    #[test]
+    fn raw_counter_artifact_is_rejected_while_envelope_counter_remains_valid() {
+        let plan = PhysicalCompiler
+            .compile(request("counter", "rate(m[1m])"), environment(10_000))
+            .unwrap();
+        plan.precompute_plan.validate().unwrap();
+        let mut raw = plan.precompute_plan;
+        raw.ingest.protocol = IngestProtocol::PrometheusRemoteWriteV1;
+        raw.ingest.endpoint_path = "/api/v1/write".into();
+        raw.ingest.timestamp_unit = TimestampUnit::UnixMilliseconds;
+        raw.ingest.require_plan_identity = false;
+        raw.ingest.require_materialization_identity = false;
+        raw.ingest.require_registered_producer = false;
+        raw.producers.clear();
+        assert!(matches!(
+            raw.validate(),
+            Err(PrecomputePlanError::UnsupportedRawCounter(_))
+        ));
+    }
+
+    // Counter fallback does not disable an independent safe summary in the same workload.
+    #[test]
+    fn counter_fallback_preserves_other_workload_summaries() {
+        let mut workload = request("counter", "sum(rate(m[1m]))");
+        workload
+            .queries
+            .extend(request("gauge", "sum_over_time(g[1m])").queries);
+        let mut environment = environment(10_000);
+        environment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
+        environment.collector_ids.clear();
+        let plan = PhysicalCompiler.compile(workload, environment).unwrap();
+        assert_eq!(plan.precompute_plan.materializations.len(), 1);
+        assert!(plan
+            .query_plan
+            .lookup("sum(rate(m[1m]))")
+            .unwrap()
+            .materialization_bindings()
+            .is_empty());
+        assert_eq!(
+            plan.query_plan
+                .lookup("sum_over_time(g[1m])")
+                .unwrap()
+                .materialization_bindings()
+                .len(),
+            1
+        );
+    }
+
+    // Capability normalization precedes candidate enumeration, avoiding duplicate exact quotes.
+    #[test]
+    fn counter_only_snapshot_has_one_exact_cost_alternative() {
+        let mut snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+        let entry = &mut snapshot.query_workload.repeating_queries.as_mut().unwrap()[0];
+        entry.query = Query("rate(m[1m])".into());
+        entry.requirements.accuracy = AccuracyRequirement::Explicit(AccuracyTarget::Exact);
+        let (request, _) = snapshot.planning_request().unwrap();
+        assert_eq!(
+            super::super::workload_cost::with_exact_alternative(request)
+                .unwrap()
+                .len(),
+            1
+        );
+    }
+
     // Count and value rankings must configure different state update contracts.
     #[test]
     fn temporal_topk_binds_planner_update_weight() {
@@ -3407,9 +3553,20 @@ mod tests {
             let mut deployment = environment(10_000);
             deployment.target = PhysicalDeploymentTarget::BackendLocalRemoteWrite;
             deployment.collector_ids.clear();
-            let plan = PhysicalCompiler
-                .compile(request(query_id, promql), deployment)
-                .unwrap_or_else(|error| panic!("{promql} must compile: {error}"));
+            let compiled = PhysicalCompiler.compile(request(query_id, promql), deployment);
+            if matches!(
+                expected_readout,
+                crate::query_plan::ExactReadout::Rate | crate::query_plan::ExactReadout::Increase
+            ) {
+                let plan = compiled.unwrap();
+                assert!(plan.precompute_plan.materializations.is_empty());
+                assert!(plan.query_plan.entries.values().all(|entry| matches!(
+                    entry.nodes[&entry.root],
+                    crate::query_plan::QueryPlanNode::ExactFallback { .. }
+                )));
+                continue;
+            }
+            let plan = compiled.unwrap_or_else(|error| panic!("{promql} must compile: {error}"));
             assert_eq!(plan.backend_plan.materializations.len(), 1, "{promql}");
             assert_eq!(plan.query_plan.entries.len(), 1, "{promql}");
             assert!(plan.collector_plans.is_empty(), "{promql}");
@@ -3462,7 +3619,7 @@ mod tests {
         assert!(plan.collector_plans.is_empty());
         assert!(plan.transmission_plan.rules.is_empty());
         assert_eq!(plan.query_plan.entries.len(), 6);
-        assert_eq!(plan.precompute_plan.materializations.len(), 5);
+        assert_eq!(plan.precompute_plan.materializations.len(), 4);
         for query in [
             "rate(asap_demo_counter_total[5s])",
             "increase(asap_demo_counter_total[5s])",
diff --git a/data_plane/src/precompute_engine/worker.rs b/data_plane/src/precompute_engine/worker.rs
index 34ce3c56..4cff1b52 100644
--- a/data_plane/src/precompute_engine/worker.rs
+++ b/data_plane/src/precompute_engine/worker.rs
@@ -3259,6 +3259,54 @@ aggregations:
     // sketch (`sketch_panes`) paths.
     // -----------------------------------------------------------------------
 
+    // This single-series updater cannot implement a grouped sum of counter increases.
+    // The physical compiler rejects raw counter producers until series state is preserved.
+    #[test]
+    fn pooled_counter_samples_lose_independent_same_timestamp_reset() {
+        use crate::precompute_engine::operators::IncreaseAccumulator;
+        let config = make_agg_config(
+            1,
+            "requests_total",
+            AggregationType::SingleSubpopulation,
+            "Increase",
+            10,
+            0,
+            vec![],
+        );
+        let sink = Arc::new(CapturingOutputSink::new());
+        let mut worker = make_worker(
+            HashMap::from([(1, config)]),
+            sink.clone(),
+            false,
+            0,
+            LateDataPolicy::Drop,
+        );
+        worker
+            .process_group_samples(
+                1,
+                PolicyFingerprint(1),
+                "",
+                vec![
+                    ("requests_total{instance=\"a\"}".into(), 1000, 100.0),
+                    ("requests_total{instance=\"b\"}".into(), 1000, 50.0),
+                    ("requests_total{instance=\"a\"}".into(), 2000, 110.0),
+                    ("requests_total{instance=\"b\"}".into(), 2000, 5.0),
+                ],
+            )
+            .unwrap();
+        worker.force_close_all().unwrap();
+        let captured = sink.drain();
+        let accumulator = captured[0]
+            .1
+            .as_any()
+            .downcast_ref::<IncreaseAccumulator>()
+            .unwrap();
+        assert_eq!(accumulator.total_increase, 10.0);
+        let independent_increases = (110.0 - 100.0) + 5.0;
+        assert_eq!(independent_increases, 15.0);
+        assert_ne!(accumulator.total_increase, independent_increases);
+    }
+
     #[test]
     fn shutdown_force_close_emits_trailing_sample_window() {
         // 10s tumbling window; make_worker uses grace=0, isolating the
diff --git a/data_plane/tests/asapquery_compatibility_process_e2e.rs b/data_plane/tests/asapquery_compatibility_process_e2e.rs
index 57791460..398dcf52 100644
--- a/data_plane/tests/asapquery_compatibility_process_e2e.rs
+++ b/data_plane/tests/asapquery_compatibility_process_e2e.rs
@@ -1,7 +1,7 @@
 //! Black-box acceptance test for the collector-free ASAPQuery profile.
 //!
 //! Starts the production binary from a canonical workload snapshot, ingests
-//! only Prometheus Remote Write v1, exercises every declared warm query family
+//! only Prometheus Remote Write v1, exercises safe warm families and counter fallback
 //! through instant and range APIs, and verifies exact fallback request parity.
 
 use std::collections::HashMap;
@@ -490,6 +490,16 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
                     (base + 9_400, 12.0),
                 ],
             ),
+            series_with_labels(
+                "asap_demo_counter_total",
+                &[("instance", "independent")],
+                &[
+                    (base + 500, 50.0),
+                    (base + 1_700, 60.0),
+                    (base + 2_900, 5.0),
+                    (base + 4_200, 15.0),
+                ],
+            ),
             series_with_labels(
                 "asap_demo_gauge",
                 &[("job", "api")],
@@ -564,22 +574,43 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let first_eval = (base + 5_000) as f64 / 1_000.0;
     let second_eval = (base + 10_000) as f64 / 1_000.0;
     let backend_log = output_dir.path().join("query_engine.log");
-    let rate = wait_for_warm_instant(
-        &client,
-        &backend,
+    // Counter roots retain the complete exact request until independent series
+    // reset/timestamp state is represented by the backend raw producer.
+    for query in [
         "rate(asap_demo_counter_total[5s])",
-        first_eval,
-        &backend_log,
-    )
-    .await;
-    let increase = wait_for_warm_instant(
-        &client,
-        &backend,
         "increase(asap_demo_counter_total[5s])",
-        first_eval,
-        &backend_log,
-    )
-    .await;
+    ] {
+        let instant: Value = client
+            .get(format!("{backend}/api/v1/query"))
+            .query(&[
+                ("query", query.to_string()),
+                ("time", first_eval.to_string()),
+            ])
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        assert_eq!(instant["data"]["result"][0]["metric"]["fallback"], "true");
+        assert!(!is_warm(&instant));
+        let range: Value = client
+            .get(format!("{backend}/api/v1/query_range"))
+            .query(&[
+                ("query", query.to_string()),
+                ("start", first_eval.to_string()),
+                ("end", second_eval.to_string()),
+                ("step", "5".into()),
+            ])
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        assert_eq!(range["data"]["result"][0]["metric"]["fallback"], "true");
+        assert!(!is_warm(&range));
+    }
     let sum = wait_for_warm_instant(
         &client,
         &backend,
@@ -635,9 +666,6 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         topk_count["data"]["result"][0]["metric"]["item"],
         "asap_demo_gauge{job=\"api\"}"
     );
-    let rate_value = first_value(&rate, "value").expect("rate value");
-    let increase_value = first_value(&increase, "value").expect("increase value");
-    assert!((rate_value * 5.0 - increase_value).abs() < 1e-9);
     assert!((first_value(&sum, "value").expect("sum value") - 240.0).abs() < 1e-9);
     let quantile_value = first_value(&quantile, "value").expect("quantile value");
     assert!(
@@ -646,8 +674,6 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     );
 
     for query in [
-        "rate(asap_demo_counter_total[5s])",
-        "increase(asap_demo_counter_total[5s])",
         "sum_over_time(asap_demo_gauge[5s])",
         "quantile_over_time(0.5, asap_demo_latency_ms[5s])",
         "topk(1, sum_over_time(asap_demo_gauge[5s]))",
@@ -870,11 +896,84 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
     let materializations = status["materializations"]
         .as_array()
         .expect("materialization statuses");
-    assert_eq!(materializations.len(), 5);
+    assert_eq!(materializations.len(), 4);
     assert!(materializations
         .iter()
         .all(|entry| entry["phase"] == "serving"));
 
+    // A previously generated raw counter plan must be rejected before staging,
+    // while the current safe generation remains available to readers.
+    let mut snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
+        serde_json::from_str(include_str!(
+            "../../docs/examples/asapquery-planning-snapshot.json"
+        ))
+        .unwrap();
+    snapshot.query_workload.repeating_queries.as_mut().unwrap()[0].query =
+        planner_types::workload::Query("rate(asap_demo_counter_total[1m])".into());
+    let (mut legacy_request, mut environment) = snapshot.planning_request().unwrap();
+    let query = &mut legacy_request.queries[0];
+    let parsed = control_plane::query_parser::parse_query_expr_canonical(
+        &query.query_string,
+        query.accuracy.clone(),
+    )
+    .unwrap();
+    query.post_asap = control_plane::physical::compiler::select_post_asap(
+        &parsed,
+        query.accuracy.clone(),
+        &query.lifecycle,
+        None,
+    )
+    .unwrap();
+    environment.target =
+        control_plane::physical::compiler::PhysicalDeploymentTarget::DistributedCollectors;
+    environment.collector_ids = vec!["legacy-counter-source".into()];
+    environment.plan_version = 2;
+    let mut legacy = control_plane::physical::compiler::PhysicalCompiler
+        .compile(legacy_request, environment)
+        .unwrap();
+    legacy.precompute_plan.ingest.protocol =
+        control_plane::physical::compiler::IngestProtocol::PrometheusRemoteWriteV1;
+    legacy.precompute_plan.ingest.endpoint_path = "/api/v1/write".into();
+    legacy.precompute_plan.ingest.timestamp_unit =
+        control_plane::physical::compiler::TimestampUnit::UnixMilliseconds;
+    legacy.precompute_plan.ingest.require_plan_identity = false;
+    legacy
+        .precompute_plan
+        .ingest
+        .require_materialization_identity = false;
+    legacy.precompute_plan.ingest.require_registered_producer = false;
+    legacy.precompute_plan.producers.clear();
+    legacy.transmission_plan.rules.clear();
+    let artifact = serde_json::json!({"precompute_plan": legacy.precompute_plan,
+        "transmission_plan": legacy.transmission_plan, "backend_plan": legacy.backend_plan.encode_to_vec(),
+        "query_plan": legacy.query_plan, "storage_routing": null, "adaptation_evidence": []});
+    let built = data_plane::drivers::query::servers::http::build_active_physical_plan(
+        serde_json::from_value(artifact.clone()).unwrap(),
+        std::sync::Arc::new(data_plane::storage_engines::types::BackendStorageRouting::empty()),
+    );
+    assert!(matches!(built, Err(error) if error.contains("raw counter state")));
+    // HTTP may reject this old generation at the earlier successor authorization
+    // boundary; either way it must leave the installed generation unchanged.
+    let rejected = client
+        .post(format!("{backend}/api/v1/physical-plan"))
+        .json(&artifact)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(rejected.status().as_u16(), 422);
+    let after: Value = client
+        .get(format!("{backend}/api/v1/physical-plan/status"))
+        .send()
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(
+        after["plans"], status["plans"],
+        "rejected artifact changed active generation"
+    );
+
     let metrics = client
         .get(format!("{backend}/metrics"))
         .send()
@@ -884,7 +983,7 @@ async fn collector_free_profile_serves_complete_matrix_and_falls_back_exactly()
         .await
         .expect("metrics body");
     assert!(metrics.contains("asap_remote_write_requests_total 4"));
-    assert!(metrics.contains("asap_remote_write_samples_total 32"));
-    assert!(metrics.contains("asap_remote_write_duplicates_total 29"));
+    assert!(metrics.contains("asap_remote_write_samples_total 36"));
+    assert!(metrics.contains("asap_remote_write_duplicates_total 33"));
     assert!(metrics.contains("asap_remote_write_rejected_requests_total 1"));
 }
diff --git a/docs/user_guide/asapquery-profile.md b/docs/user_guide/asapquery-profile.md
index e61ca2e6..9b6af24e 100644
--- a/docs/user_guide/asapquery-profile.md
+++ b/docs/user_guide/asapquery-profile.md
@@ -102,6 +102,13 @@ Query serving uses only the installed `QueryPlan` DAG. A query absent from that
 DAG is a capability miss and goes to the exact Prometheus fallback; the backend
 does not search materialization candidates while serving.
 
+Counter `rate` and `increase` queries currently use the exact Prometheus
+fallback in this raw-ingest profile. The backend does not install pooled counter
+state because independent series can reset or arrive at the same timestamp.
+Other supported queries in the workload can still use warm summaries. Previously
+generated artifacts containing raw counter state are rejected; regenerate them
+from their workload snapshots.
+
 Activation and warm readiness are deliberately separate. A newly activated
 generation exposes each materialization as `materializing`; the QueryPlan path
 promotes it through `ready` to `serving` only after its closed-window coverage