diff --git a/sentry_sdk/integrations/celery/__init__.py b/sentry_sdk/integrations/celery/__init__.py
index 1b1eb4f380..fe832f159e 100644
--- a/sentry_sdk/integrations/celery/__init__.py
+++ b/sentry_sdk/integrations/celery/__init__.py
@@ -231,6 +231,12 @@ def _update_celery_task_headers(
                 if key.startswith("sentry-"):
                     updated_headers["headers"][key] = value
 
+            # Preserve user-provided custom headers in the inner "headers" dict
+            # so they survive to task.request.headers on the worker (celery#4875).
+            for key, value in original_headers.items():
+                if key != "headers" and key not in updated_headers["headers"]:
+                    updated_headers["headers"][key] = value
+
     return updated_headers
 
 
diff --git a/tests/integrations/celery/test_celery.py b/tests/integrations/celery/test_celery.py
index b8fc2bb3e8..42ae6ea14f 100644
--- a/tests/integrations/celery/test_celery.py
+++ b/tests/integrations/celery/test_celery.py
@@ -847,6 +847,35 @@ def test_send_task_wrapped(
     assert span["trace_id"] == kwargs["headers"]["sentry-trace"].split("-")[0]
 
 
+def test_user_custom_headers_accessible_in_task(init_celery):
+    """
+    Regression test for https://github.com/getsentry/sentry-python/issues/5566
+
+    User-provided custom headers passed to apply_async() must be accessible
+    via task.request.headers on the worker side.
+    """
+    celery = init_celery(traces_sample_rate=1.0)
+
+    @celery.task(name="custom_headers_task", bind=True)
+    def custom_headers_task(self):
+        return dict(self.request.headers or {})
+
+    custom_headers = {
+        "my_custom_key": "my_value",
+        "correlation_id": "abc-123",
+        "tenant_id": "tenant-42",
+    }
+
+    with start_transaction(name="test"):
+        result = custom_headers_task.apply_async(headers=custom_headers)
+
+    received_headers = result.get()
+    for key, value in custom_headers.items():
+        assert received_headers.get(key) == value, (
+            f"Custom header {key!r} not found in task.request.headers"
+        )
+
+
 @pytest.mark.skip(reason="placeholder so that forked test does not come last")
 def test_placeholder():
     """Forked tests must not come last in the module.