chore(config): preserve provenance in typed configuration (#2279)

## Human Summary

This PR prepares us to properly close #1965. However we need to migrate the affected components to typed configuration to pick up the fix. #1965 will close when affected components migrate. It is a bit much to do in a single PR so I want to land the mechanism first.

In this PR we preserve a binary interpretation of the `source` field in the remote Agent config stream. When that stream declares the `source` as `default` or `schema`, we record `Provenance::Default`, otherwise `Provenance::Explicit`. For algorithms such as overriding `site` with `dd_url`, this information is required because the Agent stream populates the fields with their default values and we need to know which one should take precedence.

**Important**: without such guidance in my prompt, the coding agent decided that Agent defaults should **not** override explicit local settings (i.e. in the config file or environment variables that ADP reads during startup). This is a behavioral change and, to be honest, I am not sure what the right behavior is. This seems fine? But also, I could see us making the argument that the config stream is definitive even to the extent of overriding local values. In practice this should have little or no effect since we do not expect operators to configure ADP directly beyond the minimum configuration required to connect to the Agent.

**Reviewer's Guide**
- `lib/saluki-config/src/dynamic/event.rs` introduces types to carry provenance in the Agent updates.
- `lib/agent-data-plane-config/src/provenance.rs` introduces types to carry provenance in typed config.
- `bin/agent-data-plane/src/internal/remote_agent.rs` is where we preserve the information in the config stream.
- `lib/agent-data-plane-config-system/src/translators/datadog_translator.rs` is where we preserve the information for those fields where we need it in typed config.

Most of the code churn is in tests creating config stream events for themselves, which now have to add the provenance metadata.

## AI Summary

The Core Agent configuration stream publishes a complete reified configuration, including values materialized from schema defaults, and source metadata for each key. Once that configuration is deserialized into a value-only model, an explicitly supplied value equal to the schema default is indistinguishable from a defaulted value. That distinction is required for source-aware translation and is the key to resolving #1965.

This PR carries the two-state provenance (`Default`/`Explicit`) from the stream through the typed configuration system and exposes it to `SalukiConfiguration` consumers with `ConfigValue<T>`. Translators and consumers can now determine whether a setting was supplied without hardcoding or comparing schema-default values. Provenance is preserved across local bootstrap values, snapshots, partial updates, resets, and live changes while retaining the existing serialized configuration shape.

As a side effect, default-sourced Agent values no longer shadow explicit local file or environment values in the typed path; explicit Agent values still override them. This is useful for local and test configuration, but is secondary to making source-aware configuration decisions possible. The legacy `GenericConfiguration` compatibility view remains value-only, and affected components still need to migrate before #1965 is fully resolved.

### Follow-up

The components that need typed-config plumbing for the #1965 endpoint fix are below. This PR also carries provenance for the forwarder retry-queue settings; the retry-related rows are needed to consume that additional fix, but are not required for the `site`/`dd_url` endpoint fix itself.

```
 ┌────────────────────────────────────────┬─────────────────────────────────────────────────────────────────────────────────────────────────┐
 │ Component                              │ Why it must migrate                                                                             │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ DatadogMetricsConfiguration /          │ It currently deserializes `site` and `dd_url` from the raw map and decides which endpoint wins  │
 │ dd_metrics_encode                      │ in `configured_primary_endpoint` and `any_series_endpoint_matches`. It must use                 │
 │                                        │ `Endpoints.dd_url.is_explicit()` and the typed site value rather than treating the default URL  │
 │                                        │ as absent. This also covers the MRF and autoscaling-failover metrics encoder instances.         │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ DatadogForwarderConfiguration /        │ Its `EndpointConfiguration` currently independently deserializes `site` and `dd_url`. If only.  │
 │ dd_out (and `mrf_dd_out`)              │ the metrics encoder migrates, the encoder and forwarder can disagree about the primary          │
 │                                        │ endpoint.                                                                                       │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ ForwarderConfiguration                 │ This is the common adapter used by both forwarders. It needs to overlay the typed endpoint and  │
 │                                        │ retry values onto the remaining raw/deserialized forwarder settings.                            │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ EndpointConfiguration                  │ This is where primary endpoint resolution currently happens. It needs to represent a defaulted  │
 │                                        │ `dd_url` as “not an override” and an explicit `dd_url`—even `https://app.datadoghq.com`—as an   │
 │                                        │ override.                                                                                       │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ RetryConfiguration                     │ `queue_max_size_bytes()` currently uses `Option` presence. It needs to prefer                   │
 │                                        │ `retry_queue_payloads_max_size` only when that value is explicit, then fall back to             │
 │                                        │ `retry_queue_max_size` only when that value is explicit.                                        │
 ├────────────────────────────────────────┼─────────────────────────────────────────────────────────────────────────────────────────────────┤
 │ ClusterAgentForwarderConfiguration /   │ Its destination URL is explicitly supplied programmatically, so it does not need global         │
 │ cluster_agent_out                      │ `site`/`dd_url`. However, it uses the shared `ForwarderConfiguration` and calls                 │
 │                                        │ `queue_max_size_bytes()`, so it must receive the provenance-aware retry settings.               │
 └────────────────────────────────────────┴─────────────────────────────────────────────────────────────────────────────────────────────────┘
```

These are shared configuration adapters as well as components. The topology constructors in `bin/agent-data-plane/src/cli/run.rs` must pass the typed shared endpoint configuration through for the primary, MRF, and autoscaling-failover paths. The MRF and Cluster Agent endpoint overrides remain explicit programmatic overrides.

## Change Type
- [x] Non-functional (chore, refactoring, docs)

## How did you test this PR?

- `cargo check --workspace --tests`
- `cargo test -p agent-data-plane-config -p agent-data-plane-config-system -p saluki-config -p agent-data-plane`
- `git diff --check`

#### Human Note

Not tested at the integration level yet because we need to migrate to typed config. But I recommend that we test at least one example of this behavior in the integration suite when we migrate to this mechanism.

## References

- Progresses #1965
- Related to #2169

Co-authored-by: matt.briggs <matt.briggs@datadoghq.com>

Author: webern

.claude/skills/config-system/SKILL.md

@@ -56,6 +56,10 @@ into `SalukiConfiguration`.
 | Figment provider wrapping that reader                           | `lib/datadog-agent/config/src/env_provider.rs`                             |
 | Same provider, plus Saluki-only keys                            | `lib/agent-data-plane-config-system/src/env_provider.rs`                   |
 
+In `SalukiConfiguration`, you may use `ConfigValue<T>` when we need to know the difference between a
+value that was explicitly set by the user, or where the value is a default. The Agent API provides a
+`source` field which we are simplifying into `Provinence`, which is either `Default` or `Explicit`.
+
 Paths and type names can move. Notify the user when this skill needs an update.
 
 ### Architectural dependency boundaries
@@ -109,10 +113,11 @@ generated tables instead:
 - the by-key path via `EnvironmentProvider`, a Figment provider wrapping those same readers.
 
 **Every source therefore delivers the Agent's canonical shape.** A struct deserialized from
-`GenericConfiguration` **MUST** read that shape. Do not add a `#[serde(rename)]` or `#[serde(alias)]`
-that maps a nested key onto a flattened spelling, and do not reintroduce a key-alias or
-environment-remapping table. Those existed once, only ran on file load, and so never applied to the
-Datadog Agent's configuration stream — which is the authority in the Core Agent deployment.
+`GenericConfiguration` **MUST** read that shape. Do not add a `#[serde(rename)]` or
+`#[serde(alias)]` that maps a nested key onto a flattened spelling, and do not reintroduce a
+key-alias or environment-remapping table. Those existed once, only ran on file load, and so never
+applied to the Datadog Agent's configuration stream — which is the authority in the Core Agent
+deployment.
 
 Reserve `#[serde(flatten)]` for a struct that genuinely groups several *top-level* Agent keys (for
 example, the forwarder's `forwarder_*` retry settings). Name a Rust field after its canonical
@@ -132,16 +137,17 @@ Keep the authoritative Datadog path and delete the duplicate Saluki-only path.
 
 Exactly one layer owns each default:
 
-| Source class | Model type  | Default owner                                  | Translation behavior              |
-|--------------|-------------|------------------------------------------------|-----------------------------------|
-| Saluki-only  | `Option<T>` | No default                                     | `seed` preserves `None`           |
-| Saluki-only  | `T`         | `agent-data-plane-config/src/defaults.rs`      | `seed` assigns the resolved value |
-| Witnessed    | `T`         | Generated Datadog schema default               | `drive` always writes it          |
-| Witnessed    | `Option<T>` | No default                                     | `drive` preserves `None`          |
+| Source class | Model type  | Default owner                             | Translation behavior              |
+|--------------|-------------|-------------------------------------------|-----------------------------------|
+| Saluki-only  | `Option<T>` | No default                                | `seed` preserves `None`           |
+| Saluki-only  | `T`         | `agent-data-plane-config/src/defaults.rs` | `seed` assigns the resolved value |
+| Witnessed    | `T`         | Generated Datadog schema default          | `drive` always writes it          |
+| Witnessed    | `Option<T>` | No default                                | `drive` preserves `None`          |
 
 Define each Saluki-only default once in `lib/agent-data-plane-config/src/defaults.rs`; source and
 model defaults must reference that definition rather than restating its value. If the component
-requires a value, model `T`; use `Option<T>` only when absence is meaningful, not to defer a default.
+requires a value, model `T`; use `Option<T>` only when absence is meaningful, not to defer a
+default.
 
 Push source parsing, defaults, and input validation to the configuration boundary. Components keep
 only validation that is truly business logic.

bin/agent-data-plane/src/components/dogstatsd_post_aggregate_filter/mod.rs

@@ -325,7 +325,10 @@ mod tests {
     use std::time::Duration;
 
     use metrics::set_default_local_recorder;
-    use saluki_config::{dynamic::ConfigUpdate, ConfigurationLoader};
+    use saluki_config::{
+        dynamic::{ConfigSetting, ConfigUpdate},
+        ConfigurationLoader,
+    };
     use saluki_context::Context;
     use saluki_core::{
         data_model::event::{metric::Metric, Event},
@@ -560,10 +563,7 @@ mod tests {
     async fn runtime_updates_rebuild_the_effective_matcher() {
         let (config, sender) = ConfigurationLoader::for_tests(Some(serde_json::json!({})), None, true).await;
         let sender = sender.expect("sender should exist");
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
         config.ready().await;
 
         let mut filter = noop_filter(vec!["request.duration.max"], false, vec![], false);
@@ -574,10 +574,10 @@ mod tests {
 
         let mut filterlist_watcher = config.watch_for_updates("metric_filterlist");
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_filterlist".to_string(),
-                value: serde_json::json!(["request.duration.avg"]),
-            })
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_filterlist",
+                serde_json::json!(["request.duration.avg"]),
+            )))
             .await
             .unwrap();
 

bin/agent-data-plane/src/components/dogstatsd_prefix_filter/mod.rs

@@ -344,7 +344,10 @@ impl Transform for DogStatsDPrefixFilter {
 #[cfg(test)]
 mod tests {
     use metrics::set_default_local_recorder;
-    use saluki_config::{dynamic::ConfigUpdate, ConfigurationLoader};
+    use saluki_config::{
+        dynamic::{ConfigSetting, ConfigUpdate},
+        ConfigurationLoader,
+    };
     use saluki_metrics::{test::TestRecorder, MetricsBuilder};
 
     use super::*;
@@ -499,10 +502,7 @@ mod tests {
     async fn blocklist_and_filterlist_dynamic_updates_are_applied() {
         let (cfg, sender) = ConfigurationLoader::for_tests(Some(serde_json::json!({})), None, true).await;
         let sender = sender.expect("sender should exist");
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
 
         cfg.ready().await;
 
@@ -527,10 +527,10 @@ mod tests {
         let mut blocklist_watcher = cfg.watch_for_updates("statsd_metric_blocklist");
 
         sender
-            .send(ConfigUpdate::Partial {
-                key: "statsd_metric_blocklist".to_string(),
-                value: serde_json::json!(["foo".to_string()]),
-            })
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "statsd_metric_blocklist",
+                serde_json::json!(["foo".to_string()]),
+            )))
             .await
             .unwrap();
 
@@ -557,10 +557,10 @@ mod tests {
 
         let mut metric_filterlist_watcher = cfg.watch_for_updates("metric_filterlist");
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_filterlist".to_string(),
-                value: serde_json::json!(["baz".to_string()]),
-            })
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_filterlist",
+                serde_json::json!(["baz".to_string()]),
+            )))
             .await
             .unwrap();
 
