From d9a3478b6ccdb8e2c7ee836ab1b1ce92b03250dd Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 21 Aug 2026 15:52:16 -0700
Subject: [PATCH 01/10] Allow Variable.get to reuse the caller's database
 session
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`MetastoreBackend.get_variable` is decorated with `@provide_session`. Called
without a session it goes through `create_session()`, which for a scoped session
returns *the caller's own session* and commits it on exit. So any code that reads
a Variable while holding a transaction gets that transaction committed underneath
it — detaching its objects, or raising `UNEXPECTED COMMIT` under the scheduler's
`prohibit_commit` guard, where the error is then swallowed per-backend and
surfaces as a missing Variable.

`Variable.get` and `Variable.get_variable_from_secrets` now take an optional
keyword-only `session`, forwarded only to `MetastoreBackend`. `Variable.update`
forwards its own.

Affected today, all reached from `_create_dagruns_for_dags` inside the guard:

* Deadline Alerts using `VariableInterval`
* Custom timetables reading a Variable in `next_dagrun_info` — `next_dagrun` is
left NULL, so the Dag is never eligible and never runs, with nothing logged
* Dag sync via `update_dags`; `Variable.update`; `Variable.setdefault`

Scope: this fixes the core read path only. `airflow.sdk.Variable.get` and
`Connection.get_connection_from_secrets` share the defect and are unchanged, so
callers going through those are still affected. #68917 can drop its duplicated
backend walk once this lands.

closes: #71801
---
 airflow-core/src/airflow/models/variable.py   | 32 ++++++++--
 .../tests/unit/models/test_variable.py        | 61 +++++++++++++++++++
 2 files changed, 89 insertions(+), 4 deletions(-)

diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index bcbc07a9703e3..cbee47aac8a4e 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -156,6 +156,8 @@ def get(
         default_var: Any = __NO_DEFAULT_SENTINEL,
         deserialize_json: bool = False,
         team_name: str | None = None,
+        *,
+        session: Session | None = None,
     ) -> Any:
         """
         Get a value for an Airflow Variable Key.
@@ -164,6 +166,8 @@ def get(
         :param default_var: Default value of the Variable if the Variable doesn't exist
         :param deserialize_json: Deserialize the value to a Python dict
         :param team_name: Team name associated to the task trying to access the variable (if any)
+        :param session: Existing session to reuse for the metadata database lookup. Callers holding an
+            open transaction (the scheduler under ``prohibit_commit``, for example) must pass it.
         """
         # TODO: This is not the best way of having compat, but it's "better than erroring" for now. This still
         # means SQLA etc is loaded, but we can't avoid that unless/until we add import shims as a big
