test(ffe): cover delayed capability refresh through a real Agent

.github/workflows/run-end-to-end.yml

@@ -311,6 +311,12 @@ jobs:
     - name: Run FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS scenario
       if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS"')
       run: ./run.sh FEATURE_FLAGGING_AND_EXPERIMENTATION_AGENTLESS
+    - name: Run FEATURE_FLAGGING_CAPABILITY_REFRESH scenario
+      if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_CAPABILITY_REFRESH"')
+      run: ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH
+    - name: Run FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED scenario
+      if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED"')
+      run: ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED
     - name: Run FEATURE_FLAGGING_EVP_STRESS scenario
       if: steps.build.outcome == 'success' && !cancelled() && contains(inputs.scenarios, '"FEATURE_FLAGGING_EVP_STRESS"')
       run: ./run.sh FEATURE_FLAGGING_EVP_STRESS

manifests/cpp.yml

@@ -25,6 +25,7 @@ manifest:
   tests/debugger/test_debugger_probe_snapshot.py::Test_Debugger_Line_Probe_Snaphots::test_process_tags_snapshot_svc: missing_feature (Not yet implemented)
   tests/ffe/test_evaluation_telemetry_opt_out.py: irrelevant (only parametric tests are run for cpp)
   tests/ffe/test_flag_eval_evp.py: irrelevant (only parametric tests are run for cpp)
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_sqs.py::Test_SQS_PROPAGATION_VIA_AWS_XRAY_HEADERS: irrelevant (Localstack SQS does not support AWS Xray Header parsing)
   tests/integrations/test_base_service.py::Test_BaseService_SqlSpan: irrelevant (/rasp/sqli endpoint is not available)

manifests/cpp_nginx.yml

@@ -295,6 +295,7 @@ manifest:
   tests/ffe/test_exposures_datadog_agent.py: missing_feature
   tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)
   tests/ffe/test_flag_eval_metrics.py: missing_feature
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: missing_feature
   tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature

manifests/dotnet.yml

@@ -891,6 +891,7 @@ manifest:
       excluded_weblog: [poc, uds]
   tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)
   tests/ffe/test_flag_eval_metrics.py: v3.44.0
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: v2.0.0-prerelease
   tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature

manifests/golang.yml

@@ -1874,6 +1874,7 @@ manifest:
   tests/ffe/test_flag_eval_metrics.py: v2.8.0
   tests/ffe/test_flag_eval_metrics.py::Test_FFE_Eval_Metric_Parse_Error_Invalid_Regex: irrelevant (Go validates regex at config load time)
   tests/ffe/test_flag_eval_metrics.py::Test_FFE_Eval_Nested_Attributes_Ignored: irrelevant (FFL-1980)
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka:
     - weblog_declaration:

manifests/java.yml

@@ -3537,6 +3537,7 @@ manifest:
     - weblog_declaration:
         "*": irrelevant
         spring-boot: v1.63.0
+  tests/ffe/test_rc_capability_refresh.py: missing_feature (Provisional dependent-flag capability coverage; enable after the SDK release)
   tests/integration_frameworks/llm/anthropic/test_anthropic_ai_guard.py: missing_feature (APPSEC-68977, AI Guard/Anthropic integration not yet wired into INTEGRATION_FRAMEWORKS)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integration_frameworks/llm/openai/test_openai_ai_guard.py: missing_feature (APPSEC-68977, AI Guard/OpenAI integration not yet wired into INTEGRATION_FRAMEWORKS)

manifests/nodejs.yml

@@ -1993,6 +1993,7 @@ manifest:
       weblog: [express5]
     - declaration: bug (FFL-1730)
       excluded_weblog: [express5]
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_ai_guard.py::TestAnthropicAiGuard:  # TODO: a lower version might be supported
     - declaration: missing_feature (APPSEC-69630, AI Guard/Anthropic integration not yet wired into INTEGRATION_FRAMEWORKS)
       component_version: <6.19.0

manifests/php.yml

@@ -1326,6 +1326,7 @@ manifest:
         "*": v1.21.0-dev
         laravel11x: incomplete_test_app
         symfony7x: incomplete_test_app
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka: missing_feature
   tests/integrations/crossed_integrations/test_kinesis.py::Test_Kinesis_PROPAGATION_VIA_MESSAGE_ATTRIBUTES: missing_feature

manifests/python.yml

@@ -2375,6 +2375,7 @@ manifest:
     - declaration: irrelevant (FFL-1980)
     - weblog_declaration:
         aiohttp: missing_feature (new weblog, tests are enabled with easy wins)
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks:
     - weblog_declaration:
         aiohttp: missing_feature (new weblog, tests are enabled with easy wins)

manifests/ruby.yml

@@ -2197,6 +2197,7 @@ manifest:
     - weblog_declaration:
         "*": irrelevant
         rails72: v2.32.0
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/test_kafka.py::Test_Kafka:
     - weblog_declaration:

manifests/rust.yml

