diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceAttributes.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceAttributes.java
index 1a56ad2216e..988307c6b03 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceAttributes.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceAttributes.java
@@ -36,9 +36,11 @@ private OtlpResourceAttributes() {}
               "telemetry.sdk.version",
               "telemetry.sdk.language"));
 
-  /** Visits each resource attribute key/value pair with {@code visitor}. */
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
   static void visitResourceAttributes(
-      Config config, Map<String, String> extraAttributes, BiConsumer<String, String> visitor) {
+      Config config, Map<String, Object> extraAttributes, BiConsumer<String, Object> visitor) {
     String serviceName = config.getServiceName();
     String env = config.getEnv();
     String version = config.getVersion();
@@ -64,8 +66,9 @@ static void visitResourceAttributes(
         .getGlobalTags()
         .forEach(
             (key, value) -> {
-              // ignore datadog tags and their otel equivalents that we map above
-              if (!IGNORED_GLOBAL_TAGS.contains(key.toLowerCase(Locale.ROOT))) {
+              // ignore global tags replaced by canonical or extra resource attributes
+              if (!IGNORED_GLOBAL_TAGS.contains(key.toLowerCase(Locale.ROOT))
+                  && !extraAttributes.containsKey(key)) {
                 visitor.accept(key, value);
               }
             });
@@ -73,34 +76,31 @@ static void visitResourceAttributes(
     extraAttributes.forEach(visitor);
   }
 
+  private static final String PROCESS_TAGS_KEY = DATADOG_PREFIX + "process_tags";
+
   /**
    * Builds the extra resource attributes for the OTLP trace export: the {@code _dd.stats_computed}
    * marker when the SDK is computing OTLP span metrics, so a downstream Agent does not recompute
    * them from the exported spans.
    */
-  static Map<String, String> traceResourceAttributes(Config config) {
-    Map<String, String> attributes = new LinkedHashMap<>();
+  static Map<String, Object> traceResourceAttributes(Config config) {
+    Map<String, Object> attributes = new LinkedHashMap<>();
     if (config.isOtelTracesSpanMetricsEnabled()) {
       attributes.put(STATS_COMPUTED_KEY, "true");
     }
     return attributes;
   }
 
-  static Map<String, String> datadogResourceAttributes(Config config) {
-    Map<String, String> attributes = new LinkedHashMap<>();
+  static Map<String, Object> datadogResourceAttributes(Config config) {
+    Map<String, Object> attributes = new LinkedHashMap<>();
     String runtimeId = config.getRuntimeId();
     if (runtimeId != null && !runtimeId.isEmpty()) {
       attributes.put(DATADOG_PREFIX + "runtime_id", runtimeId);
     }
-    // Process tags arrive as "key:value" pairs; emit each as datadog.<key> = value.
+    // Mirrors SerializingMetricWriter's v0.6 ProcessTags shape; keep both in sync if that changes.
     List<String> processTags = ProcessTags.getTagsAsStringList();
-    if (processTags != null) {
-      for (String tag : processTags) {
-        int colon = tag.indexOf(':');
-        if (colon > 0) {
-          attributes.put(DATADOG_PREFIX + tag.substring(0, colon), tag.substring(colon + 1));
-        }
-      }
+    if (processTags != null && !processTags.isEmpty()) {
+      attributes.put(PROCESS_TAGS_KEY, processTags);
     }
     return attributes;
   }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceJson.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceJson.java
index 4731bcd592a..f6b8cfc6227 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceJson.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceJson.java
@@ -1,5 +1,6 @@
 package datadog.trace.core.otlp.common;
 
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 import static datadog.trace.core.otlp.common.OtlpCommonJson.writeAttribute;
 import static datadog.trace.core.otlp.common.OtlpResourceAttributes.datadogResourceAttributes;
@@ -9,6 +10,7 @@
 import datadog.json.JsonWriter;
 import datadog.trace.api.Config;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 /** Provides a canned JSON fragment for OpenTelemetry's "resource.proto" JSON encoding. */
@@ -21,8 +23,7 @@ private OtlpResourceJson() {}
 
   /**
    * Resource that additionally carries {@code datadog.runtime_id} and process tags (each prefixed
-   * {@code datadog.}). Used by the default-mode SDK trace-metrics export; omitted in OTel-semantics
-   * mode.
+   * {@code datadog.}). Used by the SDK trace-metrics export.
    */
   public static final String RESOURCE_FRAGMENT_WITH_DATADOG_ATTRS =
       buildResourceFragment(Config.get(), datadogResourceAttributes(Config.get()));
@@ -35,7 +36,7 @@ private OtlpResourceJson() {}
   public static final String TRACE_RESOURCE_FRAGMENT =
       buildResourceFragment(Config.get(), traceResourceAttributes(Config.get()));
 
-  static String buildResourceFragment(Config config, Map<String, String> extraAttributes) {
+  static String buildResourceFragment(Config config, Map<String, Object> extraAttributes) {
     try (JsonWriter writer = new JsonWriter()) {
       writer.beginObject();
       writer.name("attributes").beginArray();
@@ -49,7 +50,14 @@ static String buildResourceFragment(Config config, Map<String, String> extraAttr
     }
   }
 
-  private static void writeResourceAttribute(JsonWriter writer, String key, String value) {
-    writeAttribute(writer, STRING_ATTRIBUTE, key, value);
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
+  private static void writeResourceAttribute(JsonWriter writer, String key, Object value) {
+    if (value instanceof List) {
+      writeAttribute(writer, STRING_ARRAY_ATTRIBUTE, key, value);
+    } else {
+      writeAttribute(writer, STRING_ATTRIBUTE, key, value);
+    }
   }
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java
index 91a6cf7193c..3a2d4c0c460 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java
@@ -1,5 +1,6 @@
 package datadog.trace.core.otlp.common;
 
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 import static datadog.trace.core.otlp.common.OtlpCommonProto.LEN_WIRE_TYPE;
 import static datadog.trace.core.otlp.common.OtlpCommonProto.writeAttribute;
@@ -12,6 +13,7 @@
 import datadog.communication.serialization.StreamingBuffer;
 import datadog.trace.api.Config;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 /** Provides a canned message for OpenTelemetry's "resource.proto" wire protocol. */
@@ -24,8 +26,7 @@ private OtlpResourceProto() {}
 
   /**
    * Resource that additionally carries {@code datadog.runtime_id} and process tags (each prefixed
-   * {@code datadog.}). Used by the default-mode SDK trace-metrics export; omitted in OTel-semantics
-   * mode.
+   * {@code datadog.}). Used by the SDK trace-metrics export.
    */
   public static final byte[] RESOURCE_MESSAGE_WITH_DATADOG_ATTRS =
       buildResourceMessage(Config.get(), datadogResourceAttributes(Config.get()));
@@ -38,7 +39,7 @@ private OtlpResourceProto() {}
   public static final byte[] TRACE_RESOURCE_MESSAGE =
       buildResourceMessage(Config.get(), traceResourceAttributes(Config.get()));
 
-  static byte[] buildResourceMessage(Config config, Map<String, String> extraAttributes) {
+  static byte[] buildResourceMessage(Config config, Map<String, Object> extraAttributes) {
     GrowableBuffer buf = new GrowableBuffer(512);
 
     visitResourceAttributes(
@@ -52,8 +53,15 @@ static byte[] buildResourceMessage(Config config, Map<String, String> extraAttri
     return resourceMessage;
   }
 
-  private static void writeResourceAttribute(StreamingBuffer buf, String key, String value) {
+  /**
+   * {@code value} is a {@link String}, except {@code datadog.process_tags}: a {@code List<String>}.
+   */
+  private static void writeResourceAttribute(StreamingBuffer buf, String key, Object value) {
     writeTag(buf, 1, LEN_WIRE_TYPE);
-    writeAttribute(buf, STRING_ATTRIBUTE, key, value);
+    if (value instanceof List) {
+      writeAttribute(buf, STRING_ARRAY_ATTRIBUTE, key, value);
+    } else {
+      writeAttribute(buf, STRING_ATTRIBUTE, key, value);
+    }
   }
 }
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
index d7b7f56ac94..5e69d20e1f0 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
@@ -1,7 +1,9 @@
 package datadog.trace.core.otlp.metrics;
 
 import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.HISTOGRAM;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.BOOLEAN_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG_ATTRIBUTE;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ARRAY_ATTRIBUTE;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
 
 import datadog.metrics.api.Histogram;
@@ -9,6 +11,7 @@
 import datadog.trace.api.config.OtlpConfig;
 import datadog.trace.api.telemetry.OtlpTelemetry;
 import datadog.trace.api.time.SystemTimeSource;
+import datadog.trace.bootstrap.instrumentation.api.Tags;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
@@ -49,17 +52,24 @@ public final class OtlpStatsMetricWriter implements MetricWriter {
   private static final String HTTP_ROUTE = "http.route";
   private static final String RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code";
   private static final String STATUS_CODE = "status.code";
-  private static final String STATUS_CODE_ERROR = "ERROR";
+  private static final String STATUS_CODE_OK = "STATUS_CODE_OK";
+  private static final String STATUS_CODE_ERROR = "STATUS_CODE_ERROR";
   private static final String DATADOG_OPERATION_NAME = "datadog.operation.name";
   private static final String DATADOG_SPAN_TYPE = "datadog.span.type";
   private static final String DATADOG_SPAN_TOP_LEVEL = "datadog.span.top_level";
+  private static final String DATADOG_IS_TRACE_ROOT = "datadog.is_trace_root";
+  private static final String DATADOG_SERVICE_SOURCE = "datadog.svc_src";
   private static final String DATADOG_ORIGIN = "datadog.origin";
+  private static final String DATADOG_PEER_TAGS = "datadog.peer_tags";
   private static final String SYNTHETICS_ORIGIN = "synthetics";
 
-  @Nullable private final OtlpSender sender;
-  private final boolean otelSemanticsMode;
+  private static final String SPAN_KIND_SERVER = "SPAN_KIND_SERVER";
+  private static final String SPAN_KIND_CLIENT = "SPAN_KIND_CLIENT";
+  private static final String SPAN_KIND_PRODUCER = "SPAN_KIND_PRODUCER";
+  private static final String SPAN_KIND_CONSUMER = "SPAN_KIND_CONSUMER";
+  private static final String SPAN_KIND_INTERNAL = "SPAN_KIND_INTERNAL";
 
-  @Nullable private final String defaultService;
+  @Nullable private final OtlpSender sender;
 
   // own single-thread collector; forced to DELTA since trace-stats buckets are per-interval deltas.
   private final OtlpMetricsCollector collector;
@@ -87,44 +97,26 @@ private static final class PendingPoint {
 
   public OtlpStatsMetricWriter(Config config) {
     // shared protocol-based sender selection so both OTLP metrics export paths agree
-    this(
-        OtlpMetricsSenderFactory.create(config),
-        config.getOtlpMetricsProtocol(),
-        config.isTraceOtelSemanticsEnabled(),
-        config.getServiceName());
+    this(OtlpMetricsSenderFactory.create(config), config.getOtlpMetricsProtocol());
   }
 
-  // visible for testing: lets tests inject a capturing sender to decode the emitted payload and
-  // control the semantics mode and default service
-  OtlpStatsMetricWriter(
-      @Nullable OtlpSender sender, boolean otelSemanticsMode, @Nullable String defaultService) {
-    this(sender, OtlpConfig.Protocol.HTTP_PROTOBUF, otelSemanticsMode, defaultService);
+  // visible for testing: lets tests inject a capturing sender to decode the emitted payload
+  OtlpStatsMetricWriter(@Nullable OtlpSender sender) {
+    this(sender, OtlpConfig.Protocol.HTTP_PROTOBUF);
   }
 
-  private OtlpStatsMetricWriter(
-      @Nullable OtlpSender sender,
-      OtlpConfig.Protocol protocol,
-      boolean otelSemanticsMode,
-      @Nullable String defaultService) {
+  private OtlpStatsMetricWriter(@Nullable OtlpSender sender, OtlpConfig.Protocol protocol) {
     this.sender = sender;
-    this.otelSemanticsMode = otelSemanticsMode;
-    this.defaultService = defaultService;
-    // Default mode carries datadog.runtime_id / process tags on the Resource; OTel-semantics mode
-    // uses the plain vendor-neutral resource (no datadog.*).
     this.collector =
         protocol == OtlpConfig.Protocol.HTTP_JSON
             ? new OtlpMetricsJsonCollector(
                 SystemTimeSource.INSTANCE,
                 true,
-                otelSemanticsMode
-                    ? OtlpResourceJson.RESOURCE_FRAGMENT
-                    : OtlpResourceJson.RESOURCE_FRAGMENT_WITH_DATADOG_ATTRS)
+                OtlpResourceJson.RESOURCE_FRAGMENT_WITH_DATADOG_ATTRS)
             : new OtlpMetricsProtoCollector(
                 SystemTimeSource.INSTANCE,
                 true,
-                otelSemanticsMode
-                    ? OtlpResourceProto.RESOURCE_MESSAGE
-                    : OtlpResourceProto.RESOURCE_MESSAGE_WITH_DATADOG_ATTRS);
+                OtlpResourceProto.RESOURCE_MESSAGE_WITH_DATADOG_ATTRS);
   }
 
   @Override
@@ -205,17 +197,10 @@ private void emit(OtlpMetricsVisitor visitor) {
 
   private void emitDataPointAttributes(
       OtlpMetricVisitor metric, AggregateEntry entry, boolean error, boolean allTopLevel) {
-    if (error) {
-      emitStringAttribute(metric, STATUS_CODE, STATUS_CODE_ERROR);
-    }
-    // OTel semconv attrs are emitted in both modes
+    emitStringAttribute(metric, STATUS_CODE, error ? STATUS_CODE_ERROR : STATUS_CODE_OK);
     emitStringAttribute(metric, SPAN_NAME, entry.getResource());
-    emitStringAttribute(metric, SPAN_KIND, entry.getSpanKind());
-    // service.name on the point only when the span's service differs from the resource's default
-    UTF8BytesString service = entry.getService();
-    if (service != null && service.length() > 0 && !service.toString().equals(defaultService)) {
-      emitStringAttribute(metric, SERVICE_NAME, service);
-    }
+    emitStringAttribute(metric, SPAN_KIND, canonicalSpanKind(entry.getSpanKind()));
+    emitStringAttribute(metric, SERVICE_NAME, entry.getService());
     if (entry.hasHttpMethod()) {
       emitStringAttribute(metric, HTTP_REQUEST_METHOD, entry.getHttpMethod());
     }
@@ -228,40 +213,60 @@ private void emitDataPointAttributes(
     if (entry.hasGrpcStatusCode()) {
       emitStringAttribute(metric, RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode());
     }
-    // Additional metric tags: user-configured span-derived dimensions, carried as packed
-    // "key:value" UTF8 strings in schema order. Emitted in both modes as plain OTLP string
-    // attributes keyed by the tag name. NOTE: the attribute-key representation (raw tag name vs a
-    // datadog.* namespace) is an open cross-team question with the OTLP/agent side -- see the PR.
+    // additional_metric_tags support is still evolving/TBD across most tracer SDKs.
     for (UTF8BytesString additionalTag : entry.getAdditionalTags()) {
       emitAdditionalTag(metric, additionalTag);
     }
-    // Default (Datadog) mode: emit datadog.* per-point attributes
-    if (!otelSemanticsMode) {
-      emitStringAttribute(metric, DATADOG_OPERATION_NAME, entry.getOperationName());
-      emitStringAttribute(metric, DATADOG_SPAN_TYPE, entry.getType());
-      emitLongAttribute(metric, DATADOG_SPAN_TOP_LEVEL, allTopLevel ? 1L : 0L);
-      if (entry.isSynthetics()) {
-        emitStringAttribute(metric, DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
-      }
+    emitStringAttribute(metric, DATADOG_OPERATION_NAME, entry.getOperationName());
+    emitStringAttribute(metric, DATADOG_SPAN_TYPE, entry.getType());
+    emitBooleanAttribute(metric, DATADOG_SPAN_TOP_LEVEL, allTopLevel);
+    emitBooleanAttribute(metric, DATADOG_IS_TRACE_ROOT, entry.isTraceRoot());
+    if (entry.hasServiceSource()) {
+      emitStringAttribute(metric, DATADOG_SERVICE_SOURCE, entry.getServiceSource());
+    }
+    if (entry.isSynthetics()) {
+      emitStringAttribute(metric, DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
+    }
+    emitPeerTags(metric, entry.getPeerTags());
+  }
+
+  private static void emitPeerTags(OtlpMetricVisitor metric, List<UTF8BytesString> peerTags) {
+    if (peerTags.isEmpty()) {
+      return;
+    }
+    List<String> peerTagValues = new ArrayList<>(peerTags.size());
+    for (UTF8BytesString peerTag : peerTags) {
+      peerTagValues.add(peerTag.toString());
+    }
+    metric.visitAttribute(STRING_ARRAY_ATTRIBUTE, DATADOG_PEER_TAGS, peerTagValues);
+  }
+
+  private static String canonicalSpanKind(CharSequence spanKind) {
+    if (spanKind == null) {
+      return SPAN_KIND_INTERNAL;
+    } else if (Tags.SPAN_KIND_SERVER.contentEquals(spanKind)) {
+      return SPAN_KIND_SERVER;
+    } else if (Tags.SPAN_KIND_CLIENT.contentEquals(spanKind)) {
+      return SPAN_KIND_CLIENT;
+    } else if (Tags.SPAN_KIND_PRODUCER.contentEquals(spanKind)) {
+      return SPAN_KIND_PRODUCER;
+    } else if (Tags.SPAN_KIND_CONSUMER.contentEquals(spanKind)) {
+      return SPAN_KIND_CONSUMER;
+    } else {
+      return SPAN_KIND_INTERNAL;
     }
   }
 
-  // Splits a packed "key:value" additional-tag string at the first ':' (keys cannot contain ':',
-  // values may) and emits it as an OTLP string attribute. Skips only malformed slots with no ':' or
-  // an empty key. An empty value ("key:") is emitted as key="": the aggregation path treats an
-  // explicitly-empty tag as a distinct dimension from an absent one, so dropping it here would
-  // export two separately-aggregated rows with identical OTLP attribute sets.
   private static void emitAdditionalTag(OtlpMetricVisitor metric, UTF8BytesString additionalTag) {
     String packed = additionalTag.toString();
     int separator = packed.indexOf(':');
     if (separator <= 0) {
       return;
     }
-    metric.visitAttribute(
-        STRING_ATTRIBUTE, packed.substring(0, separator), packed.substring(separator + 1));
+    String key = packed.substring(0, separator);
+    metric.visitAttribute(STRING_ATTRIBUTE, key, packed.substring(separator + 1));
   }
 
-  // accepts both String literals and UTF8BytesString (both CharSequence); skips null values
   private static void emitStringAttribute(
       OtlpMetricVisitor metric, String key, @Nullable CharSequence value) {
     if (value != null) {
@@ -272,4 +277,8 @@ private static void emitStringAttribute(
   private static void emitLongAttribute(OtlpMetricVisitor metric, String key, long value) {
     metric.visitAttribute(LONG_ATTRIBUTE, key, value);
   }
+
+  private static void emitBooleanAttribute(OtlpMetricVisitor metric, String key, boolean value) {
+    metric.visitAttribute(BOOLEAN_ATTRIBUTE, key, value);
+  }
 }
diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceJsonTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceJsonTest.java
index 84ef6a9c3ae..e73311993cf 100644
--- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceJsonTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceJsonTest.java
@@ -2,6 +2,7 @@
 
 import static datadog.communication.ddagent.TracerVersion.TRACER_VERSION;
 import static datadog.trace.api.config.GeneralConfig.ENV;
+import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.SERVICE_NAME;
 import static datadog.trace.api.config.GeneralConfig.TAGS;
 import static datadog.trace.api.config.GeneralConfig.VERSION;
@@ -15,13 +16,16 @@
 
 import datadog.json.JsonMapper;
 import datadog.trace.api.Config;
+import datadog.trace.api.ProcessTags;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -43,8 +47,8 @@ private static Properties props(String... keyValues) {
     return props;
   }
 
-  private static Map<String, String> attrs(String... keyValues) {
-    Map<String, String> map = new LinkedHashMap<>();
+  private static Map<String, Object> attrs(String... keyValues) {
+    Map<String, Object> map = new LinkedHashMap<>();
     for (int i = 0; i < keyValues.length; i += 2) {
       map.put(keyValues[i], keyValues[i + 1]);
     }
@@ -54,6 +58,11 @@ private static Map<String, String> attrs(String... keyValues) {
     return map;
   }
 
+  @AfterEach
+  void resetProcessTags() {
+    ProcessTags.reset(Config.get());
+  }
+
   static Stream<Arguments> resourceFragmentCases() {
     return Stream.of(
         Arguments.of(
@@ -129,28 +138,24 @@ static Stream<Arguments> resourceFragmentCases() {
   @ParameterizedTest(name = "{0}")
   @MethodSource("resourceFragmentCases")
   void testBuildResourceFragment(
-      String caseName, Properties properties, Map<String, String> expectedAttributes)
+      String caseName, Properties properties, Map<String, Object> expectedAttributes)
       throws IOException {
     Config config = Config.get(properties);
     String fragment = OtlpResourceJson.buildResourceFragment(config, Collections.emptyMap());
 
-    Map<String, String> actualAttributes = parseResourceAttributes(fragment);
+    Map<String, Object> actualAttributes = parseResourceAttributes(fragment);
     assertEquals(expectedAttributes, actualAttributes, "For case: " + caseName);
   }
 
-  /**
-   * The datadog-attrs variant carries {@code datadog.runtime_id}; the plain variant omits it.
-   * (Process tags are emitted only when the experimental process-tag propagation is enabled, so
-   * they aren't asserted here.)
-   */
+  /** The datadog-attrs variant carries {@code datadog.runtime_id}; the plain variant omits it. */
   @Test
   void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException {
     Config config = Config.get(props(SERVICE_NAME, "my-service"));
 
-    Map<String, String> withDatadog =
+    Map<String, Object> withDatadog =
         parseResourceAttributes(
             OtlpResourceJson.buildResourceFragment(config, datadogResourceAttributes(config)));
-    Map<String, String> plain =
+    Map<String, Object> plain =
         parseResourceAttributes(
             OtlpResourceJson.buildResourceFragment(config, Collections.emptyMap()));
 
@@ -164,17 +169,41 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException {
     assertFalse(plain.containsKey("datadog.runtime_id"), "plain variant omits datadog.runtime_id");
   }
 
+  @Test
+  void datadogResourceAttributesOverrideCollidingGlobalProcessTag() throws IOException {
+    Config config =
+        Config.get(
+            props(
+                SERVICE_NAME,
+                "my-service",
+                TAGS,
+                "datadog.process_tags:user-value",
+                EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED,
+                "true"));
+    ProcessTags.reset(config);
+    ProcessTags.addTag("entrypoint.name", "app");
+    ProcessTags.addTag("entrypoint.type", "web");
+
+    Map<String, Object> withDatadog =
+        parseResourceAttributes(
+            OtlpResourceJson.buildResourceFragment(config, datadogResourceAttributes(config)));
+
+    Object processTags = withDatadog.get("datadog.process_tags");
+    assertTrue(processTags instanceof List, "datadog.process_tags is a single arrayValue");
+    assertEquals(ProcessTags.getTagsAsStringList(), processTags);
+  }
+
   @Test
   void statsComputedVariantCarriesMarker() throws IOException {
     Config withMetrics =
         Config.get(props(SERVICE_NAME, "my-service", OTEL_TRACES_SPAN_METRICS_ENABLED, "true"));
     Config withoutMetrics = Config.get(props(SERVICE_NAME, "my-service"));
 
-    Map<String, String> withMarker =
+    Map<String, Object> withMarker =
         parseResourceAttributes(
             OtlpResourceJson.buildResourceFragment(
                 withMetrics, traceResourceAttributes(withMetrics)));
-    Map<String, String> without =
+    Map<String, Object> without =
         parseResourceAttributes(
             OtlpResourceJson.buildResourceFragment(
                 withoutMetrics, traceResourceAttributes(withoutMetrics)));
@@ -200,27 +229,43 @@ void cannedFragmentsMatchTheirProtoCounterparts() throws IOException {
   // ── parsing helpers ───────────────────────────────────────────────────────
 
   @SuppressWarnings("unchecked")
-  private static Map<String, String> parseResourceAttributes(String fragment) throws IOException {
+  private static Map<String, Object> parseResourceAttributes(String fragment) throws IOException {
     Map<String, Object> resource = JsonMapper.fromJsonToMap(fragment);
     List<Object> attributes = (List<Object>) resource.get("attributes");
 
-    Map<String, String> result = new LinkedHashMap<>();
+    Map<String, Object> result = new LinkedHashMap<>();
     for (Object attribute : attributes) {
       Map<String, Object> keyValue = (Map<String, Object>) attribute;
       Map<String, Object> value = (Map<String, Object>) keyValue.get("value");
-      result.put((String) keyValue.get("key"), (String) value.get("stringValue"));
+      String key = (String) keyValue.get("key");
+      assertFalse(result.containsKey(key), "duplicate resource attribute key: " + key);
+      result.put(key, readAnyValue(value));
     }
     return result;
   }
 
-  private static Map<String, String> parseResourceAttributesFromProto(byte[] bytes)
+  @SuppressWarnings("unchecked")
+  private static Object readAnyValue(Map<String, Object> value) {
+    if (value.containsKey("arrayValue")) {
+      Map<String, Object> arrayValue = (Map<String, Object>) value.get("arrayValue");
+      List<Object> elements = (List<Object>) arrayValue.get("values");
+      List<String> strings = new ArrayList<>();
+      for (Object element : elements) {
+        strings.add((String) readAnyValue((Map<String, Object>) element));
+      }
+      return strings;
+    }
+    return value.get("stringValue");
+  }
+
+  private static Map<String, Object> parseResourceAttributesFromProto(byte[] bytes)
       throws IOException {
     com.google.protobuf.CodedInputStream outer =
         com.google.protobuf.CodedInputStream.newInstance(bytes);
     outer.readTag();
     com.google.protobuf.CodedInputStream resource = outer.readBytes().newCodedInput();
 
-    Map<String, String> attributes = new LinkedHashMap<>();
+    Map<String, Object> attributes = new LinkedHashMap<>();
     while (!resource.isAtEnd()) {
       resource.readTag();
       com.google.protobuf.CodedInputStream kv = resource.readBytes().newCodedInput();
@@ -228,10 +273,23 @@ private static Map<String, String> parseResourceAttributesFromProto(byte[] bytes
       String key = kv.readString();
       kv.readTag();
       com.google.protobuf.CodedInputStream av = kv.readBytes().newCodedInput();
-      av.readTag();
-      String value = av.readString();
-      attributes.put(key, value);
+      attributes.put(key, readAnyValueFromProto(av));
     }
     return attributes;
   }
+
+  private static Object readAnyValueFromProto(com.google.protobuf.CodedInputStream av)
+      throws IOException {
+    int tag = av.readTag();
+    if (com.google.protobuf.WireFormat.getTagFieldNumber(tag) == 5) { // array_value
+      com.google.protobuf.CodedInputStream arrayValue = av.readBytes().newCodedInput();
+      List<String> values = new ArrayList<>();
+      while (!arrayValue.isAtEnd()) {
+        arrayValue.readTag(); // ArrayValue.values (field 1, repeated AnyValue)
+        values.add((String) readAnyValueFromProto(arrayValue.readBytes().newCodedInput()));
+      }
+      return values;
+    }
+    return av.readString(); // string_value (field 1)
+  }
 }
diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java
index d9bfe5d9616..24dda2ac60a 100644
--- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java
@@ -2,6 +2,7 @@
 
 import static datadog.communication.ddagent.TracerVersion.TRACER_VERSION;
 import static datadog.trace.api.config.GeneralConfig.ENV;
+import static datadog.trace.api.config.GeneralConfig.EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED;
 import static datadog.trace.api.config.GeneralConfig.SERVICE_NAME;
 import static datadog.trace.api.config.GeneralConfig.TAGS;
 import static datadog.trace.api.config.GeneralConfig.VERSION;
@@ -16,12 +17,16 @@
 import com.google.protobuf.CodedInputStream;
 import com.google.protobuf.WireFormat;
 import datadog.trace.api.Config;
+import datadog.trace.api.ProcessTags;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -55,8 +60,8 @@ private static Properties props(String... keyValues) {
     return props;
   }
 
-  private static Map<String, String> attrs(String... keyValues) {
-    Map<String, String> map = new LinkedHashMap<>();
+  private static Map<String, Object> attrs(String... keyValues) {
+    Map<String, Object> map = new LinkedHashMap<>();
     for (int i = 0; i < keyValues.length; i += 2) {
       map.put(keyValues[i], keyValues[i + 1]);
     }
@@ -66,6 +71,11 @@ private static Map<String, String> attrs(String... keyValues) {
     return map;
   }
 
+  @AfterEach
+  void resetProcessTags() {
+    ProcessTags.reset(Config.get());
+  }
+
   static Stream<Arguments> resourceMessageCases() {
     return Stream.of(
         // service not set: should use the auto-detected name
@@ -150,28 +160,27 @@ static Stream<Arguments> resourceMessageCases() {
   @ParameterizedTest(name = "{0}")
   @MethodSource("resourceMessageCases")
   void testBuildResourceMessage(
-      String caseName, Properties properties, Map<String, String> expectedAttributes)
+      String caseName, Properties properties, Map<String, Object> expectedAttributes)
       throws IOException {
     Config config = Config.get(properties);
     byte[] bytes = OtlpResourceProto.buildResourceMessage(config, Collections.emptyMap());
 
-    Map<String, String> actualAttributes = parseResourceAttributes(bytes);
+    Map<String, Object> actualAttributes = parseResourceAttributes(bytes);
     assertEquals(expectedAttributes, actualAttributes, "For case: " + caseName);
   }
 
   /**
    * The datadog-attrs variant ({@code buildResourceMessage(config, datadogResourceAttributes)})
-   * carries {@code datadog.runtime_id}; the plain variant omits it. (Process tags are emitted only
-   * when the experimental process-tag propagation is enabled, so they aren't asserted here.)
+   * carries {@code datadog.runtime_id}; the plain variant omits it.
    */
   @Test
   void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException {
     Config config = Config.get(props(SERVICE_NAME, "my-service"));
 
-    Map<String, String> withDatadog =
+    Map<String, Object> withDatadog =
         parseResourceAttributes(
             OtlpResourceProto.buildResourceMessage(config, datadogResourceAttributes(config)));
-    Map<String, String> plain =
+    Map<String, Object> plain =
         parseResourceAttributes(
             OtlpResourceProto.buildResourceMessage(config, Collections.emptyMap()));
 
@@ -185,17 +194,41 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException {
     assertFalse(plain.containsKey("datadog.runtime_id"), "plain variant omits datadog.runtime_id");
   }
 
+  @Test
+  void datadogResourceAttributesOverrideCollidingGlobalProcessTag() throws IOException {
+    Config config =
+        Config.get(
+            props(
+                SERVICE_NAME,
+                "my-service",
+                TAGS,
+                "datadog.process_tags:user-value",
+                EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED,
+                "true"));
+    ProcessTags.reset(config);
+    ProcessTags.addTag("entrypoint.name", "app");
+    ProcessTags.addTag("entrypoint.type", "web");
+
+    Map<String, Object> withDatadog =
+        parseResourceAttributes(
+            OtlpResourceProto.buildResourceMessage(config, datadogResourceAttributes(config)));
+
+    Object processTags = withDatadog.get("datadog.process_tags");
+    assertTrue(processTags instanceof List, "datadog.process_tags is a single arrayValue");
+    assertEquals(ProcessTags.getTagsAsStringList(), processTags);
+  }
+
   @Test
   void statsComputedVariantCarriesMarker() throws IOException {
     Config withMetrics =
         Config.get(props(SERVICE_NAME, "my-service", OTEL_TRACES_SPAN_METRICS_ENABLED, "true"));
     Config withoutMetrics = Config.get(props(SERVICE_NAME, "my-service"));
 
-    Map<String, String> withMarker =
+    Map<String, Object> withMarker =
         parseResourceAttributes(
             OtlpResourceProto.buildResourceMessage(
                 withMetrics, traceResourceAttributes(withMetrics)));
-    Map<String, String> without =
+    Map<String, Object> without =
         parseResourceAttributes(
             OtlpResourceProto.buildResourceMessage(
                 withoutMetrics, traceResourceAttributes(withoutMetrics)));
@@ -214,9 +247,10 @@ void statsComputedVariantCarriesMarker() throws IOException {
    * <p>{@code buildResourceMessage} returns a length-prefixed message with an outer tag (field 1,
    * LEN wire type) followed by the Resource body size and body. Read the outer tag, then iterate
    * over all {@code Resource.attributes} (field 1, LEN wire type). Each attribute is a {@code
-   * KeyValue} whose {@code value} is an {@code AnyValue} containing a {@code string_value}.
+   * KeyValue} whose {@code value} is an {@code AnyValue} containing either a {@code string_value}
+   * (field 1) or, for {@code datadog.process_tags}, an {@code array_value} (field 5).
    */
-  private static Map<String, String> parseResourceAttributes(byte[] bytes) throws IOException {
+  private static Map<String, Object> parseResourceAttributes(byte[] bytes) throws IOException {
     // Read the outer tag (field 1, LEN wire type) that wraps the Resource body
     CodedInputStream outer = CodedInputStream.newInstance(bytes);
     int outerTag = outer.readTag();
@@ -224,7 +258,7 @@ private static Map<String, String> parseResourceAttributes(byte[] bytes) throws
     assertEquals(WireFormat.WIRETYPE_LENGTH_DELIMITED, WireFormat.getTagWireType(outerTag));
     CodedInputStream resource = outer.readBytes().newCodedInput();
 
-    Map<String, String> attributes = new LinkedHashMap<>();
+    Map<String, Object> attributes = new LinkedHashMap<>();
     while (!resource.isAtEnd()) {
       // Each attribute is Resource.attributes (field 1, LEN wire type)
       int tag = resource.readTag();
@@ -236,15 +270,10 @@ private static Map<String, String> parseResourceAttributes(byte[] bytes) throws
 
       String key = readKeyField(kv);
       CodedInputStream av = readAnyValueField(kv);
-
-      // Read AnyValue.string_value (field 1, LEN)
-      int avTag = av.readTag();
-      assertEquals(1, WireFormat.getTagFieldNumber(avTag), "AnyValue.string_value is field 1");
-      assertEquals(WireFormat.WIRETYPE_LENGTH_DELIMITED, WireFormat.getTagWireType(avTag));
-      String value = av.readString();
-      assertTrue(av.isAtEnd(), "no extra fields in AnyValue");
+      Object value = readAnyValueBody(av);
       assertTrue(kv.isAtEnd(), "no extra fields in KeyValue");
 
+      assertFalse(attributes.containsKey(key), "duplicate resource attribute key: " + key);
       attributes.put(key, value);
     }
     return attributes;
@@ -268,4 +297,31 @@ private static CodedInputStream readAnyValueField(CodedInputStream kv) throws IO
     assertEquals(WireFormat.WIRETYPE_LENGTH_DELIMITED, WireFormat.getTagWireType(tag));
     return kv.readBytes().newCodedInput();
   }
+
+  /** Reads {@code AnyValue.string_value} (field 1) or {@code AnyValue.array_value} (field 5). */
+  private static Object readAnyValueBody(CodedInputStream av) throws IOException {
+    int avTag = av.readTag();
+    int field = WireFormat.getTagFieldNumber(avTag);
+    assertEquals(WireFormat.WIRETYPE_LENGTH_DELIMITED, WireFormat.getTagWireType(avTag));
+    Object value;
+    if (field == 5) {
+      value = readArrayValue(av.readBytes().newCodedInput());
+    } else {
+      assertEquals(1, field, "AnyValue.string_value is field 1");
+      value = av.readString();
+    }
+    assertTrue(av.isAtEnd(), "no extra fields in AnyValue");
+    return value;
+  }
+
+  /** Reads {@code ArrayValue.values} (field 1, repeated {@code AnyValue}) into a string list. */
+  private static List<String> readArrayValue(CodedInputStream arrayValue) throws IOException {
+    List<String> values = new ArrayList<>();
+    while (!arrayValue.isAtEnd()) {
+      int tag = arrayValue.readTag();
+      assertEquals(1, WireFormat.getTagFieldNumber(tag), "ArrayValue.values is field 1");
+      values.add((String) readAnyValueBody(arrayValue.readBytes().newCodedInput()));
+    }
+    return values;
+  }
 }
diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
index 3d5eee87b60..e20f8713b84 100644
--- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
@@ -21,6 +21,7 @@
 import datadog.trace.core.otlp.common.OtlpSender;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -161,7 +162,7 @@ private static DecodedMetric decode(byte[] payload) throws IOException {
   /**
    * Decodes the {@code Resource.attributes} ({@code ResourceMetrics.resource = 1} → {@code
    * Resource.attributes = 1}) into a key→value map, for asserting the {@code datadog.*} resource
-   * attributes emitted in default mode.
+   * attributes emitted by the trace-metrics exporter.
    */
   private static Map<String, Object> decodeResourceAttributes(byte[] payload) throws IOException {
     CodedInputStream metricsData = CodedInputStream.newInstance(payload);
@@ -295,9 +296,15 @@ private static Object readAnyValue(CodedInputStream any) throws IOException {
         case 1: // string_value
           value = any.readString();
           break;
+        case 2: // bool_value
+          value = any.readBool();
+          break;
         case 3: // int_value
           value = any.readInt64();
           break;
+        case 5: // array_value
+          value = readArrayValue(any.readBytes().newCodedInput());
+          break;
         default:
           any.skipField(tag);
       }
@@ -305,6 +312,22 @@ private static Object readAnyValue(CodedInputStream any) throws IOException {
     return value;
   }
 
+  /**
+   * Decodes an {@code ArrayValue.values} (field 1, repeated {@code AnyValue}) into a string list.
+   */
+  private static List<String> readArrayValue(CodedInputStream arrayValue) throws IOException {
+    List<String> values = new ArrayList<>();
+    while (!arrayValue.isAtEnd()) {
+      int tag = arrayValue.readTag();
+      if (WireFormat.getTagFieldNumber(tag) == 1) {
+        values.add((String) readAnyValue(arrayValue.readBytes().newCodedInput()));
+      } else {
+        arrayValue.skipField(tag);
+      }
+    }
+    return values;
+  }
+
   // ── writer driver ─────────────────────────────────────────────────────────
 
   /**
@@ -312,10 +335,9 @@ private static Object readAnyValue(CodedInputStream any) throws IOException {
    * entry} over the fixed {@link #BUCKET_START}/{@link #BUCKET_DURATION} window, asserts that
    * exactly one payload was sent, and returns the decoded metric.
    */
-  private static DecodedMetric writeAndDecode(boolean otelSemanticsMode, AggregateEntry entry)
-      throws IOException {
+  private static DecodedMetric writeAndDecode(AggregateEntry entry) throws IOException {
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, otelSemanticsMode, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
     writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
     writer.add(entry);
     writer.finishBucket();
@@ -327,7 +349,7 @@ private static DecodedMetric writeAndDecode(boolean otelSemanticsMode, Aggregate
 
   @Test
   void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
-    DecodedMetric metric = writeAndDecode(false, okEntry(SECONDS.toNanos(1), 3));
+    DecodedMetric metric = writeAndDecode(okEntry(SECONDS.toNanos(1), 3));
 
     assertEquals("traces.span.sdk.metrics.duration", metric.name);
     assertEquals("s", metric.unit);
@@ -338,7 +360,7 @@ void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
     assertEquals(BUCKET_START, dp.start, "start_time_unix_nano == startBucket start");
     assertEquals(BUCKET_START + BUCKET_DURATION, dp.end, "time_unix_nano == start + duration");
     assertEquals(3L, dp.count);
-    assertFalse(dp.attributes.containsKey("status.code"), "ok point carries no status.code");
+    assertEquals("STATUS_CODE_OK", dp.attributes.get("status.code"), "ok point → STATUS_CODE_OK");
   }
 
   @Test
@@ -348,7 +370,7 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
     AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(2)); // ok
     AggregateEntryTestUtils.recordError(e, SECONDS.toNanos(3)); // error
 
-    DecodedMetric metric = writeAndDecode(false, e);
+    DecodedMetric metric = writeAndDecode(e);
     assertEquals(2, metric.dataPoints.size(), "ok+error → two data points");
 
     long okCount = 0;
@@ -356,7 +378,7 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
     DataPoint errorPoint = null;
     DataPoint okPoint = null;
     for (DataPoint dp : metric.dataPoints) {
-      if ("ERROR".equals(dp.attributes.get("status.code"))) {
+      if ("STATUS_CODE_ERROR".equals(dp.attributes.get("status.code"))) {
         errorPoint = dp;
         errorCount = dp.count;
       } else {
@@ -364,8 +386,9 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
         okCount = dp.count;
       }
     }
-    assertNotNull(errorPoint, "one data point must carry status.code=ERROR");
-    assertNotNull(okPoint, "one data point must omit status.code");
+    assertNotNull(errorPoint, "one data point must carry status.code=STATUS_CODE_ERROR");
+    assertNotNull(okPoint, "one data point must carry status.code=STATUS_CODE_OK");
+    assertEquals("STATUS_CODE_OK", okPoint.attributes.get("status.code"));
     assertEquals(e.getOkLatencies().getCount(), (double) okCount, 1e-9);
     assertEquals(e.getErrorLatencies().getCount(), (double) errorCount, 1e-9);
   }
@@ -373,7 +396,7 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
   @Test
   void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOException {
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
 
     // Bucket 1: the entry sees an error, so its error histogram is allocated and emits a point.
     AggregateEntry e = entry("GET /users", false, 0, null, null, null);
@@ -387,8 +410,8 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
     assertEquals(2, bucket1.dataPoints.size(), "bucket with an error → ok+error data points");
     assertTrue(
         bucket1.dataPoints.stream()
-            .anyMatch(dp -> "ERROR".equals(dp.attributes.get("status.code"))),
-        "bucket 1 must carry a status.code=ERROR point");
+            .anyMatch(dp -> "STATUS_CODE_ERROR".equals(dp.attributes.get("status.code"))),
+        "bucket 1 must carry a status.code=STATUS_CODE_ERROR point");
 
     // Bucket 2: same entry, reset then only OK hits. errorLatencies survives clear() (cleared, not
     // nulled), so a non-null-but-empty histogram must NOT emit a phantom zero-count error series.
@@ -400,9 +423,10 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
     writer.finishBucket();
     DecodedMetric bucket2 = decode(sender.lastPayload);
     assertEquals(1, bucket2.dataPoints.size(), "ok-only bucket → exactly one data point");
-    assertFalse(
-        bucket2.dataPoints.get(0).attributes.containsKey("status.code"),
-        "recovered entry must not emit a lingering status.code=ERROR series");
+    assertEquals(
+        "STATUS_CODE_OK",
+        bucket2.dataPoints.get(0).attributes.get("status.code"),
+        "recovered entry must not emit a lingering status.code=STATUS_CODE_ERROR series");
   }
 
   @Test
@@ -410,7 +434,7 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
     AggregateEntry e = entry("GET /users/{id}", false, 200, "GET", "/users/{id}", "0");
     AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
-    DecodedMetric metric = writeAndDecode(false, e);
+    DecodedMetric metric = writeAndDecode(e);
     assertEquals(1, metric.dataPoints.size());
     Map<String, Object> attrs = metric.dataPoints.get(0).attributes;
 
@@ -421,7 +445,7 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
 
     // a bare entry has none of these
     Map<String, Object> bareAttrs =
-        writeAndDecode(false, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
+        writeAndDecode(okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
     assertFalse(bareAttrs.containsKey("http.request.method"));
     assertFalse(bareAttrs.containsKey("http.response.status_code"));
     assertFalse(bareAttrs.containsKey("http.route"));
@@ -432,7 +456,7 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
   void additionalMetricTagsEmittedAsStringAttributes() throws IOException {
     // Additional tags arrive on the entry pre-packed as "key:value" UTF8 strings in schema order;
     // the writer splits each at the first ':' and emits it as a plain OTLP string attribute keyed
-    // by the tag name, in both semantics modes.
+    // by the tag name.
     AggregateEntry e =
         AggregateEntryTestUtils.of(
             "GET /users",
@@ -450,42 +474,16 @@ void additionalMetricTagsEmittedAsStringAttributes() throws IOException {
             null,
             new UTF8BytesString[] {
               UTF8BytesString.create("region:us-east-1"),
-              UTF8BytesString.create("tenant_id:acme:corp")
+              UTF8BytesString.create("tenant_id:acme:corp"),
+              UTF8BytesString.create("datadog.custom:visible")
             });
     AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
-    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
     assertEquals("us-east-1", attrs.get("region"));
     // value may itself contain ':' — only the first ':' separates key from value
     assertEquals("acme:corp", attrs.get("tenant_id"));
-  }
-
-  @Test
-  void additionalMetricTagsEmittedInOtelSemanticsMode() throws IOException {
-    // Unlike datadog.* attributes, additional tags are user-configured dimensions and are emitted
-    // in otel-semantics mode too.
-    AggregateEntry e =
-        AggregateEntryTestUtils.of(
-            "GET /users",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            0,
-            false,
-            true,
-            "server",
-            null,
-            null,
-            null,
-            null,
-            new UTF8BytesString[] {UTF8BytesString.create("region:us-east-1")});
-    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
-
-    Map<String, Object> attrs = writeAndDecode(true, e).dataPoints.get(0).attributes;
-    assertEquals("us-east-1", attrs.get("region"));
-    assertFalse(
-        attrs.containsKey("datadog.operation.name"), "datadog.* still absent in otel-semantics");
+    assertEquals("visible", attrs.get("datadog.custom"));
   }
 
   @Test
@@ -517,7 +515,7 @@ void emptyValueEmittedButMalformedSlotsSkipped() throws IOException {
             });
     AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
-    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
     assertEquals("us-east-1", attrs.get("region"), "well-formed tag still emitted");
     assertFalse(attrs.containsKey("noseparator"), "no-separator slot skipped");
     assertFalse(attrs.containsKey(""), "empty-key slot skipped");
@@ -526,39 +524,35 @@ void emptyValueEmittedButMalformedSlotsSkipped() throws IOException {
   }
 
   @Test
-  void serviceNameEmittedOnlyForNonDefaultService() throws IOException {
+  void serviceNameAlwaysEmittedOnDataPoint() throws IOException {
     CapturingSender sender = new CapturingSender();
-    // The configured default service ("web") is reported on the resource; only a span whose service
-    // differs from it repeats service.name on its own data point.
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false, "web");
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
 
     long start = SECONDS.toNanos(1_700_000_000L);
     writer.startBucket(2, start, SECONDS.toNanos(10));
-    writer.add(serviceEntry("web.request", "web")); // default service
-    writer.add(serviceEntry("db.query", "postgres")); // custom service
+    writer.add(serviceEntry("web.request", "web"));
+    writer.add(serviceEntry("db.query", "postgres"));
     writer.finishBucket();
 
     DecodedMetric metric = decode(sender.lastPayload);
     assertEquals(2, metric.dataPoints.size());
 
-    Map<String, Object> defaultAttrs = null;
-    Map<String, Object> customAttrs = null;
+    Map<String, Object> webAttrs = null;
+    Map<String, Object> postgresAttrs = null;
     for (DataPoint dp : metric.dataPoints) {
       if ("db.query".equals(dp.attributes.get("datadog.operation.name"))) {
-        customAttrs = dp.attributes;
+        postgresAttrs = dp.attributes;
       } else {
-        defaultAttrs = dp.attributes;
+        webAttrs = dp.attributes;
       }
     }
-    assertNotNull(customAttrs, "custom-service data point present");
-    assertNotNull(defaultAttrs, "default-service data point present");
+    assertNotNull(postgresAttrs, "postgres data point present");
+    assertNotNull(webAttrs, "web data point present");
+    assertEquals("postgres", postgresAttrs.get("service.name"));
     assertEquals(
-        "postgres",
-        customAttrs.get("service.name"),
-        "non-default service is carried on its own data point");
-    assertFalse(
-        defaultAttrs.containsKey("service.name"),
-        "default service must not be repeated on its data point");
+        "web",
+        webAttrs.get("service.name"),
+        "service.name is emitted unconditionally, even matching the tracer's own default service");
   }
 
   /** An ok-only entry on the given service and operation name, recording a single 1s hit. */
@@ -585,7 +579,7 @@ private static AggregateEntry serviceEntry(String operationName, String service)
   @Test
   void emptyBucketSendsNothing() {
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
 
     writer.startBucket(0, BUCKET_START, BUCKET_DURATION);
     writer.finishBucket(); // no add()
@@ -597,7 +591,7 @@ void emptyBucketSendsNothing() {
   @Test
   void nullSenderDoesNotThrowOnNonEmptyBucket() {
     // mirrors the HTTP_JSON path where createSender(config) returns null.
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(null, false, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter((OtlpSender) null);
     writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
     writer.add(okEntry(SECONDS.toNanos(1), 2));
     try {
@@ -607,39 +601,163 @@ void nullSenderDoesNotThrowOnNonEmptyBucket() {
     }
   }
 
-  @Test
-  void defaultModeCarriesDatadogAttributes() throws IOException {
-    // use an entry where all hits are top-level: OR in TOP_LEVEL_TAG
+  @ParameterizedTest
+  @CsvSource({"true", "false"})
+  void carriesDatadogAttributes(boolean topLevel) throws IOException {
     AggregateEntry e = entry("servlet.request", false, 0, null, null, null);
-    AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
+    if (topLevel) {
+      AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
+    } else {
+      AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+    }
 
-    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
-    assertTrue(
-        attrs.containsKey("datadog.operation.name"), "operation name present in default mode");
-    assertTrue(attrs.containsKey("datadog.span.type"), "span type present in default mode");
-    assertTrue(
-        attrs.containsKey("datadog.span.top_level"), "span top-level present in default mode");
-    assertEquals(1L, attrs.get("datadog.span.top_level"), "all hits top-level → 1");
-    // OTel-semconv attrs are present in both modes
-    assertTrue(attrs.containsKey("span.name"), "span.name present in both modes");
-    // datadog.origin presence/absence is covered by defaultModeEmitsSyntheticOrigin
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertTrue(attrs.containsKey("datadog.operation.name"));
+    assertTrue(attrs.containsKey("datadog.span.type"));
+    assertTrue(attrs.containsKey("datadog.span.top_level"));
+    assertTrue(attrs.get("datadog.span.top_level") instanceof Boolean);
+    assertEquals(topLevel, attrs.get("datadog.span.top_level"));
+    assertTrue(attrs.containsKey("span.name"));
+  }
+
+  @ParameterizedTest
+  @CsvSource({"true", "false"})
+  void emitsIsTraceRoot(boolean traceRoot) throws IOException {
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            traceRoot,
+            "server",
+            null,
+            null,
+            null,
+            null);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertTrue(attrs.get("datadog.is_trace_root") instanceof Boolean);
+    assertEquals(traceRoot, attrs.get("datadog.is_trace_root"));
+  }
+
+  @Test
+  void serviceSourceEmittedOnlyWhenSet() throws IOException {
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            "component",
+            "web",
+            0,
+            false,
+            true,
+            "server",
+            null,
+            null,
+            null,
+            null);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertEquals("component", attrs.get("datadog.svc_src"));
+    assertTrue(attrs.get("datadog.svc_src") instanceof String);
+
+    Map<String, Object> absentAttrs =
+        writeAndDecode(okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
+    assertFalse(absentAttrs.containsKey("datadog.svc_src"));
+  }
+
+  @Test
+  void emitsPeerTags() throws IOException {
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            true,
+            "client",
+            Arrays.asList(
+                UTF8BytesString.create("peer.service:downstream"),
+                UTF8BytesString.create("net.peer.name:downstream.example.com")),
+            null,
+            null,
+            null);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertEquals(
+        Arrays.asList("peer.service:downstream", "net.peer.name:downstream.example.com"),
+        attrs.get("datadog.peer_tags"));
+  }
+
+  @Test
+  void omitsPeerTagsWhenEmpty() throws IOException {
+    AggregateEntry e = entry("GET /users", false, 0, null, null, null);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertFalse(attrs.containsKey("datadog.peer_tags"));
+  }
+
+  @ParameterizedTest
+  @CsvSource(
+      value = {
+        "server, SPAN_KIND_SERVER",
+        "client, SPAN_KIND_CLIENT",
+        "producer, SPAN_KIND_PRODUCER",
+        "consumer, SPAN_KIND_CONSUMER",
+        "broker, SPAN_KIND_INTERNAL",
+        "'', SPAN_KIND_INTERNAL",
+        "NULL, SPAN_KIND_INTERNAL",
+      },
+      nullValues = "NULL")
+  void spanKindIsCanonicalizedToUppercaseEnumName(String spanKind, String expected)
+      throws IOException {
+    AggregateEntry e =
+        AggregateEntryTestUtils.of(
+            "GET /users",
+            "web",
+            "servlet.request",
+            null,
+            "web",
+            0,
+            false,
+            true,
+            spanKind,
+            null,
+            null,
+            null,
+            null);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
+
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
+    assertEquals(expected, attrs.get("span.kind"));
   }
 
   /**
-   * In default mode a synthetic entry emits {@code datadog.origin = "synthetics"}; a non-synthetic
-   * entry omits the attribute. Origin has collapsed to a boolean {@code synthetic} flag upstream,
-   * so {@code "synthetics"} is the only origin value that can reach the writer.
+   * A synthetic entry emits {@code datadog.origin = "synthetics"}; a non-synthetic entry omits the
+   * attribute. Origin has collapsed to a boolean {@code synthetic} flag upstream, so {@code
+   * "synthetics"} is the only origin value that can reach the writer.
    */
   @ParameterizedTest(name = "synthetic={0} → datadog.origin={1}")
   @CsvSource(
       nullValues = "NULL",
       value = {"false, NULL", "true, synthetics"})
-  void defaultModeEmitsSyntheticOrigin(boolean synthetic, String expectedOrigin)
-      throws IOException {
+  void emitsSyntheticOrigin(boolean synthetic, String expectedOrigin) throws IOException {
     AggregateEntry e = entry("servlet.request", synthetic, 0, null, null, null);
     AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
-    Map<String, Object> attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+    Map<String, Object> attrs = writeAndDecode(e).dataPoints.get(0).attributes;
     if (expectedOrigin == null) {
       assertFalse(attrs.containsKey("datadog.origin"), "non-synthetic → datadog.origin absent");
     } else {
@@ -647,23 +765,6 @@ void defaultModeEmitsSyntheticOrigin(boolean synthetic, String expectedOrigin)
     }
   }
 
-  @Test
-  void otelSemanticsModeOmitsDatadogAttributes() throws IOException {
-    // otelSemanticsMode = true → datadog.* must be absent
-    Map<String, Object> attrs =
-        writeAndDecode(true, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
-    assertFalse(
-        attrs.containsKey("datadog.operation.name"),
-        "operation name absent in otel-semantics mode");
-    assertFalse(attrs.containsKey("datadog.span.type"), "span type absent in otel-semantics mode");
-    assertFalse(
-        attrs.containsKey("datadog.span.top_level"),
-        "span top-level absent in otel-semantics mode");
-    assertFalse(attrs.containsKey("datadog.origin"), "origin absent in otel-semantics mode");
-    // OTel-semconv attrs must still be present
-    assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode");
-  }
-
   @Test
   void snapshotsEntryDataBeforeAggregatorClearsIt() throws IOException {
     // The aggregator clears each entry's per-interval data immediately after add() returns
@@ -671,7 +772,7 @@ void snapshotsEntryDataBeforeAggregatorClearsIt() throws IOException {
     // (and the top-level count) at add() time; if it deferred reading to finishBucket() it would
     // encode the already-cleared (empty, zero-count) entry.
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
 
     AggregateEntry e = entry("servlet.request", false, 0, null, null, null);
     AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
@@ -689,46 +790,26 @@ void snapshotsEntryDataBeforeAggregatorClearsIt() throws IOException {
     DataPoint dp = metric.dataPoints.get(0);
     assertEquals(3L, dp.count, "count must reflect the pre-clear snapshot, not the cleared entry");
     assertEquals(
-        1L, dp.attributes.get("datadog.span.top_level"), "all pre-clear hits were top-level");
+        Boolean.TRUE,
+        dp.attributes.get("datadog.span.top_level"),
+        "all pre-clear hits were top-level");
   }
 
   // ── resource attributes (datadog.runtime_id / process tags) ────────────────
 
   @Test
-  void defaultModeResourceCarriesRuntimeId() throws IOException {
-    // runtime-id is enabled by default, so default-mode payloads carry datadog.runtime_id on the
-    // Resource.
+  void resourceCarriesRuntimeId() throws IOException {
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false, null);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
     writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
     writer.add(okEntry(SECONDS.toNanos(1), 1));
     writer.finishBucket();
 
     Map<String, Object> resourceAttrs = decodeResourceAttributes(sender.lastPayload);
     assertTrue(
-        resourceAttrs.containsKey("datadog.runtime_id"),
-        "default mode resource carries datadog.runtime_id");
+        resourceAttrs.containsKey("datadog.runtime_id"), "resource carries datadog.runtime_id");
     Object runtimeId = resourceAttrs.get("datadog.runtime_id");
     assertNotNull(runtimeId, "runtime id value present");
     assertFalse(runtimeId.toString().isEmpty(), "runtime id value non-empty");
   }
-
-  @Test
-  void otelSemanticsModeResourceOmitsDatadogAttributes() throws IOException {
-    CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, true, null);
-    writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
-    writer.add(okEntry(SECONDS.toNanos(1), 1));
-    writer.finishBucket();
-
-    Map<String, Object> resourceAttrs = decodeResourceAttributes(sender.lastPayload);
-    assertFalse(
-        resourceAttrs.containsKey("datadog.runtime_id"),
-        "otel-semantics mode resource omits datadog.runtime_id");
-    for (String key : resourceAttrs.keySet()) {
-      assertFalse(
-          key.startsWith("datadog."),
-          "otel-semantics mode resource has no datadog.* attrs: " + key);
-    }
-  }
 }