diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index b4460b21e7563..26662d523a3ff 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -1031,6 +1031,7 @@ milli
 millis
 milton
 minikube
+misconfiguration
 misconfigured
 Mixin
 mixin
diff --git a/providers/elasticsearch/docs/changelog.rst b/providers/elasticsearch/docs/changelog.rst
index 47f9b3bcbeaa2..5e6a4f695a510 100644
--- a/providers/elasticsearch/docs/changelog.rst
+++ b/providers/elasticsearch/docs/changelog.rst
@@ -35,6 +35,20 @@ a dictionary key in the task-log output when log-hits did not carry a
 ``host`` field. The Elasticsearch client is still connected using the
 full URL, so authentication is unaffected.
 
+A new ``[elasticsearch] es_compat_with`` config option lets operators pin
+the ``compatible-with`` HTTP content-negotiation level used by the
+Elasticsearch client. Since 6.5.1 the provider depends on
+``elasticsearch>=8.10,<10``, and a default install resolves to an
+``elasticsearch>=9`` client which unconditionally negotiates
+``compatible-with=9`` on every request. Elasticsearch 8.x servers reject
+that with HTTP 400 ``media_type_header_exception`` (regression introduced
+by #64070), breaking remote task log ingestion and the SQL/Python hooks
+against ES 8 clusters. Setting ``es_compat_with = "8"`` rewrites the
+client transport so every outbound request carries
+``compatible-with=8`` (and the matching ``+x-ndjson`` form for bulk
+requests), restoring compatibility without dropping ES 9 support. When
+unset, behavior is unchanged.
+
 6.5.3
 .....
 
diff --git a/providers/elasticsearch/docs/logging/index.rst b/providers/elasticsearch/docs/logging/index.rst
index 3254bf5406ccf..5cfae8d6230da 100644
--- a/providers/elasticsearch/docs/logging/index.rst
+++ b/providers/elasticsearch/docs/logging/index.rst
@@ -93,6 +93,48 @@ Additionally, in the ``elasticsearch_configs`` section, you can pass any paramet
     api_key = "SOMEAPIKEY"
     verify_certs = True
 
+Pinning the ``compatible-with`` content-negotiation level
+'''''''''''''''''''''''''''''''''''''''''''''''''''''''''
+
+Since provider 6.5.1, the Elasticsearch dependency is ``elasticsearch>=8.10,<10``,
+which means a default install resolves to an ``elasticsearch>=9`` Python client.
+That client unconditionally negotiates ``compatible-with=9`` on every request,
+which Elasticsearch 8.x servers reject with HTTP 400
+``media_type_header_exception``. Both the task log writer and the
+``ElasticsearchSQLHook`` / ``ElasticsearchPythonHook`` are affected.
+
+If you need to keep a single Airflow image compatible with an
+``elasticsearch<9`` server, set ``[elasticsearch] es_compat_with`` to the server
+major version. The provider then rewrites the client transport so every outbound
+request carries ``Accept`` / ``Content-Type:
+application/vnd.elasticsearch+json; compatible-with=<major>`` (and the matching
+``+x-ndjson`` form for bulk requests):
+
+.. code-block:: ini
+
+    [elasticsearch]
+    es_compat_with = 8
+
+Only a positive integer major version is accepted (``"7"``, ``"8"``, ``"9"``);
+any other value (e.g. ``"v8"``, ``"8.0"``) fails fast with an
+``AirflowConfigException`` at client construction time so the misconfiguration
+is obvious in the worker startup log instead of producing a per-request 400
+storm.
+
+.. note::
+
+   The fix is installed at the **transport layer** (a wrapper around
+   ``client.transport.perform_request``) and therefore overrides the
+   per-API-method ``Accept`` / ``Content-Type`` headers that elasticsearch-py
+   negotiates from its own client major. Constructor-level ``headers=`` on
+   ``Elasticsearch.__init__`` and the ``elasticsearch_configs`` section do
+   **not** work for this purpose — elasticsearch-py re-applies its own
+   ``compatible-with=<client_major>`` headers right before the request goes
+   out, after any constructor headers.
+
+When the option is unset the client behaves as before (negotiating its own
+major version).
+
 .. _elasticsearch-document-schema:
 
 Expected Elasticsearch document schema
diff --git a/providers/elasticsearch/provider.yaml b/providers/elasticsearch/provider.yaml
index 5fff75d12efec..822580e58c881 100644
--- a/providers/elasticsearch/provider.yaml
+++ b/providers/elasticsearch/provider.yaml
@@ -221,6 +221,20 @@ config:
         type: string
         example: ~
         default: "1000"
+      es_compat_with:
+        description: |
+          Pin the ``compatible-with`` HTTP content-negotiation level used by the
+          Elasticsearch client. Accepts a server major version string (e.g. ``"7"``,
+          ``"8"``, ``"9"``). When unset, the elasticsearch-py client negotiates its
+          own major version, which makes an ``elasticsearch>=9`` client (the default
+          for fresh installs) incompatible with Elasticsearch 8.x servers — every
+          request is rejected with HTTP 400 ``media_type_header_exception``.
+          Setting this option keeps a single Airflow image compatible with both
+          ``elasticsearch<9`` and ``elasticsearch>=9`` servers.
+        version_added: 6.5.4
+        type: string
+        example: "8"
+        default: ""
   elasticsearch_configs:
     description: ~
     options:
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
new file mode 100644
index 0000000000000..ca888db8a6f99
--- /dev/null
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -0,0 +1,119 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+Helpers shared between the Elasticsearch hooks and log handler.
+
+Currently this exposes a single helper, :func:`apply_compat_with`, that lets the
+provider keep working against an Elasticsearch server whose major version does
+not match the installed ``elasticsearch`` Python client major. See the helper
+docstring for the regression context.
+"""
+
+from __future__ import annotations
+
+import functools
+import re
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+    import elasticsearch
+
+
+# Matches the JSON / NDJSON / mapbox vector-tile mimetypes the ``elasticsearch``
+# client negotiates, in either form: the raw ``application/json`` (what the
+# generated client code writes into ``__headers``) and the already-rewritten
+# ``application/vnd.elasticsearch+json; compatible-with=<N>`` (what
+# ``mimetype_header_to_compat`` in ``elasticsearch/_sync/client/_base.py``
+# emits before handing the request to the transport). Both forms must be
+# rewritten to ``compatible-with=<configured>``; anything else (notably
+# ``text/plain`` used by the cat APIs) is left intact, mirroring upstream's
+# selective substitution behaviour.
+_COMPAT_MIMETYPE_RE = re.compile(
+    r"application/(?:vnd\.elasticsearch\+)?(json|x-ndjson|vnd\.mapbox-vector-tile)"
+    r"(?:\s*;\s*compatible-with=\d+)?"
+)
+_COMPAT_MAJOR_RE = re.compile(r"^\d+$")
+
+
+def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elasticsearch:
+    """
+    Pin the ``compatible-with`` HTTP content-negotiation level for ``client``.
+
+    The ``elasticsearch`` Python client always negotiates ``compatible-with=<client_major>``
+    on every request; an Elasticsearch server with a different major version rejects
+    the request with HTTP 400 ``media_type_header_exception``. This is what happens
+    when an ``elasticsearch>=9`` client (the current default) talks to an
+    Elasticsearch 8.x server, which broke remote logging for ES 8 deployments
+    starting with provider 6.5.1.
+
+    When ``[elasticsearch] es_compat_with`` is set to a major version string
+    (e.g. ``"7"``, ``"8"``, ``"9"``) this helper wraps the client's transport
+    so every outbound request rewrites the ``compatible-with=<N>`` parameter on
+    the JSON / NDJSON / mapbox vector-tile parts of the ``Accept`` and
+    ``Content-Type`` headers. Non-JSON parts (notably ``text/plain`` used by
+    the cat APIs) are preserved verbatim, mirroring how elasticsearch-py's own
+    ``mimetype_header_to_compat`` handles content negotiation.
+
+    When the option is unset the client is returned unchanged and behaves
+    exactly as before.
+    """
+    raw = conf.get("elasticsearch", "es_compat_with", fallback=None)
+    compat = (raw or "").strip()
+    if not compat:
+        return client
+    if not _COMPAT_MAJOR_RE.match(compat):
+        raise AirflowConfigException(
+            "[elasticsearch] es_compat_with must be a positive integer major version "
+            f"(e.g. '7', '8', '9'); got {raw!r}."
+        )
+
+    transport = client.transport
+    if "perform_request" in transport.__dict__:
+        # Already wrapped on this transport instance — no-op so repeated calls
+        # to ``apply_compat_with`` (e.g. across hook reuse) stay idempotent.
+        return client
+
+    sub = rf"application/vnd.elasticsearch+\g<1>; compatible-with={compat}"
+    original_perform_request = transport.perform_request
+
+    # Accept ``*args, **kwargs`` so the wrapper survives future elastic_transport
+    # ``perform_request`` signature changes (new keyword-only params, reordered
+    # positionals). ``functools.wraps`` preserves the original ``__name__`` /
+    # ``__doc__`` / ``__wrapped__`` for tooling and introspection.
+    @functools.wraps(original_perform_request)
+    def perform_request(*args, **kwargs):  # type: ignore[no-untyped-def]
+        headers = kwargs.get("headers")
+        if not headers:
+            return original_perform_request(*args, **kwargs)
+        # Walk every key case-insensitively so a future elastic_transport that
+        # forwards PascalCase headers does not silently bypass the rewrite.
+        merged = dict(headers)
+        for key in list(merged):
+            if key.lower() in ("accept", "content-type") and merged[key]:
+                merged[key] = _COMPAT_MIMETYPE_RE.sub(sub, merged[key])
+        kwargs["headers"] = merged
+        return original_perform_request(*args, **kwargs)
+
+    # ``setattr`` instead of direct attribute assignment so mypy does not flag a
+    # ``method-assign`` error — we are *intentionally* shadowing the bound method
+    # at the instance level (the idempotency guard above checks the instance
+    # ``__dict__``).
+    setattr(transport, "perform_request", perform_request)
+    return client
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
index 2d357cecb1944..b0853a98580cd 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
@@ -151,6 +151,13 @@ def get_provider_info():
                         "example": None,
                         "default": "1000",
                     },
+                    "es_compat_with": {
+                        "description": 'Pin the ``compatible-with`` HTTP content-negotiation level used by the\nElasticsearch client. Accepts a server major version string (e.g. ``"7"``,\n``"8"``, ``"9"``). When unset, the elasticsearch-py client negotiates its\nown major version, which makes an ``elasticsearch>=9`` client (the default\nfor fresh installs) incompatible with Elasticsearch 8.x servers — every\nrequest is rejected with HTTP 400 ``media_type_header_exception``.\nSetting this option keeps a single Airflow image compatible with both\n``elasticsearch<9`` and ``elasticsearch>=9`` servers.\n',
+                        "version_added": "6.5.4",
+                        "type": "string",
+                        "example": "8",
+                        "default": "",
+                    },
                 },
             },
             "elasticsearch_configs": {
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
index e8f0cfad00ce6..e8d1610d64709 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
@@ -27,6 +27,7 @@
 
 from airflow.providers.common.compat.sdk import BaseHook
 from airflow.providers.common.sql.hooks.sql import DbApiHook
+from airflow.providers.elasticsearch._compat import apply_compat_with
 
 if TYPE_CHECKING:
     from elastic_transport import ObjectApiResponse
@@ -135,9 +136,9 @@ def __init__(
         netloc = f"{host}:{port}"
         self.url = parse.urlunparse((scheme, netloc, "/", None, None, None))
         if user and password:
-            self.es = Elasticsearch(self.url, basic_auth=(user, password), **kwargs)
+            self.es = apply_compat_with(Elasticsearch(self.url, basic_auth=(user, password), **kwargs))
         else:
-            self.es = Elasticsearch(self.url, **kwargs)
+            self.es = apply_compat_with(Elasticsearch(self.url, **kwargs))
 
     def cursor(self) -> ElasticsearchSQLCursor:
         return ElasticsearchSQLCursor(self.es, **self.kwargs)
@@ -247,7 +248,7 @@ def __init__(self, hosts: list[Any], es_conn_args: dict | None = None):
 
     def _get_elastic_connection(self):
         """Return the Elasticsearch client."""
-        client = Elasticsearch(self.hosts, **self.es_conn_args)
+        client = apply_compat_with(Elasticsearch(self.hosts, **self.es_conn_args))
 
         return client
 
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
index eb531452df824..402d64e5bf112 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -43,6 +43,7 @@
 import airflow.logging_config as alc
 from airflow.models.dagrun import DagRun
 from airflow.providers.common.compat.sdk import conf
+from airflow.providers.elasticsearch._compat import apply_compat_with
 from airflow.providers.elasticsearch.log.es_json_formatter import ElasticsearchJSONFormatter
 from airflow.providers.elasticsearch.log.es_response import ElasticSearchResponse, Hit, resolve_nested
 from airflow.providers.elasticsearch.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS
@@ -269,7 +270,7 @@ def __init__(
         )
         self.closed = False
 
-        self.client = elasticsearch.Elasticsearch(self.host, **es_kwargs)
+        self.client = apply_compat_with(elasticsearch.Elasticsearch(self.host, **es_kwargs))
         # in airflow.cfg, host of elasticsearch has to be http://dockerhostXxxx:9200
 
         self.frontend = frontend
@@ -651,7 +652,7 @@ class ElasticsearchRemoteLogIO(LoggingMixin):  # noqa: D101
 
     def __attrs_post_init__(self):
         es_kwargs = get_es_kwargs_from_config()
-        self.client = elasticsearch.Elasticsearch(self.host, **es_kwargs)
+        self.client = apply_compat_with(elasticsearch.Elasticsearch(self.host, **es_kwargs))
         self.index_patterns_callable = conf.get("elasticsearch", "index_patterns_callable", fallback="")
         self.PAGE = 0
         self.MAX_LINE_PER_PAGE = conf.getint("elasticsearch", "max_lines_per_page", fallback=1000)
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test__compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test__compat.py
new file mode 100644
index 0000000000000..7f81f9b92f099
--- /dev/null
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test__compat.py
@@ -0,0 +1,240 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Wire-level tests for ``airflow.providers.elasticsearch._compat.apply_compat_with``.
+
+These tests intercept ``elastic_transport.Transport.perform_request`` so they
+observe the exact ``Accept`` / ``Content-Type`` headers the helper produces.
+Asserting on ``client._headers`` (or any other in-memory state on the client)
+would not be enough: the elasticsearch-py client always re-applies its own
+per-API-method content-negotiation headers right before the request goes out,
+which is the very behaviour this helper has to override. Only what the
+``Transport`` sees on the wire matters.
+"""
+
+from __future__ import annotations
+
+import contextlib
+
+import pytest
+from elastic_transport import Transport
+from elasticsearch import Elasticsearch
+
+from airflow.providers.common.compat.sdk import AirflowConfigException
+from airflow.providers.elasticsearch._compat import apply_compat_with
+
+from tests_common.test_utils.config import conf_vars
+
+
+def _trigger_calls(client: Elasticsearch) -> None:
+    """Drive ``search``, ``info`` and ``bulk`` against the spy transport.
+
+    Each call hits the spy and raises; we swallow that and rely on the spy's
+    ``captured`` list to make assertions.
+    """
+    for action in (
+        lambda: client.search(index="airflow-logs", query={"match_all": {}}),
+        lambda: client.info(),
+        lambda: client.bulk(operations=[{"index": {"_index": "x"}}, {"hello": "world"}]),
+    ):
+        with contextlib.suppress(RuntimeError):
+            action()
+
+
+@pytest.fixture
+def wire_capture(monkeypatch):
+    """Replace ``Transport.perform_request`` with a recording spy.
+
+    The spy is installed at the *class* level (not on an instance) so it is
+    picked up by both the original transport and by the wrapper produced by
+    :func:`apply_compat_with`. The wrapper resolves the original
+    ``perform_request`` at wrap time, so the spy must already be in place when
+    ``apply_compat_with`` runs.
+    """
+    captured: list[dict] = []
+
+    def spy(self, method, target, *, body=None, headers=None, **kwargs):
+        captured.append({"method": method, "target": target, "headers": dict(headers or {})})
+        raise RuntimeError("captured")
+
+    monkeypatch.setattr(Transport, "perform_request", spy)
+    return captured
+
+
+@pytest.mark.parametrize("unset_value", ["", None])
+def test_apply_compat_with_unset_does_not_wrap_transport(unset_value):
+    """When the option is unset the helper returns the client untouched.
+
+    The wrap installs ``perform_request`` as an instance attribute on the
+    transport, so the most precise way to assert the helper is a no-op is to
+    check that no such instance attribute was set (i.e. lookup still resolves
+    to the class method). Identity (``is``) comparison on
+    ``transport.perform_request`` would not work even in the no-op case
+    because attribute access on a bound method produces a fresh wrapper each
+    time.
+
+    Note: ``conf_vars`` removes the override when the value is ``None``, so
+    both parametrized cases ultimately resolve to the provider yaml default
+    (``""``). The parametrize is kept to document that callers passing either
+    sentinel get the no-op path; the actual ``None`` branch in the helper is
+    covered by ``test_apply_compat_with_unset_via_missing_conf`` below.
+    """
+    client = Elasticsearch("http://localhost:9200")
+    assert "perform_request" not in client.transport.__dict__
+    with conf_vars({("elasticsearch", "es_compat_with"): unset_value}):
+        same = apply_compat_with(client)
+    assert same is client
+    assert "perform_request" not in client.transport.__dict__
+
+
+def test_apply_compat_with_unset_via_missing_conf(monkeypatch):
+    """Cover the ``conf.get`` returning ``None`` branch directly."""
+    from airflow.providers.elasticsearch import _compat
+
+    monkeypatch.setattr(_compat.conf, "get", lambda *args, **kwargs: None)
+    client = Elasticsearch("http://localhost:9200")
+    assert apply_compat_with(client) is client
+    assert "perform_request" not in client.transport.__dict__
+
+
+def test_apply_compat_with_pins_compatible_with_8(wire_capture):
+    """With ``es_compat_with = "8"`` every outbound call carries ``compatible-with=8``."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    _trigger_calls(client)
+
+    assert {c["method"] for c in wire_capture} == {"POST", "GET", "PUT"}
+    expected_json = "application/vnd.elasticsearch+json; compatible-with=8"
+    expected_ndjson = "application/vnd.elasticsearch+x-ndjson; compatible-with=8"
+
+    by_method = {c["method"]: c for c in wire_capture}
+    assert by_method["POST"]["headers"]["accept"] == expected_json
+    assert by_method["POST"]["headers"]["content-type"] == expected_json
+    assert by_method["GET"]["headers"]["accept"] == expected_json
+    # ``info()`` does not send a body, so content-type is absent on the wire
+    assert by_method["GET"]["headers"].get("content-type") in (None, "")
+    assert by_method["PUT"]["headers"]["accept"] == expected_json
+    # bulk preserves the ndjson form so the server can stream the body
+    assert by_method["PUT"]["headers"]["content-type"] == expected_ndjson
+
+
+def test_apply_compat_with_pins_compatible_with_7(wire_capture):
+    """The helper accepts arbitrary major version strings, not just ``"8"``."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "7"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    _trigger_calls(client)
+    assert wire_capture, "spy should have captured at least one call"
+    assert all(
+        c["headers"]["accept"] == "application/vnd.elasticsearch+json; compatible-with=7"
+        for c in wire_capture
+    )
+
+
+def test_apply_compat_with_preserves_text_plain_for_cat_apis(wire_capture):
+    """Cat APIs send ``Accept: text/plain[,application/json]``; the wrapper must
+    preserve the ``text/plain`` part. Earlier revisions of the helper unconditionally
+    rewrote ``accept`` to ``application/vnd.elasticsearch+json; compatible-with=N``,
+    which silently turned every ``cat.*`` response into JSON instead of plain text.
+
+    We mirror elasticsearch-py's own ``mimetype_header_to_compat`` (only
+    ``application/(json|x-ndjson|vnd.mapbox-vector-tile)`` parts get the
+    ``compatible-with`` suffix), so this test fails fast if anyone reverts to the
+    blanket overwrite.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    for action in (lambda: client.cat.help(), lambda: client.cat.indices()):
+        with contextlib.suppress(RuntimeError):
+            action()
+
+    accepts = [c["headers"].get("accept") for c in wire_capture]
+    # ``cat.help`` ships ``text/plain`` only; it must come through verbatim.
+    assert "text/plain" in accepts, accepts
+    # ``cat.indices`` ships ``text/plain,application/json``; the JSON half gets
+    # the ``compatible-with=8`` suffix, the text/plain half stays put.
+    assert any(
+        a and a.startswith("text/plain,application/vnd.elasticsearch+json; compatible-with=8")
+        for a in accepts
+    ), accepts
+
+
+def test_apply_compat_with_handles_pascal_case_headers(monkeypatch):
+    """Defensive: if ``elastic_transport`` ever forwards PascalCase header keys,
+    the rewrite must still apply (a lowercase-only ``dict.get`` would silently
+    no-op and let ``compatible-with=<client_major>`` ship to the server).
+    """
+    seen: dict = {}
+
+    def spy(self, method, target, *, body=None, headers=None, **kwargs):
+        seen["headers"] = dict(headers or {})
+        raise RuntimeError("captured")
+
+    monkeypatch.setattr(Transport, "perform_request", spy)
+
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    # Drive the wrapper with PascalCase keys directly — bypassing the
+    # ``_BaseClient.perform_request`` normalization.
+    with contextlib.suppress(RuntimeError):
+        client.transport.perform_request(
+            "GET",
+            "/",
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
+        )
+
+    assert seen["headers"]["Accept"] == "application/vnd.elasticsearch+json; compatible-with=8"
+    assert seen["headers"]["Content-Type"] == "application/vnd.elasticsearch+json; compatible-with=8"
+
+
+def test_apply_compat_with_strips_whitespace_in_config(wire_capture):
+    """Operators occasionally write ``es_compat_with = " 8"``; the helper must
+    strip whitespace before interpolating into the wire header, otherwise the
+    server returns 400 and the helper fails open in a confusing way.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): " 8 "}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    with contextlib.suppress(RuntimeError):
+        client.search(index="airflow-logs", query={"match_all": {}})
+
+    assert wire_capture[-1]["headers"]["accept"] == "application/vnd.elasticsearch+json; compatible-with=8"
+
+
+@pytest.mark.parametrize("bad_value", ["v8", "8.0", "abc", "8;9"])
+def test_apply_compat_with_rejects_non_numeric_major(bad_value):
+    """A non-numeric ``es_compat_with`` would otherwise produce malformed wire
+    headers (``compatible-with=v8``) and a per-request 400 storm. Fail fast at
+    construction time with a config exception so the misconfiguration is
+    obvious in the worker startup log.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): bad_value}):
+        with pytest.raises(AirflowConfigException, match="es_compat_with"):
+            apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+
+def test_apply_compat_with_is_idempotent():
+    """Calling ``apply_compat_with`` twice on the same client only wraps once."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+        first_wrapper = client.transport.__dict__["perform_request"]
+        apply_compat_with(client)
+        second_wrapper = client.transport.__dict__["perform_request"]
+    assert first_wrapper is second_wrapper