diff --git a/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapper.java b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapper.java
new file mode 100644
index 000000000..18f1488a0
--- /dev/null
+++ b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapper.java
@@ -0,0 +1,92 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+/** JSON-backed source wrapper for a single-policy object. */
+public final class JsonSourceWrapper implements SourceWrapper {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private final JsonNode source;
+
+  public JsonSourceWrapper(JsonNode source) {
+    this.source = Objects.requireNonNull(source, "source cannot be null");
+  }
+
+  @Override
+  public SourceFormat getFormat() {
+    return SourceFormat.JSON;
+  }
+
+  @Override
+  @Nullable
+  public String getPolicyType() {
+    JsonNode node = asJsonNode();
+    if (!node.isObject() || node.size() != 1) {
+      return null;
+    }
+    return node.fieldNames().next();
+  }
+
+  public JsonNode asJsonNode() {
+    return source;
+  }
+
+  /**
+   * Parses JSON source into one wrapper per top-level policy object.
+   *
+   * @return an empty list if the source is an empty JSON array; a non-empty list of wrappers if the
+   *     source is a valid single-policy object or array thereof; or {@code null} if the shape is
+   *     unsupported or the source is not valid JSON.
+   * @throws NullPointerException if source is null
+   */
+  @Nullable
+  public static List<SourceWrapper> parse(String source) {
+    Objects.requireNonNull(source, "source cannot be null");
+    try {
+      JsonNode parsed = MAPPER.readTree(source);
+      if (!isSupportedJsonShape(parsed)) {
+        return null;
+      }
+      if (parsed.isObject()) {
+        return Collections.singletonList(new JsonSourceWrapper(parsed));
+      }
+      List<SourceWrapper> wrappers = new ArrayList<>();
+      for (JsonNode element : parsed) {
+        wrappers.add(new JsonSourceWrapper(element));
+      }
+      return Collections.unmodifiableList(wrappers);
+    } catch (JsonProcessingException e) {
+      return null;
+    }
+  }
+
+  private static boolean isSupportedJsonShape(JsonNode node) {
+    if (node.isObject()) {
+      return isSinglePolicyObject(node);
+    }
+    if (!node.isArray()) {
+      return false;
+    }
+    for (JsonNode element : node) {
+      if (!isSinglePolicyObject(element)) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private static boolean isSinglePolicyObject(JsonNode node) {
+    return node.isObject() && node.size() == 1;
+  }
+}
diff --git a/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapper.java b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapper.java
new file mode 100644
index 000000000..cceff8060
--- /dev/null
+++ b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapper.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+
+/** KEYVALUE-backed source wrapper for a single key/value policy entry. */
+public final class KeyValueSourceWrapper implements SourceWrapper {
+  private final String key;
+  private final String value;
+
+  private static final Pattern LINE_SEPARATOR = Pattern.compile("\\R");
+
+  public KeyValueSourceWrapper(String key, String value) {
+    this.key = Objects.requireNonNull(key, "key cannot be null");
+    this.value = Objects.requireNonNull(value, "value cannot be null");
+  }
+
+  @Override
+  public SourceFormat getFormat() {
+    return SourceFormat.KEYVALUE;
+  }
+
+  @Override
+  @Nullable
+  public String getPolicyType() {
+    return key;
+  }
+
+  public String getKey() {
+    return key;
+  }
+
+  public String getValue() {
+    return value;
+  }
+
+  /**
+   * Parses KEYVALUE source into one wrapper per non-empty line.
+   *
+   * @return an empty list if the source contains no non-blank lines; a non-empty list of wrappers
+   *     if all non-blank lines are valid key=value pairs; or {@code null} if any line is malformed.
+   * @throws NullPointerException if source is null
+   */
+  @Nullable
+  public static List<SourceWrapper> parse(String source) {
+    Objects.requireNonNull(source, "source cannot be null");
+    String[] lines = LINE_SEPARATOR.split(source, -1);
+    List<SourceWrapper> wrappers = new ArrayList<>();
+    for (String rawLine : lines) {
+      String trimmedLine = rawLine.trim();
+      if (trimmedLine.isEmpty()) {
+        continue;
+      }
+      KeyValueSourceWrapper wrapper = parseSingleKeyValue(trimmedLine);
+      if (wrapper == null) {
+        return null;
+      }
+      wrappers.add(wrapper);
+    }
+    if (wrappers.isEmpty()) {
+      return Collections.emptyList();
+    }
+    return Collections.unmodifiableList(wrappers);
+  }
+
+  @Nullable
+  private static KeyValueSourceWrapper parseSingleKeyValue(String line) {
+    int separatorIndex = line.indexOf('=');
+    if (separatorIndex <= 0) {
+      return null;
+    }
+    String key = line.substring(0, separatorIndex).trim();
+    String value = line.substring(separatorIndex + 1).trim();
+    if (key.isEmpty()) {
+      return null;
+    }
+    return new KeyValueSourceWrapper(key, value);
+  }
+}
diff --git a/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormat.java b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormat.java
new file mode 100644
index 000000000..0b2557026
--- /dev/null
+++ b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormat.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import com.google.errorprone.annotations.Immutable;
+import java.util.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+/** Supported source formats and their parser dispatch. */
+public enum SourceFormat {
+  KEYVALUE("keyvalue", KeyValueSourceWrapper::parse),
+  JSON("json", JsonSourceWrapper::parse);
+
+  private final String configValue;
+  private final SourceParser parser;
+
+  SourceFormat(String configValue, SourceParser parser) {
+    this.configValue = configValue;
+    this.parser = parser;
+  }
+
+  public String configValue() {
+    return configValue;
+  }
+
+  /**
+   * Parses source text into normalized wrappers for this format.
+   *
+   * @return an empty list if the source is valid but contains no policies; a non-empty list of
+   *     wrappers if one or more policies were parsed successfully; or {@code null} if the source is
+   *     malformed or does not conform to the expected shape for this format.
+   * @throws NullPointerException if source is null
+   */
+  @Nullable
+  public List<SourceWrapper> parse(String source) {
+    Objects.requireNonNull(source, "source cannot be null");
+    return parser.parse(source);
+  }
+
+  @Immutable
+  @FunctionalInterface
+  private interface SourceParser {
+    @Nullable
+    List<SourceWrapper> parse(String source);
+  }
+}
diff --git a/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceWrapper.java b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceWrapper.java
new file mode 100644
index 000000000..02e16668a
--- /dev/null
+++ b/dynamic-control/src/main/java/io/opentelemetry/contrib/dynamic/policy/source/SourceWrapper.java
@@ -0,0 +1,16 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import javax.annotation.Nullable;
+
+/** Parsed source payload paired with its source format. */
+public interface SourceWrapper {
+  SourceFormat getFormat();
+
+  @Nullable
+  String getPolicyType();
+}
diff --git a/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapperTest.java b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapperTest.java
new file mode 100644
index 000000000..9c9d26089
--- /dev/null
+++ b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/JsonSourceWrapperTest.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class JsonSourceWrapperTest {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+
+  @Test
+  void parseSupportsSingleObject() {
+    List<SourceWrapper> parsed = JsonSourceWrapper.parse("{\"trace-sampling\": 0.5}");
+
+    assertThat(parsed).hasSize(1);
+    assertThat(parsed.get(0)).isInstanceOf(JsonSourceWrapper.class);
+    assertThat(parsed.get(0).getPolicyType()).isEqualTo("trace-sampling");
+  }
+
+  @Test
+  void parseSupportsArrayOfObjects() {
+    List<SourceWrapper> parsed =
+        JsonSourceWrapper.parse("[{\"other-policy\": 1}, {\"trace-sampling\": 0.5}]");
+
+    assertThat(parsed).hasSize(2);
+    assertThat(parsed.get(0).getPolicyType()).isEqualTo("other-policy");
+    assertThat(parsed.get(1).getPolicyType()).isEqualTo("trace-sampling");
+  }
+
+  @Test
+  void parseSupportsEmptyArray() {
+    assertThat(JsonSourceWrapper.parse("[]")).isEmpty();
+  }
+
+  @Test
+  void parseArrayResultIsImmutable() {
+    List<SourceWrapper> parsed = JsonSourceWrapper.parse("[{\"trace-sampling\": 0.5}]");
+
+    assertThatThrownBy(() -> parsed.add(new JsonSourceWrapper(MAPPER.readTree("{\"x\":1}"))))
+        .isInstanceOf(UnsupportedOperationException.class);
+  }
+
+  @Test
+  void getPolicyTypeReturnsNullWhenObjectHasMultipleFields() throws Exception {
+    JsonSourceWrapper wrapper = new JsonSourceWrapper(MAPPER.readTree("{\"a\": 1, \"b\": 2}"));
+
+    assertThat(wrapper.getPolicyType()).isNull();
+  }
+
+  @Test
+  void parseRejectsUnsupportedJsonShapes() {
+    assertThat(JsonSourceWrapper.parse("{}")).isNull();
+    assertThat(JsonSourceWrapper.parse("{\"a\": 1, \"b\": 2}")).isNull();
+    assertThat(JsonSourceWrapper.parse("[1, 2, 3]")).isNull();
+    assertThat(JsonSourceWrapper.parse("[{\"trace-sampling\": 0.5}, {}]")).isNull();
+    assertThat(JsonSourceWrapper.parse("[{\"a\": 1, \"b\": 2}]")).isNull();
+    assertThat(JsonSourceWrapper.parse("\"text\"")).isNull();
+    assertThat(JsonSourceWrapper.parse("{invalid-json")).isNull();
+  }
+
+  @Test
+  void parseRejectsNullInput() {
+    assertThatThrownBy(() -> JsonSourceWrapper.parse(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("source cannot be null");
+  }
+}
diff --git a/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapperTest.java b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapperTest.java
new file mode 100644
index 000000000..d3bf32320
--- /dev/null
+++ b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/KeyValueSourceWrapperTest.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class KeyValueSourceWrapperTest {
+
+  @Test
+  void parseSupportsSingleKeyValue() {
+    List<SourceWrapper> parsed = KeyValueSourceWrapper.parse("trace-sampling=0.5");
+
+    assertThat(parsed).hasSize(1);
+    KeyValueSourceWrapper wrapper = (KeyValueSourceWrapper) parsed.get(0);
+    assertThat(wrapper.getFormat()).isEqualTo(SourceFormat.KEYVALUE);
+    assertThat(wrapper.getPolicyType()).isEqualTo("trace-sampling");
+    assertThat(wrapper.getKey()).isEqualTo("trace-sampling");
+    assertThat(wrapper.getValue()).isEqualTo("0.5");
+  }
+
+  @Test
+  void parseSupportsMultipleLinesAndSkipsBlanks() {
+    String source = "\ntrace-sampling=0.5\r\nother-policy=1.0\n   \n";
+
+    List<SourceWrapper> parsed = KeyValueSourceWrapper.parse(source);
+
+    assertThat(parsed).hasSize(2);
+    assertThat(parsed.get(0).getPolicyType()).isEqualTo("trace-sampling");
+    assertThat(parsed.get(1).getPolicyType()).isEqualTo("other-policy");
+  }
+
+  @Test
+  void parseSupportsEmptyInput() {
+    assertThat(KeyValueSourceWrapper.parse("")).isEmpty();
+    assertThat(KeyValueSourceWrapper.parse("\n   \r\n")).isEmpty();
+  }
+
+  @Test
+  void parseResultIsImmutable() {
+    List<SourceWrapper> parsed = KeyValueSourceWrapper.parse("trace-sampling=0.5");
+
+    assertThatThrownBy(() -> parsed.add(new KeyValueSourceWrapper("other-policy", "1.0")))
+        .isInstanceOf(UnsupportedOperationException.class);
+  }
+
+  @Test
+  void parseReturnsNullForInvalidInput() {
+    assertThat(KeyValueSourceWrapper.parse("missing-separator")).isNull();
+    assertThat(KeyValueSourceWrapper.parse("=value")).isNull();
+    assertThat(KeyValueSourceWrapper.parse("# comment\ntrace-sampling=0.5")).isNull();
+  }
+
+  @Test
+  void parseRejectsNullInput() {
+    assertThatThrownBy(() -> KeyValueSourceWrapper.parse(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("source cannot be null");
+  }
+
+  @Test
+  void parseTrimsKeyAndValue() {
+    List<SourceWrapper> parsed = KeyValueSourceWrapper.parse("  trace-sampling  = 0.25 ");
+
+    assertThat(parsed).hasSize(1);
+    KeyValueSourceWrapper wrapper = (KeyValueSourceWrapper) parsed.get(0);
+    assertThat(wrapper.getKey()).isEqualTo("trace-sampling");
+    assertThat(wrapper.getValue()).isEqualTo("0.25");
+  }
+}
diff --git a/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormatTest.java b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormatTest.java
new file mode 100644
index 000000000..3cdeb0323
--- /dev/null
+++ b/dynamic-control/src/test/java/io/opentelemetry/contrib/dynamic/policy/source/SourceFormatTest.java
@@ -0,0 +1,80 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.contrib.dynamic.policy.source;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class SourceFormatTest {
+
+  @Test
+  void configValuesAreStable() {
+    assertThat(SourceFormat.JSON.configValue()).isEqualTo("json");
+    assertThat(SourceFormat.KEYVALUE.configValue()).isEqualTo("keyvalue");
+  }
+
+  @Test
+  void parseDelegatesToJsonParser() {
+    List<SourceWrapper> parsed = SourceFormat.JSON.parse("{\"trace-sampling\": 0.5}");
+
+    assertThat(parsed).hasSize(1);
+    assertThat(parsed.get(0)).isInstanceOf(JsonSourceWrapper.class);
+    assertThat(parsed.get(0).getPolicyType()).isEqualTo("trace-sampling");
+  }
+
+  @Test
+  void parseDelegatesToKeyValueParser() {
+    List<SourceWrapper> parsed = SourceFormat.KEYVALUE.parse("trace-sampling=0.5");
+
+    assertThat(parsed).hasSize(1);
+    assertThat(parsed.get(0)).isInstanceOf(KeyValueSourceWrapper.class);
+    assertThat(parsed.get(0).getPolicyType()).isEqualTo("trace-sampling");
+  }
+
+  @Test
+  void parseSupportsEmptyInputAcrossFormats() {
+    assertThat(SourceFormat.JSON.parse("[]")).isEmpty();
+    assertThat(SourceFormat.KEYVALUE.parse("")).isEmpty();
+    assertThat(SourceFormat.KEYVALUE.parse("\n   \r\n")).isEmpty();
+  }
+
+  @Test
+  void parseReturnsNullForInvalidInput() {
+    assertThat(SourceFormat.JSON.parse("{invalid-json")).isNull();
+    assertThat(SourceFormat.JSON.parse("{}")).isNull();
+    assertThat(SourceFormat.JSON.parse("{\"a\": 1, \"b\": 2}")).isNull();
+    assertThat(SourceFormat.JSON.parse("[{\"trace-sampling\": 0.5}, {}]")).isNull();
+    assertThat(SourceFormat.KEYVALUE.parse("not-key-value")).isNull();
+  }
+
+  @Test
+  void parseReturnsImmutableListsAcrossFormats() {
+    List<SourceWrapper> jsonParsed = SourceFormat.JSON.parse("{\"trace-sampling\": 0.5}");
+    List<SourceWrapper> keyValueParsed = SourceFormat.KEYVALUE.parse("trace-sampling=0.5");
+
+    assertThatThrownBy(() -> jsonParsed.add(parsed("other-policy", "1.0")))
+        .isInstanceOf(UnsupportedOperationException.class);
+    assertThatThrownBy(() -> keyValueParsed.add(parsed("other-policy", "1.0")))
+        .isInstanceOf(UnsupportedOperationException.class);
+  }
+
+  @Test
+  void parseRejectsNullInputAcrossFormats() {
+    assertThatThrownBy(() -> SourceFormat.JSON.parse(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("source cannot be null");
+    assertThatThrownBy(() -> SourceFormat.KEYVALUE.parse(null))
+        .isInstanceOf(NullPointerException.class)
+        .hasMessage("source cannot be null");
+  }
+
+  private static SourceWrapper parsed(String key, String value) {
+    return new KeyValueSourceWrapper(key, value);
+  }
+}