fix(celery): Propagate user-set headers (#5581)

### Description

The SDK interferes with Celery logic that puts custom headers into
`request.headers` in the worker. Currently, we only make internal sentry
headers available there. Make sure to copy over any user-set headers as
well.

#### Issues
Closes #5566

#### Reminders
- Please add tests to validate your changes, and lint your code using
`tox -e linters`.
- Add GH Issue ID _&_ Linear ID (if applicable)
- PR title should use [conventional
commit](https://develop.sentry.dev/engineering-practices/commit-messages/#type)
style (`feat:`, `fix:`, `ref:`, `meta:`)
- For external contributors:
[CONTRIBUTING.md](https://github.com/getsentry/sentry-python/blob/master/CONTRIBUTING.md),
[Sentry SDK development docs](https://develop.sentry.dev/sdk/), [Discord
community](https://discord.gg/Ww9hbqr)

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: sentrivana

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
 
 

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.