diff --git a/ddtrace/appsec/_asm_request_context.py b/ddtrace/appsec/_asm_request_context.py
index 0e2ba7ad4eb..452abe270f2 100644
--- a/ddtrace/appsec/_asm_request_context.py
+++ b/ddtrace/appsec/_asm_request_context.py
@@ -400,8 +400,12 @@ def finalize_asm_env(env: ASM_Environment) -> None:
                 entry_span._set_attribute(APPSEC.EVENT_RULE_ERROR_COUNT, info.failed)
             except Exception:
                 logger.debug("asm_context::finalize_asm_env::exception", extra=log_extra, exc_info=True)
-        if asm_config._rc_client_id is not None:
-            entry_span.set_tag(APPSEC.RC_CLIENT_ID, asm_config._rc_client_id)
+        if asm_config._rc_client_id_enabled:
+            from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
+
+            rc_client_id = remoteconfig_poller._client.id
+            if rc_client_id is not None:
+                entry_span.set_tag(APPSEC.RC_CLIENT_ID, rc_client_id)
         waf_adresses = env.waf_addresses
         req_headers = waf_adresses.get(SPAN_DATA_NAMES.REQUEST_HEADERS_NO_COOKIES, {})
         if req_headers:
diff --git a/ddtrace/appsec/_remoteconfiguration.py b/ddtrace/appsec/_remoteconfiguration.py
index f7085e99d95..7fe2da28515 100644
--- a/ddtrace/appsec/_remoteconfiguration.py
+++ b/ddtrace/appsec/_remoteconfiguration.py
@@ -80,10 +80,13 @@ def enable_appsec_rc(callback: "AppSecCallback") -> None:
 
     if asm_config._asm_enabled:
         telemetry_writer.product_activated(TELEMETRY_APM_PRODUCT.APPSEC, True)
-    asm_config._rc_client_id = remoteconfig_poller._client.id
+
+    asm_config._rc_client_id_enabled = True
 
 
 def disable_appsec_rc() -> None:
+    asm_config._rc_client_id_enabled = False
+
     for product_name in APPSEC_PRODUCTS:
         remoteconfig_poller.unregister_callback(product_name)
         remoteconfig_poller.disable_product(product_name)
diff --git a/ddtrace/debugging/_probe/status.py b/ddtrace/debugging/_probe/status.py
index 47aeae81225..c646d0b0031 100644
--- a/ddtrace/debugging/_probe/status.py
+++ b/ddtrace/debugging/_probe/status.py
@@ -7,9 +7,10 @@
 from ddtrace.debugging._metrics import metrics
 from ddtrace.debugging._probe.model import Probe
 from ddtrace.debugging._uploader import build_debugger_sender
-from ddtrace.internal import runtime
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.native import DebuggerTrackType
+from ddtrace.internal.runtime import get_ancestor_runtime_id
+from ddtrace.internal.runtime import get_runtime_id
 from ddtrace.internal.utils.retry import fibonacci_backoff_with_jitter
 
 
@@ -47,8 +48,8 @@ def _payload(
                 "diagnostics": {
                     "probeId": probe.probe_id,
                     "probeVersion": probe.version,
-                    "runtimeId": runtime.get_runtime_id(),
-                    "parentId": runtime.get_ancestor_runtime_id(),
+                    "runtimeId": get_runtime_id(),
+                    "parentId": get_ancestor_runtime_id(),
                     "status": status,
                 }
             },
diff --git a/ddtrace/internal/core/crashtracking.py b/ddtrace/internal/core/crashtracking.py
index 2c2f541792c..941bc39b486 100644
--- a/ddtrace/internal/core/crashtracking.py
+++ b/ddtrace/internal/core/crashtracking.py
@@ -14,6 +14,7 @@
 from ddtrace.internal.compat import ensure_text
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.runtime import get_runtime_id
+from ddtrace.internal.runtime import on_runtime_id_change
 from ddtrace.internal.settings import env
 from ddtrace.internal.settings._agent import config as agent_config
 from ddtrace.internal.settings.crashtracker import config as crashtracker_config
@@ -33,6 +34,7 @@
     from ddtrace.internal.native._native import StacktraceCollection
     from ddtrace.internal.native._native import crashtracker_init
     from ddtrace.internal.native._native import crashtracker_on_fork
+    from ddtrace.internal.native._native import crashtracker_reconfigure
     from ddtrace.internal.native._native import crashtracker_report_unhandled_exception
     from ddtrace.internal.native._native import crashtracker_status
 
@@ -41,6 +43,13 @@
     is_available = False
 
 
