diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java
index 28151f88864..e5a2edb856b 100644
--- a/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java
+++ b/dd-trace-api/src/main/java/datadog/trace/api/config/FeatureFlaggingConfig.java
@@ -3,4 +3,13 @@
 public class FeatureFlaggingConfig {
 
   public static final String FLAGGING_PROVIDER_ENABLED = "experimental.flagging.provider.enabled";
+
+  /**
+   * Opt-in gate for APM span enrichment with feature-flag evaluation metadata. The dot-form maps to
+   * {@code DD_EXPERIMENTAL_FLAGGING_PROVIDER_SPAN_ENRICHMENT_ENABLED} via the dot-to-underscore +
+   * {@code DD_} prefix normalization rule. This is DISTINCT from {@link #FLAGGING_PROVIDER_ENABLED}
+   * and is OFF by default — enabling the provider does not enable span enrichment.
+   */
+  public static final String SPAN_ENRICHMENT_ENABLED =
+      "experimental.flagging.provider.span.enrichment.enabled";
 }
diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json
index e2d171c3c3f..096b5c5b936 100644
--- a/metadata/supported-configurations.json
+++ b/metadata/supported-configurations.json
@@ -1497,6 +1497,14 @@
         "aliases": []
       }
     ],
+    "DD_EXPERIMENTAL_FLAGGING_PROVIDER_SPAN_ENRICHMENT_ENABLED": [
+      {
+        "version": "A",
+        "type": "boolean",
+        "default": "false",
+        "aliases": []
+      }
+    ],
     "DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED": [
       {
         "version": "B",
diff --git a/products/feature-flagging/feature-flagging-api/build.gradle.kts b/products/feature-flagging/feature-flagging-api/build.gradle.kts
index 1c368d51b51..ee5b950dc08 100644
--- a/products/feature-flagging/feature-flagging-api/build.gradle.kts
+++ b/products/feature-flagging/feature-flagging-api/build.gradle.kts
@@ -45,11 +45,18 @@ dependencies {
 
   compileOnly(project(":products:feature-flagging:feature-flagging-bootstrap"))
   compileOnly(project(":utils:config-utils"))
+  // Span enrichment: TraceInterceptor / GlobalTracer / AgentTracer / AgentSpan for the
+  // write tier + active-root-span lookup. compileOnly because the agent runtime
+  // (feature-flagging-agent depends on :internal-api) provides these classes; the published
+  // dd-openfeature jar must not bundle the tracer.
+  compileOnly(project(":internal-api"))
   compileOnly("io.opentelemetry:opentelemetry-api:1.47.0")
   compileOnly("io.opentelemetry:opentelemetry-sdk-metrics:1.47.0")
   compileOnly("io.opentelemetry:opentelemetry-exporter-otlp:1.47.0")
 
   testImplementation(project(":products:feature-flagging:feature-flagging-bootstrap"))
+  testImplementation(project(":internal-api"))
+  testImplementation(project(":utils:config-utils"))
   testImplementation("io.opentelemetry:opentelemetry-api:1.47.0")
   testImplementation("io.opentelemetry:opentelemetry-sdk-metrics:1.47.0")
   testImplementation("io.opentelemetry:opentelemetry-exporter-otlp:1.47.0")
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java
index 91c0aafdc7a..88fa7d4812f 100644
--- a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/DDEvaluator.java
@@ -392,6 +392,14 @@ private static <T> ProviderEvaluation<T> resolveVariant(
             .addString("flagKey", flag.key)
             .addString("variationType", flag.variationType.name())
             .addString("allocationKey", allocation.key);
+    // Surface the UFC split's serial id and the allocation's doLog flag for APM span enrichment.
+    // __dd_split_serial_id is omitted when the split carries no serial id; __dd_do_log is always
+    // present so the span-enrichment hook can decide whether to record the subject.
+    if (split.serialId != null) {
+      metadataBuilder.addString("__dd_split_serial_id", split.serialId.toString());
+    }
+    metadataBuilder.addString(
+        "__dd_do_log", String.valueOf(allocation.doLog != null && allocation.doLog));
     final ProviderEvaluation<T> result =
         ProviderEvaluation.<T>builder()
             .value(mappedValue)
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/Provider.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/Provider.java
index c492ef49c69..158e3e76842 100644
--- a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/Provider.java
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/Provider.java
@@ -2,6 +2,8 @@
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 
+import datadog.trace.api.GlobalTracer;
+import datadog.trace.config.inversion.ConfigHelper;
 import de.thetaphi.forbiddenapis.SuppressForbidden;
 import dev.openfeature.sdk.ErrorCode;
 import dev.openfeature.sdk.EvaluationContext;
@@ -16,6 +18,7 @@
 import dev.openfeature.sdk.exceptions.OpenFeatureError;
 import dev.openfeature.sdk.exceptions.ProviderNotReadyError;
 import java.lang.reflect.Constructor;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -28,6 +31,15 @@ public class Provider extends EventProvider implements Metadata {
   private static final Logger log = LoggerFactory.getLogger(Provider.class);
   static final String METADATA = "datadog-openfeature-provider";
   private static final String EVALUATOR_IMPL = "datadog.trace.api.openfeature.DDEvaluator";
+
+  /**
+   * Environment variable form of {@link
+   * datadog.trace.api.config.FeatureFlaggingConfig#SPAN_ENRICHMENT_ENABLED}. Distinct from the
+   * provider-enabled gate; OFF by default (experimental opt-in).
+   */
+  static final String SPAN_ENRICHMENT_ENABLED_ENV =
+      "DD_EXPERIMENTAL_FLAGGING_PROVIDER_SPAN_ENRICHMENT_ENABLED";
+
   private static final Options DEFAULT_OPTIONS = new Options().initTimeout(30, SECONDS);
   private volatile Evaluator evaluator;
   private final Options options;
@@ -35,6 +47,12 @@ public class Provider extends EventProvider implements Metadata {
       new AtomicReference<>(InitializationState.NOT_STARTED);
   private final FlagEvalMetrics flagEvalMetrics;
   private final FlagEvalHook flagEvalHook;
+  // Span enrichment: null unless the gate is on, so the feature has no idle overhead when off.
+  private final SpanEnrichmentHook spanEnrichmentHook;
+  private final SpanEnrichmentStates spanEnrichmentStates;
+  // Precomputed hook list returned by getProviderHooks() on every evaluation. Immutable and built
+  // once so gate-off evaluation allocates nothing on this hot path.
+  private final List<Hook> providerHooks;
 
   public Provider() {
     this(DEFAULT_OPTIONS, null);
@@ -45,6 +63,44 @@ public Provider(final Options options) {
   }
 
   Provider(final Options options, final Evaluator evaluator) {
+    this(options, evaluator, null);
+  }
+
+  /**
+   * Registers a {@link SpanEnrichmentInterceptor} with the running tracer, returning {@code true}
+   * when it was added and {@code false} when the tracer rejected it (e.g. the interceptor is
+   * already registered, or the global tracer is the no-op placeholder). Injectable so tests can
+   * drive registration deterministically without mutating the global tracer (mirrors the {@code
+   * spanEnrichmentEnabledOverride} seam).
+   */
+  interface TraceInterceptorRegistrar {
+    boolean register(SpanEnrichmentInterceptor interceptor);
+  }
+
+  /**
+   * @param spanEnrichmentEnabledOverride when non-null, forces the span-enrichment gate (test
+   *     seam); when null, the gate is read from {@link #SPAN_ENRICHMENT_ENABLED_ENV}.
+   */
+  Provider(
+      final Options options,
+      final Evaluator evaluator,
+      final Boolean spanEnrichmentEnabledOverride) {
+    this(
+        options,
+        evaluator,
+        spanEnrichmentEnabledOverride,
+        interceptor -> GlobalTracer.get().addTraceInterceptor(interceptor));
+  }
+
+  /**
+   * @param registrar registers the span-enrichment interceptor with the tracer; injectable for
+   *     tests (see {@link TraceInterceptorRegistrar}).
+   */
+  Provider(
+      final Options options,
+      final Evaluator evaluator,
+      final Boolean spanEnrichmentEnabledOverride,
+      final TraceInterceptorRegistrar registrar) {
     this.options = options;
     this.evaluator = evaluator;
     FlagEvalMetrics metrics = null;
@@ -59,6 +115,56 @@ public Provider(final Options options) {
     }
     this.flagEvalMetrics = metrics;
     this.flagEvalHook = hook;
+
+    // Span enrichment is wired ONLY when the gate is on. When off, no hook/state is constructed and
+    // there is no idle per-evaluation or per-span overhead.
+    final boolean spanEnrichmentEnabled =
+        spanEnrichmentEnabledOverride != null
+            ? spanEnrichmentEnabledOverride
+            : isSpanEnrichmentEnabled();
+    SpanEnrichmentHook seHook = null;
+    SpanEnrichmentStates seStates = null;
+    if (spanEnrichmentEnabled) {
+      try {
+        // Per-provider state store, shared with this provider's capture hook. The single,
+        // process-wide interceptor is registered once (reconfiguration-safe) and rebound to this
+        // provider's store. A later gate-on provider rebinds it to its own store; this provider's
+        // shutdown only unbinds if it is still the active provider, so reconfiguration never
+        // permanently disables enrichment and providers never clobber each other's live state.
+        seStates = new SpanEnrichmentStates();
+        SpanEnrichmentInterceptor.ensureRegistered(registrar);
+        SpanEnrichmentInterceptor.INSTANCE.bind(seStates);
+        seHook = new SpanEnrichmentHook(seStates);
+      } catch (LinkageError | Exception e) {
+        // Tracer classes absent (e.g. API-only classpath): degrade to no span enrichment.
+        log.warn("Span enrichment unavailable — tracer classes not on classpath", e);
+        seHook = null;
+        seStates = null;
+      }
+    }
+    this.spanEnrichmentHook = seHook;
+    this.spanEnrichmentStates = seStates;
+
+    // Precompute the immutable hook list once so getProviderHooks() (called on every evaluation)
+    // allocates nothing, including when the gate is off.
+    final List<Hook> hooks = new ArrayList<>(2);
+    if (flagEvalHook != null) {
+      hooks.add(flagEvalHook);
+    }
+    if (seHook != null) {
+      hooks.add(seHook);
+    }
+    this.providerHooks =
+        hooks.isEmpty() ? Collections.emptyList() : Collections.unmodifiableList(hooks);
+  }
+
+  private static boolean isSpanEnrichmentEnabled() {
+    try {
+      final String value = ConfigHelper.env(SPAN_ENRICHMENT_ENABLED_ENV);
+      return "true".equalsIgnoreCase(value) || "1".equals(value);
+    } catch (final Throwable t) {
+      return false; // never let config reading break provider construction
+    }
   }
 
   @Override
@@ -168,10 +274,7 @@ private Evaluator buildEvaluator() throws Exception {
 
   @Override
   public List<Hook> getProviderHooks() {
-    if (flagEvalHook == null) {
-      return Collections.emptyList();
-    }
-    return Collections.singletonList(flagEvalHook);
+    return providerHooks;
   }
 
   @Override
@@ -179,11 +282,28 @@ public void shutdown() {
     if (flagEvalMetrics != null) {
       flagEvalMetrics.shutdown();
     }
+    // Provider-close cleanup for span enrichment: the tracer has no interceptor-removal API, so we
+    // unbind this provider's store from the process-wide interceptor (which clears the store and
+    // makes the interceptor inert until a new provider rebinds it). The unbind is a no-op if a
+    // newer provider has already rebound the interceptor, so we never wipe another provider's
+    // in-flight state.
+    if (spanEnrichmentStates != null) {
+      SpanEnrichmentInterceptor.INSTANCE.unbind(spanEnrichmentStates);
+    }
     if (evaluator != null) {
       evaluator.shutdown();
     }
   }
 
+  // Visible for tests: expose whether span enrichment is wired (gate-on) without leaking the impls.
+  SpanEnrichmentHook spanEnrichmentHook() {
+    return spanEnrichmentHook;
+  }
+
+  SpanEnrichmentStates spanEnrichmentStates() {
+    return spanEnrichmentStates;
+  }
+
   @Override
   public Metadata getMetadata() {
     return this;
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentAccumulator.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentAccumulator.java
new file mode 100644
index 00000000000..76265a89c50
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentAccumulator.java
@@ -0,0 +1,353 @@
+package datadog.trace.api.openfeature;
+
+import dev.openfeature.sdk.Structure;
+import dev.openfeature.sdk.Value;
+import java.time.Instant;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+
+/**
+ * Per-local-root-span accumulator for APM feature-flag span enrichment.
+ *
+ * <p>Holds the serial ids, hashed subjects, and runtime defaults captured during flag evaluation
+ * for a single local trace fragment. The limits, dedupe semantics, truncation, and output tag
+ * shapes are FROZEN against the Node reference ({@code dd-trace-js#8343}) — see {@link
+ * ULeb128Encoder}.
+ *
+ * <p>Instances are created lazily and held in a {@link SpanEnrichmentStates} store, keyed by the
+ * local-root span's full trace id. The capture hook ({@link SpanEnrichmentHook}) writes; the write
+ * interceptor ({@link SpanEnrichmentInterceptor}) reads and clears. When the span-enrichment gate
+ * is off, no store and no accumulator are ever created, so there is no idle per-span overhead.
+ *
+ * <p>Output tag shapes:
+ *
+ * <ul>
+ *   <li>{@code ffe_flags_enc} — a bare base64 string (delta-varint of the serial ids)
+ *   <li>{@code ffe_subjects_enc} — a JSON object string {@code {"<sha256hex>": "<base64>", ...}}
+ *   <li>{@code ffe_runtime_defaults} — a JSON object string {@code {"<flagKey>": "<value>", ...}}
+ * </ul>
+ */
+final class SpanEnrichmentAccumulator {
+
+  static final int MAX_SERIAL_IDS = 200;
+  static final int MAX_SUBJECTS = 10;
+  static final int MAX_EXPERIMENTS_PER_SUBJECT = 20;
+  static final int MAX_DEFAULTS = 5;
+  static final int MAX_DEFAULT_VALUE_LENGTH = 64;
+
+  static final String TAG_FLAGS_ENC = "ffe_flags_enc";
+  static final String TAG_SUBJECTS_ENC = "ffe_subjects_enc";
+  static final String TAG_RUNTIME_DEFAULTS = "ffe_runtime_defaults";
+
+  // dedupe is structural (a Set); sorted for deterministic encoding.
+  private final TreeSet<Integer> serialIds = new TreeSet<>();
+  // sha256hex(targetingKey) -> serial ids. LinkedHashMap for stable iteration order.
+  private final Map<String, TreeSet<Integer>> subjects = new LinkedHashMap<>();
+  // flagKey -> value string (first-wins, truncated to MAX_DEFAULT_VALUE_LENGTH).
+  private final Map<String, String> defaults = new LinkedHashMap<>();
+
+  /** Adds a serial id, dropping silently once {@link #MAX_SERIAL_IDS} is reached. */
+  synchronized void addSerialId(final int id) {
+    if (serialIds.size() >= MAX_SERIAL_IDS && !serialIds.contains(id)) {
+      return;
+    }
+    serialIds.add(id);
+  }
+
+  /**
+   * Records that the given targeting key was exposed to the experiment identified by {@code id}.
+   * The targeting key is SHA-256-hashed before storage. Enforces both the subject cap ({@link
+   * #MAX_SUBJECTS}) and the per-subject experiment cap ({@link #MAX_EXPERIMENTS_PER_SUBJECT}).
+   */
+  synchronized void addSubject(final String targetingKey, final int id) {
+    if (targetingKey == null) {
+      return;
+    }
+    final String hashed = ULeb128Encoder.hashTargetingKey(targetingKey);
+    final TreeSet<Integer> existing = subjects.get(hashed);
+    if (existing != null) {
+      if (existing.size() >= MAX_EXPERIMENTS_PER_SUBJECT && !existing.contains(id)) {
+        return;
+      }
+      existing.add(id);
+      return;
+    }
+    if (subjects.size() >= MAX_SUBJECTS) {
+      return;
+    }
+    final TreeSet<Integer> ids = new TreeSet<>();
+    ids.add(id);
+    subjects.put(hashed, ids);
+  }
+
+  /**
+   * Records a runtime-default value for {@code flagKey} (first-wins). Object values are serialized
+   * to JSON (NOT {@code toString()}); the result is truncated to {@link #MAX_DEFAULT_VALUE_LENGTH}.
+   */
+  synchronized void addDefault(final String flagKey, final Object value) {
+    if (flagKey == null) {
+      return;
+    }
+    if (defaults.containsKey(flagKey)) {
+      return; // first-wins
+    }
+    if (defaults.size() >= MAX_DEFAULTS) {
+      return;
+    }
+    String valueStr = stringifyDefault(value);
+    if (valueStr.length() > MAX_DEFAULT_VALUE_LENGTH) {
+      valueStr = utf8SafeTruncate(valueStr, MAX_DEFAULT_VALUE_LENGTH);
+    }
+    defaults.put(flagKey, valueStr);
+  }
+
+  /**
+   * @return true when there is at least one serial id or runtime default to write. Subjects are not
+   *     checked because a subject is never recorded without its serial id.
+   */
+  synchronized boolean hasData() {
+    return !serialIds.isEmpty() || !defaults.isEmpty();
+  }
+
+  /**
+   * Builds the {@code ffe_*} span tags from the accumulated state. Empty groups are omitted.
+   *
+   * @return a map of tag name to tag value (a subset of {@code ffe_flags_enc}, {@code
+   *     ffe_subjects_enc}, {@code ffe_runtime_defaults})
+   */
+  synchronized Map<String, String> toSpanTags() {
+    final Map<String, String> tags = new LinkedHashMap<>();
+    if (!serialIds.isEmpty()) {
+      final String encoded = ULeb128Encoder.encodeDeltaVarint(serialIds);
+      if (!encoded.isEmpty()) {
+        tags.put(TAG_FLAGS_ENC, encoded);
+      }
+    }
+    if (!subjects.isEmpty()) {
+      final Map<String, String> encodedSubjects = new LinkedHashMap<>();
+      for (final Map.Entry<String, TreeSet<Integer>> entry : subjects.entrySet()) {
+        encodedSubjects.put(entry.getKey(), ULeb128Encoder.encodeDeltaVarint(entry.getValue()));
+      }
+      tags.put(TAG_SUBJECTS_ENC, toJsonObject(encodedSubjects));
+    }
+    if (!defaults.isEmpty()) {
+      tags.put(TAG_RUNTIME_DEFAULTS, toJsonObject(defaults));
+    }
+    return tags;
+  }
+
+  // ---- helpers (visible for tests) ----
+
+  /**
+   * Mirrors the Node {@code (typeof value === 'object' && value !== null) ? JSON.stringify(value) :
+   * String(value)} rule: structured values (objects, arrays) are JSON-stringified; scalars use
+   * their string form; {@code null} becomes the bare {@code null}.
+   *
+   * <p>On the real OpenFeature object-evaluation path the runtime-default arrives wrapped in a
+   * {@link Value}; we unwrap it to its native representation first so a structured default
+   * serializes to JSON (matching Node's {@code JSON.stringify} of the equivalent JS object) instead
+   * of {@code Value.toString()} (which is {@code "Value(innerObject=...)"}). The scalar cases
+   * ({@code String}/{@code Boolean}/number) collapse to the same string form Node produces.
+   */
+  static String stringifyDefault(final Object value) {
+    final Object unwrapped = value instanceof Value ? unwrapValue((Value) value) : value;
+    if (unwrapped == null) {
+      return "null";
+    }
+    if (unwrapped instanceof Map
+        || unwrapped instanceof Iterable
+        || unwrapped.getClass().isArray()) {
+      return toJsonValue(unwrapped);
+    }
+    if (unwrapped instanceof CharSequence || unwrapped instanceof Character) {
+      return unwrapped.toString();
+    }
+    // Numbers / booleans / Instant — their string form matches what Node's String(value) emits for
+    // these scalar cases.
+    return String.valueOf(unwrapped);
+  }
+
+  /**
+   * Recursively unwraps an OpenFeature {@link Value} into its native Java representation:
+   * structures become {@code Map<String, Object>}, lists become {@code List<Object>}, and scalars
+   * become their boxed value (or {@code null}). Nested {@link Value}s are unwrapped at every level
+   * so a structure containing further structures/lists serializes correctly.
+   */
+  private static Object unwrapValue(final Value value) {
+    if (value == null || value.isNull()) {
+      return null;
+    }
+    if (value.isStructure()) {
+      final Structure structure = value.asStructure();
+      final Map<String, Object> map = new LinkedHashMap<>();
+      if (structure != null) {
+        for (final String key : structure.keySet()) {
+          map.put(key, unwrapValue(structure.getValue(key)));
+        }
+      }
+      return map;
+    }
+    if (value.isList()) {
+      final java.util.List<Value> list = value.asList();
+      final java.util.List<Object> out = new java.util.ArrayList<>(list == null ? 0 : list.size());
+      if (list != null) {
+        for (final Value element : list) {
+          out.add(unwrapValue(element));
+        }
+      }
+      return out;
+    }
+    if (value.isBoolean()) {
+      return value.asBoolean();
+    }
+    if (value.isString()) {
+      return value.asString();
+    }
+    if (value.isNumber()) {
+      // Preserve integral vs fractional so the rendered JSON number matches Node.
+      final Double d = value.asDouble();
+      if (d != null && d == Math.rint(d) && !Double.isInfinite(d)) {
+        final Integer i = value.asInteger();
+        if (i != null) {
+          return i;
+        }
+      }
+      return d;
+    }
+    final Instant instant = value.asInstant();
+    if (instant != null) {
+      return instant.toString();
+    }
+    // Unknown shape: fall back to the wrapped object's own representation.
+    return value.asObject();
+  }
+
+  /** UTF-8-safe truncation: never split a surrogate pair at the {@code maxChars} boundary. */
+  static String utf8SafeTruncate(final String value, final int maxChars) {
+    if (value.length() <= maxChars) {
+      return value;
+    }
+    int end = maxChars;
+    if (Character.isHighSurrogate(value.charAt(end - 1))) {
+      end--; // drop the dangling high surrogate rather than emit a broken pair
+    }
+    return value.substring(0, end);
+  }
+
+  /** Serializes a String->String map to a compact JSON object string (keys + values escaped). */
+  static String toJsonObject(final Map<String, String> map) {
+    final StringBuilder sb = new StringBuilder();
+    sb.append('{');
+    boolean first = true;
+    for (final Map.Entry<String, String> entry : map.entrySet()) {
+      if (!first) {
+        sb.append(',');
+      }
+      first = false;
+      appendJsonString(sb, entry.getKey());
+      sb.append(':');
+      appendJsonString(sb, entry.getValue());
+    }
+    sb.append('}');
+    return sb.toString();
+  }
+
+  private static String toJsonValue(final Object value) {
+    final StringBuilder sb = new StringBuilder();
+    appendJsonValue(sb, value);
+    return sb.toString();
+  }
+
+  @SuppressWarnings("unchecked")
+  private static void appendJsonValue(final StringBuilder sb, final Object rawValue) {
+    // Unwrap any OpenFeature Value to its native representation first so nested structures/lists
+    // serialize as JSON rather than via Value.toString().
+    final Object value = rawValue instanceof Value ? unwrapValue((Value) rawValue) : rawValue;
+    if (value == null) {
+      sb.append("null");
+    } else if (value instanceof Map) {
+      sb.append('{');
+      boolean first = true;
+      for (final Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
+        if (!first) {
+          sb.append(',');
+        }
+        first = false;
+        appendJsonString(sb, String.valueOf(entry.getKey()));
+        sb.append(':');
+        appendJsonValue(sb, entry.getValue());
+      }
+      sb.append('}');
+    } else if (value instanceof Iterable) {
+      sb.append('[');
+      boolean first = true;
+      for (final Object element : (Iterable<Object>) value) {
+        if (!first) {
+          sb.append(',');
+        }
+        first = false;
+        appendJsonValue(sb, element);
+      }
+      sb.append(']');
+    } else if (value instanceof CharSequence || value instanceof Character) {
+      appendJsonString(sb, value.toString());
+    } else if (value instanceof Number || value instanceof Boolean) {
+      sb.append(String.valueOf(value));
+    } else {
+      appendJsonString(sb, String.valueOf(value));
+    }
+  }
+
+  private static void appendJsonString(final StringBuilder sb, final String value) {
+    sb.append('"');
+    for (int i = 0; i < value.length(); i++) {
+      final char c = value.charAt(i);
+      switch (c) {
+        case '"':
+          sb.append("\\\"");
+          break;
+        case '\\':
+          sb.append("\\\\");
+          break;
+        case '\n':
+          sb.append("\\n");
+          break;
+        case '\r':
+          sb.append("\\r");
+          break;
+        case '\t':
+          sb.append("\\t");
+          break;
+        case '\b':
+          sb.append("\\b");
+          break;
+        case '\f':
+          sb.append("\\f");
+          break;
+        default:
+          if (c < 0x20) {
+            sb.append(String.format("\\u%04x", (int) c));
+          } else {
+            sb.append(c);
+          }
+      }
+    }
+    sb.append('"');
+  }
+
+  // ---- test-only accessors ----
+
+  synchronized Set<Integer> serialIdsView() {
+    return new TreeSet<>(serialIds);
+  }
+
+  synchronized int subjectCount() {
+    return subjects.size();
+  }
+
+  synchronized int defaultCount() {
+    return defaults.size();
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentHook.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentHook.java
new file mode 100644
index 00000000000..9d469efee2b
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentHook.java
@@ -0,0 +1,136 @@
+package datadog.trace.api.openfeature;
+
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import dev.openfeature.sdk.EvaluationContext;
+import dev.openfeature.sdk.FlagEvaluationDetails;
+import dev.openfeature.sdk.Hook;
+import dev.openfeature.sdk.HookContext;
+import dev.openfeature.sdk.ImmutableMetadata;
+import java.util.Map;
+
+/**
+ * OpenFeature {@code finally} hook that captures feature-flag evaluation metadata into
+ * per-local-root span state for APM span enrichment. This is the CAPTURE half of the
+ * capture-vs-write split — the WRITE half is {@link SpanEnrichmentInterceptor}, which flushes the
+ * accumulated tags onto the local root span when the trace completes.
+ *
+ * <p>Mirrors {@link FlagEvalHook}: registered via {@link Provider#getProviderHooks()} (only when
+ * the span-enrichment gate is on) and reading {@code details.getFlagMetadata()}. It resolves the
+ * active local-root span via {@link AgentTracer#activeSpan()} and keys the {@link
+ * SpanEnrichmentStates} store (shared with the {@link SpanEnrichmentInterceptor}) by that root's
+ * full trace id (hex), so the interceptor running later on the write thread can recover the same
+ * state from the completed span collection. The full hex key avoids merging two distinct 128-bit
+ * traces that happen to share their low-order 64 bits.
+ *
+ * <p>Capture branch (frozen Node reference):
+ *
+ * <ul>
+ *   <li>serial id present → addSerialId, plus addSubject when {@code __dd_do_log} AND a targeting
+ *       key
+ *   <li>else variant missing (runtime default) → addDefault(flagKey, value)
+ * </ul>
+ *
+ * <p>All work is wrapped in try/catch — enrichment must NEVER break flag evaluation.
+ */
+class SpanEnrichmentHook implements Hook<Object> {
+
+  static final String METADATA_SERIAL_ID = "__dd_split_serial_id";
+  static final String METADATA_DO_LOG = "__dd_do_log";
+
+  /**
+   * Resolves the local-root span for the active trace. Injectable so tests need no static mocks.
+   */
+  interface RootSpanResolver {
+    AgentSpan activeLocalRoot();
+  }
+
+  private static final RootSpanResolver DEFAULT_RESOLVER =
+      () -> {
+        final AgentSpan active = AgentTracer.activeSpan();
+        if (active == null) {
+          return null;
+        }
+        final AgentSpan localRoot = active.getLocalRootSpan();
+        return localRoot != null ? localRoot : active;
+      };
+
+  private final RootSpanResolver rootSpanResolver;
+  // State store shared with the interceptor. The hook writes here; the interceptor reads + removes.
+  private final SpanEnrichmentStates states;
+
+  SpanEnrichmentHook(final SpanEnrichmentStates states) {
+    this(DEFAULT_RESOLVER, states);
+  }
+
+  SpanEnrichmentHook(final RootSpanResolver rootSpanResolver, final SpanEnrichmentStates states) {
+    this.rootSpanResolver = rootSpanResolver;
+    this.states = states;
+  }
+
+  @Override
+  public void finallyAfter(
+      final HookContext<Object> ctx,
+      final FlagEvaluationDetails<Object> details,
+      final Map<String, Object> hints) {
+    if (details == null) {
+      return;
+    }
+    try {
+      final AgentSpan root = rootSpanResolver.activeLocalRoot();
+      if (root == null || root.getTraceId() == null) {
+        return; // no active span → nothing to enrich
+      }
+      // Key by the full trace id (hex), not toLong(): the low-order 64 bits alone would merge two
+      // distinct 128-bit traces that share their low bits.
+      final String traceKey = root.getTraceId().toHexString();
+      capture(traceKey, ctx, details);
+    } catch (final Throwable t) {
+      // Never let span enrichment break flag evaluation.
+    }
+  }
+
+  /**
+   * Applies the frozen Node capture branch against the state keyed by {@code traceKey}. Package
+   * private so it can be driven deterministically in tests without stubbing the static tracer.
+   */
+  void capture(
+      final String traceKey,
+      final HookContext<Object> ctx,
+      final FlagEvaluationDetails<Object> details) {
+    final ImmutableMetadata metadata = details.getFlagMetadata();
+    final String serialIdStr = metadata != null ? metadata.getString(METADATA_SERIAL_ID) : null;
+    final String doLogStr = metadata != null ? metadata.getString(METADATA_DO_LOG) : null;
+    final boolean doLog = "true".equalsIgnoreCase(doLogStr);
+    final String targetingKey = targetingKey(ctx);
+
+    if (serialIdStr != null) {
+      final int serialId;
+      try {
+        serialId = Integer.parseInt(serialIdStr);
+      } catch (final NumberFormatException e) {
+        return; // malformed serial id — drop, never break eval
+      }
+      final SpanEnrichmentAccumulator state = states.getOrCreate(traceKey);
+      state.addSerialId(serialId);
+      if (doLog && targetingKey != null) {
+        state.addSubject(targetingKey, serialId);
+      }
+    } else if (details.getVariant() == null) {
+      // Runtime-default detection = MISSING VARIANT (never a reason enum).
+      states.getOrCreate(traceKey).addDefault(details.getFlagKey(), details.getValue());
+    }
+  }
+
+  private static String targetingKey(final HookContext<Object> ctx) {
+    if (ctx == null) {
+      return null;
+    }
+    final EvaluationContext evaluationContext = ctx.getCtx();
+    if (evaluationContext == null) {
+      return null;
+    }
+    final String key = evaluationContext.getTargetingKey();
+    return (key == null || key.isEmpty()) ? null : key;
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentInterceptor.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentInterceptor.java
new file mode 100644
index 00000000000..8e6bc9d45a1
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentInterceptor.java
@@ -0,0 +1,199 @@
+package datadog.trace.api.openfeature;
+
+import datadog.trace.api.DDTraceId;
+import datadog.trace.api.interceptor.MutableSpan;
+import datadog.trace.api.interceptor.TraceInterceptor;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Process-wide {@link TraceInterceptor} that writes the accumulated {@code ffe_*} tags onto the
+ * local root span when a trace <em>actually completes</em>. This is the WRITE half of the
+ * capture-vs-write split — the CAPTURE half is {@link SpanEnrichmentHook}, which fills the active
+ * {@link SpanEnrichmentStates} store during flag evaluation.
+ *
+ * <h2>Reconfiguration safety</h2>
+ *
+ * <p>The tracer holds interceptors in an add-only, priority-keyed set with no public removal API: a
+ * given priority can be occupied by exactly one interceptor for the life of the tracer. If each
+ * provider registered its own interceptor at the same priority, the first registration would win
+ * forever; after that provider closed, every later gate-on provider would be rejected as a
+ * duplicate and enrichment would be permanently disabled.
+ *
+ * <p>To survive provider close/reopen we therefore register a single, long-lived delegating
+ * interceptor ({@link #INSTANCE}) exactly once, and {@linkplain #bind(SpanEnrichmentStates) rebind}
+ * it to whichever provider is currently active. A provider {@linkplain
+ * #unbind(SpanEnrichmentStates) unbinds} on close. When no provider is bound the interceptor is
+ * inert; a fresh provider rebinds it and enrichment resumes — no second registration is ever
+ * attempted.
+ *
+ * <h2>Partial-flush correctness</h2>
+ *
+ * <p>{@code dd-trace-core} runs {@code onTraceComplete} on <b>every</b> flush, not only on final
+ * trace completion: {@code CoreTracer.write(SpanList)} → {@code interceptCompleteTrace(...)} fires
+ * for both partial flushes ({@code PendingTrace.partialFlush()} → {@code write(true)}) and the
+ * final write ({@code write(false)}). A <b>partial</b> flush deliberately <b>excludes the
+ * still-open local root</b> — {@code PendingTrace} holds the root back ({@code rootSpanWritten})
+ * and {@code getRootSpan()} returns null on a partial fragment, so {@code CoreTracer.write} does
+ * not invoke {@code onRootSpanFinished} for it.
+ *
+ * <p>Therefore this interceptor flushes+removes state <b>only when the local root span is present
+ * in the flushed collection</b> (i.e. this is the final write for the trace). On a partial flush
+ * the root is absent, so we return early and <b>keep the accumulator intact</b>, preserving every
+ * flag evaluated before the flush boundary. Without this guard, the first partial flush would drain
+ * the accumulator and write tags onto a not-yet-finished root, silently dropping all pre-flush
+ * enrichment — exactly the long-running-trace data-loss bug.
+ *
+ * <h2>State ownership</h2>
+ *
+ * <p>State lives in the per-provider {@link SpanEnrichmentStates} store that is currently bound.
+ * {@link #unbind(SpanEnrichmentStates)} clears only the store it unbinds, so one provider's close
+ * can never wipe another's in-flight state. The store is hard-bounded, so a trace that never
+ * reaches this interceptor cannot leak unboundedly.
+ *
+ * <p>All work is wrapped in try/catch — enrichment must NEVER break trace finish.
+ */
+final class SpanEnrichmentInterceptor implements TraceInterceptor {
+
+  /**
+   * Unique priority in the "trace data enrichment" band, after {@code GIT_METADATA} (3) and before
+   * the custom-sampling band ({@code Integer.MAX_VALUE - 2}). Distinct from every value in {@code
+   * AbstractTraceInterceptor.Priority} and from the CI Visibility interceptors.
+   */
+  static final int PRIORITY = 4;
+
+  /** The single, long-lived interceptor registered with the tracer (reconfiguration safety). */
+  static final SpanEnrichmentInterceptor INSTANCE = new SpanEnrichmentInterceptor();
+
+  // Whether INSTANCE has been accepted by a real tracer. Registration can legitimately fail when
+  // the global tracer is still the no-op (not yet installed); in that case we leave this false so a
+  // later provider retries. Once true we never re-attempt (the interceptor is in for good).
+  private final AtomicBoolean registered = new AtomicBoolean(false);
+
+  // The store of the currently-active provider. null when no gate-on provider is bound, in which
+  // case the interceptor is inert. Volatile: the eval/bind threads write, the trace-write thread
+  // reads.
+  private volatile SpanEnrichmentStates activeStates;
+
+  private SpanEnrichmentInterceptor() {}
+
+  /**
+   * Idempotently registers {@link #INSTANCE} with the tracer via {@code registrar}. Safe to call
+   * from every gate-on provider: the first successful registration wins and subsequent calls are
+   * no-ops. If registration fails because the tracer is not yet installed, a later call retries.
+   */
+  static void ensureRegistered(final Provider.TraceInterceptorRegistrar registrar) {
+    if (INSTANCE.registered.get()) {
+      return;
+    }
+    if (registrar.register(INSTANCE)) {
+      INSTANCE.registered.set(true);
+    }
+  }
+
+  /** Binds the active store to {@code states}, displacing any previously-bound provider. */
+  void bind(final SpanEnrichmentStates states) {
+    this.activeStates = states;
+  }
+
+  /**
+   * Unbinds {@code states} if it is still the active store and clears it (cleanup on provider
+   * close). If a newer provider has already rebound the interceptor, this is a no-op so the new
+   * provider's in-flight state is left untouched.
+   */
+  void unbind(final SpanEnrichmentStates states) {
+    if (this.activeStates == states) {
+      this.activeStates = null;
+    }
+    if (states != null) {
+      states.clear();
+    }
+  }
+
+  /** The currently-bound store, or {@code null} when the interceptor is inert. */
+  SpanEnrichmentStates activeStates() {
+    return activeStates;
+  }
+
+  boolean isRegistered() {
+    return registered.get();
+  }
+
+  @Override
+  public Collection<? extends MutableSpan> onTraceComplete(
+      final Collection<? extends MutableSpan> trace) {
+    try {
+      final SpanEnrichmentStates states = this.activeStates;
+      if (states == null || trace == null || trace.isEmpty()) {
+        return trace;
+      }
+      // Resolve the local root for this fragment, then require that the root is actually PRESENT in
+      // this collection. A partial flush excludes the still-open root, so its absence means "not
+      // the
+      // final write" — keep the accumulator and bail.
+      final MutableSpan localRoot = findLocalRootInFragment(trace);
+      if (!(localRoot instanceof AgentSpan)) {
+        return trace; // partial flush, or no resolvable in-fragment root: keep state untouched
+      }
+      final DDTraceId traceId = ((AgentSpan) localRoot).getTraceId();
+      if (traceId == null) {
+        return trace; // no trace id (e.g. Noop span) — cannot key state; keep it
+      }
+      // Key by the full trace id (hex) to match the capture-side keying.
+      final String traceKey = traceId.toHexString();
+      final SpanEnrichmentAccumulator state = states.remove(traceKey);
+      if (state == null || !state.hasData()) {
+        return trace;
+      }
+      for (final Map.Entry<String, String> tag : state.toSpanTags().entrySet()) {
+        final String value = tag.getValue();
+        if (value != null && !value.isEmpty()) {
+          localRoot.setTag(tag.getKey(), value);
+        }
+      }
+    } catch (final Throwable t) {
+      // Never let span enrichment break trace finish.
+    }
+    return trace;
+  }
+
+  /**
+   * Resolves the local root span for this fragment and returns it ONLY if it is actually present in
+   * the fragment by reference identity. Returns {@code null} when the root is not in the collection
+   * (a partial flush excludes the still-open root) or when no root can be safely identified.
+   *
+   * <p>We never guess: a non-root span is never returned. If the first span reports a non-null
+   * local root, we accept it only after confirming that exact object is in the fragment; otherwise
+   * we look for a span that is provably its own local root and present.
+   */
+  private static MutableSpan findLocalRootInFragment(
+      final Collection<? extends MutableSpan> trace) {
+    final MutableSpan first = trace.iterator().next();
+    final MutableSpan candidate = first.getLocalRootSpan();
+    if (candidate != null) {
+      // Accept the reported local root only if it is genuinely part of THIS fragment. On a partial
+      // flush the root is reachable by reference but NOT in the collection → reject (keep state).
+      for (final MutableSpan span : trace) {
+        if (span == candidate) {
+          return candidate;
+        }
+      }
+      return null; // root excluded from this fragment → partial flush, do not flush/remove
+    }
+    // Local root unknown for the first span: only accept a span that is provably its own local root
+    // and present here. Never fall back to an arbitrary span.
+    for (final MutableSpan span : trace) {
+      if (span.getLocalRootSpan() == span) {
+        return span;
+      }
+    }
+    return null;
+  }
+
+  @Override
+  public int priority() {
+    return PRIORITY;
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentStates.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentStates.java
new file mode 100644
index 00000000000..df120d8a6f2
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/SpanEnrichmentStates.java
@@ -0,0 +1,115 @@
+package datadog.trace.api.openfeature;
+
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Bounded, instance-owned store of per-trace {@link SpanEnrichmentAccumulator} state, keyed by the
+ * local-root span's full trace id (the lower-case zero-padded hex string).
+ *
+ * <p>The full hex key matters for 128-bit trace ids: keying by {@code DDTraceId.toLong()} (the
+ * low-order 64 bits only) would merge two distinct 128-bit traces that share their low bits into a
+ * single accumulator, cross-contaminating enrichment between unrelated traces. The full hex string
+ * is unique across all 128 bits and is cached on the id, so it is cheap to obtain.
+ *
+ * <p>Each {@link SpanEnrichmentHook}/{@link SpanEnrichmentInterceptor} pair shares one store
+ * instance, so:
+ *
+ * <ul>
+ *   <li><b>Unbounded leak:</b> the store is hard-capped at {@link #MAX_TRACES}. A trace that never
+ *       reaches the interceptor (dropped, Noop tracer, never-finishing root) can no longer leak
+ *       unboundedly — once the cap is reached, the <em>oldest</em> in-flight entry is evicted (FIFO
+ *       by insertion order) so the map size is strictly bounded regardless of trace completion.
+ *   <li><b>Isolation:</b> because the store is instance-owned rather than a shared static, one
+ *       provider's cleanup clears only its own state and can never wipe another (still-active)
+ *       provider's in-flight entries.
+ * </ul>
+ *
+ * <p>Bounded eviction is intentionally lossy under pathological pressure: dropping the oldest
+ * accumulator degrades enrichment for that one (likely already-abandoned) trace rather than
+ * exhausting the heap. Enrichment correctness is best-effort by contract; heap safety is not.
+ *
+ * <p>Thread-safety: all access is guarded by the intrinsic lock on this instance. The hook writes
+ * (eval thread) and the interceptor reads+removes (trace-write thread) concurrently, so every
+ * mutator and accessor synchronizes. Contention is low — operations are O(1) map touches.
+ */
+final class SpanEnrichmentStates {
+
+  private static final Logger log = LoggerFactory.getLogger(SpanEnrichmentStates.class);
+
+  /**
+   * Hard cap on the number of concurrently-tracked traces. Sized well above any realistic count of
+   * simultaneously in-flight traces with active flag evaluations on a single JVM, so the live path
+   * never evicts; the cap exists purely to bound a leak of never-completing traces.
+   */
+  static final int MAX_TRACES = 4096;
+
+  // Insertion-ordered so the eldest entry is the natural eviction victim (FIFO). accessOrder=false
+  // (the default) — we evict by age of creation, not by recency of use, so a long-running but
+  // never-completing trace cannot pin the map by being repeatedly touched.
+  private final LinkedHashMap<String, SpanEnrichmentAccumulator> states = new LinkedHashMap<>();
+
+  // One-shot guard so a sustained leak logs once at WARN rather than on every eviction.
+  private boolean evictionWarned = false;
+
+  /**
+   * Returns the accumulator for {@code traceKey}, creating (and inserting) it if absent. When
+   * insertion would exceed {@link #MAX_TRACES}, the eldest entry is evicted first so the store
+   * stays bounded.
+   */
+  synchronized SpanEnrichmentAccumulator getOrCreate(final String traceKey) {
+    SpanEnrichmentAccumulator existing = states.get(traceKey);
+    if (existing != null) {
+      return existing;
+    }
+    if (states.size() >= MAX_TRACES) {
+      evictEldest();
+    }
+    final SpanEnrichmentAccumulator created = new SpanEnrichmentAccumulator();
+    states.put(traceKey, created);
+    return created;
+  }
+
+  /** Removes and returns the accumulator for {@code traceKey}, or {@code null} if absent. */
+  synchronized SpanEnrichmentAccumulator remove(final String traceKey) {
+    return states.remove(traceKey);
+  }
+
+  /** Clears all tracked state (cleanup on provider close / unbind). */
+  synchronized void clear() {
+    states.clear();
+  }
+
+  synchronized int size() {
+    return states.size();
+  }
+
+  synchronized boolean isEmpty() {
+    return states.isEmpty();
+  }
+
+  // ---- test-only accessor ----
+
+  synchronized SpanEnrichmentAccumulator peek(final String traceKey) {
+    return states.get(traceKey);
+  }
+
+  private void evictEldest() {
+    final Iterator<Map.Entry<String, SpanEnrichmentAccumulator>> it = states.entrySet().iterator();
+    if (it.hasNext()) {
+      it.next();
+      it.remove();
+    }
+    if (!evictionWarned) {
+      evictionWarned = true;
+      log.warn(
+          "Span-enrichment state cap ({}) reached; evicting oldest in-flight trace state. "
+              + "This indicates traces with flag evaluations that never complete; enrichment for "
+              + "evicted traces is dropped to bound memory.",
+          MAX_TRACES);
+    }
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/ULeb128Encoder.java b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/ULeb128Encoder.java
new file mode 100644
index 00000000000..c256a829150
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/main/java/datadog/trace/api/openfeature/ULeb128Encoder.java
@@ -0,0 +1,119 @@
+package datadog.trace.api.openfeature;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Base64;
+import java.util.Set;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+/**
+ * ULEB128 delta-varint + base64 codec for APM feature-flag span enrichment.
+ *
+ * <p>Ported VERBATIM from the frozen Node reference ({@code dd-trace-js#8343}). The tag names,
+ * encoding, and golden vectors are FROZEN against that contract — backend/Trino decode and the
+ * parametric system-tests assertions depend on exact parity, so this MUST NOT be re-derived.
+ *
+ * <p>Algorithm: dedupe (via {@link Set}) → sort ascending → emit the delta from the previous id as
+ * an unsigned LEB128 varint (7 bits/byte, MSB = continuation) → base64-encode the byte buffer. The
+ * empty set encodes to the empty string (the caller then omits the tag).
+ *
+ * <p>Golden vector: {@code {100, 108, 128, 130}} → deltas {@code [100, 8, 20, 2]} → bytes {@code
+ * [0x64, 0x08, 0x14, 0x02]} → base64 {@code "ZAgUAg=="}.
+ */
+final class ULeb128Encoder {
+
+  private ULeb128Encoder() {}
+
+  /**
+   * ULEB128 delta-varint encodes the given serial ids into a bare base64 string.
+   *
+   * @param serialIds the serial ids to encode (deduped + sorted ascending internally)
+   * @return the base64-encoded delta-varint bytes, or the empty string when {@code serialIds} is
+   *     empty or null
+   */
+  static String encodeDeltaVarint(final Set<Integer> serialIds) {
+    if (serialIds == null || serialIds.isEmpty()) {
+      // Empty set encodes to the empty string; the caller omits the tag.
+      return "";
+    }
+    final SortedSet<Integer> sorted =
+        serialIds instanceof SortedSet ? (SortedSet<Integer>) serialIds : new TreeSet<>(serialIds);
+    // Worst case: 5 bytes per 32-bit varint.
+    final byte[] buffer = new byte[sorted.size() * 5];
+    int length = 0;
+    int previous = 0;
+    for (final Integer id : sorted) {
+      long delta =
+          ((long) id) - previous; // long to stay safe; deltas are non-negative (sorted asc)
+      previous = id;
+      while (delta > 0x7FL) {
+        buffer[length++] = (byte) ((delta & 0x7FL) | 0x80L);
+        delta >>>= 7;
+      }
+      buffer[length++] = (byte) (delta & 0x7FL);
+    }
+    final byte[] out = new byte[length];
+    System.arraycopy(buffer, 0, out, 0, length);
+    return Base64.getEncoder().encodeToString(out);
+  }
+
+  /**
+   * Decodes a delta-varint base64 string back into the original ascending serial ids. Used by the
+   * codec round-trip oracle (the decode side mirrors the L2 system-tests decoder).
+   *
+   * @param encoded a base64 string produced by {@link #encodeDeltaVarint(Set)}
+   * @return the decoded serial ids in ascending order (empty for the empty string)
+   */
+  static SortedSet<Integer> decodeDeltaVarint(final String encoded) {
+    final SortedSet<Integer> result = new TreeSet<>();
+    if (encoded == null || encoded.isEmpty()) {
+      return result;
+    }
+    final byte[] bytes = Base64.getDecoder().decode(encoded);
+    int previous = 0;
+    int index = 0;
+    while (index < bytes.length) {
+      long value = 0;
+      int shift = 0;
+      while (true) {
+        final byte b = bytes[index++];
+        value |= ((long) (b & 0x7F)) << shift;
+        if ((b & 0x80) == 0) {
+          break;
+        }
+        shift += 7;
+      }
+      previous += (int) value; // delta from previous
+      result.add(previous);
+    }
+    return result;
+  }
+
+  /**
+   * Lower-case hex SHA-256 of the given string. Used to hash subject targeting keys before they are
+   * emitted (privacy: subject keys are never emitted in clear text).
+   *
+   * @param value the value to hash
+   * @return the lower-case hex SHA-256 digest
+   */
+  static String hashTargetingKey(final String value) {
+    try {
+      final MessageDigest digest = MessageDigest.getInstance("SHA-256");
+      final byte[] hash = digest.digest(value.getBytes(StandardCharsets.UTF_8));
+      final StringBuilder hex = new StringBuilder(hash.length * 2);
+      for (final byte b : hash) {
+        final int v = b & 0xFF;
+        if (v < 0x10) {
+          hex.append('0');
+        }
+        hex.append(Integer.toHexString(v));
+      }
+      return hex.toString();
+    } catch (final NoSuchAlgorithmException e) {
+      // SHA-256 is mandated by the JLS to be present on every JVM; this is unreachable in practice.
+      throw new IllegalStateException("SHA-256 algorithm not available", e);
+    }
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java
index bb86c409bad..db2d791d3bd 100644
--- a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java
+++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/DDEvaluatorTest.java
@@ -641,7 +641,7 @@ private ServerConfiguration createTestConfiguration() {
   private Flag createSimpleFlag(String key, ValueType type, Object value, String variantKey) {
     final Map<String, Variant> variants = new HashMap<>();
     variants.put(variantKey, new Variant(variantKey, value));
-    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null));
+    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null, null));
     final List<Allocation> allocations =
         singletonList(new Allocation("alloc1", null, null, null, splits, false));
     return new Flag(key, true, type, variants, allocations);
@@ -657,12 +657,12 @@ private Flag createRuleBasedFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.MATCHES, "email", "@company\\.com$"));
     final List<Rule> premiumRules = singletonList(new Rule(premiumConditions));
-    final List<Split> premiumSplits = singletonList(new Split(emptyList(), "premium", null));
+    final List<Split> premiumSplits = singletonList(new Split(emptyList(), "premium", null, null));
     final Allocation premiumAllocation =
         new Allocation("premium-alloc", premiumRules, null, null, premiumSplits, false);
 
     // Fallback allocation for basic
-    final List<Split> basicSplits = singletonList(new Split(emptyList(), "basic", null));
+    final List<Split> basicSplits = singletonList(new Split(emptyList(), "basic", null, null));
     final Allocation basicAllocation =
         new Allocation("basic-alloc", null, null, null, basicSplits, false);
 
@@ -680,12 +680,12 @@ private Flag createNumericRuleFlag() {
     final List<ConditionConfiguration> vipConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.GTE, "score", 800));
     final List<Rule> vipRules = singletonList(new Rule(vipConditions));
-    final List<Split> vipSplits = singletonList(new Split(emptyList(), "vip", null));
+    final List<Split> vipSplits = singletonList(new Split(emptyList(), "vip", null, null));
     final Allocation vipAllocation =
         new Allocation("vip-alloc", vipRules, null, null, vipSplits, false);
 
     // Fallback
-    final List<Split> regularSplits = singletonList(new Split(emptyList(), "regular", null));
+    final List<Split> regularSplits = singletonList(new Split(emptyList(), "regular", null, null));
     final Allocation regularAllocation =
         new Allocation("regular-alloc", null, null, null, regularSplits, false);
 
@@ -706,12 +706,12 @@ private Flag createNullCheckFlag() {
     final List<ConditionConfiguration> noBetaConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.IS_NULL, "beta_feature", true));
     final List<Rule> noBetaRules = singletonList(new Rule(noBetaConditions));
-    final List<Split> noBetaSplits = singletonList(new Split(emptyList(), "no-beta", null));
+    final List<Split> noBetaSplits = singletonList(new Split(emptyList(), "no-beta", null, null));
     final Allocation noBetaAllocation =
         new Allocation("no-beta-alloc", noBetaRules, null, null, noBetaSplits, false);
 
     // Fallback
-    final List<Split> hasBetaSplits = singletonList(new Split(emptyList(), "has-beta", null));
+    final List<Split> hasBetaSplits = singletonList(new Split(emptyList(), "has-beta", null, null));
     final Allocation hasBetaAllocation =
         new Allocation("has-beta-alloc", null, null, null, hasBetaSplits, false);
 
@@ -734,12 +734,13 @@ private Flag createOneOfRuleFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.ONE_OF, "region", allowedRegions));
     final List<Rule> regionalRules = singletonList(new Rule(regionalConditions));
-    final List<Split> regionalSplits = singletonList(new Split(emptyList(), "regional", null));
+    final List<Split> regionalSplits =
+        singletonList(new Split(emptyList(), "regional", null, null));
     final Allocation regionalAllocation =
         new Allocation("regional-alloc", regionalRules, null, null, regionalSplits, false);
 
     // Fallback
-    final List<Split> globalSplits = singletonList(new Split(emptyList(), "global", null));
+    final List<Split> globalSplits = singletonList(new Split(emptyList(), "global", null, null));
     final Allocation globalAllocation =
         new Allocation("global-alloc", null, null, null, globalSplits, false);
 
@@ -755,7 +756,7 @@ private Flag createTimeBasedFlag() {
     final Map<String, Variant> variants = new HashMap<>();
     variants.put("time-limited", new Variant("time-limited", "time-limited"));
 
-    final List<Split> splits = singletonList(new Split(emptyList(), "time-limited", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "time-limited", null, null));
 
     // Allocation that ended in 2022 (should be inactive)
     final List<Allocation> allocations =
@@ -779,7 +780,7 @@ private Flag createShardBasedFlag() {
     final List<ShardRange> ranges = singletonList(new ShardRange(0, 50)); // 0-49 out of 100
     final List<Shard> shards = singletonList(new Shard("test-salt", ranges, 100));
 
-    final List<Split> splits = singletonList(new Split(shards, "shard-variant", null));
+    final List<Split> splits = singletonList(new Split(shards, "shard-variant", null, null));
 
     final List<Allocation> allocations =
         singletonList(new Allocation("shard-alloc", null, null, null, splits, false));
@@ -792,7 +793,7 @@ private Flag createBrokenFlag() {
     final Map<String, Variant> variants = new HashMap<>();
     variants.put("existing", new Variant("existing", "value"));
 
-    final List<Split> splits = singletonList(new Split(emptyList(), "missing-variant", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "missing-variant", null, null));
 
     final List<Allocation> allocations =
         singletonList(new Allocation("alloc1", null, null, null, splits, false));
@@ -814,7 +815,7 @@ private Flag createComparisonFlag(
     final List<ConditionConfiguration> conditions =
         singletonList(new ConditionConfiguration(operator, attribute, threshold));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null));
+    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null, null));
     final Allocation allocation = new Allocation(allocKey, rules, null, null, splits, false);
 
     return new Flag(flagKey, true, ValueType.STRING, variants, singletonList(allocation));
@@ -849,7 +850,7 @@ private Flag createNotOperatorFlag(
     final List<ConditionConfiguration> conditions =
         singletonList(new ConditionConfiguration(operator, attribute, value));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null));
+    final List<Split> splits = singletonList(new Split(emptyList(), variantKey, null, null));
     final Allocation allocation = new Allocation(allocKey, rules, null, null, splits, false);
 
     return new Flag(flagKey, true, ValueType.STRING, variants, singletonList(allocation));
@@ -886,7 +887,7 @@ private Flag createDoubleEqualsFlag() {
     final List<ConditionConfiguration> piConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.MATCHES, "rate", "3.14159"));
     final List<Rule> piRules = singletonList(new Rule(piConditions));
-    final List<Split> piSplits = singletonList(new Split(emptyList(), "pi", null));
+    final List<Split> piSplits = singletonList(new Split(emptyList(), "pi", null, null));
     final Allocation piAllocation =
         new Allocation("pi-alloc", piRules, null, null, piSplits, false);
 
@@ -903,7 +904,7 @@ private Flag createNestedAttributeFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.MATCHES, "user.profile.level", "premium"));
     final List<Rule> premiumRules = singletonList(new Rule(premiumConditions));
-    final List<Split> premiumSplits = singletonList(new Split(emptyList(), "premium", null));
+    final List<Split> premiumSplits = singletonList(new Split(emptyList(), "premium", null, null));
     final Allocation premiumAllocation =
         new Allocation("premium-nested-alloc", premiumRules, null, null, premiumSplits, false);
 
@@ -915,7 +916,7 @@ private Flag createExposureFlag() {
     final Map<String, Variant> variants = new HashMap<>();
     variants.put("tracked", new Variant("tracked", "tracked-value"));
 
-    final List<Split> splits = singletonList(new Split(emptyList(), "tracked", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "tracked", null, null));
     // Create allocation with doLog=true to trigger exposure logging
     final List<Allocation> allocations =
         singletonList(new Allocation("exposure-alloc", null, null, null, splits, true));
@@ -931,7 +932,7 @@ private Flag createDoubleComparisonFlag() {
     final List<ConditionConfiguration> exactConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.LTE, "score", 3.14159));
     final List<Rule> exactRules = singletonList(new Rule(exactConditions));
-    final List<Split> exactSplits = singletonList(new Split(emptyList(), "exact", null));
+    final List<Split> exactSplits = singletonList(new Split(emptyList(), "exact", null, null));
     final Allocation exactAllocation =
         new Allocation("exact-alloc", exactRules, null, null, exactSplits, false);
 
@@ -947,7 +948,7 @@ private Flag createExposureLoggingFlag() {
     final List<ConditionConfiguration> loggedConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.MATCHES, "feature", "premium"));
     final List<Rule> loggedRules = singletonList(new Rule(loggedConditions));
-    final List<Split> loggedSplits = singletonList(new Split(emptyList(), "logged", null));
+    final List<Split> loggedSplits = singletonList(new Split(emptyList(), "logged", null, null));
     // Create allocation with doLog=true to trigger exposure logging and allocationKey method
     final Allocation loggedAllocation =
         new Allocation("logged-alloc", loggedRules, null, null, loggedSplits, true);
@@ -966,7 +967,8 @@ private Flag createNumericOneOfFlag() {
     final List<ConditionConfiguration> numericConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.ONE_OF, "score", numericValues));
     final List<Rule> numericRules = singletonList(new Rule(numericConditions));
-    final List<Split> numericSplits = singletonList(new Split(emptyList(), "numeric-match", null));
+    final List<Split> numericSplits =
+        singletonList(new Split(emptyList(), "numeric-match", null, null));
     final Allocation numericAllocation =
         new Allocation("numeric-alloc", numericRules, null, null, numericSplits, false);
 
@@ -985,7 +987,8 @@ private Flag createNumericNotOneOfFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_ONE_OF, "score", excludedValues));
     final List<Rule> excludedRules = singletonList(new Rule(excludedConditions));
-    final List<Split> excludedSplits = singletonList(new Split(emptyList(), "excluded", null));
+    final List<Split> excludedSplits =
+        singletonList(new Split(emptyList(), "excluded", null, null));
     final Allocation excludedAllocation =
         new Allocation("excluded-alloc", excludedRules, null, null, excludedSplits, false);
 
@@ -1005,7 +1008,7 @@ private Flag createIsNullFalseFlag() {
     final List<ConditionConfiguration> notNullConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.IS_NULL, "attr", false));
     final List<Rule> notNullRules = singletonList(new Rule(notNullConditions));
-    final List<Split> notNullSplits = singletonList(new Split(emptyList(), "not-null", null));
+    final List<Split> notNullSplits = singletonList(new Split(emptyList(), "not-null", null, null));
     final Allocation notNullAllocation =
         new Allocation("not-null-alloc", notNullRules, null, null, notNullSplits, false);
 
@@ -1022,7 +1025,7 @@ private Flag createIsNullNonBooleanFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.IS_NULL, "missing_attr", "string"));
     final List<Rule> nullRules = singletonList(new Rule(nullConditions));
-    final List<Split> nullSplits = singletonList(new Split(emptyList(), "null-match", null));
+    final List<Split> nullSplits = singletonList(new Split(emptyList(), "null-match", null, null));
     final Allocation nullAllocation =
         new Allocation("null-alloc", nullRules, null, null, nullSplits, false);
 
@@ -1042,7 +1045,8 @@ private Flag createNullAttributeFlag() {
     final List<ConditionConfiguration> nullAttrConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.MATCHES, null, "test"));
     final List<Rule> nullAttrRules = singletonList(new Rule(nullAttrConditions));
-    final List<Split> nullAttrSplits = singletonList(new Split(emptyList(), "fallback", null));
+    final List<Split> nullAttrSplits =
+        singletonList(new Split(emptyList(), "fallback", null, null));
     final Allocation nullAttrAllocation =
         new Allocation("null-attr-alloc", nullAttrRules, null, null, nullAttrSplits, false);
 
@@ -1059,7 +1063,8 @@ private Flag createNotMatchesPositiveFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_MATCHES, "email", "@company\\.com"));
     final List<Rule> externalRules = singletonList(new Rule(externalConditions));
-    final List<Split> externalSplits = singletonList(new Split(emptyList(), "external", null));
+    final List<Split> externalSplits =
+        singletonList(new Split(emptyList(), "external", null, null));
     final Allocation externalAllocation =
         new Allocation("external-alloc", externalRules, null, null, externalSplits, false);
 
@@ -1081,7 +1086,7 @@ private Flag createNotOneOfPositiveFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_ONE_OF, "region", excludedRegions));
     final List<Rule> otherRules = singletonList(new Rule(otherConditions));
