@@ -3,6 +3,7 @@
 import datadog.trace.api.config.TracerConfig;
 import datadog.trace.api.sampling.SamplingMechanism;
 import datadog.trace.api.sampling.SamplingRule;
+import datadog.trace.common.writer.RemoteResponseListener;
 import datadog.trace.core.CoreSpan;
 import datadog.trace.core.util.SimpleRateLimiter;
 import java.util.ArrayList;
@@ -12,7 +13,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class RuleBasedTraceSampler<T extends CoreSpan<T>> implements Sampler, PrioritySampler {
+public class RuleBasedTraceSampler<T extends CoreSpan<T>>
+    implements Sampler, PrioritySampler, RemoteResponseListener {
 
   private static final Logger log = LoggerFactory.getLogger(RuleBasedTraceSampler.class);
   private final List<RateSamplingRule> samplingRules;
@@ -47,6 +49,31 @@ public static RuleBasedTraceSampler build(
       final List<? extends SamplingRule.TraceSamplingRule> traceSamplingRules,
       final Double defaultRate,
       final int rateLimit) {
+    return build(
+        serviceRules,
+        operationRules,
+        traceSamplingRules,
+        defaultRate,
+        rateLimit,
+        new RateByServiceTraceSampler());
+  }
+
+  /**
+   * Builds a rule based sampler that delegates to {@code fallbackSampler} for spans that match no
+   * rule.
+   *
+   * <p>The fallback is supplied by the caller so that a single agent rate sampler instance can be
+   * shared across sampler rebuilds triggered by remote configuration. That instance is the one
+   * registered to receive agent sampling rates, so reusing it keeps the rates that have already
+   * been learned and keeps the rule miss path connected to the agent.
+   */
+  public static RuleBasedTraceSampler build(
+      @Deprecated final Map<String, String> serviceRules,
+      @Deprecated final Map<String, String> operationRules,
+      final List<? extends SamplingRule.TraceSamplingRule> traceSamplingRules,
+      final Double defaultRate,
+      final int rateLimit,
+      final PrioritySampler fallbackSampler) {
 
     final List<RateSamplingRule> samplingRules = new ArrayList<>();
 
@@ -113,7 +140,31 @@ public static RuleBasedTraceSampler build(
       samplingRules.add(samplingRule);
     }
 
-    return new RuleBasedTraceSampler(samplingRules, rateLimit, new RateByServiceTraceSampler());
+    return new RuleBasedTraceSampler(samplingRules, rateLimit, fallbackSampler);
+  }
+
+  /**
+   * Returns the sampler applying agent published rates to spans that match no rule, or {@code null}
+   * when the fallback does not use agent rates.
+   */
+  public RateByServiceTraceSampler agentSampler() {
+    return fallbackSampler instanceof RateByServiceTraceSampler
+        ? (RateByServiceTraceSampler) fallbackSampler
+        : null;
+  }
+
+  /**
+   * Forwards agent sampling rates to the fallback sampler.
+   *
+   * <p>Without this the inner {@link RateByServiceTraceSampler} would never observe the rates
+   * published by the agent, and every span matching no rule would be kept at a rate of 1.0.
+   */
+  @Override
+  public void onResponse(
+      final String endpoint, final Map<String, Map<String, Number>> responseJson) {
+    if (fallbackSampler instanceof RemoteResponseListener) {
+      ((RemoteResponseListener) fallbackSampler).onResponse(endpoint, responseJson);
+    }
   }
 
   private static byte samplingMechanism(SamplingRule.Provenance provenance) {
 
@@ -34,6 +34,21 @@ final class Builder {
     private static final Logger log = LoggerFactory.getLogger(Builder.class);
 
     public static Sampler forConfig(final Config config, final TraceConfig traceConfig) {
+      return forConfig(config, traceConfig, new RateByServiceTraceSampler());
+    }
+
+    /**
+     * Builds the sampler described by {@code config} and {@code traceConfig}, using {@code
+     * agentSampler} wherever agent published sampling rates apply.
+     *
+     * <p>Callers that rebuild the sampler when remote configuration changes must pass the same
+     * {@code agentSampler} instance every time: it is the instance registered to receive agent
+     * sampling rates, and a fresh one would silently start over at a rate of 1.0.
+     */
+    public static Sampler forConfig(
+        final Config config,
+        final TraceConfig traceConfig,
+        final RateByServiceTraceSampler agentSampler) {
       Sampler sampler;
       if (config != null) {
         if (!config.isApmTracingEnabled() && isAsmEnabled(config)) {
@@ -75,7 +90,8 @@ public static Sampler forConfig(final Config config, final TraceConfig traceConf
                     operationRules,
                     traceSamplingRules,
                     traceSampleRate,
-                    config.getTraceRateLimit());
+                    config.getTraceRateLimit(),
+                    agentSampler);
           } catch (final IllegalArgumentException e) {
             log.error("Invalid sampler configuration. Using AllSampler", e);
             sampler = new AllSampler();
@@ -95,7 +111,7 @@ public static Sampler forConfig(final Config config, final TraceConfig traceConf
                 "OTLP traces export enabled. Using ParentBasedAlwaysOnSampler instead of RateByServiceTraceSampler.");
             sampler = new ParentBasedAlwaysOnSampler();
           } else {
-            sampler = new RateByServiceTraceSampler();
+            sampler = agentSampler;
           }
         } else if (config.isTraceOtlpExporterEnabled()) {
           // AllSampler does not emit a sampling priority; OTLP export requires one.
 
@@ -85,6 +85,8 @@
 import datadog.trace.common.GitMetadataTraceInterceptor;
 import datadog.trace.common.metrics.MetricsAggregator;
 import datadog.trace.common.metrics.NoOpMetricsAggregator;
+import datadog.trace.common.sampling.RateByServiceTraceSampler;
+import datadog.trace.common.sampling.RuleBasedTraceSampler;
 import datadog.trace.common.sampling.Sampler;
 import datadog.trace.common.sampling.SingleSpanSampler;
 import datadog.trace.common.sampling.SpanSamplingRules;
@@ -181,6 +183,17 @@ public static CoreTracerBuilder builder() {
   /** Sampler defines the sampling policy in order to reduce the number of traces for instance */
   final Sampler initialSampler;
 
+  /**
+   * The sampler applying the rates published by the agent, shared by every sampler this tracer
+   * builds.
+   *
+   * <p>Only the initial sampler is registered to receive agent rates, and that registration is
+   * never repeated. Reusing this instance when the sampler is rebuilt on a remote configuration
+   * change keeps the rebuilt sampler connected to those rates, and keeps the rates already learned.
+   * May be {@code null} when the tracer was given a sampler that does not use agent rates.
+   */
+  final RateByServiceTraceSampler agentSampler;
+
   /** Scope manager is in charge of managing the scopes from which spans are created */
   final ContinuableScopeManager scopeManager;
 
@@ -679,6 +692,7 @@ private CoreTracer(
     this.initialConfig = config;
     this.apmTracingEnabled = config.isApmTracingEnabled();
     this.initialSampler = sampler;
+    this.agentSampler = agentSamplerOf(sampler);
 
     // Get initial Trace Sampling Rules from config
     String traceSamplingRulesJson = config.getTraceSamplingRules();
@@ -1641,6 +1655,22 @@ private static <K, V> Map<V, K> invertMap(Map<K, V> map) {
     return Collections.unmodifiableMap(inverted);
   }
 
+  /**
+   * Extracts the sampler applying agent published rates from the sampler this tracer was built
+   * with, so that it can be reused whenever the sampler is rebuilt.
+   *
+   * @return the agent rate sampler in use, or {@code null} if this tracer does not use one.
+   */
+  private static RateByServiceTraceSampler agentSamplerOf(final Sampler sampler) {
+    if (sampler instanceof RateByServiceTraceSampler) {
+      return (RateByServiceTraceSampler) sampler;
+    }
+    if (sampler instanceof RuleBasedTraceSampler) {
+      return ((RuleBasedTraceSampler<?>) sampler).agentSampler();
+    }
+    return null;
+  }
+
   /** Spans are built using this builder */
   public abstract static class CoreSpanBuilder implements AgentTracer.SpanBuilder {
     protected static final boolean USE_SCOPE = false;
@@ -2547,7 +2577,13 @@ protected ConfigSnapshot(
           && Objects.equals(getTraceSamplingRules(), oldSnapshot.getTraceSamplingRules())) {
         sampler = oldSnapshot.sampler;
       } else {
-        sampler = Sampler.Builder.forConfig(CoreTracer.this.initialConfig, this);
+        // Reuse the agent sampler so the rebuilt sampler keeps receiving agent rates: the response
+        // listener is registered once, against the initial sampler, and is never re-registered.
+        RateByServiceTraceSampler agentSampler = CoreTracer.this.agentSampler;
+        sampler =
+            agentSampler == null
+                ? Sampler.Builder.forConfig(CoreTracer.this.initialConfig, this)
+                : Sampler.Builder.forConfig(CoreTracer.this.initialConfig, this, agentSampler);
       }
 
       if (null == oldSnapshot) {
 
@@ -0,0 +1,180 @@
+package datadog.trace.common.sampling;
+
+import static datadog.trace.api.config.TracerConfig.TRACE_SAMPLE_RATE;
+import static datadog.trace.api.config.TracerConfig.TRACE_SAMPLING_RULES;
+import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_DROP;
+import static datadog.trace.api.sampling.PrioritySampling.SAMPLER_KEEP;
+import static datadog.trace.api.sampling.PrioritySampling.USER_KEEP;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import datadog.trace.api.Config;
+import datadog.trace.common.writer.ListWriter;
+import datadog.trace.common.writer.RemoteResponseListener;
+import datadog.trace.core.CoreTracer;
+import datadog.trace.core.DDCoreJavaSpecification;
+import datadog.trace.core.DDSpan;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Agent published sampling rates must still apply to spans that match no sampling rule.
+ *
+ * <p>Only one sampler is registered to receive those rates, so a rule based sampler has to forward
+ * them to the fallback it delegates to.
+ */
+class RuleBasedSamplerAgentRatesTest extends DDCoreJavaSpecification {
+
+  private static final String RULE_MISSING_SERVICE =
+      "[{\"service\": \"other-service\", \"sample_rate\": 1}]";
+
+  @Test
+  void ruleBasedSamplerForwardsAgentRatesToFallback() {
+    Properties properties = new Properties();
+    properties.setProperty(TRACE_SAMPLING_RULES, RULE_MISSING_SERVICE);
+    Sampler sampler = Sampler.Builder.forConfig(properties);
+    assertInstanceOf(RuleBasedTraceSampler.class, sampler);
+
+    // The agent asks for everything to be dropped for this service.
+    assertInstanceOf(RemoteResponseListener.class, sampler);
+    ((RemoteResponseListener) sampler).onResponse("traces", rateByService("service", "bar", 0.0));
+
+    CoreTracer tracer = tracerBuilder().writer(new ListWriter()).build();
+    try {
+      DDSpan span = buildSpan(tracer, "service");
+      ((PrioritySampler) sampler).setSamplingPriority(span);
+
+      assertEquals(SAMPLER_DROP, (int) span.getSamplingPriority());
+      assertEquals(0.0, span.getTag(RateByServiceTraceSampler.SAMPLING_AGENT_RATE));
+    } finally {
+      tracer.close();
+    }
+  }
+
+  @Test
+  void agentRatesDoNotOverrideMatchedRules() {
+    Properties properties = new Properties();
+    properties.setProperty(
+        TRACE_SAMPLING_RULES, "[{\"service\": \"service\", \"sample_rate\": 1}]");
+    Sampler sampler = Sampler.Builder.forConfig(properties);
+
+    ((RemoteResponseListener) sampler).onResponse("traces", rateByService("service", "bar", 0.0));
+
+    CoreTracer tracer = tracerBuilder().writer(new ListWriter()).build();
+    try {
+      DDSpan span = buildSpan(tracer, "service");
+      ((PrioritySampler) sampler).setSamplingPriority(span);
+
+      assertEquals(USER_KEEP, (int) span.getSamplingPriority());
+      assertEquals(1.0, span.getTag(RuleBasedTraceSampler.SAMPLING_RULE_RATE));
+      assertNull(span.getTag(RateByServiceTraceSampler.SAMPLING_AGENT_RATE));
+    } finally {
+      tracer.close();
+    }
+  }
+
+  @Test
+  void defaultRateKeepsAgentRatesUnused() {
+    Properties properties = new Properties();
+    properties.setProperty(TRACE_SAMPLING_RULES, RULE_MISSING_SERVICE);
+    // A default rate matches every span, so the fallback is never reached.
+    properties.setProperty(TRACE_SAMPLE_RATE, "1");
+    Sampler sampler = Sampler.Builder.forConfig(properties);
+
+    ((RemoteResponseListener) sampler).onResponse("traces", rateByService("service", "bar", 0.0));
+
+    CoreTracer tracer = tracerBuilder().writer(new ListWriter()).build();
+    try {
+      DDSpan span = buildSpan(tracer, "service");
+      ((PrioritySampler) sampler).setSamplingPriority(span);
+
+      assertEquals(USER_KEEP, (int) span.getSamplingPriority());
+      assertNull(span.getTag(RateByServiceTraceSampler.SAMPLING_AGENT_RATE));
+    } finally {
+      tracer.close();
+    }
+  }
+
+  @Test
+  void rulesKeepTheSuppliedAgentSampler() {
+    RateByServiceTraceSampler agentSampler = new RateByServiceTraceSampler();
+    Properties properties = new Properties();
+    properties.setProperty(TRACE_SAMPLING_RULES, RULE_MISSING_SERVICE);
+
+    Sampler sampler = Sampler.Builder.forConfig(Config.get(properties), null, agentSampler);
+
+    assertInstanceOf(RuleBasedTraceSampler.class, sampler);
+    assertSame(agentSampler, ((RuleBasedTraceSampler<?>) sampler).agentSampler());
+  }
+
+  @Test
+  void noRulesReturnsTheSuppliedAgentSampler() {
+    RateByServiceTraceSampler agentSampler = new RateByServiceTraceSampler();
+
+    Sampler sampler = Sampler.Builder.forConfig(Config.get(new Properties()), null, agentSampler);
+
+    assertSame(agentSampler, sampler);
+  }
+
+  @Test
+  void ratesLearnedBeforeARebuildStillApplyAfterIt() {
+    RateByServiceTraceSampler agentSampler = new RateByServiceTraceSampler();
+    agentSampler.onResponse("traces", rateByService("service", "bar", 0.0));
+
+    // Rebuilding the sampler with the same agent sampler keeps the rates it has already learned.
+    Properties properties = new Properties();
+    properties.setProperty(TRACE_SAMPLING_RULES, RULE_MISSING_SERVICE);
+    Sampler sampler = Sampler.Builder.forConfig(Config.get(properties), null, agentSampler);
+
+    CoreTracer tracer = tracerBuilder().writer(new ListWriter()).build();
+    try {
+      DDSpan span = buildSpan(tracer, "service");
+      ((PrioritySampler) sampler).setSamplingPriority(span);
+
+      assertEquals(SAMPLER_DROP, (int) span.getSamplingPriority());
+    } finally {
+      tracer.close();
+    }
+  }
+
+  @Test
+  void unknownServiceIsKeptUntilTheAgentReportsARate() {
+    Properties properties = new Properties();
+    properties.setProperty(TRACE_SAMPLING_RULES, RULE_MISSING_SERVICE);
+    Sampler sampler = Sampler.Builder.forConfig(properties);
+
+    CoreTracer tracer = tracerBuilder().writer(new ListWriter()).build();
+    try {
+      DDSpan span = buildSpan(tracer, "service");
+      ((PrioritySampler) sampler).setSamplingPriority(span);
+
+      assertEquals(SAMPLER_KEEP, (int) span.getSamplingPriority());
+      assertEquals(1.0, span.getTag(RateByServiceTraceSampler.SAMPLING_AGENT_RATE));
+    } finally {
+      tracer.close();
+    }
+  }
+
+  private static DDSpan buildSpan(CoreTracer tracer, String serviceName) {
+    return (DDSpan)
+        tracer
+            .buildSpan("datadog", "operation")
+            .withServiceName(serviceName)
+            .withTag("env", "bar")
+            .ignoreActiveSpan()
+            .start();
+  }
+
+  private static Map<String, Map<String, Number>> rateByService(
+      String service, String env, double rate) {
+    Map<String, Number> byService = new HashMap<>();
+    byService.put("service:" + service + ",env:" + env, rate);
+    Map<String, Map<String, Number>> response = new HashMap<>();
+    response.put("rate_by_service", byService);
+    return response;
+  }
+}