+_identity_refresh_additional_tags: Optional[dict[str, str]] = None
+
+
+def _on_identity_refresh(_new_runtime_id: str) -> None:
+    _reconfigure_for_identity_refresh(_identity_refresh_additional_tags)
+
+
 def _get_tags(additional_tags: Optional[dict[str, str]]) -> dict[str, str]:
     tags = {
         "language": "python",
@@ -211,7 +220,20 @@ def is_started() -> bool:
     return crashtracker_status() == CrashtrackerStatus.Initialized
 
 
+def _reconfigure_for_identity_refresh(additional_tags: Optional[dict[str, str]]) -> None:
+    if not is_started():
+        return
+
+    config, receiver_config, metadata = _get_args(additional_tags)
+    if config is None or receiver_config is None or metadata is None:
+        log.error("Failed to reconfigure crashtracker after identity refresh: failed to construct configuration")
+        return
+    crashtracker_reconfigure(config, receiver_config, metadata)
+
+
 def start(additional_tags: Optional[dict[str, str]] = None) -> bool:
+    global _identity_refresh_additional_tags
+
     if not is_available:
         return False
     if not crashtracker_config.enabled:
@@ -256,6 +278,8 @@ def crashtracker_fork_handler():
             crashtracker_on_fork(config, receiver_config, metadata)
 
         forksafe.register(crashtracker_fork_handler)
+        _identity_refresh_additional_tags = additional_tags
+        on_runtime_id_change(_on_identity_refresh)
     except Exception:
         log.exception("Failed to start crashtracker")
         return False
diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi
index 56ba431263f..b60c9519a96 100644
--- a/ddtrace/internal/native/_native.pyi
+++ b/ddtrace/internal/native/_native.pyi
@@ -122,6 +122,9 @@ def crashtracker_init(
 def crashtracker_on_fork(
     config: CrashtrackerConfiguration, receiver_config: CrashtrackerReceiverConfig, metadata: CrashtrackerMetadata
 ) -> None: ...
+def crashtracker_reconfigure(
+    config: CrashtrackerConfiguration, receiver_config: CrashtrackerReceiverConfig, metadata: CrashtrackerMetadata
+) -> None: ...
 def crashtracker_status() -> CrashtrackerStatus: ...
 def crashtracker_receiver() -> None: ...
 def crashtracker_report_unhandled_exception(
diff --git a/ddtrace/internal/remoteconfig/client.py b/ddtrace/internal/remoteconfig/client.py
index 93604bb71f9..3056fc1a7e3 100644
--- a/ddtrace/internal/remoteconfig/client.py
+++ b/ddtrace/internal/remoteconfig/client.py
@@ -9,7 +9,6 @@
 import ddtrace
 from ddtrace.internal import gitmetadata
 from ddtrace.internal import process_tags
-from ddtrace.internal import runtime
 from ddtrace.internal.hostname import get_hostname
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.packages import is_distribution_available
@@ -17,6 +16,8 @@
 from ddtrace.internal.remoteconfig import Payload
 from ddtrace.internal.remoteconfig import PayloadType
 from ddtrace.internal.remoteconfig import RCCallback
+from ddtrace.internal.runtime import get_runtime_id
+from ddtrace.internal.runtime import on_runtime_id_change
 from ddtrace.internal.settings._agent import config as agent_config
 from ddtrace.internal.settings._core import DDConfig
 from ddtrace.internal.telemetry import telemetry_writer
@@ -98,6 +99,8 @@ def __init__(self) -> None:
         self._native: Optional[Any] = None
         self._reader: Optional[Any] = None
 
+        on_runtime_id_change(self._on_identity_refresh)
+
     def ensure_native(self) -> Any:
         if self._native is None:
             from ddtrace.internal.native import RemoteConfigClient as _NativeClient
@@ -109,7 +112,7 @@ def ensure_native(self) -> Any:
                 agent_url=str(self.agent_url),
                 tracer_version=tracer_version,
                 client_id=self.id,
-                runtime_id=runtime.get_runtime_id(),
+                runtime_id=get_runtime_id(),
                 service=ddtrace.config.service or "",
                 env=ddtrace.config.env or "",
                 app_version=ddtrace.config.version or "",
@@ -125,6 +128,15 @@ def ensure_native(self) -> Any:
     def renew_id(self) -> None:
         self.id = str(uuid.uuid4())
 
+    def _on_identity_refresh(self, new_runtime_id: str) -> None:
+        # Regenerate the client id and drop the native client, which bakes both ids in
+        # as immutable constructor arguments (get_client_id() is documented "stable for
+        # the process lifetime"). The next ensure_native() call rebuilds it bound to the
+        # fresh ids. Safe across threads: request() captures self._native into a local
+        # before calling .poll(), so an in-flight poll on the old client is unaffected.
+        self.renew_id()
+        self._native = None
+
     def register_callback(self, product_name: "RemoteConfigProduct", callback: RCCallback) -> None:
         self._product_callbacks[product_name] = callback
         log.debug("[%s][P: %s] Registered callback for product %s", os.getpid(), os.getppid(), product_name)
diff --git a/ddtrace/internal/runtime/runtime_metrics.py b/ddtrace/internal/runtime/runtime_metrics.py
index 17ef94bfec4..05c99a220a7 100644
--- a/ddtrace/internal/runtime/runtime_metrics.py
+++ b/ddtrace/internal/runtime/runtime_metrics.py
@@ -4,6 +4,7 @@
 
 from ddtrace.internal import atexit
 from ddtrace.internal.constants import EXPERIMENTAL_FEATURES
+from ddtrace.internal.runtime import on_runtime_id_change
 from ddtrace.internal.settings._agent import config as agent_config
 from ddtrace.internal.settings._config import config
 from ddtrace.internal.threads import Lock
@@ -93,11 +94,10 @@ def __init__(self, interval=DEFAULT_RUNTIME_METRICS_INTERVAL, dogstatsd_url=None
         else:
             self.send_metric = self._dogstatsd_client.distribution
 
+        self._platform_tags = self._build_platform_tags()
         if config._runtime_metrics_runtime_id_enabled:
-            # Enables tagging runtime metrics with runtime-id (as well as all the v1 tags)
-            self._platform_tags = self._format_tags(PlatformTagsV2())
-        else:
-            self._platform_tags = self._format_tags(PlatformTags())
+            # refresh ids to ensure the tags are up to date upon MicroVM instance starts.
+            on_runtime_id_change(self._on_identity_refresh)
 
         self._process_tags: list[str] = list(ProcessTags())
         # Only dd.internal.entity_id needs preserving here: service/env/version are already
@@ -108,6 +108,15 @@ def __init__(self, interval=DEFAULT_RUNTIME_METRICS_INTERVAL, dogstatsd_url=None
             tag for tag in (self._dogstatsd_client.constant_tags or []) if tag.startswith(entity_id_prefix)
         ]
 
+    def _build_platform_tags(self) -> list[str]:
+        if config._runtime_metrics_runtime_id_enabled:
+            # Enables tagging runtime metrics with runtime-id (as well as all the v1 tags)
+            return self._format_tags(PlatformTagsV2())
+        return self._format_tags(PlatformTags())
+
+    def _on_identity_refresh(self, _new_runtime_id: str) -> None:
+        self._platform_tags = self._build_platform_tags()
+
     @classmethod
     def disable(cls) -> None:
         with cls._lock:
diff --git a/ddtrace/internal/settings/asm.py b/ddtrace/internal/settings/asm.py
index c129617c3f8..cba47f46631 100644
--- a/ddtrace/internal/settings/asm.py
+++ b/ddtrace/internal/settings/asm.py
@@ -268,7 +268,9 @@ class ASMConfig(DDConfig):
         sys.platform.startswith("win") or sys.platform.startswith("cygwin")
     )
 
-    _rc_client_id: Optional[str] = None
+    # Set by enable_appsec_rc()/disable_appsec_rc(); gates _dd.rc.client_id span tagging so it's
+    # only emitted while AppSec RC is actually enabled, not just whenever a live RC client exists.
+    _rc_client_id_enabled: bool = False
 
     def __init__(self):
         super().__init__()
diff --git a/ddtrace/internal/symbol_db/symbols.py b/ddtrace/internal/symbol_db/symbols.py
index b8b9daa35e6..39cfecd6753 100644
--- a/ddtrace/internal/symbol_db/symbols.py
+++ b/ddtrace/internal/symbol_db/symbols.py
@@ -38,6 +38,7 @@
 from ddtrace.internal.periodic import Timer
 from ddtrace.internal.runtime import get_ancestor_runtime_id
 from ddtrace.internal.runtime import get_runtime_id
+from ddtrace.internal.runtime import on_runtime_id_change
 from ddtrace.internal.safety import _isinstance
 from ddtrace.internal.settings._agent import config as agent_config
 from ddtrace.internal.settings.dynamic_instrumentation import config as di_config
@@ -561,6 +562,9 @@ def __init__(self, scopes: t.Optional[list[Scope]] = None) -> None:
         }
 
         forksafe.register(self._reset_on_fork)
+        # Same rebuild, triggered by an explicit identity refresh (e.g. an AWS Lambda
+        # MicroVM /run hook) rather than an actual fork.
+        on_runtime_id_change(self._on_identity_refresh)
 
     @cached_property
     def _sender(self) -> SymDBSender:
@@ -577,6 +581,11 @@ def _reset_on_fork(self) -> None:
         self._event_data["runtimeId"] = get_runtime_id()
         self._event_data["parentId"] = get_ancestor_runtime_id()
 
+    def _on_identity_refresh(self, new_runtime_id: str) -> None:
+        # Same rebuild as _reset_on_fork(): runtimeId is baked into _event_data, so it must be
+        # refreshed here too or every batch keeps reporting the pre-refresh snapshot's ID.
+        self._reset_on_fork()
+
     def _set_timer(self) -> None:
         with self._timer_lock:
             if self._timer is None:
diff --git a/ddtrace/internal/telemetry/writer.py b/ddtrace/internal/telemetry/writer.py
index e5ab691ed75..863da58eafb 100644
--- a/ddtrace/internal/telemetry/writer.py
+++ b/ddtrace/internal/telemetry/writer.py
@@ -22,6 +22,7 @@
 from ..runtime import get_ancestor_runtime_id
 from ..runtime import get_parent_runtime_id
 from ..runtime import get_runtime_id
+from ..runtime import on_runtime_id_change
 from ..utils.formats import get_test_session_token
 from ..utils.version import version as tracer_version
 from .constants import TELEMETRY_APM_PRODUCT
@@ -203,6 +204,9 @@ def __init__(self, agentless: Optional[bool] = None) -> None:
             # runtime's after_fork_child hook, which get_native_runtime() registered
             # during enable(), so the shared runtime is restarted before we rebuild).
             forksafe.register(self._fork_writer)
+            # Same rebuild, triggered by an explicit identity refresh (e.g. an AWS Lambda
+            # MicroVM /run hook) rather than an actual fork.
+            on_runtime_id_change(self._on_identity_refresh)
             get_logger("ddtrace").addHandler(DDTelemetryErrorHandler(self))
 
     def _build_worker(self) -> "TelemetryWorker":
@@ -915,6 +919,18 @@ def _fork_writer(self) -> None:
         # Re-discover dependencies from scratch so the child reports its own imports.
         self._dependency_tracker.reset()
 
+    def _on_identity_refresh(self, new_runtime_id: str) -> None:
+        # Same rebuild as _fork_writer(): the native worker bakes in get_runtime_id() at
+        # construction, so it must be dropped and lazily rebuilt on the next telemetry call.
+        # Unlike after a fork, the worker is still alive here -- it must be stopped (not just
+        # dropped) or it keeps heartbeating with the stale runtime ID until process shutdown.
+        if self._worker is not None:
+            try:
+                self._worker.stop(send_app_closing=False)
+            except Exception:
+                log.debug("Failed to stop the native telemetry worker during identity refresh", exc_info=True)
+        self._fork_writer()
+
     def _telemetry_excepthook(self, tp, value, root_traceback) -> None:
         if root_traceback is not None:
             # Get the frame which raised the exception
diff --git a/ddtrace/internal/writer/writer.py b/ddtrace/internal/writer/writer.py
index 652cca4b15c..4772fd293ff 100644
--- a/ddtrace/internal/writer/writer.py
+++ b/ddtrace/internal/writer/writer.py
@@ -20,6 +20,7 @@
 from ddtrace.internal.native._native import SpanData
 from ddtrace.internal.native_runtime import get_native_runtime
 from ddtrace.internal.runtime import get_runtime_id
+from ddtrace.internal.runtime import on_runtime_id_change
 from ddtrace.internal.settings import env
 from ddtrace.internal.settings._agent import config as agent_config
 from ddtrace.internal.settings._config import config
@@ -815,6 +816,7 @@ def __init__(
         self._stats_opt_out = stats_opt_out
 
         self._exporter = self._create_exporter()
+        on_runtime_id_change(self._on_identity_refresh)
 
     @staticmethod
     def _parse_otlp_headers(raw: str) -> list:
@@ -930,6 +932,19 @@ def shutdown_exporter(self) -> None:
         except Exception:
             _safelog(log.warning, "failed to shutdown exporter", exc_info=True)
 
+    def _on_identity_refresh(self, new_runtime_id: str) -> None:
+        # Rebuild the exporter so it picks up the new runtime_id (baked in at construction
+        # via enable_telemetry()), without touching the span buffer: unlike a fork, no spans
+        # were lost, so anything already buffered should still flush once the new exporter's
+        # connection is up. Modeled on set_test_session_token(), not recreate()/fork, which
+        # replace the whole writer and drop the buffer.
+        old_exporter = self._exporter
+        self._exporter = self._create_exporter()
+        try:
+            old_exporter.shutdown(3_000_000_000)
+        except Exception:
+            _safelog(log.warning, "failed to shutdown exporter", exc_info=True)
+
     def recreate(
         self,
         appsec_enabled: Optional[bool] = None,
diff --git a/src/native/crashtracker.rs b/src/native/crashtracker.rs
index e2fb8149de7..ad1a35ce6fb 100644
--- a/src/native/crashtracker.rs
+++ b/src/native/crashtracker.rs
@@ -320,6 +320,19 @@ pub fn crashtracker_on_fork<'py>(
     libdd_crashtracker::on_fork(inner_config, inner_receiver_config, inner_metadata)
 }
 
+#[pyfunction(name = "crashtracker_reconfigure")]
+pub fn crashtracker_reconfigure<'py>(
+    mut config: PyRefMut<'py, CrashtrackerConfigurationPy>,
+    mut receiver_config: PyRefMut<'py, CrashtrackerReceiverConfigPy>,
+    mut metadata: PyRefMut<'py, CrashtrackerMetadataPy>,
+) -> anyhow::Result<()> {
+    let inner_config = (*config).take_inner_or_err()?;
+    let inner_receiver_config = (*receiver_config).take_inner_or_err()?;
+    let inner_metadata = (*metadata).take_inner_or_err()?;
+
+    libdd_crashtracker::reconfigure(inner_config, inner_receiver_config, inner_metadata)
+}
+
 #[pyfunction(name = "crashtracker_status")]
 pub fn crashtracker_status() -> anyhow::Result<CrashtrackerStatus> {
     CrashtrackerStatus::try_from(CRASHTRACKER_STATUS.load(Ordering::SeqCst))
diff --git a/src/native/lib.rs b/src/native/lib.rs
index 61cb6005a9f..0ca9d530751 100644
--- a/src/native/lib.rs
+++ b/src/native/lib.rs
@@ -56,6 +56,7 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> {
         m.add_class::<crashtracker::CrashtrackerStatus>()?;
         m.add_function(wrap_pyfunction!(crashtracker::crashtracker_init, m)?)?;
         m.add_function(wrap_pyfunction!(crashtracker::crashtracker_on_fork, m)?)?;
+        m.add_function(wrap_pyfunction!(crashtracker::crashtracker_reconfigure, m)?)?;
         m.add_function(wrap_pyfunction!(crashtracker::crashtracker_status, m)?)?;
         m.add_function(wrap_pyfunction!(crashtracker::crashtracker_receiver, m)?)?;
         m.add_function(wrap_pyfunction!(
diff --git a/tests/appsec/appsec/test_asm_request_context.py b/tests/appsec/appsec/test_asm_request_context.py
index c42a8768652..94241ee87b5 100644
--- a/tests/appsec/appsec/test_asm_request_context.py
+++ b/tests/appsec/appsec/test_asm_request_context.py
@@ -14,6 +14,20 @@
 config_asm = {"_asm_enabled": True}
 
 
+@pytest.mark.parametrize("auto_enable_crashtracking", [False])
+def test_import_does_not_load_remoteconfig_worker(run_python_code_in_subprocess, auto_enable_crashtracking):
+    code = """
+import sys
+
+import ddtrace.appsec._asm_request_context  # noqa: F401
+
+assert "ddtrace.internal.remoteconfig.worker" not in sys.modules
+"""
+
+    _, stderr, status, _ = run_python_code_in_subprocess(code)
+    assert status == 0, stderr
+
+
 def test_context_set_and_reset():
     with asm_context(
         ip_addr=_TEST_IP,
diff --git a/tests/appsec/appsec/test_remoteconfiguration.py b/tests/appsec/appsec/test_remoteconfiguration.py
index f0c7f66dd17..c169626ebba 100644
--- a/tests/appsec/appsec/test_remoteconfiguration.py
+++ b/tests/appsec/appsec/test_remoteconfiguration.py
@@ -270,6 +270,46 @@ def test_rc_activation_validate_client_id(tracer, rc_poller, appsec_callback):
     disable_appsec_rc()
 
 
+def test_rc_client_id_tag_reflects_live_value_not_a_stale_cache(tracer, rc_poller, appsec_callback):
+    """_dd.rc.client_id must be read live at span-tagging time, not cached once at RC enable time.
+
+    Otherwise the tag would go stale after e.g. an AWS Lambda MicroVM identity refresh
+    regenerates the real client id.
+    """
+    from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
+
+    with override_global_config(dict(_asm_enabled=True, _remote_config_enabled=True, api_version="v0.4")):
+        tracer.configure(appsec_enabled=True)
+        enable_appsec_rc(appsec_callback)
+
+        with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"):
+            with asm_context(tracer) as span:
+                set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200")
+            assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-one"
+
+        with mock.patch.object(remoteconfig_poller._client, "id", "client-id-two"):
+            with asm_context(tracer) as span:
+                set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200")
+            assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-two"
+    disable_appsec_rc()
+
+
+def test_rc_client_id_tag_not_set_when_rc_disabled(tracer):
+    """_dd.rc.client_id must not be tagged when AppSec RC was never enabled, even though a live
+    RC client (and id) exists process-wide -- otherwise every ASM-tracked span would get tagged
+    with an id from a Remote Config subscription AppSec never activated.
+    """
+    from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
+
+    with override_global_config(dict(_asm_enabled=True, api_version="v0.4")):
+        tracer.configure(appsec_enabled=True)
+
+        with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"):
+            with asm_context(tracer) as span:
+                set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200")
+            assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) is None
+
+
 @pytest.mark.parametrize(
     "env_rules, expected",
     [
diff --git a/tests/crashtracker/test_crashtracker.py b/tests/crashtracker/test_crashtracker.py
index 2c23bba2037..677c6aac461 100644
--- a/tests/crashtracker/test_crashtracker.py
+++ b/tests/crashtracker/test_crashtracker.py
@@ -98,6 +98,46 @@ def test_crashtracker_started():
         pytest.fail("contents of stdout.log: %s, stderr.log: %s" % (stdout_msg, stderr_msg))
 
 
+@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only")
+@pytest.mark.subprocess(err=None)
+@pytest.mark.parametrize("auto_enable_crashtracking", [False])
+def test_crashtracker_identity_refresh_reconfigures_metadata():
+    from contextlib import ExitStack
+
+    import mock
+
+    from ddtrace.internal.core import crashtracking
+    import ddtrace.internal.runtime as runtime
+
+    init_args = (object(), object(), object())
+    refresh_args = (object(), object(), object())
+    tags = {"service": "identity-refresh"}
+    initialized = object()
+    status = type("CrashtrackerStatus", (), {"Initialized": initialized})
+    get_args = mock.Mock(side_effect=[init_args, refresh_args])
+    init = mock.Mock()
+    reconfigure = mock.Mock()
+
+    with ExitStack() as stack:
+        stack.enter_context(mock.patch.object(crashtracking, "is_available", True))
+        stack.enter_context(mock.patch.object(crashtracking, "crashtracker_config", mock.Mock(enabled=True)))
+        stack.enter_context(mock.patch.object(crashtracking, "CrashtrackerStatus", status, create=True))
+        stack.enter_context(mock.patch.object(crashtracking, "_identity_refresh_additional_tags", None))
+        stack.enter_context(mock.patch.object(crashtracking, "_get_args", get_args))
+        stack.enter_context(mock.patch.object(crashtracking, "crashtracker_init", init, create=True))
+        stack.enter_context(mock.patch.object(crashtracking, "crashtracker_reconfigure", reconfigure, create=True))
+        stack.enter_context(
+            mock.patch.object(crashtracking, "crashtracker_status", mock.Mock(return_value=initialized), create=True)
+        )
+
+        assert crashtracking.start(tags)
+        runtime.refresh_identity()
+
+    assert get_args.call_args_list == [mock.call(tags), mock.call(tags)]
+    init.assert_called_once_with(*init_args)
+    reconfigure.assert_called_once_with(*refresh_args)
+
+
 @pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only")
 @pytest.mark.subprocess()
 def test_crashtracker_receiver_not_in_path():
diff --git a/tests/internal/remoteconfig/test_remoteconfig_native.py b/tests/internal/remoteconfig/test_remoteconfig_native.py
index f1813266321..ff646a167f2 100644
--- a/tests/internal/remoteconfig/test_remoteconfig_native.py
+++ b/tests/internal/remoteconfig/test_remoteconfig_native.py
@@ -418,3 +418,46 @@ def test_enable_builds_native_runtime_before_registering_fork_hook(monkeypatch):
         assert poller.enable() is True
 
     assert order == ["native", "before_fork", "start"], order
+
+
+def test_identity_refresh_renews_client_id_and_drops_native():
+    # get_client_id() on the native client is documented "stable for the process lifetime",
+    # so refreshing must drop it (not mutate it in place) for the id to actually change.
+    client = RemoteConfigClient()
+    old_id = client.id
+    client.ensure_native()
+    assert client._native is not None
+
+    client._on_identity_refresh("some-new-runtime-id")
+
+    assert client.id != old_id
+    assert client._native is None
+
+
+def test_identity_refresh_rebuilds_native_client_with_fresh_id():
+    client = RemoteConfigClient()
+    native_before = client.ensure_native()
+    old_native_client_id = native_before.get_client_id()
+
+    client._on_identity_refresh("some-new-runtime-id")
+    native_after = client.ensure_native()
+
+    assert native_after is not native_before
+    assert native_after.get_client_id() == client.id
+    assert native_after.get_client_id() != old_native_client_id
+
+
+@pytest.mark.subprocess
+def test_identity_refresh_wired_to_runtime_id_change():
+    """A RemoteConfigClient subscribes itself at construction; refresh_identity() reaches it."""
+    from ddtrace.internal import runtime
+    from ddtrace.internal.remoteconfig.client import RemoteConfigClient
+
+    client = RemoteConfigClient()
+    old_id = client.id
+    client.ensure_native()
+
+    runtime.refresh_identity()
+
+    assert client.id != old_id
+    assert client._native is None
diff --git a/tests/internal/symbol_db/test_symbols.py b/tests/internal/symbol_db/test_symbols.py
index c1c5b075312..920ef5cba70 100644
--- a/tests/internal/symbol_db/test_symbols.py
+++ b/tests/internal/symbol_db/test_symbols.py
@@ -526,6 +526,31 @@ def test_symbols_fork_uploads():
         assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, f"child {pid} exited with status {status}"
 
 
+@pytest.mark.subprocess(ddtrace_run=True, err=None)
+def test_symbols_identity_refresh_updates_runtime_id():
+    """A non-fork identity refresh (e.g. an AWS Lambda MicroVM /run hook) must also refresh the
+    ScopeContext's cached runtimeId, the same way _reset_on_fork() does after an actual fork --
+    otherwise every upload after a MicroVM /run keeps reporting the pre-refresh snapshot's ID.
+    """
+    import typing as t
+
+    import ddtrace.internal.runtime as runtime
+    from ddtrace.internal.symbol_db.symbols import SymbolDatabaseUploader
+
+    SymbolDatabaseUploader.install()
+
+    context = t.cast(SymbolDatabaseUploader, SymbolDatabaseUploader._instance)._context
+    old_runtime_id = runtime.get_runtime_id()
+    old_upload_id = context._upload_id
+    assert context._event_data["runtimeId"] == old_runtime_id
+
+    runtime.refresh_identity()
+
+    assert runtime.get_runtime_id() != old_runtime_id
+    assert context._event_data["runtimeId"] == runtime.get_runtime_id()
+    assert context._upload_id != old_upload_id
+
+
 @pytest.mark.subprocess(ddtrace_run=True, err=None)
 def test_symbols_fork_forces_reenable_and_install():
     """
diff --git a/tests/runtime/test_runtime_metrics_api.py b/tests/runtime/test_runtime_metrics_api.py
index 963d8deeae9..b0e19778190 100644
--- a/tests/runtime/test_runtime_metrics_api.py
+++ b/tests/runtime/test_runtime_metrics_api.py
@@ -232,6 +232,28 @@ def test_runtime_metrics_experimental_runtime_tag():
         )
 
 
+@pytest.mark.subprocess(env={"DD_RUNTIME_METRICS_RUNTIME_ID_ENABLED": "true"}, err=None)
+def test_runtime_metrics_runtime_id_tag_refreshes_on_identity_refresh():
+    from ddtrace.internal import runtime
+    from ddtrace.internal.runtime.runtime_metrics import RuntimeWorker
+
+    try:
+        RuntimeWorker.enable()
+        assert RuntimeWorker._instance is not None
+
+        worker_instance = RuntimeWorker._instance
+        runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}"
+        assert runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags
+
+        runtime.refresh_identity()
+
+        refreshed_runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}"
+        assert refreshed_runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags
+        assert runtime_id_tag not in worker_instance._platform_tags, worker_instance._platform_tags
+    finally:
+        RuntimeWorker.disable()
+
+
 @pytest.mark.subprocess(
     parametrize={"DD_TRACE_EXPERIMENTAL_FEATURES_ENABLED": ["DD_RUNTIME_METRICS_ENABLED,someotherfeature", ""]},
     err=None,
diff --git a/tests/telemetry/test_writer.py b/tests/telemetry/test_writer.py
index 7bbede8e4c6..ce5b71a4c98 100644
--- a/tests/telemetry/test_writer.py
+++ b/tests/telemetry/test_writer.py
@@ -570,6 +570,61 @@ def test_telemetry_writer_agent_setup():
         assert new_telemetry_writer._agentless is False
 
 
+def test_identity_refresh_rebuilds_native_worker():
+    """Same rebuild as after a fork: the native worker bakes in get_runtime_id() at construction."""
+    with override_global_config(
+        {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False}
+    ):
+        writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False)
+        assert writer._worker is not None
+
+        writer._on_identity_refresh("some-new-runtime-id")
+
+        assert writer._worker is None
+        assert writer.started is False
+
+
+def test_identity_refresh_stops_live_worker_before_dropping():
+    """Unlike after a fork, the worker is still alive here and must be explicitly stopped, or it
+    keeps heartbeating with the stale runtime ID until process shutdown.
+    """
+    with override_global_config(
+        {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False}
+    ):
+        writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False)
+        assert writer._worker is not None
+
+        # TelemetryWorker is a native extension type -- its methods can't be patched in place,
+        # so swap in a mock to observe the stop() call instead.
+        fake_worker = mock.Mock()
+        writer._worker = fake_worker
+
+        writer._on_identity_refresh("some-new-runtime-id")
+
+        fake_worker.stop.assert_called_once_with(send_app_closing=False)
+        assert writer._worker is None
+
+
+@pytest.mark.subprocess(
+    env={"DD_SITE": "datad0g.com", "DD_API_KEY": "foobarkey", "DD_CIVISIBILITY_AGENTLESS_ENABLED": "false"}
+)
+def test_identity_refresh_wired_to_runtime_id_change():
+    """Drives the refresh through runtime.refresh_identity() instead of calling
+    _on_identity_refresh directly (as the test above does), so a dropped
+    on_runtime_id_change() subscription would actually fail this.
+    """
+    from ddtrace.internal import runtime
+    import ddtrace.internal.telemetry
+
+    writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False)
+    assert writer._worker is not None
+
+    runtime.refresh_identity()
+
+    assert writer._worker is None
+    assert writer.started is False
+
+
 @pytest.mark.parametrize(
     "env_agentless,arg_agentless",
     [
diff --git a/tests/tracer/test_writer.py b/tests/tracer/test_writer.py
index cc90b6403b2..d9f07704e74 100644
--- a/tests/tracer/test_writer.py
+++ b/tests/tracer/test_writer.py
@@ -419,6 +419,20 @@ def test_on_shutdown_before_start(self):
         # Call shutdown without ever calling start()
         writer.on_shutdown()
 
+    def test_identity_refresh_rebuilds_exporter_without_recreating_writer(self):
+        """Same trigger as an AWS Lambda MicroVM /run hook: rebuild the exporter (it bakes in
+        get_runtime_id() at construction) without touching the writer/buffer, unlike recreate()
+        (used on fork), which replaces the whole writer and drops anything already written.
+        """
+        writer = NativeWriter("http://dne:1234")
+        old_exporter = writer._exporter
+        old_clients = writer._clients
+
+        writer._on_identity_refresh("some-new-runtime-id")
+
+        assert writer._exporter is not old_exporter
+        assert writer._clients is old_clients
+
     # Http related metrics are sent by the native code
     def test_drop_reason_bad_endpoint(self):
         pytest.skip()
@@ -428,6 +442,25 @@ def test_gzip_compression_exception_logging_and_metrics(self):
         pytest.skip()
 
 
+@pytest.mark.subprocess
+def test_native_writer_identity_refresh_wired_to_runtime_id_change():
+    """Drives the refresh through runtime.refresh_identity() instead of calling
+    _on_identity_refresh directly (as the test above does), so a dropped
+    on_runtime_id_change() subscription would actually fail this.
+    """
+    from ddtrace.internal import runtime
+    from ddtrace.internal.writer import NativeWriter
+
+    writer = NativeWriter("http://dne:1234")
+    old_exporter = writer._exporter
+    old_clients = writer._clients
+
+    runtime.refresh_identity()
+
+    assert writer._exporter is not old_exporter
+    assert writer._clients is old_clients
+
+
 class CIVisibilityWriterTests(NativeWriterTests):
     WRITER_CLASS = CIVisibilityWriter