-    final List<Split> otherSplits = singletonList(new Split(emptyList(), "other", null));
+    final List<Split> otherSplits = singletonList(new Split(emptyList(), "other", null, null));
     final Allocation otherAllocation =
         new Allocation("other-alloc", otherRules, null, null, otherSplits, false);
 
@@ -1101,7 +1106,8 @@ private Flag createFalseNumericComparisonsFlag() {
     final List<ConditionConfiguration> highScoreConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.GTE, "score", 800));
     final List<Rule> highScoreRules = singletonList(new Rule(highScoreConditions));
-    final List<Split> highScoreSplits = singletonList(new Split(emptyList(), "high-score", null));
+    final List<Split> highScoreSplits =
+        singletonList(new Split(emptyList(), "high-score", null, null));
     final Allocation highScoreAllocation =
         new Allocation("high-score-alloc", highScoreRules, null, null, highScoreSplits, false);
 
@@ -1131,7 +1137,7 @@ private Flag createEmptyConditionsFlag() {
 
     // Rule with empty conditions list - this will be skipped, causing allocation to not match
     final Rule emptyConditionsRule = new Rule(emptyList());
-    final List<Split> splits = singletonList(new Split(emptyList(), "default", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "default", null, null));
     final Allocation allocation =
         new Allocation(
             "empty-conditions-alloc",
@@ -1153,7 +1159,7 @@ private Flag createShardMatchingFlag() {
     final List<ShardRange> ranges =
         singletonList(new ShardRange(0, 100)); // Full range to ensure match
     final List<Shard> shards = singletonList(new Shard("test-salt", ranges, 100));
-    final List<Split> splits = singletonList(new Split(shards, "matched", null));
+    final List<Split> splits = singletonList(new Split(shards, "matched", null, null));
     final Allocation allocation =
         new Allocation("shard-matching-alloc", null, null, null, splits, false);
 
@@ -1165,7 +1171,7 @@ private Flag createFutureAllocationFlag() {
     final Map<String, Variant> variants = new HashMap<>();
     variants.put("future", new Variant("future", "future-value"));
 
-    final List<Split> splits = singletonList(new Split(emptyList(), "future", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "future", null, null));
 
     // Allocation that starts in the future (2050)
     final Allocation allocation =
@@ -1185,7 +1191,7 @@ private Flag createIdAttributeFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.MATCHES, "id", "user-special-id"));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "id-match", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "id-match", null, null));
     final Allocation allocation = new Allocation("id-attr-alloc", rules, null, null, splits, false);
 
     return new Flag(
@@ -1200,7 +1206,7 @@ private Flag createNonIterableConditionFlag() {
     final List<ConditionConfiguration> conditions =
         singletonList(new ConditionConfiguration(ConditionOperator.ONE_OF, "attr", "single-value"));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "no-match", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "no-match", null, null));
     final Allocation allocation =
         new Allocation("non-iterable-alloc", rules, null, null, splits, false);
 