@@ -172,6 +176,11 @@ def get(
         # If this is set it means we are in some kind of execution context (Task, Dag Parse or Triggerer perhaps)
         # and should use the Task SDK API server path
         if hasattr(sys.modules.get("airflow.sdk.execution_time.task_runner"), "SUPERVISOR_COMMS"):
+            if session is not None:
+                raise ValueError(
+                    "Variable.get() cannot use a metadata database session from an execution context; "
+                    "reads there go through the Execution API. Use airflow.sdk.Variable.get() instead."
+                )
             warnings.warn(
                 "Using Variable.get from `airflow.models` is deprecated."
                 "Please use `get` on Variable from sdk(`airflow.sdk.Variable`) instead",
@@ -192,7 +201,7 @@ def get(
                 "Multi-team mode is not configured in the Airflow environment but the task trying to access the variable belongs to a team"
             )
 
-        var_val = Variable.get_variable_from_secrets(key=key, team_name=team_name)
+        var_val = Variable.get_variable_from_secrets(key=key, team_name=team_name, session=session)
         if var_val is None:
             if default_var is not cls.__NO_DEFAULT_SENTINEL:
                 return default_var
@@ -344,7 +353,7 @@ def update(
 
         Variable.check_for_write_conflict(key=key, team_name=team_name)
 
-        if Variable.get_variable_from_secrets(key=key, team_name=team_name) is None:
+        if Variable.get_variable_from_secrets(key=key, team_name=team_name, session=session) is None:
             raise KeyError(f"Variable {key} does not exist")
 
         ctx: contextlib.AbstractContextManager
@@ -469,12 +478,18 @@ def check_for_write_conflict(key: str, team_name: str | None = None) -> None:
         return None
 
     @staticmethod
-    def get_variable_from_secrets(key: str, team_name: str | None = None) -> str | None:
+    def get_variable_from_secrets(
+        key: str, team_name: str | None = None, *, session: Session | None = None
+    ) -> str | None:
         """
         Get Airflow Variable by iterating over all Secret Backends.
 
         :param key: Variable Key
         :param team_name: Team name associated to the task trying to access the variable (if any)
+        :param session: Existing session to reuse for the metadata database lookup. Callers that
+            already hold a transaction must pass it, otherwise ``MetastoreBackend`` opens the same
+            scoped session and commits it, which detaches the caller's objects and is rejected
+            outright under ``prohibit_commit``.
         :return: Variable Value
         """
         from airflow.sdk import SecretCache
@@ -491,7 +506,16 @@ def get_variable_from_secrets(key: str, team_name: str | None = None) -> str | N
         for secrets_backend in ensure_secrets_loaded():
             try:
                 var_val = call_secrets_backend_method(
-                    secrets_backend.get_variable, team_name=team_name, key=key
+                    secrets_backend.get_variable,
+                    team_name=team_name,
+                    key=key,
+                    # Only the metastore backend touches the metadata database, and it is the only
+                    # one whose signature accepts a session.
+                    **(
+                        {"session": session}
+                        if session is not None and isinstance(secrets_backend, MetastoreBackend)
+                        else {}
+                    ),
                 )
                 if var_val is not None:
                     break
diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py
index cc2f9f94b9367..5d861a518241f 100644
--- a/airflow-core/tests/unit/models/test_variable.py
+++ b/airflow-core/tests/unit/models/test_variable.py
@@ -30,6 +30,7 @@
 from airflow.sdk import SecretCache
 from airflow.secrets import BaseSecretsBackend
 from airflow.secrets.metastore import MetastoreBackend
+from airflow.utils.sqlalchemy import prohibit_commit
 
 from tests_common.test_utils import db
 from tests_common.test_utils.config import conf_vars
@@ -253,6 +254,66 @@ def test_update_forwards_team_name_to_write_conflict_check(self, mock_check, tes
         Variable.update(key="key", value="new-value", team_name=testing_team.name, session=session)
 
         assert mock_check.call_args.kwargs["team_name"] == testing_team.name
+    @mock.patch.object(MetastoreBackend, "get_variable", autospec=True)
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_get_forwards_session_to_metastore_backend(self, mock_ensure_secrets, mock_get_variable, session):
+        mock_get_variable.return_value = "from_db"
+        mock_ensure_secrets.return_value = [MetastoreBackend()]
+
+        assert Variable.get("some_key", session=session) == "from_db"
+        assert mock_get_variable.call_args.kwargs["session"] is session
+
+    @mock.patch.object(MetastoreBackend, "get_variable", autospec=True)
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_get_without_session_omits_session_kwarg(self, mock_ensure_secrets, mock_get_variable):
+        mock_get_variable.return_value = "from_db"
+        mock_ensure_secrets.return_value = [MetastoreBackend()]
+
+        assert Variable.get("some_key") == "from_db"
+        assert "session" not in mock_get_variable.call_args.kwargs
+
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_get_does_not_forward_session_to_other_backends(self, mock_ensure_secrets, session):
+        """Only the metastore backend reads the metadata database, so only it accepts a session."""
+        mock_backend = mock.Mock()
+        mock_backend.get_variable.return_value = "from_backend"
+        mock_backend.__class__.__name__ = "MockSecretsBackend"
+        mock_ensure_secrets.return_value = [mock_backend]
+
+        assert Variable.get("some_key", session=session) == "from_backend"
+        assert "session" not in mock_backend.get_variable.call_args.kwargs
+
+    def test_get_with_session_does_not_commit_under_prohibit_commit(self, session):
+        """
+        A caller holding an open transaction can read a Variable without its session being committed.
+
+        Without the session being forwarded, ``MetastoreBackend.get_variable``'s ``provide_session``
+        takes the same scoped session and commits it, which the guard rejects.
+        """
+        Variable.set(key="interval_key", value="60", session=session)
+        session.commit()
+        SecretCache.invalidate_variable("interval_key")
+
+        with prohibit_commit(session):
+            assert Variable.get("interval_key", session=session) == "60"
+
+    def test_update_with_session_does_not_commit_under_prohibit_commit(self, session):
+        """``update`` verifies existence through the secrets chain, which must reuse the session too."""
+        Variable.set(key="interval_key", value="60", session=session)
+        session.commit()
+        SecretCache.invalidate_variable("interval_key")
+
+        with prohibit_commit(session):
+            Variable.update(key="interval_key", value="120", session=session)
+
+    def test_get_rejects_session_in_execution_context(self):
+        """Reads from an execution context go via the Execution API, where a session is meaningless."""
+        task_runner = mock.Mock(SUPERVISOR_COMMS=mock.Mock())
+        with (
+            mock.patch.dict("sys.modules", {"airflow.sdk.execution_time.task_runner": task_runner}),
+            pytest.raises(ValueError, match="cannot use a metadata database session"),
+        ):
+            Variable.get("some_key", session=mock.Mock())
 
     def test_variable_set_get_round_trip_json(self):
         value = {"a": 17, "b": 47}

From ceee252c84ed775766b00e38c8ba53c808bf73a7 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 21 Aug 2026 16:07:09 -0700
Subject: [PATCH 02/10] Add newsfragment

---
 airflow-core/newsfragments/71968.bugfix.rst | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 airflow-core/newsfragments/71968.bugfix.rst

diff --git a/airflow-core/newsfragments/71968.bugfix.rst b/airflow-core/newsfragments/71968.bugfix.rst
new file mode 100644
index 0000000000000..2123a6b91b13f
--- /dev/null
+++ b/airflow-core/newsfragments/71968.bugfix.rst
@@ -0,0 +1 @@
+``Variable.get`` and ``Variable.get_variable_from_secrets`` now accept an optional keyword-only ``session``, which is forwarded to the metastore secrets backend so the lookup reuses the caller's transaction instead of committing it. Callers that read a Variable while holding an open session -- most notably code running inside the scheduler's ``prohibit_commit`` guard -- should pass it; previously the backend opened the same scoped session and committed it, detaching the caller's objects or failing the lookup outright.

From 30a22ddbacc6827ccdb1fd422380752985a87139 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 21 Aug 2026 16:41:10 -0700
Subject: [PATCH 03/10] Allow Variable.setdefault to reuse the caller's
 database session

---
 airflow-core/newsfragments/71968.bugfix.rst     |  8 +++++++-
 airflow-core/src/airflow/models/variable.py     | 15 +++++++++++----
 airflow-core/tests/unit/models/test_variable.py | 17 +++++++++++++++++
 3 files changed, 35 insertions(+), 5 deletions(-)

diff --git a/airflow-core/newsfragments/71968.bugfix.rst b/airflow-core/newsfragments/71968.bugfix.rst
index 2123a6b91b13f..fab9a98070e2d 100644
--- a/airflow-core/newsfragments/71968.bugfix.rst
+++ b/airflow-core/newsfragments/71968.bugfix.rst
@@ -1 +1,7 @@
-``Variable.get`` and ``Variable.get_variable_from_secrets`` now accept an optional keyword-only ``session``, which is forwarded to the metastore secrets backend so the lookup reuses the caller's transaction instead of committing it. Callers that read a Variable while holding an open session -- most notably code running inside the scheduler's ``prohibit_commit`` guard -- should pass it; previously the backend opened the same scoped session and committed it, detaching the caller's objects or failing the lookup outright.
+``Variable.get``, ``Variable.get_variable_from_secrets`` and ``Variable.setdefault`` now accept an
+optional keyword-only ``session``, which is forwarded to the metastore secrets backend so the lookup
+reuses the caller's transaction instead of committing it.  Callers that read a Variable while holding
+an open session (most notably code running inside the scheduler's ``prohibit_commit`` guard) should
+pass it; previously the backend opened the same scoped session and committed it, detaching the caller's
+objects or failing the lookup outright.  ``Variable.update`` now forwards the session it was given to
+its own existence check, which was affected by the same problem.
diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index cbee47aac8a4e..e4cc22c63793f 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -126,7 +126,7 @@ def val(cls):
         return synonym("_val", descriptor=property(cls.get_val, cls.set_val))
 
     @classmethod
-    def setdefault(cls, key, default, description=None, deserialize_json=False):
+    def setdefault(cls, key, default, description=None, deserialize_json=False, *, session=None):
         """
         Return the current value for a key or store the default value and return it.
 
@@ -138,13 +138,20 @@ def setdefault(cls, key, default, description=None, deserialize_json=False):
         :param description: Default value to set Description of the Variable
         :param deserialize_json: Store this as a JSON encoded value in the DB
             and un-encode it when retrieving a value
-        :param session: Session
+        :param session: Existing session to reuse for the metadata database read and write.
+            Callers holding an open transaction must pass it.
         :return: Mixed
         """
-        obj = Variable.get(key, default_var=None, deserialize_json=deserialize_json)
+        obj = Variable.get(key, default_var=None, deserialize_json=deserialize_json, session=session)
         if obj is None:
             if default is not None:
-                Variable.set(key=key, value=default, description=description, serialize_json=deserialize_json)
+                Variable.set(
+                    key=key,
+                    value=default,
+                    description=description,
+                    serialize_json=deserialize_json,
+                    session=session,
+                )
                 return default
             raise ValueError("Default Value must be set")
         return obj
diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py
index 5d861a518241f..8012be2edcd27 100644
--- a/airflow-core/tests/unit/models/test_variable.py
+++ b/airflow-core/tests/unit/models/test_variable.py
@@ -306,6 +306,23 @@ def test_update_with_session_does_not_commit_under_prohibit_commit(self, session
         with prohibit_commit(session):
             Variable.update(key="interval_key", value="120", session=session)
 
+    def test_setdefault_with_session_does_not_commit_under_prohibit_commit(self, session):
+        """``setdefault`` reads through the secrets chain before deciding whether to write."""
+        Variable.set(key="interval_key", value="60", session=session)
+        session.commit()
+        SecretCache.invalidate_variable("interval_key")
+
+        with prohibit_commit(session):
+            assert Variable.setdefault("interval_key", "120", session=session) == "60"
+
+    def test_setdefault_writes_default_with_session_under_prohibit_commit(self, session):
+        """The write half must reuse the session too, so the miss path stays inside the transaction."""
+        with prohibit_commit(session):
+            assert Variable.setdefault("absent_key", "30", session=session) == "30"
+        session.commit()
+
+        assert Variable.get("absent_key", session=session) == "30"
+
     def test_get_rejects_session_in_execution_context(self):
         """Reads from an execution context go via the Execution API, where a session is meaningless."""
         task_runner = mock.Mock(SUPERVISOR_COMMS=mock.Mock())

From df3480698f779f96738e0b9c11e5eaa9f7acfed2 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Mon, 24 Aug 2026 09:48:47 -0700
Subject: [PATCH 04/10] PR fixes; shorten newsfragment and Vincent request

---
 airflow-core/newsfragments/71968.bugfix.rst |  8 +-------
 airflow-core/src/airflow/models/variable.py | 13 ++++++-------
 2 files changed, 7 insertions(+), 14 deletions(-)

diff --git a/airflow-core/newsfragments/71968.bugfix.rst b/airflow-core/newsfragments/71968.bugfix.rst
index fab9a98070e2d..0713c4772903b 100644
--- a/airflow-core/newsfragments/71968.bugfix.rst
+++ b/airflow-core/newsfragments/71968.bugfix.rst
@@ -1,7 +1 @@
-``Variable.get``, ``Variable.get_variable_from_secrets`` and ``Variable.setdefault`` now accept an
-optional keyword-only ``session``, which is forwarded to the metastore secrets backend so the lookup
-reuses the caller's transaction instead of committing it.  Callers that read a Variable while holding
-an open session (most notably code running inside the scheduler's ``prohibit_commit`` guard) should
-pass it; previously the backend opened the same scoped session and committed it, detaching the caller's
-objects or failing the lookup outright.  ``Variable.update`` now forwards the session it was given to
-its own existence check, which was affected by the same problem.
+``Variable.get``, ``Variable.get_variable_from_secrets`` and ``Variable.setdefault`` now accept an optional keyword-only ``session`` that is forwarded to the metastore secrets backend, so a lookup made while holding an open session (most notably inside the scheduler's ``prohibit_commit`` guard) reuses the caller's transaction instead of opening a scoped session and committing it; ``Variable.update`` now forwards its session to its own existence check, which had the same problem.
diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index e4cc22c63793f..0a0379ab63d83 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -511,18 +511,17 @@ def get_variable_from_secrets(
         var_val = None
         # iterate over backends if not in cache (or expired)
         for secrets_backend in ensure_secrets_loaded():
+            # Only the metastore backend touches the metadata database, and it is the only
+            # one whose signature accepts a session.
+            session_kwargs: dict[str, Session] = {}
+            if session is not None and isinstance(secrets_backend, MetastoreBackend):
+                session_kwargs["session"] = session
             try:
                 var_val = call_secrets_backend_method(
                     secrets_backend.get_variable,
                     team_name=team_name,
                     key=key,
-                    # Only the metastore backend touches the metadata database, and it is the only
-                    # one whose signature accepts a session.
-                    **(
-                        {"session": session}
-                        if session is not None and isinstance(secrets_backend, MetastoreBackend)
-                        else {}
-                    ),
+                    **session_kwargs,
                 )
                 if var_val is not None:
                     break

From 8262d772ef82f10d87554f49463fd3a97fb744ca Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Mon, 24 Aug 2026 11:21:31 -0700
Subject: [PATCH 05/10] fix type hints

---
 airflow-core/src/airflow/models/variable.py | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index 0a0379ab63d83..4f2cbdbeaa698 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -126,7 +126,15 @@ def val(cls):
         return synonym("_val", descriptor=property(cls.get_val, cls.set_val))
 
     @classmethod
-    def setdefault(cls, key, default, description=None, deserialize_json=False, *, session=None):
+    def setdefault(
+        cls,
+        key: str,
+        default: Any,
+        description: str | None = None,
+        deserialize_json: bool = False,
+        *,
+        session: Session | None = None,
+    ) -> Any:
         """
         Return the current value for a key or store the default value and return it.
 

From 4a299c92e993c93cb865fcee98f5353f9d59db5c Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Mon, 24 Aug 2026 16:17:30 -0700
Subject: [PATCH 06/10] doc tweaks for Niko

---
 airflow-core/newsfragments/71968.bugfix.rst | 2 +-
 airflow-core/src/airflow/models/variable.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow-core/newsfragments/71968.bugfix.rst b/airflow-core/newsfragments/71968.bugfix.rst
index 0713c4772903b..a312ab2909112 100644
--- a/airflow-core/newsfragments/71968.bugfix.rst
+++ b/airflow-core/newsfragments/71968.bugfix.rst
@@ -1 +1 @@
-``Variable.get``, ``Variable.get_variable_from_secrets`` and ``Variable.setdefault`` now accept an optional keyword-only ``session`` that is forwarded to the metastore secrets backend, so a lookup made while holding an open session (most notably inside the scheduler's ``prohibit_commit`` guard) reuses the caller's transaction instead of opening a scoped session and committing it; ``Variable.update`` now forwards its session to its own existence check, which had the same problem.
+``Variable.get``, ``Variable.get_variable_from_secrets`` and ``Variable.setdefault`` now accept an optional keyword-only ``session`` that is forwarded to the metastore secrets backend.  A lookup made while holding an open session (most notably inside the scheduler's ``prohibit_commit`` guard) now reuses the caller's transaction instead of opening a scoped session and committing it.  ``Variable.update`` now forwards its session to its own existence check, which had the same problem.
diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index 4f2cbdbeaa698..3708128cbced1 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -519,7 +519,7 @@ def get_variable_from_secrets(
         var_val = None
         # iterate over backends if not in cache (or expired)
         for secrets_backend in ensure_secrets_loaded():
-            # Only the metastore backend touches the metadata database, and it is the only
+            # Only the metastore Variable backend touches the metadata database, and it is the only
             # one whose signature accepts a session.
             session_kwargs: dict[str, Session] = {}
             if session is not None and isinstance(secrets_backend, MetastoreBackend):

From a7df3049004cecc0cd50b8894c1767c8034ad85f Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Mon, 24 Aug 2026 17:14:26 -0700
Subject: [PATCH 07/10] Only forward session to backend overrides that accept
 it

---
 airflow-core/src/airflow/models/variable.py   | 14 +++++---
 .../tests/unit/models/test_variable.py        | 35 +++++++++++++++++++
 .../airflow_shared/secrets_backend/base.py    | 25 +++++++------
 .../tests/secrets_backend/test_base.py        | 25 ++++++++++++-
 4 files changed, 81 insertions(+), 18 deletions(-)

diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index 3708128cbced1..ed3afd098f0cd 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -28,7 +28,7 @@
 from sqlalchemy.dialects.mysql import MEDIUMTEXT
 from sqlalchemy.orm import Mapped, declared_attr, mapped_column, reconstructor, synonym
 
-from airflow._shared.secrets_backend.base import call_secrets_backend_method
+from airflow._shared.secrets_backend.base import accepts_kwarg, call_secrets_backend_method
 from airflow._shared.secrets_masker import mask_secret
 from airflow.configuration import conf, ensure_secrets_loaded
 from airflow.models.base import ID_LEN, Base
@@ -519,10 +519,16 @@ def get_variable_from_secrets(
         var_val = None
         # iterate over backends if not in cache (or expired)
         for secrets_backend in ensure_secrets_loaded():
-            # Only the metastore Variable backend touches the metadata database, and it is the only
-            # one whose signature accepts a session.
+            # Only the metastore Variable backend touches the metadata database, so it is the only
+            # one offered a session, and only when its own override accepts one.  A subclass that
+            # overrides get_variable without the parameter raises TypeError, which the handler
+            # below swallows into a false "not found" that then gets cached.
             session_kwargs: dict[str, Session] = {}
-            if session is not None and isinstance(secrets_backend, MetastoreBackend):
+            if (
+                session is not None
+                and isinstance(secrets_backend, MetastoreBackend)
+                and accepts_kwarg(secrets_backend.get_variable, "session")
+            ):
                 session_kwargs["session"] = session
             try:
                 var_val = call_secrets_backend_method(
diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py
index 8012be2edcd27..cb7fe9793b242 100644
--- a/airflow-core/tests/unit/models/test_variable.py
+++ b/airflow-core/tests/unit/models/test_variable.py
@@ -65,6 +65,25 @@ def __init__(self):
     def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         self.received_team_name = team_name
         return "secret_val"
+class _SessionUnawareMetastoreBackend(MetastoreBackend):
+    """A custom backend whose ``get_variable`` override predates the ``session`` keyword."""
+
+    def get_variable(self, key: str, team_name: str | None = None) -> str | None:
+        return "from_subclass"
+
+
+class _SessionAwareMetastoreBackend(MetastoreBackend):
+    """A custom backend whose ``get_variable`` override does accept ``session``."""
+
+    def __init__(self):
+        super().__init__()
+        self.received_session: Session | None = None
+
+    def get_variable(
+        self, key: str, team_name: str | None = None, *, session: Session | None = None
+    ) -> str | None:
+        self.received_session = session
+        return "from_subclass"
 
 
 class TestVariable:
@@ -283,6 +302,22 @@ def test_get_does_not_forward_session_to_other_backends(self, mock_ensure_secret
         assert Variable.get("some_key", session=session) == "from_backend"
         assert "session" not in mock_backend.get_variable.call_args.kwargs
 
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_get_omits_session_for_session_unaware_metastore_subclass(self, mock_ensure_secrets, session):
+        """Forwarding a session to an override that predates it would read as a missing Variable."""
+        mock_ensure_secrets.return_value = [_SessionUnawareMetastoreBackend()]
+
+        assert Variable.get("some_key", session=session) == "from_subclass"
+
+    @mock.patch("airflow.models.variable.ensure_secrets_loaded")
+    def test_get_forwards_session_to_session_aware_metastore_subclass(self, mock_ensure_secrets, session):
+        """A subclass that does accept a session still receives it, so it reuses the transaction."""
+        backend = _SessionAwareMetastoreBackend()
+        mock_ensure_secrets.return_value = [backend]
+
+        assert Variable.get("some_key", session=session) == "from_subclass"
+        assert backend.received_session is session
+
     def test_get_with_session_does_not_commit_under_prohibit_commit(self, session):
         """
         A caller holding an open transaction can read a Variable without its session being committed.
diff --git a/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py b/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py
index 77c08be45f1bd..d98829c118e87 100644
--- a/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py
+++ b/shared/secrets_backend/src/airflow_shared/secrets_backend/base.py
@@ -21,24 +21,23 @@
 from collections.abc import Callable
 
 
-def _accepts_team_name(method: Callable) -> bool:
+def accepts_kwarg(method: Callable, name: str) -> bool:
     """
-    Return whether a secrets-backend method accepts the ``team_name`` keyword.
-
-    Backends written before Airflow 3.2 override ``get_conn_value`` / ``get_variable`` /
-    ``get_connection`` with the legacy ``(self, conn_id)`` / ``(self, key)`` signature.
-    AIP-67 (multi-team) added a ``team_name`` keyword; forwarding it to those raises
-    ``TypeError``. A method accepts it if it declares a ``team_name`` parameter or a
-    ``**kwargs`` catch-all.
+    Return whether a secrets-backend method accepts the keyword *name*.
+
+    Backends override ``get_conn_value`` / ``get_variable`` / ``get_connection`` with
+    whatever signature was current when they were written, so a keyword added later cannot
+    be forwarded blindly: it raises ``TypeError`` inside the backend, which callers swallow
+    per-backend and report as a missing secret.  ``team_name`` (added by AIP-67 in 3.2) is
+    one example.  A method is considered to accept *name* if it explicitly declares the
+    parameter or has a ``**kwargs`` catch-all.
     """
     try:
         parameters = inspect.signature(method).parameters
     except (TypeError, ValueError):
-        # Un-introspectable callable (e.g. C-implemented): assume the 3.2+ signature.
+        # Un-introspectable callable (e.g. C-implemented): assume the current signature.
         return True
-    return "team_name" in parameters or any(
-        p.kind is inspect.Parameter.VAR_KEYWORD for p in parameters.values()
-    )
+    return name in parameters or any(p.kind is inspect.Parameter.VAR_KEYWORD for p in parameters.values())
 
 
 def call_secrets_backend_method(method: Callable, *, team_name: str | None, **kwargs):
@@ -52,7 +51,7 @@ def call_secrets_backend_method(method: Callable, *, team_name: str | None, **kw
     rather than retried without ``team_name``, which could mask the error and resolve a
     team-scoped lookup against the global scope.
     """
-    if _accepts_team_name(method):
+    if accepts_kwarg(method, "team_name"):
         return method(team_name=team_name, **kwargs)
     return method(**kwargs)
 
diff --git a/shared/secrets_backend/tests/secrets_backend/test_base.py b/shared/secrets_backend/tests/secrets_backend/test_base.py
index e374a883cc44e..60587892d8e21 100644
--- a/shared/secrets_backend/tests/secrets_backend/test_base.py
+++ b/shared/secrets_backend/tests/secrets_backend/test_base.py
@@ -19,7 +19,7 @@
 
 import pytest
 
-from airflow_shared.secrets_backend.base import BaseSecretsBackend
+from airflow_shared.secrets_backend.base import BaseSecretsBackend, accepts_kwarg
 
 
 class MockConnection:
@@ -243,3 +243,26 @@ def test_team_unaware_backend_missing_conn_returns_none(self, team_name):
         backend = _TeamUnawareConnValueBackend(conn_values={})
 
         assert backend.get_connection(conn_id="missing", team_name=team_name) is None
+
+
+class TestAcceptsKwarg:
+    @pytest.mark.parametrize(
+        ("method", "name", "expected"),
+        [
+            (lambda key: None, "session", False),
+            (lambda key, session=None: None, "session", True),
+            (lambda key, **kwargs: None, "session", True),
+            (lambda conn_id: None, "team_name", False),
+            (lambda conn_id, team_name=None: None, "team_name", True),
+            (lambda conn_id, **kwargs: None, "team_name", True),
+        ],
+    )
+    def test_declared_parameter_or_kwargs_catch_all(self, method, name, expected):
+        assert accepts_kwarg(method, name) is expected
+
+    def test_each_keyword_judged_independently(self):
+        def get_variable(key, team_name=None):
+            return None
+
+        assert accepts_kwarg(get_variable, "team_name") is True
+        assert accepts_kwarg(get_variable, "session") is False

From 4d3471b165001b28ca3a8a62540f7112b114814c Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Tue, 25 Aug 2026 16:11:54 -0700
Subject: [PATCH 08/10] Fix mypy override error in Variable session test
 fixture

---
 airflow-core/tests/unit/models/test_variable.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py
index cb7fe9793b242..c33687d92093a 100644
--- a/airflow-core/tests/unit/models/test_variable.py
+++ b/airflow-core/tests/unit/models/test_variable.py
@@ -68,7 +68,9 @@ def get_variable(self, key: str, team_name: str | None = None) -> str | None:
 class _SessionUnawareMetastoreBackend(MetastoreBackend):
     """A custom backend whose ``get_variable`` override predates the ``session`` keyword."""
 
-    def get_variable(self, key: str, team_name: str | None = None) -> str | None:
+    # The signature mismatch with the base class is the point of this fixture, so mypy's
+    # override check has to be waived here rather than fixed.
+    def get_variable(self, key: str, team_name: str | None = None) -> str | None:  # type: ignore[override]
         return "from_subclass"
 
 

From 63e655b84e073449711855f4301ce770048939f2 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Wed, 9 Sep 2026 12:39:55 -0700
Subject: [PATCH 09/10] pass session into resolve()

---
 airflow-core/src/airflow/models/variable.py          |  6 ++----
 .../src/airflow/serialization/definitions/dag.py     |  2 +-
 .../airflow/serialization/definitions/deadline.py    |  8 ++++++--
 .../unit/serialization/definitions/test_deadline.py  | 12 ++++++++++++
 4 files changed, 21 insertions(+), 7 deletions(-)

diff --git a/airflow-core/src/airflow/models/variable.py b/airflow-core/src/airflow/models/variable.py
index ed3afd098f0cd..1c53237aa1687 100644
--- a/airflow-core/src/airflow/models/variable.py
+++ b/airflow-core/src/airflow/models/variable.py
@@ -146,8 +146,7 @@ def setdefault(
         :param description: Default value to set Description of the Variable
         :param deserialize_json: Store this as a JSON encoded value in the DB
             and un-encode it when retrieving a value
-        :param session: Existing session to reuse for the metadata database read and write.
-            Callers holding an open transaction must pass it.
+        :param session: Existing SQLAlchemy Session. Callers holding an open transaction must pass it.
         :return: Mixed
         """
         obj = Variable.get(key, default_var=None, deserialize_json=deserialize_json, session=session)
@@ -181,8 +180,7 @@ def get(
         :param default_var: Default value of the Variable if the Variable doesn't exist
         :param deserialize_json: Deserialize the value to a Python dict
         :param team_name: Team name associated to the task trying to access the variable (if any)
-        :param session: Existing session to reuse for the metadata database lookup. Callers holding an
-            open transaction (the scheduler under ``prohibit_commit``, for example) must pass it.
+        :param session: Existing SQLAlchemy Session. Callers holding an open transaction must pass it.
         """
         # TODO: This is not the best way of having compat, but it's "better than erroring" for now. This still
         # means SQLA etc is loaded, but we can't avoid that unless/until we add import shims as a big
diff --git a/airflow-core/src/airflow/serialization/definitions/dag.py b/airflow-core/src/airflow/serialization/definitions/dag.py
index 5d5c476362266..e6979fc44f327 100644
--- a/airflow-core/src/airflow/serialization/definitions/dag.py
+++ b/airflow-core/src/airflow/serialization/definitions/dag.py
@@ -763,7 +763,7 @@ def _process_dagrun_deadline_alerts(
             interval = deserialized_deadline_alert.interval
 
             if isinstance(interval, SerializedVariableInterval):
-                interval = interval.resolve()
+                interval = interval.resolve(session=session)
 
             if isinstance(deserialized_deadline_alert.reference, SerializedReferenceModels.TYPES.DAGRUN):
                 deadline_time = deserialized_deadline_alert.reference.evaluate_with(
diff --git a/airflow-core/src/airflow/serialization/definitions/deadline.py b/airflow-core/src/airflow/serialization/definitions/deadline.py
index 5f462c63d1a21..e4362338cc0e0 100644
--- a/airflow-core/src/airflow/serialization/definitions/deadline.py
+++ b/airflow-core/src/airflow/serialization/definitions/deadline.py
@@ -388,10 +388,14 @@ class SerializedVariableInterval:
 
     key: str
 
-    def resolve(self) -> timedelta:
+    def resolve(self, *, session: Session | None = None) -> timedelta:
+        """
+        Get the Airflow Variable and return it as a ``timedelta``.
 
+        :param session: Existing SQLAlchemy Session. Callers holding an open transaction must pass it.
+        """
         try:
-            value = Variable.get(self.key)
+            value = Variable.get(self.key, session=session)
         except KeyError as e:
             raise ValueError(f"VariableInterval '{self.key}' not found") from e
 
diff --git a/airflow-core/tests/unit/serialization/definitions/test_deadline.py b/airflow-core/tests/unit/serialization/definitions/test_deadline.py
index 3e12fe0830c43..f0fad39a37e1a 100644
--- a/airflow-core/tests/unit/serialization/definitions/test_deadline.py
+++ b/airflow-core/tests/unit/serialization/definitions/test_deadline.py
@@ -43,6 +43,18 @@ def test_resolve_valid(self, mocker, value, expected):
 
         assert interval.resolve() == expected
 
+    def test_resolve_forwards_session(self, mocker):
+        """The scheduler resolves intervals while holding an open transaction, so the caller's
+        session has to reach ``Variable.get`` rather than open a new session."""
+        expected_seconds = 42
+        mock_get = mocker.patch.object(Variable, "get", return_value=str(expected_seconds))
+        session = mocker.MagicMock()
+
+        interval = SerializedVariableInterval(key="test_interval")
+
+        assert interval.resolve(session=session) == timedelta(seconds=expected_seconds)
+        mock_get.assert_called_once_with("test_interval", session=session)
+
     @pytest.mark.parametrize(
         ("value", "raise_missing", "match"),
         [

From 419f0a7ad3a235f2ad47324d8b7cca51d275dbe6 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Wed, 9 Sep 2026 15:06:36 -0700
Subject: [PATCH 10/10] static checks

---
 airflow-core/tests/unit/models/test_variable.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/airflow-core/tests/unit/models/test_variable.py b/airflow-core/tests/unit/models/test_variable.py
index c33687d92093a..57742aff4fa5e 100644
--- a/airflow-core/tests/unit/models/test_variable.py
+++ b/airflow-core/tests/unit/models/test_variable.py
@@ -65,6 +65,8 @@ def __init__(self):
     def get_variable(self, key: str, team_name: str | None = None) -> str | None:
         self.received_team_name = team_name
         return "secret_val"
+
+
 class _SessionUnawareMetastoreBackend(MetastoreBackend):
     """A custom backend whose ``get_variable`` override predates the ``session`` keyword."""
 
@@ -275,6 +277,7 @@ def test_update_forwards_team_name_to_write_conflict_check(self, mock_check, tes
         Variable.update(key="key", value="new-value", team_name=testing_team.name, session=session)
 
         assert mock_check.call_args.kwargs["team_name"] == testing_team.name
+
     @mock.patch.object(MetastoreBackend, "get_variable", autospec=True)
     @mock.patch("airflow.models.variable.ensure_secrets_loaded")
     def test_get_forwards_session_to_metastore_backend(self, mock_ensure_secrets, mock_get_variable, session):