@@ -594,10 +594,7 @@ mod tests {
         .await;
         let sender = sender.expect("sender should exist");
 
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
         cfg.ready().await;
 
         let mut filter = FilterBuilder::new()
@@ -616,10 +613,10 @@ mod tests {
 
         let mut match_prefix_watcher = cfg.watch_for_updates("metric_filterlist_match_prefix");
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_filterlist_match_prefix".to_string(),
-                value: serde_json::json!(true),
-            })
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_filterlist_match_prefix",
+                serde_json::json!(true),
+            )))
             .await
             .unwrap();
 

bin/agent-data-plane/src/components/tag_filterlist/mod.rs

@@ -345,7 +345,10 @@ pub fn filter_metric_tags(
 mod tests {
     use std::sync::Arc;
 
-    use saluki_config::{dynamic::ConfigUpdate, ConfigurationLoader};
+    use saluki_config::{
+        dynamic::{ConfigSetting, ConfigUpdate},
+        ConfigurationLoader,
+    };
     use saluki_context::{
         tags::{Tag, TagSet},
         Context, TagSetMutViewState,
@@ -915,21 +918,18 @@ mod tests {
     async fn dynamic_update_partial_replaces_filter() {
         let (cfg, sender) = ConfigurationLoader::for_tests(Some(serde_json::json!({})), None, true).await;
         let sender = sender.expect("sender should exist");
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
         cfg.ready().await;
 
         let mut watcher = cfg.watch_for_updates("metric_tag_filterlist");
 
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_tag_filterlist".to_string(),
-                value: serde_json::json!([
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_tag_filterlist",
+                serde_json::json!([
                     { "metric_name": "my.dist", "action": "exclude", "tags": ["host"] }
                 ]),
-            })
+            )))
             .await
             .unwrap();
 
@@ -952,21 +952,18 @@ mod tests {
     async fn dynamic_update_to_empty_clears_filter() {
         let (cfg, sender) = ConfigurationLoader::for_tests(Some(serde_json::json!({})), None, true).await;
         let sender = sender.expect("sender should exist");
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
         cfg.ready().await;
 
         let mut watcher = cfg.watch_for_updates("metric_tag_filterlist");
 
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_tag_filterlist".to_string(),
-                value: serde_json::json!([
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_tag_filterlist",
+                serde_json::json!([
                     { "metric_name": "my.dist", "action": "exclude", "tags": ["env"] }
                 ]),
-            })
+            )))
             .await
             .unwrap();
 
@@ -978,10 +975,10 @@ mod tests {
         .expect("timed out waiting for initial metric_tag_filterlist update");
 
         sender
-            .send(ConfigUpdate::Partial {
-                key: "metric_tag_filterlist".to_string(),
-                value: serde_json::json!([]),
-            })
+            .send(ConfigUpdate::Partial(ConfigSetting::explicit(
+                "metric_tag_filterlist",
+                serde_json::json!([]),
+            )))
             .await
             .unwrap();
 
@@ -1004,20 +1001,18 @@ mod tests {
     async fn dynamic_update_snapshot_applies_filter() {
         let (cfg, sender) = ConfigurationLoader::for_tests(Some(serde_json::json!({})), None, true).await;
         let sender = sender.expect("sender should exist");
-        sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({})))
-            .await
-            .unwrap();
+        sender.send(ConfigUpdate::snapshot([])).await.unwrap();
         cfg.ready().await;
 
         let mut watcher = cfg.watch_for_updates("metric_tag_filterlist");
 
         sender
-            .send(ConfigUpdate::Snapshot(serde_json::json!({
-                "metric_tag_filterlist": [
+            .send(ConfigUpdate::snapshot([ConfigSetting::explicit(
+                "metric_tag_filterlist",
+                serde_json::json!([
                     { "metric_name": "my.dist", "action": "include", "tags": ["service"] }
-                ]
-            })))
+                ]),
+            )]))
             .await
             .unwrap();