refactor(types): rename AggregationConfig::aggregation_id() to policy_fp_u64() (closes #272 step 5)

crates/asap_types/src/aggregation_config.rs

@@ -124,11 +124,11 @@ impl AggregationConfig {
     }
 
     /// `PolicyFingerprint::as_u64()` — the u64-form handle used by the
-    /// legacy aggregation-id-keyed call sites. **Always** equal to
-    /// `self.policy_fingerprint().as_u64()`. Provided so callers that
-    /// previously read `config.aggregation_id` can keep the same
-    /// arithmetic without owning the typed handle.
-    pub fn aggregation_id(&self) -> u64 {
+    /// policy-fingerprint-keyed call sites (e.g. `StreamingConfig`'s
+    /// `HashMap<u64, AggregationConfig>` keys). **Always** equal to
+    /// `self.policy_fingerprint().as_u64()`. The value is content-
+    /// addressed identity, NOT a controller-allocated counter id.
+    pub fn policy_fp_u64(&self) -> u64 {
         self.policy_fingerprint().as_u64()
     }
 
@@ -437,16 +437,16 @@ mod tests {
         );
     }
 
-    /// The `aggregation_id()` accessor is exactly the fingerprint u64.
+    /// The `policy_fp_u64()` accessor is exactly the fingerprint u64.
     #[test]
-    fn aggregation_id_accessor_equals_fingerprint_u64() {
+    fn policy_fp_u64_accessor_equals_fingerprint_u64() {
         let cfg = AggregationConfig::from_yaml_data(
             &sample_yaml(false),
             None,
             QueryLanguage::promql,
         )
         .expect("parse");
-        assert_eq!(cfg.aggregation_id(), cfg.policy_fingerprint().as_u64());
+        assert_eq!(cfg.policy_fp_u64(), cfg.policy_fingerprint().as_u64());
     }
 
     /// PR 5: `serialize_to_json` no longer emits `aggregationId`.

crates/asap_types/src/capability_matching.rs

@@ -397,7 +397,7 @@ pub fn spatial_filter_compatible(config_filter: &str, req_filter: &str) -> bool
 ///    avoids the key-aggregation hunt when both shapes serve the
 ///    statistic — which is the common case for `Statistic::Sum`
 ///    matching both `Sum` and `CountMinSketch`.
-/// 3. **Tie-break on `aggregation_id()` (the policy fingerprint).**
+/// 3. **Tie-break on `policy_fp_u64()` (the policy fingerprint).**
 ///    Deterministic across runs and hosts; fixes the
 ///    HashMap-iteration-order flake on `avg_finds_sum_and_count`.
 pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ordering {
@@ -407,7 +407,7 @@ pub fn aggregation_priority(a: &AggregationConfig, b: &AggregationConfig) -> Ord
         .cmp(&a.window_size)
         // `false < true` in Rust's bool Ord → single-pop sorts FIRST.
         .then_with(|| a_multi.cmp(&b_multi))
-        .then_with(|| a.aggregation_id().cmp(&b.aggregation_id()))
+        .then_with(|| a.policy_fp_u64().cmp(&b.policy_fp_u64()))
 }
 
 // ---------------------------------------------------------------------------
@@ -463,7 +463,7 @@ pub fn find_compatible_aggregation(
                     );
                 if !ok {
                     debug!(
-                        agg_id = c.aggregation_id(),
+                        policy_fp = c.policy_fp_u64(),
                         agg_type = %c.aggregation_type,
                         metric = %c.metric,
                         window_size_s = c.window_size,
@@ -489,7 +489,7 @@ pub fn find_compatible_aggregation(
         debug!(
             statistic = ?stat,
             num_candidates = candidates.len(),
-            chosen_agg_id = candidates[0].aggregation_id(),
+            chosen_policy_fp = candidates[0].policy_fp_u64(),
             chosen_agg_type = %candidates[0].aggregation_type,
             chosen_window_size_s = candidates[0].window_size,
             "capability matching: found candidates, chose best",
@@ -538,17 +538,17 @@ pub fn find_compatible_aggregation(
 
     debug!(
         metric = %requirements.metric,
-        value_agg_id = value_agg.aggregation_id(),
+        value_policy_fp = value_agg.policy_fp_u64(),
         value_agg_type = %value_agg.aggregation_type,
-        key_agg_id = key_agg.aggregation_id(),
+        key_policy_fp = key_agg.policy_fp_u64(),
         key_agg_type = %key_agg.aggregation_type,
         "capability matching: resolved",
     );
 
     Some(AggregationIdInfo {
-        aggregation_id_for_value: value_agg.aggregation_id(),
+        aggregation_id_for_value: value_agg.policy_fp_u64(),
         aggregation_type_for_value: value_agg.aggregation_type,
-        aggregation_id_for_key: key_agg.aggregation_id(),
+        aggregation_id_for_key: key_agg.policy_fp_u64(),
         aggregation_type_for_key: key_agg.aggregation_type,
     })
 }
@@ -620,7 +620,7 @@ mod tests {
 
     fn single_config(config: AggregationConfig) -> HashMap<u64, AggregationConfig> {
         let mut m = HashMap::new();
-        m.insert(config.aggregation_id(), config);
+        m.insert(config.policy_fp_u64(), config);
         m
     }
 
@@ -629,7 +629,7 @@ mod tests {
     #[test]
     fn basic_sum_match() {
         let cfg = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "");
-        let expected = cfg.aggregation_id();
+        let expected = cfg.policy_fp_u64();
         let configs = single_config(cfg);
         let result = find_compatible_aggregation(
             &configs,
@@ -642,7 +642,7 @@ mod tests {
     #[test]
     fn quantile_any_value_finds_kll() {
         let cfg = make_config(2, "lat", "DatasketchesKLL", "", 300, "tumbling", &[], "");
-        let expected = cfg.aggregation_id();
+        let expected = cfg.policy_fp_u64();
         let configs = single_config(cfg);
         // quantile value (0.5 or 0.9) is NOT part of QueryRequirements — both should find the same config
         let r1 = find_compatible_aggregation(
@@ -660,7 +660,7 @@ mod tests {
     #[test]
     fn quantile_matches_hydrarkll() {
         let cfg = make_config(3, "lat", "HydraKLL", "", 300, "tumbling", &[], "");
-        let expected = cfg.aggregation_id();
+        let expected = cfg.policy_fp_u64();
         let configs = single_config(cfg);
         let result = find_compatible_aggregation(
             &configs,
@@ -757,10 +757,10 @@ mod tests {
     fn window_priority_largest_wins() {
         let small = make_config(1, "cpu", "Sum", "", 300, "tumbling", &[], "");
         let large = make_config(2, "cpu", "Sum", "", 900, "tumbling", &[], "");
-        let expected = large.aggregation_id();
+        let expected = large.policy_fp_u64();
         let mut configs = HashMap::new();
-        configs.insert(small.aggregation_id(), small);
-        configs.insert(large.aggregation_id(), large);
+        configs.insert(small.policy_fp_u64(), small);
+        configs.insert(large.policy_fp_u64(), large);
         // 900_000 ms is divisible by both 300 s and 900 s — prefer 900 s
         let result = find_compatible_aggregation(
             &configs,
@@ -1025,8 +1025,8 @@ mod tests {
         let sum = make_config(1, "cpu", "Sum", "", 300, "tumbling", &["job"], "");
         let cnt = make_config(2, "cpu", "CountMinSketch", "", 300, "tumbling", &["job"], "");
         let mut configs = HashMap::new();
-        configs.insert(sum.aggregation_id(), sum);
-        configs.insert(cnt.aggregation_id(), cnt);
+        configs.insert(sum.policy_fp_u64(), sum);
+        configs.insert(cnt.policy_fp_u64(), cnt);
         let result = find_compatible_aggregation(
             &configs,
             &req(
@@ -1046,8 +1046,8 @@ mod tests {
         // Count config has different window_size — must be rejected
         let cnt = make_config(2, "cpu", "CountMinSketch", "", 900, "tumbling", &["job"], "");
         let mut configs = HashMap::new();
-        configs.insert(sum.aggregation_id(), sum);
-        configs.insert(cnt.aggregation_id(), cnt);
+        configs.insert(sum.policy_fp_u64(), sum);
+        configs.insert(cnt.policy_fp_u64(), cnt);
         let result = find_compatible_aggregation(
             &configs,
             &req(
@@ -1220,9 +1220,9 @@ mod tests {
             &[],
             "",
         );
-        let expected = cfg.aggregation_id();
+        let expected = cfg.policy_fp_u64();
         let mut configs = HashMap::new();
-        configs.insert(cfg.aggregation_id(), cfg);
+        configs.insert(cfg.policy_fp_u64(), cfg);
         let result = find_compatible_aggregation(
             &configs,
             &req(

crates/asap_types/src/policy_fingerprint.rs

@@ -261,10 +261,10 @@ mod tests {
     /// vacuously true. Kept as a doc-comment anchor; no runtime test
     /// is needed.
     #[test]
-    fn aggregation_id_accessor_equals_fingerprint_u64() {
+    fn policy_fp_u64_accessor_equals_fingerprint_u64() {
         let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, "");
         assert_eq!(
-            a.aggregation_id(),
+            a.policy_fp_u64(),
             PolicyFingerprint::from_config(&a).as_u64(),
         );
     }

crates/asap_types/src/streaming_config.rs

@@ -111,9 +111,9 @@ impl StreamingConfig {
                     QueryLanguage::promql,
                 )?;
                 // PR 5: the map key IS the policy-fingerprint u64.
-                // `AggregationConfig::aggregation_id()` is the canonical
+                // `AggregationConfig::policy_fp_u64()` is the canonical
                 // accessor for this value.
-                aggregation_configs.insert(config.aggregation_id(), config);
+                aggregation_configs.insert(config.policy_fp_u64(), config);
             }
         }
 
@@ -181,7 +181,7 @@ aggregations:\n\
         assert_eq!(cfg.aggregation_configs.len(), 1);
         let (k, v) = cfg.aggregation_configs.iter().next().unwrap();
         assert_ne!(*k, 0, "derived id is not the 0 sentinel");
-        assert_eq!(*k, v.aggregation_id(), "map key equals fingerprint u64");
+        assert_eq!(*k, v.policy_fp_u64(), "map key equals fingerprint u64");
         assert_eq!(v.metric, "cpu_seconds");
     }
 

data_plane/src/drivers/ingest/otel.rs

@@ -598,7 +598,7 @@ fn resolve_bucket_sid_for_agg_config(
     let sid = ingest_state
         .series_resolver
         .resolve(&config.metric, &fp, &agg_kind_canonical);
-    let policy_fp = asap_types::PolicyFingerprint(config.aggregation_id());
+    let policy_fp = asap_types::PolicyFingerprint(config.policy_fp_u64());
     (sid, policy_fp)
 }
 
@@ -2874,7 +2874,7 @@ mod sid_bucketing_tests {
     ///   - each sid equals what `SeriesIdResolver::lookup` records for
     ///     `(metric, "zone=<zv>;", ExactAgg-canonical)` — i.e. the
     ///     bucket identity is folded into sid via the resolver
-    ///   - policy_fp = config.aggregation_id() on every message
+    ///   - policy_fp = config.policy_fp_u64() on every message
     ///   - samples in each bucket are exactly the DPs whose `zone`
     ///     attribute matches that bucket (the GROUP-BY semantic)
     ///
@@ -2898,9 +2898,9 @@ mod sid_bucketing_tests {
 
         let metric = "cpu_seconds";
         let cfg = sum_agg_config(metric, &["zone"]);
-        let policy_fp = asap_types::PolicyFingerprint(cfg.aggregation_id());
+        let policy_fp = asap_types::PolicyFingerprint(cfg.policy_fp_u64());
         let mut configs = HashMap::new();
-        configs.insert(cfg.aggregation_id(), cfg.clone());
+        configs.insert(cfg.policy_fp_u64(), cfg.clone());
         let streaming = StreamingConfig::new(configs);
         let hot_reload = HotReloadStreamingConfig::new(streaming);
 
@@ -2957,7 +2957,7 @@ mod sid_bucketing_tests {
 
         // Both buckets carry the same policy_fp (one source config).
         for (_, pf, _, _) in &groups {
-            assert_eq!(*pf, policy_fp, "policy_fp must equal config.aggregation_id()");
+            assert_eq!(*pf, policy_fp, "policy_fp must equal config.policy_fp_u64()");
         }
 
         // sids must be non-zero (zero is reserved on the wire) and distinct.

data_plane/src/drivers/query/servers/http.rs

@@ -2725,7 +2725,7 @@ aggregations:
             // PR 5: streaming-config is keyed on the policy
             // fingerprint. Build a marker→fingerprint map so the test
             // POSTs the right id on the wire.
-            let fp = cfg.aggregation_id();
+            let fp = cfg.policy_fp_u64();
             marker_to_fp.insert(*marker, fp);
             agg_map.insert(fp, cfg);
         }

data_plane/src/precompute_engine/output_sink.rs

@@ -203,7 +203,7 @@ mod tests {
     fn sum_agg_config(_id: u64, metric: &str, grouping_keys: &[&str]) -> AggregationConfig {
         // `_id` is unused after PR 5 — identity is content-addressed
         // via `PolicyFingerprint::from_config`. Callers obtain the id
-        // via `config.aggregation_id()`.
+        // via `config.policy_fp_u64()`.
         AggregationConfig {
             aggregation_type: AggregationType::Sum,
             aggregation_sub_type: String::new(),
@@ -229,7 +229,7 @@ mod tests {
     #[test]
     fn sketch_index_sink_writes_to_index() {
         let cfg = sum_agg_config(7, "cpu_seconds", &["zone"]);
-        let agg_id = cfg.aggregation_id();
+        let agg_id = cfg.policy_fp_u64();
         let mut configs = HashMap::new();
         configs.insert(agg_id, cfg);
         let streaming = StreamingConfig::new(configs);

data_plane/src/precompute_engine/worker.rs

@@ -1285,7 +1285,7 @@ mod tests {
     ) -> AggregationConfig {
         // `_id` is unused after PR 5 — identity is content-addressed
         // via `PolicyFingerprint::from_config`. Callers below build the
-        // streaming-config map by reading `config.aggregation_id()`
+        // streaming-config map by reading `config.policy_fp_u64()`
         // from the returned value.
         let window_type = if slide_secs == 0 || slide_secs == window_secs {
             WindowType::Tumbling

data_plane/src/query_engines/asap_query_engine/engine.rs

@@ -1722,7 +1722,7 @@ mod hot_reload_phase2_tests {
     fn cfg_with_agg(id: u64, metric: &str) -> (StreamingConfig, u64) {
         let mut map = std::collections::HashMap::new();
         let cfg = dummy_agg(id, metric);
-        let fp = cfg.aggregation_id();
+        let fp = cfg.policy_fp_u64();
         map.insert(fp, cfg);
         (StreamingConfig::new(map), fp)
     }

data_plane/src/storage_engines/sketch_db/backfill/mod.rs

@@ -559,7 +559,7 @@ impl BackfillRegistry {
         windows_total: u64,
         data_retention_ms: Option<u64>,
     ) -> Result<u64, CreateError> {
-        let agg_id = config.aggregation_id();
+        let agg_id = config.policy_fp_u64();
         // Time-disjoint invariant: live ingest writes `[created_at, ∞)`
         // so backfill must stay strictly inside `[0, created_at)` or
         // touch the boundary exactly.

data_plane/src/storage_engines/sketch_db/backfill/processor.rs

@@ -439,14 +439,14 @@ mod tests {
 
     fn streaming_config_with(config: AggregationConfig) -> Arc<StreamingConfig> {
         let mut map = std::collections::HashMap::new();
-        map.insert(config.aggregation_id(), config);
+        map.insert(config.policy_fp_u64(), config);
         Arc::new(StreamingConfig::new(map))
     }
 
     #[tokio::test]
     async fn happy_path_writes_one_output_per_group() {
         let cfg = sum_config(1, "latency", vec!["svc"]);
-        let fp = cfg.aggregation_id();
+        let fp = cfg.policy_fp_u64();
         let streaming = streaming_config_with(cfg.clone());
         let hot = HotReloadStreamingConfig::from_arc(streaming.clone());
         let registry = Arc::new(BackfillRegistry::new());
@@ -514,7 +514,7 @@ mod tests {
     #[tokio::test]
     async fn empty_samples_complete_without_write() {
         let cfg = sum_config(1, "m", vec![]);
-        let fp = cfg.aggregation_id();
+        let fp = cfg.policy_fp_u64();
         let streaming = streaming_config_with(cfg);
         let hot = HotReloadStreamingConfig::from_arc(streaming.clone());
         let registry = Arc::new(BackfillRegistry::new());
@@ -536,7 +536,7 @@ mod tests {
         // Exercise the full chain: BackfillWorker drives the
         // processor over a job that covers 4 windows.
         let cfg = sum_config(1, "latency", vec!["svc"]);
-        let fp = cfg.aggregation_id();
+        let fp = cfg.policy_fp_u64();
         let streaming = streaming_config_with(cfg);
         let hot = HotReloadStreamingConfig::from_arc(streaming.clone());
         let registry = Arc::new(BackfillRegistry::new());
@@ -692,7 +692,7 @@ mod tests {
     fn create_checked_rejects_end_past_created_at() {
         use super::super::CreateError;
         let cfg = sum_config(1, "m", vec![]);
-        let expected_fp = cfg.aggregation_id();
+        let expected_fp = cfg.policy_fp_u64();
         let created = now_ms();
         let registry = BackfillRegistry::new();
 
@@ -742,7 +742,7 @@ mod tests {
     fn create_checked_rejects_start_older_than_data_retention() {
         use super::super::CreateError;
         let cfg = sum_config(1, "m", vec![]);
-        let expected_fp = cfg.aggregation_id();
+        let expected_fp = cfg.policy_fp_u64();
         let created = now_ms();
         let registry = BackfillRegistry::new();
         // Created-at is now_ms(), so data retention of 1 hour with
@@ -846,7 +846,7 @@ mod tests {
         use crate::storage_engines::sketch_db::index::{SidLookup, SketchStore};
 
         let cfg = sum_config(1, "latency", vec!["svc"]);
-        let fp = cfg.aggregation_id();
+        let fp = cfg.policy_fp_u64();
         let streaming = streaming_config_with(cfg.clone());
         let hot = HotReloadStreamingConfig::from_arc(streaming.clone());
         let registry = Arc::new(BackfillRegistry::new());