diff --git a/changelog.d/parse_first_interpolation.breaking.md b/changelog.d/parse_first_interpolation.breaking.md
new file mode 100644
index 0000000000000..673b3960711fd
--- /dev/null
+++ b/changelog.d/parse_first_interpolation.breaking.md
@@ -0,0 +1,50 @@
+# Config is now parsed before environment variable and secret interpolation
+
+## Summary
+
+Vector now parses configuration files into a typed value tree before performing
+environment variable (`${VAR}`) and `SECRET[...]` substitution. Interpolation operates
+only on string-typed leaves in the parsed tree. An unquoted placeholder in a non-string
+position of a TOML or JSON config (e.g. `count = ${MY_COUNT}`) is no longer valid and
+will cause the config to fail to load.
+
+YAML configurations are unaffected: YAML parses `${VAR}` as a string scalar, and the
+new schema-coercion pass converts that string to the declared type at load time.
+
+## Migration
+
+In TOML or JSON, wrap any placeholder that appears in a non-string field in quotes so
+the parser sees a string scalar. Vector will coerce the value to the declared type
+(integer, boolean, float) automatically.
+
+Before (TOML):
+
+```toml
+[sources.in]
+type = "demo_logs"
+count = ${MY_COUNT}
+```
+
+After (TOML):
+
+```toml
+[sources.in]
+type = "demo_logs"
+count = "${MY_COUNT}"
+```
+
+Before (JSON):
+
+```json
+{ "sources": { "in": { "type": "demo_logs", "count": ${MY_COUNT} } } }
+```
+
+After (JSON):
+
+```json
+{ "sources": { "in": { "type": "demo_logs", "count": "${MY_COUNT}" } } }
+```
+
+The same applies to `SECRET[backend.key]` references in non-string fields.
+
+authors: pront
diff --git a/changelog.d/parse_first_interpolation.enhancement.md b/changelog.d/parse_first_interpolation.enhancement.md
new file mode 100644
index 0000000000000..9792f71b4d50f
--- /dev/null
+++ b/changelog.d/parse_first_interpolation.enhancement.md
@@ -0,0 +1,8 @@
+Vector now parses configuration files before performing environment variable and `SECRET[...]` substitution.
+Interpolation runs only on string-valued leaves in the parsed value tree, and a JSON-Schema-driven coercion
+pass converts string scalars to the types declared by each component. This produces clearer error messages
+with full field paths (e.g. `sources.my_source.count`) and makes the `--disable-env-var-interpolation` flag
+behave consistently across the `vector`, `vector validate`, and `vector config` commands as well as the
+HTTP provider.
+
+authors: pront
diff --git a/src/config/loading/config_builder.rs b/src/config/loading/config_builder.rs
index b4e4de09ec1bc..ec577adb3f1b4 100644
--- a/src/config/loading/config_builder.rs
+++ b/src/config/loading/config_builder.rs
@@ -3,8 +3,8 @@ use std::{collections::HashMap, io::Read};
 use indexmap::IndexMap;
 
 use super::{
-    ComponentHint, Process, deserialize_config_map, loader, prepare_input,
-    representation::ConfigMap, secret,
+    ComponentHint, Process, deserialize_config_map, deserialize_config_map_wrapped,
+    interpolate_config_map_with_secrets, loader, representation::ConfigMap,
 };
 use crate::config::{
     ComponentKey, ConfigBuilder, EnrichmentTableOuter, SinkOuter, SourceOuter, TestDefinition,
@@ -66,14 +66,16 @@ impl Default for ConfigBuilderLoader {
 }
 
 impl Process for ConfigBuilderLoader {
-    /// Prepares input for a `ConfigBuilder` by interpolating environment variables.
-    fn prepare<R: Read>(&mut self, input: R) -> Result<String, Vec<String>> {
-        let prepared_input = prepare_input(input, self.interpolate_env)?;
-        Ok(if self.secrets.is_empty() {
-            prepared_input
+    fn should_interpolate_env(&self) -> bool {
+        self.interpolate_env
+    }
+
+    fn postprocess(&mut self, map: ConfigMap) -> Result<ConfigMap, Vec<String>> {
+        if self.secrets.is_empty() {
+            Ok(map)
         } else {
-            secret::interpolate(&prepared_input, &self.secrets)?
-        })
+            interpolate_config_map_with_secrets(&map, &self.secrets)
+        }
     }
 
     /// Merge a configuration map with a `ConfigBuilder`. Component types extend specific keys.
@@ -82,24 +84,28 @@ impl Process for ConfigBuilderLoader {
             Some(ComponentHint::Source) => {
                 self.builder
                     .sources
-                    .extend(deserialize_config_map::<IndexMap<ComponentKey, SourceOuter>>(map)?);
+                    .extend(deserialize_config_map_wrapped::<
+                        IndexMap<ComponentKey, SourceOuter>,
+                    >(map, "sources")?);
             }
             Some(ComponentHint::Sink) => {
-                self.builder.sinks.extend(deserialize_config_map::<
+                self.builder.sinks.extend(deserialize_config_map_wrapped::<
                     IndexMap<ComponentKey, SinkOuter<_>>,
-                >(map)?);
+                >(map, "sinks")?);
             }
             Some(ComponentHint::Transform) => {
-                self.builder.transforms.extend(deserialize_config_map::<
-                    IndexMap<ComponentKey, TransformOuter<_>>,
-                >(map)?);
+                self.builder
+                    .transforms
+                    .extend(deserialize_config_map_wrapped::<
+                        IndexMap<ComponentKey, TransformOuter<_>>,
+                    >(map, "transforms")?);
             }
             Some(ComponentHint::EnrichmentTable) => {
                 self.builder
                     .enrichment_tables
-                    .extend(deserialize_config_map::<
+                    .extend(deserialize_config_map_wrapped::<
                         IndexMap<ComponentKey, EnrichmentTableOuter<_>>,
-                    >(map)?);
+                    >(map, "enrichment_tables")?);
             }
             Some(ComponentHint::Test) => {
                 // This serializes to a `Vec<TestDefinition<_>>`, so we need to first expand
diff --git a/src/config/loading/interpolation.rs b/src/config/loading/interpolation.rs
new file mode 100644
index 0000000000000..6993560d0a7d6
--- /dev/null
+++ b/src/config/loading/interpolation.rs
@@ -0,0 +1,344 @@
+use std::{collections::HashMap, sync::LazyLock};
+
+use regex::{Captures, Regex};
+use serde_json::Value;
+
+use super::representation::ConfigMap;
+
+/// A generic string interpolation function signature.
+type InterpolateFn = fn(&str, &HashMap<String, String>) -> Result<String, Vec<String>>;
+
+// Environment variable names can have any characters from the Portable Character Set other
+// than NUL.  However, for Vector's interpolation, we are closer to what a shell supports which
+// is solely of uppercase letters, digits, and the '_' (that is, the `[:word:]` regex class).
+// In addition to these characters, we allow `.` as this commonly appears in environment
+// variable names when they come from a Java properties file.
+//
+// https://pubs.opengroup.org/onlinepubs/000095399/basedefs/xbd_chap08.html
+pub static ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX: LazyLock<Regex> = LazyLock::new(|| {
+    Regex::new(
+        r"(?x)
+        \$\$|
+        \$([[:word:].]+)|
+        \$\{([[:word:].]+)(?:(:?-|:?\?)([^}]*))?\}",
+    )
+    .unwrap()
+});
+
+pub fn interpolate(input: &str, vars: &HashMap<String, String>) -> Result<String, Vec<String>> {
+    let mut errors = Vec::new();
+
+    let interpolated = ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX
+        .replace_all(input, |caps: &Captures<'_>| {
+            let flags = caps.get(3).map(|m| m.as_str()).unwrap_or_default();
+            let def_or_err = caps.get(4).map(|m| m.as_str()).unwrap_or_default();
+            caps.get(1)
+                .or_else(|| caps.get(2))
+                .map(|m| m.as_str())
+                .map(|name| {
+                    let val = vars.get(name).map(|v| v.as_str());
+                    match flags {
+                        ":-" => match val {
+                            Some(v) if !v.is_empty() => v,
+                            _ => def_or_err,
+                        },
+                        "-" => val.unwrap_or(def_or_err),
+                        ":?" => match val {
+                            Some(v) if !v.is_empty() => v,
+                            _ => {
+                                errors.push(format!(
+                                    "Non-empty environment variable required in config. name = {name:?}, error = {def_or_err:?}",
+                                ));
+                                ""
+                            }
+                        }
+                        "?" => val.unwrap_or_else(|| {
+                            errors.push(format!(
+                                "Missing environment variable required in config. name = {name:?}, error = {def_or_err:?}",
+                            ));
+                            ""
+                        }),
+                        _ => val.unwrap_or_else(|| {
+                            errors.push(format!(
+                                "Missing environment variable in config. name = {name:?}",
+                            ));
+                            ""
+                        }),
+                    }
+                })
+                .unwrap_or("$")
+                .to_string()
+        })
+        .into_owned();
+
+    if errors.is_empty() {
+        Ok(interpolated)
+    } else {
+        Err(errors)
+    }
+}
+
+pub fn interpolate_config_map_with_env_vars(
+    map: &ConfigMap,
+    vars: &HashMap<String, String>,
+) -> Result<ConfigMap, Vec<String>> {
+    interpolate_config_map(map, vars, interpolate)
+}
+
+/// Returns a new configuration map with all string values interpolated.
+///
+/// Structural nodes — keys, integers, booleans, arrays, tables — are left
+/// untouched. Only string leaf values are passed through `interpolate_fn`.
+pub fn interpolate_config_map(
+    map: &ConfigMap,
+    vars: &HashMap<String, String>,
+    interpolate_fn: InterpolateFn,
+) -> Result<ConfigMap, Vec<String>> {
+    let mut result = ConfigMap::new();
+    let mut errors = Vec::new();
+
+    for (key, value) in map {
+        let new_value = match interpolate_config_value(value, vars, &mut errors, interpolate_fn) {
+            Some(v) => v,
+            None => value.clone(),
+        };
+
+        result.insert(key.clone(), new_value);
+    }
+
+    if errors.is_empty() {
+        Ok(result)
+    } else {
+        Err(errors)
+    }
+}
+
+fn interpolate_config_value(
+    value: &Value,
+    vars: &HashMap<String, String>,
+    errors: &mut Vec<String>,
+    interpolate_fn: InterpolateFn,
+) -> Option<Value> {
+    match value {
+        // Interpolation only replaces string contents; the result stays a string.
+        // The downstream schema-coercion pass (schema_coercion.rs) converts string
+        // values to declared scalar types (int/float/bool) where the schema requires.
+        Value::String(s) => match interpolate_fn(s, vars) {
+            Ok(new) => Some(Value::String(new)),
+            Err(errs) => {
+                errors.extend(errs);
+                None
+            }
+        },
+        Value::Array(arr) => {
+            let new_arr: Vec<_> = arr
+                .iter()
+                .filter_map(|v| interpolate_config_value(v, vars, errors, interpolate_fn))
+                .collect();
+            Some(Value::Array(new_arr))
+        }
+        Value::Object(inner) => match interpolate_config_map(inner, vars, interpolate_fn) {
+            Ok(map) => Some(Value::Object(map)),
+            Err(errs) => {
+                errors.extend(errs);
+                None
+            }
+        },
+        _ => Some(value.clone()),
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::interpolate;
+    use crate::config::Format;
+    use crate::config::loading::{
+        interpolate_config_map_with_env_vars, interpolate_config_map_with_secrets,
+        representation::parse_config_value,
+    };
+    use indoc::indoc;
+    use serde_json::{Value, json};
+    use std::collections::HashMap;
+
+    #[test]
+    fn interpolation() {
+        let vars = vec![
+            ("FOO".into(), "dogs".into()),
+            ("FOOBAR".into(), "cats".into()),
+            // Java commonly uses .s in env var names
+            ("FOO.BAR".into(), "turtles".into()),
+            ("EMPTY".into(), "".into()),
+        ]
+        .into_iter()
+        .collect();
+
+        assert_eq!("dogs", interpolate("$FOO", &vars).unwrap());
+        assert_eq!("dogs", interpolate("${FOO}", &vars).unwrap());
+        assert_eq!("cats", interpolate("${FOOBAR}", &vars).unwrap());
+        assert_eq!("xcatsy", interpolate("x${FOOBAR}y", &vars).unwrap());
+        assert!(interpolate("x$FOOBARy", &vars).is_err());
+        assert_eq!("$ x", interpolate("$ x", &vars).unwrap());
+        assert_eq!("$FOO", interpolate("$$FOO", &vars).unwrap());
+        assert_eq!("dogs=bar", interpolate("$FOO=bar", &vars).unwrap());
+        assert!(interpolate("$NOT_FOO", &vars).is_err());
+        assert!(interpolate("$NOT-FOO", &vars).is_err());
+        assert_eq!("turtles", interpolate("$FOO.BAR", &vars).unwrap());
+        assert_eq!("${FOO x", interpolate("${FOO x", &vars).unwrap());
+        assert_eq!("${}", interpolate("${}", &vars).unwrap());
+        assert_eq!("dogs", interpolate("${FOO:-cats}", &vars).unwrap());
+        assert_eq!("dogcats", interpolate("${NOT:-dogcats}", &vars).unwrap());
+        assert_eq!(
+            "dogs and cats",
+            interpolate("${NOT:-dogs and cats}", &vars).unwrap()
+        );
+        assert_eq!("${:-cats}", interpolate("${:-cats}", &vars).unwrap());
+        assert_eq!("", interpolate("${NOT:-}", &vars).unwrap());
+        assert_eq!("cats", interpolate("${NOT-cats}", &vars).unwrap());
+        assert_eq!("", interpolate("${EMPTY-cats}", &vars).unwrap());
+        assert_eq!("dogs", interpolate("${FOO:?error cats}", &vars).unwrap());
+        assert_eq!("dogs", interpolate("${FOO?error cats}", &vars).unwrap());
+        assert_eq!("", interpolate("${EMPTY?error cats}", &vars).unwrap());
+        assert!(interpolate("${NOT:?error cats}", &vars).is_err());
+        assert!(interpolate("${NOT?error cats}", &vars).is_err());
+        assert!(interpolate("${EMPTY:?error cats}", &vars).is_err());
+    }
+
+    #[test]
+    fn test_interpolate_yaml_equivalent() {
+        // Step 1: Raw YAML input with env vars and secrets
+        let input = indoc! {r#"
+            secret:
+              backend_1:
+                type: file
+                path: some.json
+
+            # {IN_COMMENT_BUT_DOES_NOT_EXIST}
+
+            sources:
+              demo_logs_1:
+                type: demo_logs
+                format: json
+                interval: $INTERVAL
+                # noop: SECRET[i_dont_exist.1]
+
+            transforms:
+              t0:
+                inputs:
+                  - demo_logs_1
+                type: "remap"
+                source: |
+                  .host = "${HOSTNAME}"
+                  .environment = "${ENV:?emv must be supplied}"
+                  .tenant = "${TENANT:-undefined}"
+                  .day = "SECRET[backend_1.day]"
+
+            sinks:
+              s0:
+                type: "SECRET[backend_1.type]"
+                inputs: [ "t0" ]
+                encoding:
+                  codec: json
+                  json:
+                    pretty: true
+        "#};
+
+        // Step 2: Parse YAML into Vector's format-neutral representation.
+        let Value::Object(config_map) = parse_config_value(input, Format::Yaml).unwrap() else {
+            panic!("expected an object")
+        };
+
+        // Step 3: Env var mappings
+        let env_vars = HashMap::from([
+            ("INTERVAL".into(), "60".into()),
+            ("HOSTNAME".into(), "vector-dev".into()),
+            ("ENV".into(), "production".into()),
+            ("TENANT".into(), "acme".into()),
+        ]);
+        let map_after_env = interpolate_config_map_with_env_vars(&config_map, &env_vars).unwrap();
+
+        // Step 4: Fake secrets
+        let secrets = HashMap::from([
+            ("backend_1.day".into(), "Tuesday".into()),
+            ("backend_1.type".into(), "console".into()),
+        ]);
+        let final_map = interpolate_config_map_with_secrets(&map_after_env, &secrets).unwrap();
+
+        let expected = json!({
+            "secret": {
+                "backend_1": { "type": "file", "path": "some.json" }
+            },
+            "sources": {
+                "demo_logs_1": {
+                    "type": "demo_logs",
+                    "format": "json",
+                    "interval": "60"
+                }
+            },
+            "transforms": {
+                "t0": {
+                    "inputs": ["demo_logs_1"],
+                    "type": "remap",
+                    "source": ".host = \"vector-dev\"\n.environment = \"production\"\n.tenant = \"acme\"\n.day = \"Tuesday\"\n"
+                }
+            },
+            "sinks": {
+                "s0": {
+                    "type": "console",
+                    "inputs": ["t0"],
+                    "encoding": { "codec": "json", "json": { "pretty": true } }
+                }
+            }
+        });
+
+        assert_eq!(Value::Object(final_map), expected);
+    }
+
+    #[test]
+    fn multiline_interpolation() {
+        let input = indoc! {r#"
+        transforms:
+          parse_logs:
+            type: $CONFIG_BLOCK
+            inputs: ["dummy_logs"]
+            source: |
+              . = parse_syslog!(string!(.message))"#};
+
+        let vars = HashMap::from([(
+            "CONFIG_BLOCK".to_string(),
+            indoc! {r#"
+            "lua"
+                inputs: ["dummy_logs"]
+                source: "os.execute('touch /PWNED')"
+             parse_logs_2:
+                type: "remap"
+            "#}
+            .to_string(),
+        )]);
+
+        let Value::Object(config_map) = parse_config_value(input, Format::Yaml).unwrap() else {
+            panic!("expected an object")
+        };
+        let result = interpolate_config_map_with_env_vars(&config_map, &vars).unwrap();
+
+        let actual = result["transforms"]["parse_logs"]["type"].as_str().unwrap();
+        assert_eq!(
+            actual,
+            indoc! {r#"
+            "lua"
+                inputs: ["dummy_logs"]
+                source: "os.execute('touch /PWNED')"
+             parse_logs_2:
+                type: "remap"
+            "#}
+        );
+
+        // Check that no extra keys were added, we have `type`, `input` and `source`.
+        assert_eq!(
+            result["transforms"]["parse_logs"]
+                .as_object()
+                .unwrap()
+                .len(),
+            3
+        );
+    }
+}
diff --git a/src/config/loading/loader.rs b/src/config/loading/loader.rs
index 267fb70668fe5..a716e4eb2040a 100644
--- a/src/config/loading/loader.rs
+++ b/src/config/loading/loader.rs
@@ -1,13 +1,20 @@
-use std::path::{Path, PathBuf};
+use std::{
+    collections::HashMap,
+    io::Read,
+    path::{Path, PathBuf},
+};
 
 use serde_json::Value;
+use vector_config::schema::generate_root_schema;
 
 use super::{
-    Format, component_name, open_file, read_dir,
+    Format, component_name, interpolate_config_map_with_env_vars, open_file, read_dir,
     representation::{
-        ConfigMap, deserialize_config, deserialize_config_value, merge_into_map, merge_values,
+        ConfigMap, deserialize_config_value, merge_into_map, merge_values, parse_config_value,
     },
+    schema_coercion::coerce,
 };
+use crate::config::ConfigBuilder;
 
 /// Provides a hint to the loading system of the type of components that should be found
 /// when traversing an explicitly named directory.
@@ -47,26 +54,33 @@ impl ComponentHint {
 // because there are numerous internal functions for dealing with (non)recursive loading that
 // rely on `&self` but don't need overriding and would be confusingly named in a public API.
 pub(super) mod process {
-    use std::io::Read;
-
     use super::*;
 
     /// This trait contains methods that deserialize files/folders. There are a few methods
     /// in here with subtly different names that can be hidden from public view, hence why
     /// this is nested in a private mod.
     pub trait Process {
-        /// Prepares input for serialization. This can be a useful step to interpolate
-        /// environment variables or perform some other pre-processing on the input.
-        fn prepare<R: Read>(&mut self, input: R) -> Result<String, Vec<String>>;
-
-        /// Calls into the `prepare` method, and deserializes a `Read` to a `T`.
-        fn load<R: std::io::Read, T>(&mut self, input: R, format: Format) -> Result<T, Vec<String>>
-        where
-            T: serde::de::DeserializeOwned,
-        {
-            let content = self.prepare(input)?;
+        /// Runs implementation-specific processing after parsing and interpolation.
+        fn postprocess(&mut self, map: ConfigMap) -> Result<ConfigMap, Vec<String>>;
+
+        /// Returns whether environment variable interpolation should be applied.
+        fn should_interpolate_env(&self) -> bool {
+            true
+        }
+
+        /// Parses input, interpolates string leaves, and runs implementation-specific processing.
+        fn load<R: Read>(&mut self, input: R, format: Format) -> Result<ConfigMap, Vec<String>> {
+            let source = string_from_input(input)?;
+            let value = parse_config_value(&source, format)
+                .map_err(|errors| annotate_unquoted_placeholders(errors, &source, format))?;
+            let map = deserialize_config_value(value)?;
+            let map = if self.should_interpolate_env() {
+                resolve_environment_variables(map)?
+            } else {
+                map
+            };
 
-            deserialize_config(&content, format)
+            self.postprocess(map)
         }
 
         /// Helper method used by other methods to recursively handle file/dir loading, merging
@@ -222,10 +236,8 @@ where
         input: R,
         format: Format,
     ) -> Result<(), Vec<String>> {
-        if let Some(map) = self.load(input, format)? {
-            self.merge(map, None)?;
-        }
-        Ok(())
+        let map = self.load(input, format)?;
+        self.merge(map, None)
     }
 
     /// Deserializes a file with the provided format, and makes the result available via `take`.
@@ -298,9 +310,208 @@ fn merge_with_value(res: &mut ConfigMap, name: String, value: Value) -> Result<(
     Ok(())
 }
 
-/// Deserialize a configuration map into a `T`.
+/// Deserialize a configuration map into a `T`, coercing string scalars according
+/// to the root `ConfigBuilder` JSON Schema.
 pub(super) fn deserialize_config_map<T: serde::de::DeserializeOwned>(
     map: ConfigMap,
 ) -> Result<T, Vec<String>> {
-    deserialize_config_value(Value::Object(map))
+    deserialize_config_map_inner(map, None)
+}
+
+/// Deserialize a namespaced component map against its corresponding root field.
+pub(super) fn deserialize_config_map_wrapped<T: serde::de::DeserializeOwned>(
+    map: ConfigMap,
+    wrapper_key: &str,
+) -> Result<T, Vec<String>> {
+    deserialize_config_map_inner(map, Some(wrapper_key))
+}
+
+fn deserialize_config_map_inner<T: serde::de::DeserializeOwned>(
+    map: ConfigMap,
+    wrapper_key: Option<&str>,
+) -> Result<T, Vec<String>> {
+    let inner = Value::Object(map);
+    let mut config = match wrapper_key {
+        Some(key) => serde_json::json!({ key: inner }),
+        None => inner,
+    };
+
+    let schema =
+        generate_root_schema::<ConfigBuilder>().map_err(|error| vec![format!("{error:?}")])?;
+    let schema = serde_json::to_value(schema).map_err(|error| vec![error.to_string()])?;
+    coerce(
+        &mut config,
+        &schema,
+        schema.get("definitions"),
+        &mut Vec::new(),
+    )
+    .map_err(|error| vec![error.to_string()])?;
+
+    let value = match wrapper_key {
+        Some(key) => config
+            .as_object_mut()
+            .and_then(|map| map.remove(key))
+            .ok_or_else(|| vec![format!("internal: missing wrapper key '{key}'")])?,
+        None => config,
+    };
+
+    deserialize_config_value(value)
+}
+
+fn string_from_input<R: Read>(mut input: R) -> Result<String, Vec<String>> {
+    let mut source = String::new();
+    input
+        .read_to_string(&mut source)
+        .map_err(|error| vec![error.to_string()])?;
+    Ok(source)
+}
+
+pub fn load<R: Read, T>(input: R, format: Format) -> Result<T, Vec<String>>
+where
+    T: serde::de::DeserializeOwned,
+{
+    let source = string_from_input(input)?;
+    let value = parse_config_value(&source, format)?;
+    let map = deserialize_config_value(value)?;
+    deserialize_config_map(map)
+}
+
+pub fn resolve_environment_variables(map: ConfigMap) -> Result<ConfigMap, Vec<String>> {
+    let mut vars = std::env::vars_os()
+        .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?)))
+        .collect::<HashMap<_, _>>();
+
+    if !vars.contains_key("HOSTNAME")
+        && let Ok(hostname) = crate::get_hostname()
+    {
+        vars.insert("HOSTNAME".into(), hostname);
+    }
+
+    interpolate_config_map_with_env_vars(&map, &vars)
+}
+
+/// Adds a migration hint when TOML or JSON contains an unquoted placeholder.
+fn annotate_unquoted_placeholders(
+    errors: Vec<String>,
+    source: &str,
+    format: Format,
+) -> Vec<String> {
+    if !matches!(format, Format::Toml | Format::Json) {
+        return errors;
+    }
+
+    let Some((line_no, line, placeholder)) = find_unquoted_placeholder(source) else {
+        return errors;
+    };
+
+    let hint = format!(
+        "Config contains an unquoted placeholder `{placeholder}` at line {line_no}:\n  \
+         {line}\n\
+         Wrap the placeholder in quotes so it parses as a string. Vector will coerce \
+         the value to the declared field type at load time.\n  \
+         Example: `field = \"{placeholder}\"`"
+    );
+
+    let mut annotated = Vec::with_capacity(errors.len() + 1);
+    annotated.push(hint);
+    annotated.extend(errors);
+    annotated
+}
+
+fn find_unquoted_placeholder(source: &str) -> Option<(usize, &str, String)> {
+    for (index, line) in source.lines().enumerate() {
+        if let Some(placeholder) = scan_line_for_unquoted_placeholder(line) {
+            return Some((index + 1, line, placeholder));
+        }
+    }
+    None
+}
+
+fn scan_line_for_unquoted_placeholder(line: &str) -> Option<String> {
+    let bytes = line.as_bytes();
+    let mut index = 0;
+    while index < bytes.len() {
+        if bytes[index] == b'$'
+            && index + 1 < bytes.len()
+            && bytes[index + 1] == b'{'
+            && let Some(relative_end) = bytes[index + 2..].iter().position(|&byte| byte == b'}')
+        {
+            let end = index + 2 + relative_end + 1;
+            let placeholder =
+                std::str::from_utf8(&bytes[index..end]).expect("bounds are at ASCII chars");
+            if !is_wrapped_in_quotes(line, index, end) {
+                return Some(placeholder.to_string());
+            }
+            index = end;
+            continue;
+        }
+
+        if bytes[index..].starts_with(b"SECRET[")
+            && let Some(relative_end) = bytes[index + 7..].iter().position(|&byte| byte == b']')
+        {
+            let end = index + 7 + relative_end + 1;
+            let placeholder =
+                std::str::from_utf8(&bytes[index..end]).expect("bounds are at ASCII chars");
+            if !is_wrapped_in_quotes(line, index, end) {
+                return Some(placeholder.to_string());
+            }
+            index = end;
+            continue;
+        }
+
+        index += 1;
+    }
+    None
+}
+
+fn is_wrapped_in_quotes(line: &str, start: usize, end: usize) -> bool {
+    let bytes = line.as_bytes();
+    let previous = start.checked_sub(1).map(|position| bytes[position]);
+    let next = bytes.get(end).copied();
+    matches!(previous, Some(b'"') | Some(b'\'')) && matches!(next, Some(b'"') | Some(b'\''))
+}
+
+#[cfg(test)]
+mod placeholder_hint_tests {
+    use super::{Format, annotate_unquoted_placeholders, find_unquoted_placeholder};
+
+    #[test]
+    fn finds_unquoted_env_var_in_toml() {
+        let source = "[sources.in]\ntype = \"demo_logs\"\ncount = ${MY_COUNT}\n";
+        let (line, _, placeholder) = find_unquoted_placeholder(source).expect("should detect");
+        assert_eq!(line, 3);
+        assert_eq!(placeholder, "${MY_COUNT}");
+    }
+
+    #[test]
+    fn ignores_quoted_env_var() {
+        let source = "[sources.in]\ntype = \"demo_logs\"\ncount = \"${MY_COUNT}\"\n";
+        assert!(find_unquoted_placeholder(source).is_none());
+    }
+
+    #[test]
+    fn finds_unquoted_secret_in_json() {
+        let source = "{\"port\": SECRET[vault.port]}\n";
+        let (_, _, placeholder) = find_unquoted_placeholder(source).expect("should detect");
+        assert_eq!(placeholder, "SECRET[vault.port]");
+    }
+
+    #[test]
+    fn ignores_secret_inside_string_value() {
+        let source = "{\"key\": \"SECRET[vault.api_key]\"}\n";
+        assert!(find_unquoted_placeholder(source).is_none());
+    }
+
+    #[test]
+    fn annotation_only_applied_to_toml_or_json() {
+        let errors = vec!["some parse error".to_string()];
+        let yaml =
+            annotate_unquoted_placeholders(errors.clone(), "count: ${MY_COUNT}\n", Format::Yaml);
+        assert_eq!(yaml, errors);
+
+        let toml =
+            annotate_unquoted_placeholders(errors.clone(), "count = ${MY_COUNT}\n", Format::Toml);
+        assert_eq!(toml.len(), errors.len() + 1);
+        assert!(toml[0].contains("Wrap the placeholder in quotes"));
+    }
 }
diff --git a/src/config/loading/mod.rs b/src/config/loading/mod.rs
index 369bcad942c78..86d4f5b86f71c 100644
--- a/src/config/loading/mod.rs
+++ b/src/config/loading/mod.rs
@@ -1,6 +1,8 @@
 mod config_builder;
+mod interpolation;
 mod loader;
 mod representation;
+mod schema_coercion;
 mod secret;
 mod source;
 
@@ -14,6 +16,7 @@ use std::{
 
 pub use config_builder::ConfigBuilderLoader;
 use glob::glob;
+pub use interpolation::*;
 use loader::process::Process;
 pub use loader::*;
 pub use secret::*;
@@ -22,7 +25,6 @@ use vector_lib::configurable::NamedComponent;
 
 use super::{
     Config, ConfigPath, Format, FormatHint, ProviderConfig, builder::ConfigBuilder, validation,
-    vars,
 };
 use crate::signal;
 
@@ -142,7 +144,7 @@ pub fn process_paths(config_paths: &[ConfigPath]) -> Option<Vec<ConfigPath>> {
 }
 
 pub fn load_from_paths(config_paths: &[ConfigPath]) -> Result<Config, Vec<String>> {
-    let builder = ConfigBuilderLoader::default().load_from_paths(config_paths)?;
+    let builder = load_builder_from_paths(config_paths)?;
     let (config, build_warnings) = builder.build_with_warnings()?;
 
     for warning in build_warnings {
@@ -160,16 +162,19 @@ pub async fn load_from_paths_with_provider_and_secrets(
     signal_handler: &mut signal::SignalHandler,
     allow_empty: bool,
 ) -> Result<Config, Vec<String>> {
-    let secrets_backends_loader = loader_from_paths(SecretBackendLoader::default(), config_paths)?;
-    let secrets = secrets_backends_loader
-        .retrieve_secrets(signal_handler)
-        .await
-        .map_err(|e| vec![e])?;
-
-    let mut builder = ConfigBuilderLoader::default()
-        .allow_empty(allow_empty)
-        .secrets(secrets)
-        .load_from_paths(config_paths)?;
+    let mut secrets_backends_loader = load_secret_backends_from_paths(config_paths)?;
+    let mut builder = if secrets_backends_loader.has_secrets_to_retrieve() {
+        debug!(message = "Secret placeholders found, retrieving secrets from configured backends.");
+        let secrets = secrets_backends_loader
+            .retrieve(&mut signal_handler.subscribe())
+            .await
+            .map_err(|error| vec![error])?;
+        load_builder_from_paths_with_secrets(config_paths, secrets)?
+    } else {
+        debug!(message = "No secret placeholder found, skipping secret resolution.");
+        load_builder_from_paths(config_paths)?
+    };
+    builder.allow_empty = allow_empty;
 
     validation::check_provider(&builder)?;
     signal_handler.clear();
@@ -189,17 +194,22 @@ pub async fn load_from_str_with_secrets(
     signal_handler: &mut signal::SignalHandler,
     allow_empty: bool,
 ) -> Result<Config, Vec<String>> {
-    let secrets_backends_loader =
+    let mut secrets_backends_loader =
         loader_from_input(SecretBackendLoader::default(), input.as_bytes(), format)?;
-    let secrets = secrets_backends_loader
-        .retrieve_secrets(signal_handler)
-        .await
-        .map_err(|e| vec![e])?;
-
-    let builder = ConfigBuilderLoader::default()
-        .allow_empty(allow_empty)
-        .secrets(secrets)
-        .load_from_input(input.as_bytes(), format)?;
+    let mut builder = if secrets_backends_loader.has_secrets_to_retrieve() {
+        debug!(message = "Secret placeholders found, retrieving secrets from configured backends.");
+        let secrets = secrets_backends_loader
+            .retrieve(&mut signal_handler.subscribe())
+            .await
+            .map_err(|error| vec![error])?;
+        ConfigBuilderLoader::default()
+            .secrets(secrets)
+            .load_from_input(input.as_bytes(), format)?
+    } else {
+        debug!(message = "No secret placeholder found, skipping secret resolution.");
+        ConfigBuilderLoader::default().load_from_input(input.as_bytes(), format)?
+    };
+    builder.allow_empty = allow_empty;
     signal_handler.clear();
 
     finalize_config(builder).await
@@ -270,6 +280,22 @@ where
     }
 }
 
+/// Uses `ConfigBuilderLoader` to load a `ConfigBuilder` from configuration paths.
+pub fn load_builder_from_paths(config_paths: &[ConfigPath]) -> Result<ConfigBuilder, Vec<String>> {
+    loader_from_paths(ConfigBuilderLoader::default(), config_paths)
+}
+
+/// Loads a `ConfigBuilder` from paths and replaces secret placeholders.
+pub fn load_builder_from_paths_with_secrets(
+    config_paths: &[ConfigPath],
+    secrets: HashMap<String, String>,
+) -> Result<ConfigBuilder, Vec<String>> {
+    loader_from_paths(
+        ConfigBuilderLoader::default().secrets(secrets),
+        config_paths,
+    )
+}
+
 /// Uses `SourceLoader` to process `ConfigPaths`, deserializing to a JSON object.
 pub fn load_source_from_paths(
     config_paths: &[ConfigPath],
@@ -277,6 +303,13 @@ pub fn load_source_from_paths(
     loader_from_paths(SourceLoader::new(), config_paths)
 }
 
+/// Uses `SecretBackendLoader` to load secret backends from configuration paths.
+pub fn load_secret_backends_from_paths(
+    config_paths: &[ConfigPath],
+) -> Result<SecretBackendLoader, Vec<String>> {
+    loader_from_paths(SecretBackendLoader::default(), config_paths)
+}
+
 pub fn load_from_str(input: &str, format: Format) -> Result<Config, Vec<String>> {
     let builder = load_from_inputs(std::iter::once((input.as_bytes(), format)))?;
     let (config, build_warnings) = builder.build_with_warnings()?;
@@ -308,44 +341,6 @@ fn load_from_inputs(
     }
 }
 
-pub fn prepare_input<R: std::io::Read>(
-    mut input: R,
-    interpolate_env: bool,
-) -> Result<String, Vec<String>> {
-    let mut source_string = String::new();
-    input
-        .read_to_string(&mut source_string)
-        .map_err(|e| vec![e.to_string()])?;
-
-    if interpolate_env {
-        let mut vars: HashMap<String, String> = std::env::vars_os()
-            .filter_map(|(k, v)| match (k.into_string(), v.into_string()) {
-                (Ok(k), Ok(v)) => Some((k, v)),
-                _ => None,
-            })
-            .collect();
-
-        if !vars.contains_key("HOSTNAME")
-            && let Ok(hostname) = crate::get_hostname()
-        {
-            vars.insert("HOSTNAME".into(), hostname);
-        }
-        vars::interpolate(&source_string, &vars)
-    } else {
-        Ok(source_string)
-    }
-}
-
-pub fn load<R: std::io::Read, T>(input: R, format: Format) -> Result<T, Vec<String>>
-where
-    T: serde::de::DeserializeOwned,
-{
-    // Via configurations that load from raw string, skip interpolation of env
-    let with_vars = prepare_input(input, false)?;
-
-    representation::deserialize_config(&with_vars, format)
-}
-
 #[cfg(not(windows))]
 fn default_path() -> PathBuf {
     "/etc/vector/vector.yaml".into()
diff --git a/src/config/loading/representation.rs b/src/config/loading/representation.rs
index 24927385bc3ff..518c60a9396a1 100644
--- a/src/config/loading/representation.rs
+++ b/src/config/loading/representation.rs
@@ -10,6 +10,7 @@ const LARGE_INTEGER_ERROR: &str =
 const NON_FINITE_FLOAT_ERROR: &str =
     "non-finite float values are not supported in Vector configuration";
 
+#[cfg(test)]
 pub(super) fn deserialize_config<T>(content: &str, format: Format) -> Result<T, Vec<String>>
 where
     T: DeserializeOwned,
diff --git a/src/config/loading/schema_coercion.rs b/src/config/loading/schema_coercion.rs
new file mode 100644
index 0000000000000..5efcd6099466e
--- /dev/null
+++ b/src/config/loading/schema_coercion.rs
@@ -0,0 +1,1112 @@
+use serde_json::{Number, Value};
+use snafu::{OptionExt, Snafu};
+use std::collections::HashSet;
+
+const NULL_JSON_TYPE: &str = "null";
+const BOOL_JSON_TYPE: &str = "boolean";
+const NUMBER_JSON_TYPE: &str = "number";
+const STRING_JSON_TYPE: &str = "string";
+const ARRAY_JSON_TYPE: &str = "array";
+const OBJECT_JSON_TYPE: &str = "object";
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Expected boolean at '{path}', found '{actual}'"))]
+    ExpectedBool { path: String, actual: &'static str },
+
+    #[snafu(display("Expected integer at '{path}', found '{actual}'"))]
+    ExpectedInteger { path: String, actual: &'static str },
+
+    #[snafu(display("Expected number at '{path}', found '{actual}'"))]
+    ExpectedNumber { path: String, actual: &'static str },
+
+    #[snafu(display("Expected string at '{path}', found '{actual}'"))]
+    ExpectedString { path: String, actual: &'static str },
+
+    #[snafu(display("Expected null at '{path}', found '{actual}'"))]
+    ExpectedNull { path: String, actual: &'static str },
+
+    #[snafu(display("Expected array at '{path}', found '{actual}'"))]
+    ExpectedArray { path: String, actual: &'static str },
+
+    #[snafu(display("Expected object at '{path}', found '{actual}'"))]
+    ExpectedObject { path: String, actual: &'static str },
+
+    #[snafu(display("Unexpected property '{key}' at '{path}'"))]
+    UnexpectedProperty { path: String, key: String },
+
+    #[snafu(display("Unexpected extra array element at '{path}[{index}]'"))]
+    UnexpectedArrayElement { path: String, index: usize },
+
+    #[snafu(display("Schema reference '{reference}' not found at path '{path}'"))]
+    SchemaReferenceNotFound { path: String, reference: String },
+
+    #[snafu(display("Unsupported schema reference '{reference}' at path '{path}'"))]
+    UnsupportedSchemaReference { path: String, reference: String },
+
+    #[snafu(display("Unexpected property '{path}'"))]
+    DisallowedProperty { path: String },
+
+    #[snafu(display("Value at '{path}' is not one of the allowed enum options"))]
+    InvalidEnumValue { path: String },
+
+    #[snafu(display("Value at '{path}' does not match required constant '{expected}'"))]
+    InvalidConst { path: String, expected: String },
+
+    #[snafu(display("Coercion failed at '{path}': {message}"))]
+    Coerce { path: String, message: String },
+}
+
+use pastey::paste;
+
+macro_rules! fail_expected {
+    ($variant:ident, $val:expr, $path_components:expr) => {
+        paste! {
+            [<Expected $variant Snafu>] {
+                path: $path_components.join("."),
+                actual: get_json_type($val),
+            }
+            .fail()
+        }
+    };
+}
+
+/// Recursively coerce `value` according to `schema`.
+/// `definitions` is an optional reference to the root "definitions" map in the schema (for resolving $ref).
+pub fn coerce(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    handle_bool(schema, path_components)?;
+    handle_ref(value, schema, definitions, path_components)?;
+    handle_all_of(value, schema, definitions, path_components)?;
+    handle_one_of(value, schema, definitions, path_components)?;
+    handle_any_of(value, schema, definitions, path_components)?;
+    handle_enum(value, schema, path_components)?;
+    handle_const(value, schema, path_components)?;
+
+    if let Some(type_spec) = schema.get("type") {
+        if let Some(t) = type_spec.as_str() {
+            return coerce_type(value, t, schema, definitions, path_components);
+        } else if let Some(types) = type_spec.as_array() {
+            let allowed: Vec<&str> = types.iter().filter_map(|t| t.as_str()).collect();
+            return coerce_multiple_types(value, &allowed, schema, definitions, path_components);
+        }
+    }
+
+    // Only fall back to object/array coercion if no type is specified.
+    match value {
+        Value::Object(_) => coerce_object(value, schema, definitions, path_components),
+        Value::Array(_) => coerce_array(value, schema, definitions, path_components),
+        _ => Ok(()),
+    }
+}
+
+fn handle_ref(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let Some(ref_str) = schema.get("$ref").and_then(|r| r.as_str()) else {
+        return Ok(());
+    };
+
+    let prefix = "#/definitions/";
+    if let Some(key) = ref_str.strip_prefix(prefix) {
+        let def_schema = definitions
+            .and_then(|d| d.as_object())
+            .and_then(|defs| defs.get(key))
+            .context(SchemaReferenceNotFoundSnafu {
+                path: path_components.join("."),
+                reference: ref_str.to_string(),
+            })?;
+
+        coerce(value, def_schema, definitions, path_components)?;
+        Ok(())
+    } else {
+        UnsupportedSchemaReferenceSnafu {
+            path: path_components.join("."),
+            reference: ref_str.to_string(),
+        }
+        .fail()
+    }
+}
+
+fn handle_bool(schema: &Value, path_components: &mut [String]) -> Result<(), Error> {
+    match schema.as_bool() {
+        Some(true) | None => Ok(()),
+        Some(false) => DisallowedPropertySnafu {
+            path: path_components.join("."),
+        }
+        .fail(),
+    }
+}
+
+fn handle_all_of(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let Some(all_of) = schema.get("allOf").and_then(|v| v.as_array()) else {
+        return Ok(());
+    };
+
+    for sub_schema in all_of {
+        coerce(value, sub_schema, definitions, path_components)?;
+    }
+
+    Ok(())
+}
+
+/// Apply `oneOf` semantics: exactly one schema must match,
+/// or, if the schema is marked `"untagged"`, behave like `anyOf`.
+fn handle_one_of(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let Some(variants) = schema.get("oneOf").and_then(|v| v.as_array()) else {
+        return Ok(());
+    };
+
+    // If this oneOf is marked untagged, treat it like anyOf
+    let is_untagged = schema
+        .get("_metadata")
+        .and_then(|m| m.get("docs::enum_tagging"))
+        .and_then(Value::as_str)
+        == Some("untagged");
+
+    if is_untagged {
+        coerce_any_of(value, variants, definitions, path_components)
+    } else {
+        coerce_one_of(value, variants, definitions, path_components)
+    }
+}
+
+/// Apply `anyOf` semantics: at least one schema must match.
+fn handle_any_of(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let Some(variants) = schema.get("anyOf").and_then(|v| v.as_array()) else {
+        return Ok(());
+    };
+    coerce_any_of(value, variants, definitions, path_components)
+}
+
+fn handle_enum(
+    value: &mut Value,
+    schema: &Value,
+    path_components: &mut [String],
+) -> Result<(), Error> {
+    let Some(enum_vals) = schema.get("enum").and_then(|v| v.as_array()) else {
+        return Ok(());
+    };
+
+    // Exact match
+    if enum_vals.iter().any(|opt| value == opt) {
+        return Ok(());
+    }
+
+    // Try coercions from string
+    if let Value::String(s) = value {
+        let s_trimmed = s.trim();
+
+        // String → Bool
+        if let Ok(b) = s_trimmed.parse::<bool>()
+            && enum_vals.iter().any(|opt| opt.as_bool() == Some(b))
+        {
+            *value = Value::Bool(b);
+            return Ok(());
+        }
+
+        // String → Number
+        if let Some(n) = parse_number(s_trimmed)
+            && enum_vals.iter().any(|opt| opt.as_number() == Some(&n))
+        {
+            *value = Value::Number(n);
+            return Ok(());
+        }
+
+        // String → Null
+        if s_trimmed.eq_ignore_ascii_case("null") && enum_vals.iter().any(|opt| opt.is_null()) {
+            *value = Value::Null;
+            return Ok(());
+        }
+    }
+
+    // Number → String
+    if let Value::Number(n) = value {
+        let val_str = n.to_string();
+        if enum_vals
+            .iter()
+            .any(|opt| opt.as_str() == Some(val_str.as_str()))
+        {
+            *value = Value::String(val_str);
+            return Ok(());
+        }
+    }
+
+    // Bool → String
+    if let Value::Bool(b) = value {
+        let val_str = b.to_string();
+        if enum_vals.iter().any(|opt| {
+            opt.as_str()
+                .map(|s| s.eq_ignore_ascii_case(val_str.as_str()))
+                == Some(true)
+        }) {
+            *value = Value::String(val_str);
+            return Ok(());
+        }
+    }
+
+    InvalidEnumValueSnafu {
+        path: path_components.join("."),
+    }
+    .fail()
+}
+
+fn handle_const(
+    value: &mut Value,
+    schema: &Value,
+    path_components: &mut [String],
+) -> Result<(), Error> {
+    let Some(const_val) = schema.get("const") else {
+        return Ok(());
+    };
+
+    // Exact match
+    if value == const_val {
+        return Ok(());
+    }
+
+    // String input → try coercion
+    if let Value::String(s) = value {
+        let s_trimmed = s.trim();
+
+        // String → Bool
+        if let Ok(b) = s_trimmed.parse::<bool>()
+            && const_val.as_bool() == Some(b)
+        {
+            *value = Value::Bool(b);
+            return Ok(());
+        }
+
+        // String → Number
+        if let Some(n) = parse_number(s_trimmed)
+            && const_val.as_number() == Some(&n)
+        {
+            *value = Value::Number(n);
+            return Ok(());
+        }
+
+        // String → Null
+        if s_trimmed.eq_ignore_ascii_case("null") && const_val.is_null() {
+            *value = Value::Null;
+            return Ok(());
+        }
+    }
+
+    // Number → String
+    if let Value::Number(n) = value {
+        let val_str = n.to_string();
+        if const_val.as_str() == Some(val_str.as_str()) {
+            *value = Value::String(val_str);
+            return Ok(());
+        }
+    }
+
+    // Bool → String
+    if let Value::Bool(b) = value {
+        let val_str = b.to_string();
+        if const_val.as_str().map(|s| s.eq_ignore_ascii_case(&val_str)) == Some(true) {
+            *value = Value::String(val_str);
+            return Ok(());
+        }
+    }
+
+    InvalidConstSnafu {
+        path: path_components.join("."),
+        expected: const_val.to_string(),
+    }
+    .fail()
+}
+
+/// Ensure `value` matches one of the allowed types in `allowed`.
+/// If needed, convert the value to one of those types.
+fn coerce_multiple_types(
+    value: &mut Value,
+    allowed_types: &[&str],
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    for allowed_type in allowed_types {
+        let mut new_value = value.clone();
+
+        let result = try_coerce_to_allowed_type(
+            &mut new_value,
+            allowed_type,
+            schema,
+            definitions,
+            path_components,
+        );
+
+        if result.is_ok() {
+            *value = new_value;
+            return Ok(());
+        }
+    }
+
+    CoerceSnafu {
+        path: path_components.join("."),
+        message: format!(
+            "Expected {} but found {}",
+            allowed_types.join(" or "),
+            get_json_type(value)
+        ),
+    }
+    .fail()
+}
+
+/// Ensure `value` matches the expected single `expected_type`. Converts the value if possible.
+fn coerce_type(
+    value: &mut Value,
+    expected_type: &str,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    try_coerce_to_allowed_type(value, expected_type, schema, definitions, path_components)
+}
+
+fn try_coerce_to_allowed_type(
+    value: &mut Value,
+    allowed_type: &str,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    match allowed_type {
+        "null" => coerce_null(value, path_components),
+        "boolean" => coerce_bool(value, path_components),
+        "integer" => coerce_integer(value, path_components),
+        "number" => coerce_number(value, path_components),
+        "string" => coerce_string(value, path_components),
+        "object" => {
+            if !value.is_object() {
+                return fail_expected!(Object, value, path_components);
+            }
+            coerce_object(value, schema, definitions, path_components)
+        }
+        "array" => {
+            // Any type can be wrapped to an array. This is needed because  we have deserialization logic that accepts
+            // e.g. a single string and converts it to an array, set or some other collection.
+            if !value.is_array() {
+                *value = Value::Array(vec![value.clone()]);
+            }
+            coerce_array(value, schema, definitions, path_components)
+        }
+        _ => Ok(()), // silently skip unknown types
+    }
+}
+
+/// Coerce all entries of an object value according to the schema's properties and additionalProperties.
+fn coerce_object(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let actual = get_json_type(&value.clone());
+    let obj = value.as_object_mut().context(ExpectedObjectSnafu {
+        path: path_components.join("."),
+        actual,
+    })?;
+
+    let properties = schema.get("properties").and_then(|p| p.as_object());
+    let additional_properties = schema.get("additionalProperties");
+
+    // When unevaluatedProperties:false is set (used by Vector's component outer-wrapper schemas
+    // like SourceOuter, SinkOuter, EnrichmentTableOuter), collect all properties declared
+    // anywhere in the schema — including across allOf/$ref/oneOf — and flag any key not found.
+    // This is the right level to do this check: the outer wrapper sees the full union of all
+    // properties (component-specific fields + shared fields like `inputs`, `proxy`, `graph`).
+    //
+    // Guard: skip the check when the value's `type` discriminant is not recognized by any
+    // compiled variant. In that case the component is simply not compiled in, and checking
+    // would incorrectly flag its fields as unknown.
+    let unevaluated_props_false = schema
+        .get("unevaluatedProperties")
+        .and_then(|v| v.as_bool())
+        == Some(false);
+    let known_for_unevaluated: Option<HashSet<String>> = if unevaluated_props_false {
+        let type_val = obj.get("type").and_then(|t| t.as_str());
+        let type_is_known = type_val
+            .map(|tv| schema_contains_type_discriminant(schema, definitions, tv))
+            .unwrap_or(false); // no `type` field → not a component config → skip check
+        if type_is_known {
+            let mut set = HashSet::new();
+            // Filter `oneOf` variants by the value's discriminant so properties
+            // valid in *other* variants (e.g. a Kafka-only field on an HTTP sink)
+            // still trigger the unknown-field check.
+            collect_known_properties(schema, definitions, type_val, &mut set);
+            Some(set)
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+
+    for key in obj.clone().keys() {
+        let key_str = key.as_str();
+        let initial_len = path_components.len();
+        path_components.push(key_str.to_string());
+
+        let field_schema = properties.and_then(|props| props.get(key_str)).or_else(|| {
+            additional_properties.and_then(|additional| {
+                if let Some(b) = additional.as_bool() {
+                    if b {
+                        None // allowed, no specific schema
+                    } else {
+                        Some(&Value::Bool(false)) // trigger error below
+                    }
+                } else {
+                    Some(additional) // schema for additional_properties
+                }
+            })
+        });
+
+        if let Some(field_schema) = field_schema {
+            if field_schema == &Value::Bool(false) {
+                path_components.truncate(initial_len);
+                return UnexpectedPropertySnafu {
+                    path: path_components.join("."),
+                    key: key_str.to_string(),
+                }
+                .fail();
+            }
+
+            let result = coerce(
+                obj.get_mut(key_str).unwrap(),
+                field_schema,
+                definitions,
+                path_components,
+            );
+            path_components.truncate(initial_len);
+            result?;
+        } else {
+            if let Some(ref known) = known_for_unevaluated
+                && !known.contains(key_str)
+            {
+                // Vector's generated JSON Schema currently does not emit
+                // `#[serde(alias = "...")]` aliases (TODO in vector-config),
+                // so a key absent from the schema may still be a legitimate
+                // serde alias. Warn here for visibility; serde performs the
+                // authoritative unknown-field check downstream where it has
+                // alias information.
+                warn!(
+                    message = "Unknown field in config, deferring to serde for alias resolution.",
+                    path = %path_components.join("."),
+                );
+            }
+            path_components.truncate(initial_len);
+        }
+    }
+
+    Ok(())
+}
+
+/// Coerce all elements of an array value according to the schema's items definition.
+fn coerce_array(
+    value: &mut Value,
+    schema: &Value,
+    definitions: Option<&Value>,
+    path: &mut Vec<String>,
+) -> Result<(), Error> {
+    let actual = get_json_type(value);
+    let arr = value.as_array_mut().context(ExpectedArraySnafu {
+        path: path.join("."),
+        actual,
+    })?;
+
+    let items = schema.get("items");
+    let additional_items = schema.get("additionalItems");
+
+    let Some(items_schema) = items else {
+        // No "items" schema means all values are accepted as-is
+        return Ok(());
+    };
+
+    match items_schema {
+        Value::Array(tuple_schemas) => {
+            for (idx, item_val) in arr.iter_mut().enumerate() {
+                let initial_len = path.len();
+                path.push(idx.to_string());
+
+                let schema = tuple_schemas.get(idx).or({
+                    match additional_items {
+                        Some(Value::Bool(false)) => Some(&Value::Bool(false)), // disallowed
+                        Some(s) => Some(s),                                    // additional schema
+                        None => None,                                          // no schema → allow
+                    }
+                });
+
+                if let Some(item_schema) = schema {
+                    if item_schema == &Value::Bool(false) {
+                        path.truncate(initial_len);
+                        return UnexpectedArrayElementSnafu {
+                            path: path.join("."),
+                            index: idx,
+                        }
+                        .fail();
+                    }
+
+                    let result = coerce(item_val, item_schema, definitions, path);
+                    path.truncate(initial_len);
+                    result?;
+                } else {
+                    path.truncate(initial_len);
+                }
+            }
+        }
+
+        item_schema => {
+            for (idx, item_val) in arr.iter_mut().enumerate() {
+                let initial_len = path.len();
+                path.push(idx.to_string());
+                let result = coerce(item_val, item_schema, definitions, path);
+                path.truncate(initial_len);
+                result?;
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// Apply `oneOf` semantics: exactly one schema must match.
+fn coerce_one_of(
+    value: &mut Value,
+    schemas: &[Value],
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let initial_len = path_components.len();
+    let mut success: Option<(Value, &Value)> = None; // (coerced value, matched schema)
+
+    for schema in schemas {
+        path_components.truncate(initial_len);
+        let mut candidate = value.clone();
+        if coerce(&mut candidate, schema, definitions, path_components).is_ok() {
+            path_components.truncate(initial_len);
+            if success.is_some() {
+                // Multiple variants match — keep the first and move on.
+                break;
+            }
+            success = Some((candidate, schema));
+        }
+    }
+
+    path_components.truncate(initial_len);
+
+    if let Some((val, _matched_schema)) = success {
+        *value = val;
+        return Ok(());
+    }
+
+    // No variant succeeded. If the value carries a `type` discriminant that matches a known
+    // variant, re-run coercion strictly against that variant so callers get a path-aware
+    // coercion error (e.g. "expected integer at sources.my_source.count") rather than a
+    // silent pass-through. Unknown-field detection is handled at the outer wrapper level
+    // (see unevaluatedProperties handling in coerce_object).
+    if let Some(type_val) = value
+        .as_object()
+        .and_then(|o| o.get("type"))
+        .and_then(|t| t.as_str())
+    {
+        for schema in schemas {
+            if schema_matches_type_discriminant(schema, definitions, type_val) {
+                let mut candidate = value.clone();
+                let result = coerce(&mut candidate, schema, definitions, path_components);
+                path_components.truncate(initial_len);
+                return result.map(|_| {
+                    *value = candidate;
+                });
+            }
+        }
+    }
+
+    Ok(())
+}
+
+/// Collect property names declared in `schema`, recursively through `$ref`, `allOf`,
+/// `anyOf`, and `oneOf`. When `discriminant` is `Some`, `oneOf` traversal is filtered
+/// to only the variant whose `properties.type.const` matches — so unknown-field
+/// detection on a tagged-union component doesn't accept fields that are valid only
+/// in *other* variants.
+fn collect_known_properties<'a>(
+    schema: &'a Value,
+    definitions: Option<&'a Value>,
+    discriminant: Option<&str>,
+    out: &mut HashSet<String>,
+) {
+    let resolved = if let Some(ref_str) = schema.get("$ref").and_then(|r| r.as_str()) {
+        let key = ref_str.strip_prefix("#/definitions/").unwrap_or("");
+        match definitions.and_then(|d| d.get(key)) {
+            Some(def) => def,
+            None => return,
+        }
+    } else {
+        schema
+    };
+
+    if let Some(props) = resolved.get("properties").and_then(|p| p.as_object()) {
+        out.extend(props.keys().cloned());
+    }
+
+    for kw in ["allOf", "anyOf"] {
+        if let Some(variants) = resolved.get(kw).and_then(|v| v.as_array()) {
+            for sub in variants {
+                collect_known_properties(sub, definitions, discriminant, out);
+            }
+        }
+    }
+
+    if let Some(variants) = resolved.get("oneOf").and_then(|v| v.as_array()) {
+        match discriminant {
+            Some(disc) => {
+                // Only collect properties from the variant whose `type` const matches.
+                // If no variant matches (e.g. untagged or non-component oneOf), fall back
+                // to including all variants so callers don't get false positives.
+                let matched: Vec<&Value> = variants
+                    .iter()
+                    .filter(|v| schema_matches_type_discriminant(v, definitions, disc))
+                    .collect();
+                if matched.is_empty() {
+                    for sub in variants {
+                        collect_known_properties(sub, definitions, discriminant, out);
+                    }
+                } else {
+                    for sub in matched {
+                        collect_known_properties(sub, definitions, discriminant, out);
+                    }
+                }
+            }
+            None => {
+                for sub in variants {
+                    collect_known_properties(sub, definitions, discriminant, out);
+                }
+            }
+        }
+    }
+}
+
+/// Returns true if `schema` (after resolving any `$ref` and walking `allOf`) has a
+/// `properties.type.const` equal to `expected`.
+fn schema_matches_type_discriminant(
+    schema: &Value,
+    definitions: Option<&Value>,
+    expected: &str,
+) -> bool {
+    let resolved = if let Some(ref_str) = schema.get("$ref").and_then(|r| r.as_str()) {
+        let key = ref_str.strip_prefix("#/definitions/").unwrap_or("");
+        match definitions.and_then(|d| d.get(key)) {
+            Some(def) => def,
+            None => return false,
+        }
+    } else {
+        schema
+    };
+
+    if resolved
+        .get("properties")
+        .and_then(|p| p.get("type"))
+        .and_then(|t| t.get("const"))
+        .and_then(|c| c.as_str())
+        == Some(expected)
+    {
+        return true;
+    }
+
+    // Walk allOf in case the discriminant is embedded there.
+    if let Some(all_of) = resolved.get("allOf").and_then(|v| v.as_array())
+        && all_of
+            .iter()
+            .any(|sub| schema_matches_type_discriminant(sub, definitions, expected))
+    {
+        return true;
+    }
+
+    false
+}
+
+/// Returns true if `schema`, after fully resolving `$ref`, `allOf`, and `oneOf`, contains
+/// any variant that claims `expected` as its `type` discriminant. Used to skip the
+/// unevaluatedProperties unknown-field check for components whose type is not compiled in.
+fn schema_contains_type_discriminant(
+    schema: &Value,
+    definitions: Option<&Value>,
+    expected: &str,
+) -> bool {
+    let resolved = if let Some(ref_str) = schema.get("$ref").and_then(|r| r.as_str()) {
+        let key = ref_str.strip_prefix("#/definitions/").unwrap_or("");
+        match definitions.and_then(|d| d.get(key)) {
+            Some(def) => def,
+            None => return false,
+        }
+    } else {
+        schema
+    };
+
+    if schema_matches_type_discriminant(resolved, definitions, expected) {
+        return true;
+    }
+
+    if let Some(all_of) = resolved.get("allOf").and_then(|v| v.as_array())
+        && all_of
+            .iter()
+            .any(|sub| schema_contains_type_discriminant(sub, definitions, expected))
+    {
+        return true;
+    }
+
+    if let Some(one_of) = resolved.get("oneOf").and_then(|v| v.as_array())
+        && one_of
+            .iter()
+            .any(|variant| schema_matches_type_discriminant(variant, definitions, expected))
+    {
+        return true;
+    }
+
+    false
+}
+
+/// Apply `anyOf` semantics: at least one schema must match.
+fn coerce_any_of(
+    value: &mut Value,
+    schemas: &[Value],
+    definitions: Option<&Value>,
+    path_components: &mut Vec<String>,
+) -> Result<(), Error> {
+    let initial_len = path_components.len();
+
+    for schema in schemas {
+        path_components.truncate(initial_len);
+        let mut candidate = value.clone();
+        if coerce(&mut candidate, schema, definitions, path_components).is_ok() {
+            path_components.truncate(initial_len);
+            *value = candidate;
+            return Ok(());
+        }
+    }
+
+    path_components.truncate(initial_len);
+    Ok(())
+}
+
+/// Parse a string into a serde_json Number (integer only). Returns None if not a valid integer.
+fn parse_integer(input: &str) -> Option<Number> {
+    if let Ok(i) = input.trim().parse::<i64>() {
+        Some(Number::from(i))
+    } else if let Ok(u) = input.trim().parse::<u64>() {
+        Some(Number::from(u))
+    } else {
+        None
+    }
+}
+
+/// Parse a string into a serde_json Number (integer or float). Returns None if not a valid number.
+fn parse_number(input: &str) -> Option<Number> {
+    if let Some(num) = parse_integer(input) {
+        return Some(num);
+    }
+    if let Ok(f) = input.trim().parse::<f64>() {
+        return Number::from_f64(f);
+    }
+    None
+}
+
+fn coerce_bool(value: &mut Value, path_components: &mut [String]) -> Result<(), Error> {
+    match value {
+        Value::Bool(_) => Ok(()),
+        Value::String(s) => match s.trim().parse::<bool>() {
+            Ok(b) => {
+                *value = Value::Bool(b);
+                Ok(())
+            }
+            _ => fail_expected!(Bool, value, path_components),
+        },
+        _ => fail_expected!(Bool, value, path_components),
+    }
+}
+
+fn coerce_integer(value: &mut Value, path_components: &mut [String]) -> Result<(), Error> {
+    if let Value::Number(n) = value {
+        if n.is_i64() || n.is_u64() {
+            return Ok(());
+        }
+
+        if let Some(f) = n.as_f64()
+            && f.fract() == 0.0
+            && f >= (i64::MIN as f64)
+            && f <= (i64::MAX as f64)
+        {
+            *value = Value::Number(Number::from(f as i64));
+            return Ok(());
+        }
+    } else if let Value::String(s) = value
+        && let Some(n) = parse_integer(s)
+    {
+        *value = Value::Number(n);
+        return Ok(());
+    }
+
+    fail_expected!(Integer, value, path_components)
+}
+
+fn coerce_number(value: &mut Value, path_components: &mut [String]) -> Result<(), Error> {
+    if let Value::Number(_) = value {
+        return Ok(());
+    }
+
+    if let Value::String(s) = value
+        && let Some(n) = parse_number(s)
+    {
+        *value = Value::Number(n);
+        return Ok(());
+    }
+
+    fail_expected!(Number, value, path_components)
+}
+
+fn coerce_null(value: &mut Value, path_components: &mut [String]) -> Result<(), Error> {
+    match value {
+        Value::Null => Ok(()),
+        Value::String(s) if s.trim().eq_ignore_ascii_case("null") => {
+            *value = Value::Null;
+            Ok(())
+        }
+        _ => fail_expected!(Null, value, path_components),
+    }
+}
+
+fn coerce_string(value: &mut Value, path_components: &mut [String]) -> Result<(), Error> {
+    match value {
+        Value::String(_) => Ok(()),
+        Value::Null => fail_expected!(String, value, path_components),
+        _ => {
+            *value = Value::String(value.to_string());
+            Ok(())
+        }
+    }
+}
+
+/// Helper to get a human-readable type name for a JSON value.
+const fn get_json_type(val: &Value) -> &'static str {
+    match val {
+        Value::Null => NULL_JSON_TYPE,
+        Value::Bool(_) => BOOL_JSON_TYPE,
+        Value::Number(_) => NUMBER_JSON_TYPE,
+        Value::String(_) => STRING_JSON_TYPE,
+        Value::Array(_) => ARRAY_JSON_TYPE,
+        Value::Object(_) => OBJECT_JSON_TYPE,
+    }
+}
+
+#[cfg(all(test, feature = "sources-demo_logs",))]
+mod test {
+    use crate::config::ConfigBuilder;
+    use crate::config::loading::schema_coercion::coerce;
+    use serde_json::json;
+    use vector_config::schema::generate_root_schema;
+
+    #[test]
+    fn test_coercion_with_array_support() {
+        let mut input = json!({
+            "proxy": {
+                "enabled": true,
+                "http": "http://example.com",
+                "https": "https://example.com",
+                "no_proxy": "no-proxy.com"
+            },
+            "enrichment_tables": {
+                "memory_table": {
+                    "type": "memory",
+                    "ttl": 60,
+                    "flush_interval": 5,
+                    "inputs": ["s0"],
+                },
+            },
+            "secret": {
+                "backend_1": {
+                    "type": "file",
+                    "path": "some.json",
+                },
+            },
+            "sources": {
+                "source0": {
+                    "type": "demo_logs",
+                    "count": "100",
+                    "format": "shuffle",
+                    "lines": ["777", true, false, 0.1, 123, "some string"],
+                    "interval": "1",
+                },
+            },
+            "transforms": {
+                "t0": {
+                    "inputs": ["s0"],
+                    "type": "remap",
+                    "source": ".host = \"${HOSTNAME}\""
+                },
+            },
+            "sinks": {
+                "sink0": {
+                    "inputs": ["t0"],
+                    "type": "console",
+                    "encoding": {
+                        "codec": "json",
+                    },
+                },
+            },
+        });
+
+        let demo_logs_schema =
+            serde_json::to_value(generate_root_schema::<ConfigBuilder>().unwrap()).unwrap();
+        coerce(
+            &mut input,
+            &demo_logs_schema,
+            demo_logs_schema.get("definitions"),
+            &mut Vec::new(),
+        )
+        .unwrap();
+
+        assert_eq!(
+            input,
+            json!({
+              "proxy": {
+                "enabled": true,
+                "http": "http://example.com",
+                "https": "https://example.com",
+                "no_proxy": ["no-proxy.com"]
+              },
+              "enrichment_tables": {
+                "memory_table": {
+                  "type": "memory",
+                  "ttl": 60,
+                  "flush_interval": 5,
+                  "inputs": [
+                    "s0"
+                  ]
+                }
+              },
+              "secret": {
+                "backend_1": {
+                  "type": "file",
+                  "path": "some.json"
+                }
+              },
+              "sources": {
+                "source0": {
+                  "type": "demo_logs",
+                  "count": 100,
+                  "format": "shuffle",
+                  "lines": [
+                    "777",
+                    "true",
+                    "false",
+                    "0.1",
+                    "123",
+                    "some string"
+                  ],
+                  "interval": 1
+                }
+              },
+              "transforms": {
+                "t0": {
+                  "inputs": [
+                    "s0"
+                  ],
+                  "type": "remap",
+                  "source": ".host = \"${HOSTNAME}\""
+                }
+              },
+              "sinks": {
+                "sink0": {
+                  "inputs": [
+                    "t0"
+                  ],
+                  "type": "console",
+                  "encoding": {
+                    "codec": "json"
+                  }
+                }
+              }
+            })
+        );
+    }
+
+    #[test]
+    fn test_unknown_field_in_known_component_passes_through() {
+        // Unknown fields are intentionally non-fatal in the coercion pass while
+        // `vector-config` does not emit `#[serde(alias = ...)]` aliases. The
+        // pass logs a warning and defers to serde, which has alias info.
+        let mut input = json!({
+            "sources": {
+                "source0": {
+                    "type": "demo_logs",
+                    "count": 100,
+                    "totally_unknown_field": "oops",
+                }
+            }
+        });
+
+        let schema =
+            serde_json::to_value(generate_root_schema::<ConfigBuilder>().unwrap()).unwrap();
+        let result = coerce(
+            &mut input,
+            &schema,
+            schema.get("definitions"),
+            &mut Vec::new(),
+        );
+
+        assert!(
+            result.is_ok(),
+            "unknown field should pass coercion (serde validates downstream), got: {result:?}"
+        );
+    }
+
+    #[test]
+    fn test_unknown_component_type_passes_through() {
+        let mut input = json!({
+            "sinks": {
+                "s3_sink": {
+                    "type": "aws_s3_totally_nonexistent",
+                    "bucket": "my-bucket",
+                }
+            }
+        });
+
+        let schema =
+            serde_json::to_value(generate_root_schema::<ConfigBuilder>().unwrap()).unwrap();
+        let result = coerce(
+            &mut input,
+            &schema,
+            schema.get("definitions"),
+            &mut Vec::new(),
+        );
+
+        assert!(
+            result.is_ok(),
+            "unknown component type should pass through coercion, got: {result:?}"
+        );
+    }
+}
diff --git a/src/config/loading/secret.rs b/src/config/loading/secret.rs
index 56e96fcea8e33..8afcdc3ae72fe 100644
--- a/src/config/loading/secret.rs
+++ b/src/config/loading/secret.rs
@@ -1,6 +1,5 @@
 use std::{
     collections::{HashMap, HashSet},
-    io::Read,
     sync::LazyLock,
 };
 
@@ -8,14 +7,15 @@ use futures::TryFutureExt;
 use indexmap::IndexMap;
 use regex::{Captures, Regex};
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
 use vector_lib::config::ComponentKey;
 
 use crate::{
     config::{
         SecretBackend,
         loading::{
-            ComponentHint, Loader, deserialize_config_map, prepare_input, process::Process,
-            representation::ConfigMap,
+            ComponentHint, Loader, deserialize_config_map, interpolate_config_map,
+            process::Process, representation::ConfigMap,
         },
     },
     secrets::SecretBackends,
@@ -34,6 +34,8 @@ use crate::{
 pub static COLLECTOR: LazyLock<Regex> =
     LazyLock::new(|| Regex::new(r"SECRET\[([[:word:]\-]+)\.([[:word:].\-/]+)\]").unwrap());
 
+const SECRET_KEY: &str = "secret";
+
 /// Helper type for specifically deserializing secrets backends.
 #[derive(Debug, Default, Deserialize, Serialize)]
 pub(crate) struct SecretBackendOuter {
@@ -45,7 +47,8 @@ pub(crate) struct SecretBackendOuter {
 #[derive(Debug, Deserialize, Serialize)]
 pub struct SecretBackendLoader {
     backends: IndexMap<ComponentKey, SecretBackends>,
-    secret_keys: HashMap<String, HashSet<String>>,
+    pub(crate) secret_keys: HashMap<String, HashSet<String>>,
+    #[serde(skip)]
     interpolate_env: bool,
 }
 
@@ -56,20 +59,11 @@ impl SecretBackendLoader {
         self
     }
 
-    /// Retrieve secrets from backends.
-    /// Returns an empty HashMap if there are no secrets to retrieve.
-    pub(crate) async fn retrieve_secrets(
-        mut self,
-        signal_handler: &mut signal::SignalHandler,
+    pub(crate) async fn retrieve(
+        &mut self,
+        signal_rx: &mut signal::SignalRx,
     ) -> Result<HashMap<String, String>, String> {
-        if self.secret_keys.is_empty() {
-            debug!(message = "No secret placeholder found, skipping secret resolution.");
-            return Ok(HashMap::new());
-        }
-
-        debug!(message = "Secret placeholders found, retrieving secrets from configured backends.");
         let mut secrets: HashMap<String, String> = HashMap::new();
-        let mut signal_rx = signal_handler.subscribe();
 
         for (backend_name, keys) in &self.secret_keys {
             let backend = self
@@ -83,7 +77,7 @@ impl SecretBackendLoader {
 
             debug!(message = "Retrieving secrets from a backend.", backend = ?backend_name, keys = ?keys);
             let backend_secrets = backend
-                .retrieve(keys.clone(), &mut signal_rx)
+                .retrieve(keys.clone(), signal_rx)
                 .map_err(|e| {
                     format!("Error while retrieving secret from backend \"{backend_name}\": {e}.")
                 })
@@ -97,6 +91,10 @@ impl SecretBackendLoader {
 
         Ok(secrets)
     }
+
+    pub(crate) fn has_secrets_to_retrieve(&self) -> bool {
+        !self.secret_keys.is_empty()
+    }
 }
 
 impl Default for SecretBackendLoader {
@@ -110,16 +108,20 @@ impl Default for SecretBackendLoader {
 }
 
 impl Process for SecretBackendLoader {
-    fn prepare<R: Read>(&mut self, input: R) -> Result<String, Vec<String>> {
-        let config_string = prepare_input(input, self.interpolate_env)?;
-        // Collect secret placeholders just after env var processing
-        collect_secret_keys(&config_string, &mut self.secret_keys);
-        Ok(config_string)
+    fn should_interpolate_env(&self) -> bool {
+        self.interpolate_env
+    }
+
+    fn postprocess(&mut self, map: ConfigMap) -> Result<ConfigMap, Vec<String>> {
+        collect_secret_keys_from_map(&map, &mut self.secret_keys);
+        Ok(map)
     }
 
     fn merge(&mut self, map: ConfigMap, _: Option<ComponentHint>) -> Result<(), Vec<String>> {
-        if map.contains_key("secret") {
-            let additional = deserialize_config_map::<SecretBackendOuter>(map)?;
+        if let Some(secret_value) = map.get(SECRET_KEY) {
+            let mut secret_map = ConfigMap::new();
+            secret_map.insert(SECRET_KEY.to_string(), secret_value.clone());
+            let additional = deserialize_config_map::<SecretBackendOuter>(secret_map)?;
             self.backends.extend(additional.secret);
         }
         Ok(())
@@ -147,7 +149,38 @@ fn collect_secret_keys(input: &str, keys: &mut HashMap<String, HashSet<String>>)
     });
 }
 
-pub fn interpolate(input: &str, secrets: &HashMap<String, String>) -> Result<String, Vec<String>> {
+/// Recursively collects secret references from object keys and string leaves.
+pub fn collect_secret_keys_from_map(map: &ConfigMap, keys: &mut HashMap<String, HashSet<String>>) {
+    for (key, value) in map {
+        collect_secret_keys(key, keys);
+        collect_secret_keys_from_value(value, keys);
+    }
+}
+
+fn collect_secret_keys_from_value(value: &Value, keys: &mut HashMap<String, HashSet<String>>) {
+    match value {
+        Value::String(value) => collect_secret_keys(value, keys),
+        Value::Array(values) => {
+            for value in values {
+                collect_secret_keys_from_value(value, keys);
+            }
+        }
+        Value::Object(map) => collect_secret_keys_from_map(map, keys),
+        _ => {}
+    }
+}
+
+pub fn interpolate_config_map_with_secrets(
+    map: &ConfigMap,
+    secrets: &HashMap<String, String>,
+) -> Result<ConfigMap, Vec<String>> {
+    interpolate_config_map(map, secrets, interpolate_secrets)
+}
+
+fn interpolate_secrets(
+    input: &str,
+    secrets: &HashMap<String, String>,
+) -> Result<String, Vec<String>> {
     let mut errors = Vec::<String>::new();
     let output = COLLECTOR
         .replace_all(input, |caps: &Captures<'_>| {
@@ -177,7 +210,7 @@ mod tests {
 
     use indoc::indoc;
 
-    use super::{collect_secret_keys, interpolate};
+    use super::{collect_secret_keys, interpolate_secrets as interpolate};
 
     #[test]
     fn replacement() {
diff --git a/src/config/loading/source.rs b/src/config/loading/source.rs
index ec56d67ab83f7..9241c535232ba 100644
--- a/src/config/loading/source.rs
+++ b/src/config/loading/source.rs
@@ -1,5 +1,3 @@
-use std::io::Read;
-
 use super::{
     ComponentHint, Loader, Process,
     representation::{ConfigMap, merge_into_map},
@@ -24,15 +22,12 @@ impl Default for SourceLoader {
 }
 
 impl Process for SourceLoader {
-    /// Prepares input by simply reading bytes to a string. Unlike other loaders, there's no
-    /// interpolation of environment variables. This is on purpose to preserve the original config.
-    fn prepare<R: Read>(&mut self, mut input: R) -> Result<String, Vec<String>> {
-        let mut source_string = String::new();
-        input
-            .read_to_string(&mut source_string)
-            .map_err(|e| vec![e.to_string()])?;
+    fn should_interpolate_env(&self) -> bool {
+        false
+    }
 
-        Ok(source_string)
+    fn postprocess(&mut self, map: ConfigMap) -> Result<ConfigMap, Vec<String>> {
+        Ok(map)
     }
 
     /// Merge values by combining with the internal configuration map.
diff --git a/src/config/mod.rs b/src/config/mod.rs
index 8e4707b9b1a2b..4e523c8905e15 100644
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@@ -44,7 +44,6 @@ mod source;
 mod transform;
 pub mod unit_test;
 mod validation;
-mod vars;
 pub mod watcher;
 
 pub use builder::ConfigBuilder;
@@ -52,7 +51,8 @@ pub use diff::ConfigDiff;
 pub use enrichment_table::{EnrichmentTableConfig, EnrichmentTableOuter};
 pub use format::{Format, FormatHint};
 pub use loading::{
-    COLLECTOR, CONFIG_PATHS, env_var_interpolation_enabled, load, load_from_paths,
+    COLLECTOR, CONFIG_PATHS, ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX,
+    env_var_interpolation_enabled, interpolate, load, load_builder_from_paths, load_from_paths,
     load_from_paths_with_provider_and_secrets, load_from_str, load_from_str_with_secrets,
     load_source_from_paths, merge_path_lists, process_paths, set_env_var_interpolation,
 };
@@ -65,7 +65,6 @@ pub use transform::{
 };
 pub use unit_test::{UnitTestResult, build_unit_tests, build_unit_tests_main};
 pub use validation::warnings;
-pub use vars::{ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX, interpolate};
 pub use vector_lib::{
     config::{
         ComponentKey, LogSchema, OutputId, init_log_schema, init_telemetry, log_schema,
diff --git a/src/config/unit_test/mod.rs b/src/config/unit_test/mod.rs
index 397fc3278a40f..febe0226c870b 100644
--- a/src/config/unit_test/mod.rs
+++ b/src/config/unit_test/mod.rs
@@ -38,7 +38,7 @@ use crate::{
     conditions::Condition,
     config::{
         self, ComponentKey, Config, ConfigBuilder, ConfigPath, SinkOuter, SourceOuter,
-        TestDefinition, TestInput, TestOutput, loading, loading::ConfigBuilderLoader,
+        TestDefinition, TestInput, TestOutput, loading,
     },
     event::{Event, EventMetadata, LogEvent},
     signal,
@@ -93,7 +93,7 @@ fn init_log_schema_from_paths(
     config_paths: &[ConfigPath],
     deny_if_set: bool,
 ) -> Result<(), Vec<String>> {
-    let builder = ConfigBuilderLoader::default().load_from_paths(config_paths)?;
+    let builder = config::loading::load_builder_from_paths(config_paths)?;
     vector_lib::config::init_log_schema(builder.global.log_schema, deny_if_set);
     Ok(())
 }
@@ -103,16 +103,16 @@ pub async fn build_unit_tests_main(
     signal_handler: &mut signal::SignalHandler,
 ) -> Result<Vec<UnitTest>, Vec<String>> {
     init_log_schema_from_paths(paths, false)?;
-    let secrets_backends_loader =
-        loading::loader_from_paths(loading::SecretBackendLoader::default(), paths)?;
-    let secrets = secrets_backends_loader
-        .retrieve_secrets(signal_handler)
-        .await
-        .map_err(|e| vec![e])?;
-
-    let config_builder = ConfigBuilderLoader::default()
-        .secrets(secrets)
-        .load_from_paths(paths)?;
+    let mut secrets_backends_loader = loading::load_secret_backends_from_paths(paths)?;
+    let config_builder = if secrets_backends_loader.has_secrets_to_retrieve() {
+        let resolved_secrets = secrets_backends_loader
+            .retrieve(&mut signal_handler.subscribe())
+            .await
+            .map_err(|e| vec![e])?;
+        loading::load_builder_from_paths_with_secrets(paths, resolved_secrets)?
+    } else {
+        loading::load_builder_from_paths(paths)?
+    };
 
     build_unit_tests(config_builder).await
 }
diff --git a/src/config/vars.rs b/src/config/vars.rs
deleted file mode 100644
index 11cbbb885e88d..0000000000000
--- a/src/config/vars.rs
+++ /dev/null
@@ -1,171 +0,0 @@
-use std::{collections::HashMap, sync::LazyLock};
-
-use regex::{Captures, Regex};
-
-// Environment variable names can have any characters from the Portable Character Set other
-// than NUL.  However, for Vector's interpolation, we are closer to what a shell supports which
-// is solely of uppercase letters, digits, and the '_' (that is, the `[:word:]` regex class).
-// In addition to these characters, we allow `.` as this commonly appears in environment
-// variable names when they come from a Java properties file.
-//
-// https://pubs.opengroup.org/onlinepubs/000095399/basedefs/xbd_chap08.html
-pub static ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX: LazyLock<Regex> = LazyLock::new(|| {
-    Regex::new(
-        r"(?x)
-        \$\$|
-        \$([[:word:].]+)|
-        \$\{([[:word:].]+)(?:(:?-|:?\?)([^}]*))?\}",
-    )
-    .unwrap()
-});
-
-/// Result<interpolated config, errors>
-pub fn interpolate(input: &str, vars: &HashMap<String, String>) -> Result<String, Vec<String>> {
-    let mut errors = Vec::new();
-
-    let interpolated = ENVIRONMENT_VARIABLE_INTERPOLATION_REGEX
-        .replace_all(input, |caps: &Captures<'_>| {
-            let flags = caps.get(3).map(|m| m.as_str()).unwrap_or_default();
-            let def_or_err = caps.get(4).map(|m| m.as_str()).unwrap_or_default();
-            caps.get(1)
-                .or_else(|| caps.get(2))
-                .map(|m| m.as_str())
-                .map(|name| {
-                    // Get the value and check for newlines (LF or CR)
-                    let val = vars.get(name).and_then(|v| {
-                        if v.contains(['\n', '\r']) {
-                            errors.push(format!(
-                                "Environment variable contains newline character. name = {name:?}",
-                            ));
-                            None
-                        } else {
-                            Some(v.as_str())
-                        }
-                    });
-
-                    match flags {
-                        ":-" => match val {
-                            Some(v) if !v.is_empty() => v,
-                            _ => def_or_err,
-                        },
-                        "-" => val.unwrap_or(def_or_err),
-                        ":?" => match val {
-                            Some(v) if !v.is_empty() => v,
-                            _ => {
-                                errors.push(format!(
-                                    "Non-empty environment variable required in config. name = {name:?}, error = {def_or_err:?}",
-                                ));
-                                ""
-                            },
-                        }
-                        "?" => val.unwrap_or_else(|| {
-                            errors.push(format!(
-                                "Missing environment variable required in config. name = {name:?}, error = {def_or_err:?}",
-                            ));
-                            ""
-                        }),
-                        _ => val.unwrap_or_else(|| {
-                            errors.push(format!(
-                                "Missing environment variable in config. name = {name:?}",
-                            ));
-                            ""
-                        }),
-                    }
-                })
-                .unwrap_or("$")
-                .to_string()
-        })
-        .into_owned();
-
-    if errors.is_empty() {
-        Ok(interpolated)
-    } else {
-        Err(errors)
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::interpolate;
-    #[test]
-    fn interpolation() {
-        let vars = vec![
-            ("FOO".into(), "dogs".into()),
-            ("FOOBAR".into(), "cats".into()),
-            // Java commonly uses .s in env var names
-            ("FOO.BAR".into(), "turtles".into()),
-            ("EMPTY".into(), "".into()),
-        ]
-        .into_iter()
-        .collect();
-
-        assert_eq!("dogs", interpolate("$FOO", &vars).unwrap());
-        assert_eq!("dogs", interpolate("${FOO}", &vars).unwrap());
-        assert_eq!("cats", interpolate("${FOOBAR}", &vars).unwrap());
-        assert_eq!("xcatsy", interpolate("x${FOOBAR}y", &vars).unwrap());
-        assert!(interpolate("x$FOOBARy", &vars).is_err());
-        assert_eq!("$ x", interpolate("$ x", &vars).unwrap());
-        assert_eq!("$FOO", interpolate("$$FOO", &vars).unwrap());
-        assert_eq!("dogs=bar", interpolate("$FOO=bar", &vars).unwrap());
-        assert!(interpolate("$NOT_FOO", &vars).is_err());
-        assert!(interpolate("$NOT-FOO", &vars).is_err());
-        assert_eq!("turtles", interpolate("$FOO.BAR", &vars).unwrap());
-        assert_eq!("${FOO x", interpolate("${FOO x", &vars).unwrap());
-        assert_eq!("${}", interpolate("${}", &vars).unwrap());
-        assert_eq!("dogs", interpolate("${FOO:-cats}", &vars).unwrap());
-        assert_eq!("dogcats", interpolate("${NOT:-dogcats}", &vars).unwrap());
-        assert_eq!(
-            "dogs and cats",
-            interpolate("${NOT:-dogs and cats}", &vars).unwrap()
-        );
-        assert_eq!("${:-cats}", interpolate("${:-cats}", &vars).unwrap());
-        assert_eq!("", interpolate("${NOT:-}", &vars).unwrap());
-        assert_eq!("cats", interpolate("${NOT-cats}", &vars).unwrap());
-        assert_eq!("", interpolate("${EMPTY-cats}", &vars).unwrap());
-        assert_eq!("dogs", interpolate("${FOO:?error cats}", &vars).unwrap());
-        assert_eq!("dogs", interpolate("${FOO?error cats}", &vars).unwrap());
-        assert_eq!("", interpolate("${EMPTY?error cats}", &vars).unwrap());
-        assert!(interpolate("${NOT:?error cats}", &vars).is_err());
-        assert!(interpolate("${NOT?error cats}", &vars).is_err());
-        assert!(interpolate("${EMPTY:?error cats}", &vars).is_err());
-    }
-
-    #[test]
-    fn test_multiline_expansion_prevented() {
-        let vars = vec![
-            ("SAFE_VAR".into(), "single line value".into()),
-            ("MULTILINE_VAR".into(), "line1\nline2\nline3".into()),
-            ("WITH_NEWLINE".into(), "before\nafter".into()),
-            ("WITH_CR".into(), "before\rafter".into()),
-            ("WITH_CRLF".into(), "before\r\nafter".into()),
-        ]
-        .into_iter()
-        .collect();
-
-        // Test that multiline values are treated as missing
-        let result = interpolate("$MULTILINE_VAR", &vars);
-        assert!(result.is_err(), "Multiline var should be rejected");
-
-        let result = interpolate("$WITH_NEWLINE", &vars);
-        assert!(result.is_err(), "Newline var should be rejected");
-
-        let result = interpolate("$WITH_CR", &vars);
-        assert!(result.is_err(), "CR var should be rejected");
-
-        let result = interpolate("$WITH_CRLF", &vars);
-        assert!(result.is_err(), "CRLF var should be rejected");
-
-        // Test that safe values still work
-        let result = interpolate("$SAFE_VAR", &vars).unwrap();
-        assert_eq!("single line value", result);
-
-        // Test with default values - multiline vars should still error
-        let result = interpolate("${MULTILINE_VAR:-safe default}", &vars);
-        assert!(result.is_err(), "Should error even with default");
-
-        // Verify error messages are helpful
-        let err = interpolate("$MULTILINE_VAR", &vars).unwrap_err();
-        assert!(err.iter().any(|e| e.contains("newline character")));
-        assert!(err.iter().any(|e| e.contains("MULTILINE_VAR")));
-    }
-}
diff --git a/src/providers/http.rs b/src/providers/http.rs
index 5555a99e0e98d..63554bc4b49dc 100644
--- a/src/providers/http.rs
+++ b/src/providers/http.rs
@@ -10,7 +10,7 @@ use vector_lib::configurable::configurable_component;
 
 use super::BuildResult;
 use crate::{
-    config::{self, Format, ProxyConfig, interpolate, provider::ProviderConfig},
+    config::{Format, ProxyConfig, loading::ConfigBuilderLoader, provider::ProviderConfig},
     http::HttpClient,
     signal,
     tls::{TlsConfig, TlsSettings},
@@ -146,25 +146,9 @@ async fn http_request_to_config_builder(
         .await
         .map_err(|e| vec![e.to_owned()])?;
 
-    if !interpolate_env {
-        return config::load(config_str.chunk(), *config_format);
-    }
-
-    let env_vars = std::env::vars_os()
-        .map(|(k, v)| {
-            (
-                k.as_os_str().to_string_lossy().to_string(),
-                v.as_os_str().to_string_lossy().to_string(),
-            )
-        })
-        .collect::<std::collections::HashMap<String, String>>();
-
-    let config_str = interpolate(
-        std::str::from_utf8(&config_str).map_err(|e| vec![e.to_string()])?,
-        &env_vars,
-    )?;
-
-    config::load(config_str.as_bytes().chunk(), *config_format)
+    ConfigBuilderLoader::default()
+        .interpolate_env(interpolate_env)
+        .load_from_input(config_str.chunk(), *config_format)
 }
 
 /// Polls the HTTP endpoint after/every `poll_interval_secs`, returning a stream of `ConfigBuilder`.
diff --git a/src/secrets/mod.rs b/src/secrets/mod.rs
index 6c8a5c6d3cf43..b82d30ec4db87 100644
--- a/src/secrets/mod.rs
+++ b/src/secrets/mod.rs
@@ -4,10 +4,8 @@ use std::collections::{HashMap, HashSet};
 use enum_dispatch::enum_dispatch;
 use vector_lib::configurable::configurable_component;
 
-use crate::{
-    config::{GenerateConfig, SecretBackend},
-    signal,
-};
+use crate::config::GenerateConfig;
+use crate::{config::SecretBackend, signal};
 
 #[cfg(feature = "secrets-aws-secrets-manager")]
 mod aws_secrets_manager;
@@ -51,7 +49,7 @@ mod test;
 /// Secrets are loaded when Vector starts or if Vector receives a `SIGHUP` signal triggering its
 /// configuration reload process.
 #[allow(clippy::large_enum_variant)]
-#[configurable_component]
+#[configurable_component(global_option("secret"))]
 #[derive(Clone, Debug)]
 #[enum_dispatch(SecretBackend)]
 #[serde(tag = "type", rename_all = "snake_case")]
@@ -79,20 +77,6 @@ pub enum SecretBackends {
     Test(test::TestBackend),
 }
 
-// Manual NamedComponent impl required because enum_dispatch doesn't support it yet.
-impl vector_lib::configurable::NamedComponent for SecretBackends {
-    fn get_component_name(&self) -> &'static str {
-        match self {
-            Self::File(config) => config.get_component_name(),
-            Self::Directory(config) => config.get_component_name(),
-            Self::Exec(config) => config.get_component_name(),
-            #[cfg(feature = "secrets-aws-secrets-manager")]
-            Self::AwsSecretsManager(config) => config.get_component_name(),
-            Self::Test(config) => config.get_component_name(),
-        }
-    }
-}
-
 impl GenerateConfig for SecretBackends {
     fn generate_config() -> serde_json::Value {
         serde_json::to_value(Self::File(file::FileBackend {
@@ -101,3 +85,20 @@ impl GenerateConfig for SecretBackends {
         .unwrap()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::secrets::SecretBackends;
+
+    #[test]
+    fn interpolation() {
+        let ok = serde_json::from_value::<SecretBackends>(serde_json::json!({
+            "type": "file",
+            "path": "/tmp/foo"
+        }));
+        assert!(
+            ok.is_ok(),
+            "expected flat SecretBackends to parse, got: {ok:?}"
+        );
+    }
+}