@@ -1250,7 +1256,7 @@ private Flag createNotMatchesFalseFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_MATCHES, "email", "@company\\.com"));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "internal", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "internal", null, null));
     final Allocation allocation =
         new Allocation("not-matches-false-alloc", rules, null, null, splits, false);
 
@@ -1269,7 +1275,7 @@ private Flag createNotOneOfFalseFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_ONE_OF, "region", excludedRegions));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "excluded", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "excluded", null, null));
     final Allocation allocation =
         new Allocation("not-one-of-false-alloc", rules, null, null, splits, false);
 
@@ -1285,7 +1291,7 @@ private Flag createNullContextValuesFlag() {
     final List<ConditionConfiguration> conditions =
         singletonList(new ConditionConfiguration(ConditionOperator.IS_NULL, "nullAttr", true));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "null-variant", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "null-variant", null, null));
     final Allocation allocation =
         new Allocation("null-context-alloc", rules, null, null, splits, false);
 
@@ -1303,12 +1309,12 @@ private Flag createCountryRuleFlag() {
     final List<ConditionConfiguration> usConditions =
         singletonList(new ConditionConfiguration(ConditionOperator.ONE_OF, "country", usCountries));
     final List<Rule> usRules = singletonList(new Rule(usConditions));
-    final List<Split> usSplits = singletonList(new Split(emptyList(), "us", null));
+    final List<Split> usSplits = singletonList(new Split(emptyList(), "us", null, null));
     final Allocation usAllocation =
         new Allocation("us-alloc", usRules, null, null, usSplits, false);
 
     // Fallback allocation (no rules, no shards)
-    final List<Split> globalSplits = singletonList(new Split(emptyList(), "global", null));
+    final List<Split> globalSplits = singletonList(new Split(emptyList(), "global", null, null));
     final Allocation globalAllocation =
         new Allocation("global-alloc", null, null, null, globalSplits, false);
 
@@ -1328,7 +1334,7 @@ private Flag createInvalidRegexFlag() {
     final List<ConditionConfiguration> conditions =
         singletonList(new ConditionConfiguration(ConditionOperator.MATCHES, "email", "[invalid"));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "matched", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "matched", null, null));
     final Allocation allocation =
         new Allocation("invalid-regex-alloc", rules, null, null, splits, false);
 