@@ -29,6 +29,7 @@ manifest:
   tests/ffe/test_exposures_datadog_agent.py: missing_feature
   tests/ffe/test_flag_eval_evp.py: missing_feature (FFL-2446)
   tests/ffe/test_flag_eval_metrics.py: missing_feature
+  tests/ffe/test_rc_capability_refresh.py: incomplete_test_app (Requires a provider initialized lazily after the initial FFE_FLAGS RC subscription)
   tests/integration_frameworks/llm/anthropic/test_anthropic_llmobs.py::TestAnthropicLlmObsMessages::test_create_error: bug (MLOB-1234)
   tests/integrations/crossed_integrations/: incomplete_test_app (crossed integration endpoints are not implemented)
   tests/integrations/crossed_integrations/test_sqs.py::Test_SQS_PROPAGATION_VIA_AWS_XRAY_HEADERS: irrelevant (Localstack SQS does not support AWS Xray Header parsing)

tests/ffe/README.md

@@ -419,3 +419,39 @@ The tests use `interfaces.mocked_backend.get_metrics()` to retrieve the metrics
 4. **Enable tests** in manifest file
 5. **Run locally** with `./run.sh` to iterate quickly
 6. **Push to CI** only when local tests pass
+
+
+### Delayed provider capability registration (real Agent)
+
+`FEATURE_FLAGGING_CAPABILITY_REFRESH` and `FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED`
+exercise the SDK → real Agent → mocked HTTPS RC backend path. They require Python 3.12
+and the Java `spring-boot` weblog, whose provider starts lazily on the first `/ffe` request.
+They are provisional scenarios selected explicitly, rather than scheduled for every SDK.
+
+Publish base and dependent configurations once, observe Java ACK the base with capability
+46 but without 50, then initialize the provider. The same RC client must add 50 without
+resetting its targets version, receive and ACK the companion, and evaluate a dependent
+flag successfully. The preloaded case first caches the companion through another capable
+protocol client, with a legacy client kept active so the selected file union stays
+unchanged when Java upgrades.
+
+The mocked backend renews signed director metadata on its ordinary Agent polling cadence.
+It models backend capability selection and client-ID predicates; configuration repository
+metadata, file hashes, and publication version stay fixed. A second publication is rejected.
+This guards the real Agent and SDK transition, including the Java replacement failure in
+[dd-trace-java#13258](https://github.com/DataDog/dd-trace-java/pull/13258). It does not verify
+that the production backend continues renewing director metadata; that needs a backend
+contract test or a live environment test.
+
+Use a Java build with dependent capability reporting, dependent evaluation, and the RC
+replacement fix in `binaries/` (see [custom binaries](../../docs/execute/binaries.md)):
+
+```sh
+./build.sh java -w spring-boot
+TEST_LIBRARY=java ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH -F tests/ffe/test_rc_capability_refresh.py
+TEST_LIBRARY=java ./run.sh FEATURE_FLAGGING_CAPABILITY_REFRESH_PRELOADED -F tests/ffe/test_rc_capability_refresh.py
+```
+
+`-F` makes failures blocking while the manifest remains provisional. Enable the Java
+manifest entry for the confirmed SDK release once the changes ship. Other weblogs need
+an equivalent delayed provider lifecycle before this scenario can cover them.

tests/ffe/test_rc_capability_refresh.py

@@ -0,0 +1,154 @@
+"""Delayed capability registration through a real Agent, with no UFC republish."""
+
+import base64
+import json
+import time
+from collections.abc import Callable
+from typing import TYPE_CHECKING, cast
+
+from utils import context, features, scenarios, weblog
+from utils.mocked_backend.rc_capabilities import BASE_PATH, DEPENDENT_PATH, capability_mask
+from utils.proxy.ports import ProxyPorts
+
+from .utils.rc_capability_refresh import CachePreloader, acknowledged, configurations, sdk_polls
+
+
+if TYPE_CHECKING:
+    from utils._context._scenarios.feature_flagging import FeatureFlaggingCapabilityRefresh
+
+
+def wait_until(predicate: Callable[[], bool]) -> bool:
+    deadline = time.monotonic() + 40
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.2)
+    return False
+
+
+@scenarios.feature_flagging_capability_refresh
+@scenarios.feature_flagging_capability_refresh_preloaded
+@features.feature_flags_dynamic_evaluation
+class Test_FFE_RC_Capability_Refresh:
+    def setup_delayed_provider(self) -> None:
+        scenario = cast("FeatureFlaggingCapabilityRefresh", context.scenario)
+        backend = scenario.mocked_backend_container.client
+        base, dependent = configurations()
+        self.published = backend.rc_publish_capability_variants(base, dependent)
+        preloader = None
+        legacy = None
+        self.cache_ready = True
+        if scenario is scenarios.feature_flagging_capability_refresh_preloaded:
+            port = scenario.proxy_container.runtime_container.attrs["NetworkSettings"]["Ports"][
+                f"{ProxyPorts.weblog}/tcp"
+            ][0]["HostPort"]
+            legacy = CachePreloader(f"http://127.0.0.1:{port}", dependent=False)
+            legacy.poll()
+            preloader = CachePreloader(f"http://127.0.0.1:{port}")
+            self.cache_ready = wait_until(lambda: DEPENDENT_PATH in preloader.poll())
+            preloader.poll()  # ACK the companion before Java registers its provider.
+        self.preloaded = preloader is not None
+        self.base_ready = wait_until(
+            lambda: any(
+                acknowledged(data["request"]["content"]["client"], BASE_PATH)
+                for data in sdk_polls()
+                if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-")
+            )
+        )
+        self.before_status = backend.rc_capability_status()
+        # Spring's @Lazy Client initializes the Datadog provider on the first /ffe request.
+        self.initial = weblog.post(
+            "/ffe",
+            json={
+                "flag": "parent",
+                "variationType": "STRING",
+                "defaultValue": "caller-default",
+                "targetingKey": "test-user",
+                "attributes": {},
+            },
+        )
+
+        def companion_acknowledged() -> bool:
+            # Keep both cache clients live throughout the wait. Their expiry must not
+            # change the target union and accidentally unstick a stale director.
+            for cache_client in (preloader, legacy):
+                if cache_client is not None:
+                    cache_client.poll()
+            return any(
+                acknowledged(data["request"]["content"]["client"], DEPENDENT_PATH)
+                for data in sdk_polls()
+                if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-")
+            )
+
+        self.dependent_ready = wait_until(companion_acknowledged)
+        # Keep the preloader active, so the preloaded case retains the same union of files.
+        if preloader is not None:
+            preloader.poll()
+        if legacy is not None:
+            legacy.poll()
+        self.evaluation = weblog.post(
+            "/ffe",
+            json={
+                "flag": "child",
+                "variationType": "STRING",
+                "defaultValue": "caller-default",
+                "targetingKey": "test-user",
+                "attributes": {},
+            },
+        )
+        self.after = sdk_polls()
+        self.after_status = backend.rc_capability_status()
+
+    def test_delayed_provider(self) -> None:
+        assert self.cache_ready, "The second client must cache the companion before provider initialization"
+        assert self.base_ready, "Java must ACK the base config before provider initialization"
+        assert self.dependent_ready, "Java must ACK the companion after provider initialization"
+        polls = [
+            data for data in self.after if not data["request"]["content"]["client"]["id"].startswith("cache-preloader-")
+        ]
+        clients = [data["request"]["content"]["client"] for data in polls]
+        assert len({client["id"] for client in clients}) == 1, "Provider startup must retain the RC client identity"
+        transition = next(
+            index for index, client in enumerate(clients) if capability_mask(client["capabilities"]) & (1 << 50)
+        )
+        assert transition > 0, "Must observe polls before the provider registers capability 50"
+        previous = clients[transition - 1]
+        upgraded = clients[transition]
+        assert capability_mask(previous["capabilities"]) & (1 << 46)
+        assert not capability_mask(previous["capabilities"]) & (1 << 50)
+        assert acknowledged(previous, BASE_PATH)
+        previous_version = previous["state"]["targets_version"]
+        response = polls[transition - 1]["response"].get("content", {})
+        if isinstance(response, dict) and response.get("targets"):
+            previous_version = json.loads(base64.b64decode(response["targets"]))["signed"]["version"]
+        assert upgraded["state"]["targets_version"] == previous_version > 0, (
+            "Capabilities must change without resetting SDK state"
+        )
+        versions = [client["state"]["targets_version"] for client in clients[transition:]]
+        assert versions == sorted(versions), "SDK state must not decrease to recover delivery"
+        assert all(version > 0 for version in versions), "SDK state must not reset to recover delivery"
+        assert any(acknowledged(client, DEPENDENT_PATH) for client in clients[transition:])
+        deliveries = [data["response"].get("content", {}) for data in polls[transition:]]
+        assert any(
+            isinstance(response, dict) and response.get("client_configs") == [DEPENDENT_PATH] for response in deliveries
+        )
+        assert self.initial.status_code == 200
+        assert json.loads(self.initial.text)["value"] == "parent-value"
+        assert self.evaluation.status_code == 200
+        assert json.loads(self.evaluation.text)["value"] == "dependent-match", (
+            "ACK alone does not prove the provider retained the replacement"
+        )
+        assert self.published["hashes"] == self.after_status["hashes"]
+        assert self.after_status["publications"] == self.after_status["config_version"] == 1
+        client_id = upgraded["id"]
+        before = next(
+            entry for entry in reversed(self.before_status["history"]) if entry["selected"].get(client_id) == BASE_PATH
+        )
+        after = next(
+            entry for entry in self.after_status["history"] if entry["selected"].get(client_id) == DEPENDENT_PATH
+        )
+        assert after["version"] > before["version"]
+        if self.preloaded:
+            assert set(before["selected"].values()) == set(after["selected"].values()) == {BASE_PATH, DEPENDENT_PATH}
+        else:
+            assert set(before["selected"].values()) == {BASE_PATH}