@@ -1345,7 +1351,7 @@ private Flag createInvalidRegexNotMatchesFlag() {
         singletonList(
             new ConditionConfiguration(ConditionOperator.NOT_MATCHES, "email", "[invalid"));
     final List<Rule> rules = singletonList(new Rule(conditions));
-    final List<Split> splits = singletonList(new Split(emptyList(), "excluded", null));
+    final List<Split> splits = singletonList(new Split(emptyList(), "excluded", null, null));
     final Allocation allocation =
         new Allocation("invalid-regex-not-matches-alloc", rules, null, null, splits, false);
 
diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentHookTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentHookTest.java
new file mode 100644
index 00000000000..b3d2a197d39
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentHookTest.java
@@ -0,0 +1,492 @@
+package datadog.trace.api.openfeature;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import datadog.trace.api.DDTraceId;
+import datadog.trace.api.interceptor.MutableSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import dev.openfeature.sdk.FlagEvaluationDetails;
+import dev.openfeature.sdk.FlagValueType;
+import dev.openfeature.sdk.Hook;
+import dev.openfeature.sdk.HookContext;
+import dev.openfeature.sdk.ImmutableContext;
+import dev.openfeature.sdk.ImmutableMetadata;
+import dev.openfeature.sdk.ImmutableStructure;
+import dev.openfeature.sdk.Value;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.TreeSet;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit suite for APM feature-flag span enrichment.
+ *
+ * <p>Covers the seven required validation cases plus the explicit max-200 case and the codec
+ * golden-vector round-trip. The contract (encoding, limits, tag shapes) is FROZEN against the Node
+ * reference ({@code dd-trace-js#8343}).
+ */
+class SpanEnrichmentHookTest {
+
+  // Instance-owned state store shared by the hook and interceptor under test, mirroring how a
+  // single
+  // Provider wires them.
+  private SpanEnrichmentStates states;
+
+  @BeforeEach
+  void freshState() {
+    states = new SpanEnrichmentStates();
+  }
+
+  @AfterEach
+  void clearState() {
+    states.clear();
+    // The interceptor is a process-wide singleton; leave it inert for the next test.
+    SpanEnrichmentInterceptor.INSTANCE.unbind(SpanEnrichmentInterceptor.INSTANCE.activeStates());
+  }
+
+  // ---- helpers ----
+
+  /** The store key the production code derives from a trace id (full hex). */
+  private static String key(final long traceId) {
+    return DDTraceId.from(traceId).toHexString();
+  }
+
+  private static FlagEvaluationDetails<Object> details(
+      final String flagKey,
+      final String variant,
+      final Object value,
+      final ImmutableMetadata metadata) {
+    return FlagEvaluationDetails.builder()
+        .flagKey(flagKey)
+        .variant(variant)
+        .value(value)
+        .flagMetadata(metadata)
+        .build();
+  }
+
+  private static ImmutableMetadata metadata(final Integer serialId, final boolean doLog) {
+    final ImmutableMetadata.ImmutableMetadataBuilder builder = ImmutableMetadata.builder();
+    if (serialId != null) {
+      builder.addString(SpanEnrichmentHook.METADATA_SERIAL_ID, serialId.toString());
+    }
+    builder.addString(SpanEnrichmentHook.METADATA_DO_LOG, String.valueOf(doLog));
+    return builder.build();
+  }
+
+  private static HookContext<Object> ctx(final String flagKey, final String targetingKey) {
+    return HookContext.from(
+        flagKey,
+        FlagValueType.STRING,
+        null,
+        null,
+        targetingKey == null ? new ImmutableContext() : new ImmutableContext(targetingKey),
+        "default");
+  }
+
+  /** Drives the capture branch directly (no static tracer) for a fixed trace id. */
+  private static void capture(
+      final SpanEnrichmentHook hook,
+      final long traceId,
+      final String flagKey,
+      final String targetingKey,
+      final String variant,
+      final Object value,
+      final Integer serialId,
+      final boolean doLog) {
+    hook.capture(
+        key(traceId),
+        ctx(flagKey, targetingKey),
+        details(flagKey, variant, value, metadata(serialId, doLog)));
+  }
+
+  /**
+   * Configures a mock root span to report itself as its own local root with the given trace id. The
+   * returned span, passed in a singleton collection, models a FINAL flush (the root is present in
+   * the fragment), so the interceptor flushes + removes state.
+   */
+  private static AgentSpan rootSpanCollection(final long traceId, final AgentSpan rootSpan) {
+    when(rootSpan.getLocalRootSpan()).thenReturn(rootSpan);
+    when(rootSpan.getTraceId()).thenReturn(DDTraceId.from(traceId));
+    return rootSpan;
+  }
+
+  /** Binds a fresh interceptor to the given store, models a final flush, and returns it. */
+  private static SpanEnrichmentInterceptor boundInterceptor(final SpanEnrichmentStates states) {
+    SpanEnrichmentInterceptor.INSTANCE.bind(states);
+    return SpanEnrichmentInterceptor.INSTANCE;
+  }
+
+  // ---- 1. codec golden-vector round-trip ----
+
+  @Test
+  void codecGoldenVectorAndRoundTrip() {
+    final SortedSet<Integer> ids = new TreeSet<>();
+    ids.add(100);
+    ids.add(108);
+    ids.add(128);
+    ids.add(130);
+    final String encoded = ULeb128Encoder.encodeDeltaVarint(ids);
+    assertEquals("ZAgUAg==", encoded, "golden vector must match the frozen Node contract");
+    // round-trip: decode back to the same ascending ids
+    assertEquals(ids, ULeb128Encoder.decodeDeltaVarint(encoded));
+    // empty set -> empty string (tag omitted)
+    assertEquals("", ULeb128Encoder.encodeDeltaVarint(Collections.emptySet()));
+    // dedupe is structural: a duplicate id does not change the encoding
+    final SortedSet<Integer> withDup = new TreeSet<>(ids);
+    withDup.add(100);
+    assertEquals(encoded, ULeb128Encoder.encodeDeltaVarint(withDup));
+  }
+
+  // ---- 2. no-span (no active root) ----
+
+  @Test
+  void noActiveSpanDoesNotCrashOrAccumulate() {
+    // Injected resolver returns null => no active local root (the no-span case).
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(() -> null, states);
+    // Should be a no-op, never throw.
+    hook.finallyAfter(
+        ctx("flag", "user-1"),
+        details("flag", "on", "v", metadata(100, true)),
+        Collections.emptyMap());
+    assertTrue(states.isEmpty(), "no active span => no accumulator state");
+  }
+
+  @Test
+  void finallyAfterResolvesRootViaResolverAndAccumulates() {
+    // Drives finallyAfter end-to-end with an injected root resolver (no static mocks).
+    final AgentSpan root = mock(AgentSpan.class);
+    when(root.getTraceId()).thenReturn(DDTraceId.from(0x77L));
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(() -> root, states);
+    hook.finallyAfter(
+        ctx("flag", "user-1"),
+        details("flag", "on", "v", metadata(42, true)),
+        Collections.emptyMap());
+    final SpanEnrichmentAccumulator state = states.peek(key(0x77L));
+    assertTrue(state != null && state.serialIdsView().contains(42));
+    assertEquals(1, state.subjectCount(), "doLog=true + targeting key => subject recorded");
+  }
+
+  // ---- 3. finished-root (accumulate + dedupe, then flush via interceptor) ----
+
+  @Test
+  void finishedRootFlushesFlagsEncTag() {
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final long traceId = 0xABCDL;
+    // accumulate {100,108,128,130} with a duplicate 100 to prove dedupe
+    capture(hook, traceId, "f1", "user-1", "on", "v", 100, false);
+    capture(hook, traceId, "f2", "user-1", "on", "v", 108, false);
+    capture(hook, traceId, "f3", "user-1", "on", "v", 128, false);
+    capture(hook, traceId, "f4", "user-1", "on", "v", 130, false);
+    capture(hook, traceId, "f1", "user-1", "on", "v", 100, false); // dup
+
+    final AgentSpan root = mock(AgentSpan.class);
+    rootSpanCollection(traceId, root);
+    boundInterceptor(states).onTraceComplete(Collections.singletonList(root));
+
+    verify(root).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "ZAgUAg==");
+    // state cleared after flush
+    assertTrue(states.isEmpty(), "state must be cleared on flush");
+  }
+
+  // ---- 4. error/default variant (missing variant -> ffe_runtime_defaults JSON object) ----
+
+  @Test
+  void runtimeDefaultMissingVariantWritesJsonObject() {
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final long traceId = 0x1234L;
+    // no serial id + null variant => runtime default; object value must be JSON-stringified
+    final Map<String, Object> objectValue = Collections.singletonMap("k", "val");
+    hook.capture(
+        key(traceId),
+        ctx("obj-flag", "user-1"),
+        FlagEvaluationDetails.builder()
+            .flagKey("obj-flag")
+            .variant(null)
+            .value(objectValue)
+            .flagMetadata(ImmutableMetadata.builder().build())
+            .build());
+
+    final AgentSpan root = mock(AgentSpan.class);
+    rootSpanCollection(traceId, root);
+    boundInterceptor(states).onTraceComplete(Collections.singletonList(root));
+
+    // ffe_runtime_defaults is a JSON object string, NOT [object Object]/toString.
+    verify(root)
+        .setTag(
+            SpanEnrichmentAccumulator.TAG_RUNTIME_DEFAULTS,
+            "{\"obj-flag\":\"{\\\"k\\\":\\\"val\\\"}\"}");
+    // no flags tag when there are no serial ids
+    verify(root, never()).setTag(eq(SpanEnrichmentAccumulator.TAG_FLAGS_ENC), anyString());
+  }
+
+  // ---- 4b. real OpenFeature object path: a Value structure default must serialize to JSON ----
+
+  /**
+   * The real object-evaluation path hands the runtime default in as a {@code
+   * dev.openfeature.sdk.Value}, not a raw {@code Map}. The accumulator must unwrap the {@code
+   * Value} and emit JSON (matching Node's {@code JSON.stringify}), never {@code Value.toString()}
+   * (which is {@code "Value(innerObject=...)"}).
+   */
+  @Test
+  void runtimeDefaultStructureValueSerializesAsJsonNotToString() {
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final long traceId = 0x2468L;
+
+    // Single-key structure so the exact-string assertion does not depend on the OpenFeature SDK's
+    // internal key ordering. (The multi-key / nesting behaviour is covered by the direct
+    // stringifyDefault assertions below.)
+    final Map<String, Value> inner = new LinkedHashMap<>();
+    inner.put("enabled", new Value(true));
+    final Value structureDefault = new Value(new ImmutableStructure(inner));
+
+    hook.capture(
+        key(traceId),
+        ctx("struct-flag", "user-1"),
+        FlagEvaluationDetails.builder()
+            .flagKey("struct-flag")
+            .variant(null)
+            .value(structureDefault)
+            .flagMetadata(ImmutableMetadata.builder().build())
+            .build());
+
+    final AgentSpan root = mock(AgentSpan.class);
+    rootSpanCollection(traceId, root);
+    boundInterceptor(states).onTraceComplete(Collections.singletonList(root));
+
+    // {"struct-flag":"{\"enabled\":true}"}  — note: NO "Value(innerObject=...)".
+    verify(root)
+        .setTag(
+            SpanEnrichmentAccumulator.TAG_RUNTIME_DEFAULTS,
+            "{\"struct-flag\":\"{\\\"enabled\\\":true}\"}");
+  }
+
+  /**
+   * Direct stringify of a structured {@code Value}: asserts the value is JSON (objects + nested
+   * scalars), not {@code Value.toString()}. The structure is single-key to keep ordering
+   * deterministic across OpenFeature SDK versions.
+   */
+  @Test
+  void stringifyDefaultUnwrapsValueStructureToJson() {
+    final Map<String, Value> inner = new LinkedHashMap<>();
+    inner.put("count", new Value(42));
+    final Value structureDefault = new Value(new ImmutableStructure(inner));
+    assertEquals("{\"count\":42}", SpanEnrichmentAccumulator.stringifyDefault(structureDefault));
+  }
+
+  /**
+   * A list-valued {@code Value} default serializes to a JSON array, with nested Values unwrapped.
+   */
+  @Test
+  void runtimeDefaultListValueSerializesAsJsonArray() {
+    final Value listDefault =
+        new Value(Arrays.asList(new Value("a"), new Value(2), new Value(true)));
+    // direct stringify assertion (exact bytes)
+    assertEquals("[\"a\",2,true]", SpanEnrichmentAccumulator.stringifyDefault(listDefault));
+  }
+
+  /** Scalar Values collapse to the same string form Node's String(value) produces. */
+  @Test
+  void scalarValueDefaultsMatchNodeStringForm() {
+    assertEquals("hello", SpanEnrichmentAccumulator.stringifyDefault(new Value("hello")));
+    assertEquals("true", SpanEnrichmentAccumulator.stringifyDefault(new Value(true)));
+    assertEquals("7", SpanEnrichmentAccumulator.stringifyDefault(new Value(7)));
+    assertEquals("null", SpanEnrichmentAccumulator.stringifyDefault(new Value()));
+  }
+
+  // ---- 5. per-subject cap (10 subjects / 20 experiments / doLog gating) ----
+
+  @Test
+  void subjectCapsAndDoLogGating() {
+    final SpanEnrichmentAccumulator acc = new SpanEnrichmentAccumulator();
+
+    // doLog gating: addSubject only happens when doLog true (driven via the hook branch below).
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final long traceId = 0x5L;
+    capture(hook, traceId, "f", "user-A", "on", "v", 1, false); // doLog=false => no subject
+    assertEquals(
+        0, states.peek(key(traceId)).subjectCount(), "doLog=false must not record a subject");
+    capture(hook, traceId, "f", "user-A", "on", "v", 2, true); // doLog=true => subject recorded
+    assertEquals(1, states.peek(key(traceId)).subjectCount());
+
+    // per-subject experiment cap: 20 max
+    for (int i = 0; i < 25; i++) {
+      acc.addSubject("subjectX", i);
+    }
+    final Map<String, String> tags = acc.toSpanTags();
+    final SortedSet<Integer> decoded =
+        ULeb128Encoder.decodeDeltaVarint(
+            // ffe_subjects_enc is {"<sha>":"<base64>"}; extract the single base64 value
+            tags.get(SpanEnrichmentAccumulator.TAG_SUBJECTS_ENC)
+                .replaceAll("^\\{\"[a-f0-9]+\":\"", "")
+                .replaceAll("\"\\}$", ""));
+    assertEquals(SpanEnrichmentAccumulator.MAX_EXPERIMENTS_PER_SUBJECT, decoded.size());
+
+    // subject cap: 10 max distinct subjects
+    final SpanEnrichmentAccumulator acc2 = new SpanEnrichmentAccumulator();
+    for (int i = 0; i < 15; i++) {
+      acc2.addSubject("subject-" + i, i);
+    }
+    assertEquals(SpanEnrichmentAccumulator.MAX_SUBJECTS, acc2.subjectCount());
+  }
+
+  // ---- 6. max-200 serial ids ----
+
+  @Test
+  void max200SerialIdsEnforced() {
+    final SpanEnrichmentAccumulator acc = new SpanEnrichmentAccumulator();
+    for (int i = 0; i < 300; i++) {
+      acc.addSerialId(i);
+    }
+    assertEquals(
+        SpanEnrichmentAccumulator.MAX_SERIAL_IDS,
+        acc.serialIdsView().size(),
+        "serial ids must be capped at 200");
+  }
+
+  // ---- 7. JSON/object-default (json not toString + 64-char truncation) ----
+
+  @Test
+  void objectDefaultJsonAndTruncation() {
+    // object -> JSON, not toString
+    assertEquals(
+        "{\"a\":\"b\"}",
+        SpanEnrichmentAccumulator.stringifyDefault(Collections.singletonMap("a", "b")));
+    // scalar string -> as-is
+    assertEquals("hello", SpanEnrichmentAccumulator.stringifyDefault("hello"));
+    // null -> "null"
+    assertEquals("null", SpanEnrichmentAccumulator.stringifyDefault(null));
+
+    // 64-char truncation (first-wins handled separately)
+    final StringBuilder longValue = new StringBuilder();
+    for (int i = 0; i < 100; i++) {
+      longValue.append('x');
+    }
+    final SpanEnrichmentAccumulator acc = new SpanEnrichmentAccumulator();
+    acc.addDefault("flag", longValue.toString());
+    final String tag = acc.toSpanTags().get(SpanEnrichmentAccumulator.TAG_RUNTIME_DEFAULTS);
+    // {"flag":"<64 x's>"}
+    final String expectedValue =
+        longValue.substring(0, SpanEnrichmentAccumulator.MAX_DEFAULT_VALUE_LENGTH);
+    assertEquals("{\"flag\":\"" + expectedValue + "\"}", tag);
+
+    // first-wins: a second addDefault for the same flag is ignored
+    acc.addDefault("flag", "second");
+    assertEquals(1, acc.defaultCount());
+  }
+
+  // ---- gate-off negative control (no ffe_*, no hook, no state) ----
+
+  @Test
+  void gateOffConstructsNothingAndAccumulatesNoState() {
+    // Gate OFF via the injectable override (no static config mocking).
+    final Provider provider = new Provider(new Provider.Options(), null, Boolean.FALSE);
+
+    // No hook, no state store constructed (zero idle overhead).
+    assertNull(provider.spanEnrichmentHook(), "gate off => no span-enrichment hook");
+    assertNull(provider.spanEnrichmentStates(), "gate off => no span-enrichment state store");
+    // getProviderHooks must not contain a SpanEnrichmentHook.
+    final List<Hook> hooks = provider.getProviderHooks();
+    for (final Hook hook : hooks) {
+      assertFalse(
+          hook instanceof SpanEnrichmentHook, "gate off => SpanEnrichmentHook never registered");
+    }
+  }
+
+  /**
+   * getProviderHooks() is called on every evaluation; it must return the SAME precomputed list
+   * (allocating nothing) regardless of gate state.
+   */
+  @Test
+  void getProviderHooksReturnsSameInstanceEachCall() {
+    final Provider gateOff = new Provider(new Provider.Options(), null, Boolean.FALSE);
+    assertTrue(
+        gateOff.getProviderHooks() == gateOff.getProviderHooks(),
+        "gate off => getProviderHooks allocates nothing (same instance)");
+
+    final Provider gateOn =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    assertTrue(
+        gateOn.getProviderHooks() == gateOn.getProviderHooks(),
+        "gate on => getProviderHooks allocates nothing (same instance)");
+  }
+
+  // ---- gate-on construction + provider-close cleanup ----
+
+  @Test
+  void gateOnConstructsHookAndStateThenShutdownUnbinds() {
+    final Provider provider =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    assertTrue(provider.spanEnrichmentHook() != null, "gate on => hook constructed");
+    assertTrue(provider.spanEnrichmentStates() != null, "gate on => state store constructed");
+    // the process-wide interceptor is bound to this provider's store
+    assertTrue(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates() == provider.spanEnrichmentStates(),
+        "gate on => interceptor bound to this provider's store");
+    // hook registered in provider hooks
+    boolean registered = false;
+    for (final Hook hook : provider.getProviderHooks()) {
+      if (hook instanceof SpanEnrichmentHook) {
+        registered = true;
+      }
+    }
+    assertTrue(registered, "gate on => SpanEnrichmentHook registered in getProviderHooks");
+
+    // provider close unbinds + drains ITS OWN state.
+    final SpanEnrichmentStates providerStates = provider.spanEnrichmentStates();
+    providerStates.getOrCreate(key(1L));
+    assertFalse(providerStates.isEmpty());
+    provider.shutdown();
+    assertNull(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates(), "shutdown unbinds the active store");
+    assertTrue(providerStates.isEmpty(), "shutdown drains residual state");
+  }
+
+  // ---- error isolation: enrichment never throws ----
+
+  @Test
+  void captureNeverThrowsOnNullInputs() {
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    // null details handled in finallyAfter; capture with empty metadata + null variant
+    hook.finallyAfter(null, null, null);
+    assertTrue(states.isEmpty());
+  }
+
+  // ---- interceptor inert when unbound / empty trace robustness ----
+
+  @Test
+  void interceptorNoOpsWhenUnboundOrEmpty() {
+    // empty trace
+    SpanEnrichmentInterceptor.INSTANCE.bind(states);
+    assertTrue(
+        SpanEnrichmentInterceptor.INSTANCE.onTraceComplete(Collections.emptyList()).isEmpty());
+    // unbound (inert)
+    SpanEnrichmentInterceptor.INSTANCE.unbind(states);
+    final AgentSpan root = mock(AgentSpan.class);
+    final List<MutableSpan> trace = Collections.singletonList(root);
+    SpanEnrichmentInterceptor.INSTANCE.onTraceComplete(trace);
+    verify(root, never()).setTag(anyString(), anyString());
+  }
+
+  // ---- interceptor priority uniqueness ----
+
+  @Test
+  void interceptorPriorityIsUnique() {
+    // Distinct from AbstractTraceInterceptor.Priority values (0,1,2,3, MAX-2, MAX-1, MAX).
+    assertEquals(4, SpanEnrichmentInterceptor.INSTANCE.priority());
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentLifecycleRegressionTest.java b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentLifecycleRegressionTest.java
new file mode 100644
index 00000000000..7e4d00b5875
--- /dev/null
+++ b/products/feature-flagging/feature-flagging-api/src/test/java/datadog/trace/api/openfeature/SpanEnrichmentLifecycleRegressionTest.java
@@ -0,0 +1,388 @@
+package datadog.trace.api.openfeature;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import datadog.trace.api.DD128bTraceId;
+import datadog.trace.api.DDTraceId;
+import datadog.trace.api.interceptor.MutableSpan;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import dev.openfeature.sdk.EvaluationContext;
+import dev.openfeature.sdk.FlagEvaluationDetails;
+import dev.openfeature.sdk.FlagValueType;
+import dev.openfeature.sdk.HookContext;
+import dev.openfeature.sdk.ImmutableContext;
+import dev.openfeature.sdk.ImmutableMetadata;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Regression suite for the per-root-span lifecycle and reconfiguration behaviour of span
+ * enrichment. Each test exercises a scenario the single-root, single-thread, single-provider happy
+ * path does not cover.
+ *
+ * <ul>
+ *   <li>partial flush — a fragment of child spans only (the still-open root is excluded by
+ *       dd-trace-core) must NOT drain the accumulator; pre-flush flags must survive onto the root
+ *       at final completion.
+ *   <li>unbounded leak — traces that never reach the interceptor (dropped / Noop / never-finishing)
+ *       must not leak accumulator entries unboundedly; the store is hard-bounded.
+ *   <li>reconfiguration — after a provider closes, a new gate-on provider must still enrich (the
+ *       process-wide interceptor rebinds), and a closing provider must never clobber a newer
+ *       provider's in-flight state.
+ *   <li>128-bit trace ids — two distinct 128-bit traces that share their low-order 64 bits must not
+ *       merge enrichment state.
+ * </ul>
+ */
+class SpanEnrichmentLifecycleRegressionTest {
+
+  @AfterEach
+  void resetInterceptor() {
+    // The interceptor is a process-wide singleton; leave it inert for the next test.
+    SpanEnrichmentInterceptor.INSTANCE.unbind(SpanEnrichmentInterceptor.INSTANCE.activeStates());
+  }
+
+  // ---- helpers ----
+
+  private static String key(final long traceId) {
+    return DDTraceId.from(traceId).toHexString();
+  }
+
+  private static ImmutableMetadata serialMeta(final int serialId, final boolean doLog) {
+    return ImmutableMetadata.builder()
+        .addString(SpanEnrichmentHook.METADATA_SERIAL_ID, Integer.toString(serialId))
+        .addString(SpanEnrichmentHook.METADATA_DO_LOG, String.valueOf(doLog))
+        .build();
+  }
+
+  private static FlagEvaluationDetails<Object> serialDetails(
+      final String flagKey, final int serialId, final boolean doLog) {
+    return FlagEvaluationDetails.builder()
+        .flagKey(flagKey)
+        .variant("on")
+        .value("v")
+        .flagMetadata(serialMeta(serialId, doLog))
+        .build();
+  }
+
+  private static HookContext<Object> ctx(final String flagKey, final String targetingKey) {
+    final EvaluationContext ec =
+        targetingKey == null ? new ImmutableContext() : new ImmutableContext(targetingKey);
+    return HookContext.from(flagKey, FlagValueType.STRING, null, null, ec, "default");
+  }
+
+  /** A mock child span whose local root is {@code root} but which is itself NOT the root. */
+  private static AgentSpan childOf(final AgentSpan root) {
+    final AgentSpan child = mock(AgentSpan.class);
+    when(child.getLocalRootSpan()).thenReturn(root);
+    return child;
+  }
+
+  /** A mock local-root span reporting itself as its own local root with the given trace id. */
+  private static AgentSpan rootSpan(final DDTraceId traceId) {
+    final AgentSpan root = mock(AgentSpan.class);
+    when(root.getLocalRootSpan()).thenReturn(root);
+    when(root.getTraceId()).thenReturn(traceId);
+    return root;
+  }
+
+  private static AgentSpan rootSpan(final long traceId) {
+    return rootSpan(DDTraceId.from(traceId));
+  }
+
+  // =====================================================================================
+  // partial flush must not drop captured state or misattribute tags
+  // =====================================================================================
+
+  /**
+   * Models the real dd-trace-core sequence for a long-running trace:
+   *
+   * <ol>
+   *   <li>flags are evaluated (captured into the accumulator),
+   *   <li>a PARTIAL flush fires {@code onTraceComplete} with a fragment of CHILD spans only — the
+   *       still-open local root is excluded (held back via {@code rootSpanWritten}),
+   *   <li>more flags are evaluated,
+   *   <li>the FINAL flush fires {@code onTraceComplete} with the root present.
+   * </ol>
+   *
+   * All serial ids — both pre- and post-partial-flush — must appear on the root's {@code
+   * ffe_flags_enc} tag, and nothing must be written on the partial flush.
+   */
+  @Test
+  void partialFlushExcludingRootPreservesPreFlushFlags() {
+    final SpanEnrichmentStates states = new SpanEnrichmentStates();
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final SpanEnrichmentInterceptor interceptor = SpanEnrichmentInterceptor.INSTANCE;
+    interceptor.bind(states);
+
+    final long traceId = 0x10A6L;
+    final AgentSpan root = rootSpan(traceId);
+
+    // (1) pre-partial-flush evaluations
+    hook.capture(key(traceId), ctx("f1", "user-1"), serialDetails("f1", 100, false));
+    hook.capture(key(traceId), ctx("f2", "user-1"), serialDetails("f2", 108, false));
+
+    // (2) PARTIAL flush: a fragment of children only — the open root is NOT in the collection.
+    final AgentSpan child1 = childOf(root);
+    final AgentSpan child2 = childOf(root);
+    final List<MutableSpan> partialFragment = Arrays.asList(child1, child2);
+    interceptor.onTraceComplete(partialFragment);
+
+    // No tags may be written on a partial flush, and the accumulator must survive intact.
+    verify(child1, never()).setTag(anyString(), anyString());
+    verify(child2, never()).setTag(anyString(), anyString());
+    verify(root, never()).setTag(anyString(), anyString());
+    final SpanEnrichmentAccumulator surviving = states.peek(key(traceId));
+    assertNotNull(surviving, "partial flush must NOT remove the accumulator");
+    assertTrue(
+        surviving.serialIdsView().contains(100) && surviving.serialIdsView().contains(108),
+        "pre-flush serial ids must survive a partial flush");
+
+    // (3) more evaluations after the partial flush
+    hook.capture(key(traceId), ctx("f3", "user-1"), serialDetails("f3", 128, false));
+    hook.capture(key(traceId), ctx("f4", "user-1"), serialDetails("f4", 130, false));
+
+    // (4) FINAL flush: the root is present in the fragment (alongside a late child).
+    final AgentSpan lateChild = childOf(root);
+    interceptor.onTraceComplete(Arrays.asList(lateChild, root));
+
+    // The full set {100,108,128,130} -> golden "ZAgUAg==" must land on the root.
+    verify(root).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "ZAgUAg==");
+    assertTrue(states.isEmpty(), "state must be removed only on the final flush");
+  }
+
+  /**
+   * A partial flush whose first span reports a non-null local root that is absent from the fragment
+   * must be treated as "not the final write" and leave state intact — even when there are several
+   * children. Guards the "never tag a non-root span" concern.
+   */
+  @Test
+  void partialFlushNeverWritesTagsOnAChildSpan() {
+    final SpanEnrichmentStates states = new SpanEnrichmentStates();
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final SpanEnrichmentInterceptor interceptor = SpanEnrichmentInterceptor.INSTANCE;
+    interceptor.bind(states);
+
+    final long traceId = 0xC0DEL;
+    final AgentSpan root = rootSpan(traceId);
+    hook.capture(key(traceId), ctx("f", "user-1"), serialDetails("f", 7, false));
+
+    final AgentSpan child = childOf(root);
+    interceptor.onTraceComplete(Collections.singletonList(child));
+
+    verify(child, never()).setTag(anyString(), anyString());
+    verify(root, never()).setTag(anyString(), anyString());
+    assertNotNull(states.peek(key(traceId)), "child-only fragment must keep state");
+  }
+
+  // =====================================================================================
+  // never-completing traces must not leak accumulator entries unboundedly
+  // =====================================================================================
+
+  /**
+   * Simulates a sustained stream of traces that each evaluate a flag but never reach the
+   * interceptor (dropped traces / Noop tracer / never-finishing roots). The store must stay bounded
+   * by {@link SpanEnrichmentStates#MAX_TRACES} rather than growing without limit.
+   */
+  @Test
+  void neverCompletingTracesDoNotLeakUnbounded() {
+    final SpanEnrichmentStates states = new SpanEnrichmentStates();
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+
+    final int churn = SpanEnrichmentStates.MAX_TRACES * 3;
+    for (int i = 0; i < churn; i++) {
+      // Each distinct trace id accumulates once and is NEVER flushed (no interceptor call).
+      hook.capture(key(i), ctx("f", "user-" + i), serialDetails("f", i % 1000 + 1, false));
+      assertTrue(
+          states.size() <= SpanEnrichmentStates.MAX_TRACES,
+          "state store must stay bounded for never-completing traces");
+    }
+    assertEquals(
+        SpanEnrichmentStates.MAX_TRACES, states.size(), "store saturates at the cap, never beyond");
+  }
+
+  /**
+   * Bounding is FIFO by insertion: once the cap is reached, the oldest entries are evicted so the
+   * newest in-flight traces are retained.
+   */
+  @Test
+  void boundedStoreEvictsOldestFirst() {
+    final SpanEnrichmentStates states = new SpanEnrichmentStates();
+    // Fill exactly to the cap with ids [0, MAX).
+    for (long i = 0; i < SpanEnrichmentStates.MAX_TRACES; i++) {
+      states.getOrCreate(key(i));
+    }
+    assertEquals(SpanEnrichmentStates.MAX_TRACES, states.size());
+    // One more distinct id evicts the oldest (id 0) and keeps the rest + the new one.
+    final long overflow = SpanEnrichmentStates.MAX_TRACES;
+    states.getOrCreate(key(overflow));
+    assertEquals(SpanEnrichmentStates.MAX_TRACES, states.size(), "size stays at the cap");
+    assertNull(states.peek(key(0L)), "oldest entry evicted first (FIFO)");
+    assertNotNull(states.peek(key(overflow)), "newest entry retained");
+    assertNotNull(states.peek(key(1L)), "second-oldest still present");
+  }
+
+  // =====================================================================================
+  // reconfiguration: a closing provider must not permanently disable enrichment, nor
+  // clobber a newer provider's state
+  // =====================================================================================
+
+  /**
+   * The core reconfiguration regression: a first gate-on provider shuts down, then a second gate-on
+   * provider is created. Enrichment must still work for the second provider. The process-wide
+   * interceptor is registered once and rebound, so the second provider is NOT rejected as a
+   * duplicate (which would permanently disable enrichment).
+   */
+  @Test
+  void newProviderAfterShutdownStillEnriches() {
+    // First provider registers and binds.
+    final Provider first =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    assertNotNull(first.spanEnrichmentStates(), "first provider wires a store");
+    first.shutdown();
+    assertNull(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates(),
+        "first provider's shutdown leaves the interceptor inert");
+
+    // Second provider is created AFTER the first closed. With the old per-provider interceptor
+    // model
+    // this registration would be rejected as a duplicate and enrichment would be permanently off.
+    final Provider second =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    final SpanEnrichmentStates secondStates = second.spanEnrichmentStates();
+    assertNotNull(secondStates, "second provider wires a store");
+    assertTrue(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates() == secondStates,
+        "the interceptor must rebind to the second provider's store");
+
+    // End-to-end: the second provider captures and the interceptor flushes onto the root.
+    final long traceId = 0xBEE5L;
+    second
+        .spanEnrichmentHook()
+        .capture(key(traceId), ctx("f", "user-1"), serialDetails("f", 5, false));
+    final AgentSpan root = rootSpan(traceId);
+    SpanEnrichmentInterceptor.INSTANCE.onTraceComplete(Collections.singletonList(root));
+    verify(root).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "BQ=="); // {5} -> 0x05
+  }
+
+  /**
+   * Overlapping reconfiguration: a second provider rebinds the interceptor while the first is still
+   * "open"; when the FIRST provider later shuts down, it must NOT clear the second provider's
+   * in-flight state (its unbind is a no-op because it is no longer the active provider).
+   */
+  @Test
+  void lateShutdownOfDisplacedProviderDoesNotClobberActiveProvider() {
+    final Provider first =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    final Provider second =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    final SpanEnrichmentStates secondStates = second.spanEnrichmentStates();
+
+    // Second provider is now the active one and has live state.
+    final long liveTrace = 0xA11FEL;
+    second
+        .spanEnrichmentHook()
+        .capture(key(liveTrace), ctx("f", "user-1"), serialDetails("f", 9, false));
+    assertNotNull(secondStates.peek(key(liveTrace)), "second provider has in-flight state");
+    assertTrue(SpanEnrichmentInterceptor.INSTANCE.activeStates() == secondStates);
+
+    // The DISPLACED first provider shuts down late. The active (second) provider must be untouched.
+    first.shutdown();
+    assertTrue(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates() == secondStates,
+        "late shutdown of a displaced provider must not unbind the active provider");
+    assertNotNull(
+        secondStates.peek(key(liveTrace)),
+        "late shutdown of a displaced provider must not clear the active provider's state");
+
+    // Sanity: the second provider still flushes correctly.
+    final AgentSpan root = rootSpan(liveTrace);
+    SpanEnrichmentInterceptor.INSTANCE.onTraceComplete(Collections.singletonList(root));
+    verify(root).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "CQ=="); // {9} -> 0x09
+  }
+
+  /** Each provider owns a DISTINCT state store, so they never share mutable state. */
+  @Test
+  void eachProviderOwnsADistinctStateStore() {
+    final Provider a =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    final Provider b =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    assertNotNull(a.spanEnrichmentStates());
+    assertNotNull(b.spanEnrichmentStates());
+    assertTrue(
+        a.spanEnrichmentStates() != b.spanEnrichmentStates(),
+        "providers must own distinct state stores");
+
+    a.spanEnrichmentStates().getOrCreate(key(1L));
+    assertTrue(b.spanEnrichmentStates().isEmpty(), "stores are isolated");
+  }
+
+  /**
+   * Within ONE provider, the capture hook and the bound interceptor must share the SAME store, so a
+   * capture is visible to the flush.
+   */
+  @Test
+  void hookAndInterceptorOfOneProviderShareTheSameStore() {
+    final Provider provider =
+        new Provider(new Provider.Options(), null, Boolean.TRUE, interceptor -> true);
+    final long traceId = 0xBEEFL;
+    provider
+        .spanEnrichmentHook()
+        .capture(key(traceId), ctx("f", "user-1"), serialDetails("f", 9, false));
+    assertNotNull(
+        SpanEnrichmentInterceptor.INSTANCE.activeStates().peek(key(traceId)),
+        "hook capture must be visible in the bound store (same instance)");
+  }
+
+  // =====================================================================================
+  // 128-bit trace ids that share their low-order 64 bits must not merge
+  // =====================================================================================
+
+  /**
+   * Two distinct 128-bit trace ids whose low-order 64 bits are identical (so {@code toLong()}
+   * collides) must keep SEPARATE accumulators. Keying by {@code toLong()} would merge them; keying
+   * by the full hex string keeps them apart.
+   */
+  @Test
+  void distinct128BitTraceIdsSharingLowBitsDoNotMerge() {
+    final SpanEnrichmentStates states = new SpanEnrichmentStates();
+    final SpanEnrichmentHook hook = new SpanEnrichmentHook(states);
+    final SpanEnrichmentInterceptor interceptor = SpanEnrichmentInterceptor.INSTANCE;
+    interceptor.bind(states);
+
+    // Same low 64 bits (0xABCD), different high 64 bits — toLong() is identical for both.
+    final long lowBits = 0xABCDL;
+    final DDTraceId idA = DD128bTraceId.from(0x1111L, lowBits);
+    final DDTraceId idB = DD128bTraceId.from(0x2222L, lowBits);
+    assertEquals(idA.toLong(), idB.toLong(), "precondition: low 64 bits collide");
+    assertTrue(!idA.toHexString().equals(idB.toHexString()), "precondition: full ids differ");
+
+    // Capture distinct serial ids under each full trace id.
+    hook.capture(idA.toHexString(), ctx("fa", "user-A"), serialDetails("fa", 100, false));
+    hook.capture(idB.toHexString(), ctx("fb", "user-B"), serialDetails("fb", 130, false));
+
+    // They must NOT have merged into one accumulator.
+    assertEquals(2, states.size(), "distinct 128-bit traces must not share an accumulator");
+
+    // Flush trace A: only {100} (not {100,130}).
+    final AgentSpan rootA = rootSpan(idA);
+    interceptor.onTraceComplete(Collections.singletonList(rootA));
+    verify(rootA).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "ZA=="); // {100} -> 0x64
+
+    // Flush trace B: only {130}.
+    final AgentSpan rootB = rootSpan(idB);
+    interceptor.onTraceComplete(Collections.singletonList(rootB));
+    verify(rootB).setTag(SpanEnrichmentAccumulator.TAG_FLAGS_ENC, "ggE="); // {130} -> 0x82 0x01
+  }
+}
diff --git a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/ufc/v1/Split.java b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/ufc/v1/Split.java
index 1782032278d..37f6909eeea 100644
--- a/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/ufc/v1/Split.java
+++ b/products/feature-flagging/feature-flagging-bootstrap/src/main/java/datadog/trace/api/featureflag/ufc/v1/Split.java
@@ -7,11 +7,19 @@ public class Split {
   public final List<Shard> shards;
   public final String variationKey;
   public final Map<String, String> extraLogging;
+  // Nullable Integer (not primitive int): the serialId is absent in some UFC shapes. Populated by
+  // Moshi reflective deserialization from the UFC "serialId" JSON field. Surfaced as
+  // __dd_split_serial_id in eval metadata for APM span enrichment.
+  public final Integer serialId;
 
   public Split(
-      final List<Shard> shards, final String variationKey, final Map<String, String> extraLogging) {
+      final List<Shard> shards,
+      final String variationKey,
+      final Map<String, String> extraLogging,
+      final Integer serialId) {
     this.shards = shards;
     this.variationKey = variationKey;
     this.extraLogging = extraLogging;
+    this.serialId = serialId;
   }
 }