@@ -220,7 +220,7 @@ def duration(self) -> float | None:
     @duration.expression  # type: ignore[no-redef]
     def duration(cls) -> Any:
         @provide_session
-        def _get_dialect(*, session: Session = NEW_SESSION) -> str:
+        def _get_dialect(*, session: Session = NEW_SESSION) -> str | None:
             return get_dialect_name(session=session)
 
         dialect_name = _get_dialect()
 
@@ -83,6 +83,7 @@
 
     from dateutil.relativedelta import relativedelta
     from sqlalchemy.orm.state import InstanceState
+    from sqlalchemy.sql.elements import ColumnElement
 
     from airflow.sdk import Context
     from airflow.serialization.definitions.assets import (
@@ -650,15 +651,15 @@ def dag_display_name(self) -> str:
         return self._dag_display_property_value or self.dag_id
 
     @dag_display_name.expression  # type: ignore[no-redef]
-    def dag_display_name(self) -> str:
+    def dag_display_name(cls: type[DagModel]) -> ColumnElement[Any]:
         """
         Expression part of the ``dag_display`` name hybrid property.
 
         :meta private:
         """
         return case(
-            (self._dag_display_property_value.is_not(None), self._dag_display_property_value),
-            else_=self.dag_id,
+            (cls._dag_display_property_value.is_not(None), cls._dag_display_property_value),
+            else_=cls.dag_id,
         )
 
     @classmethod
 
@@ -122,7 +122,7 @@
     from pydantic import NonNegativeInt
     from sqlalchemy.engine import ScalarResult
     from sqlalchemy.orm import Session
-    from sqlalchemy.sql.elements import Case, ColumnElement
+    from sqlalchemy.sql.elements import ColumnElement
     from sqlalchemy.sql.selectable import Select
 
     from airflow.api_fastapi.execution_api.datamodels.taskinstance import (
@@ -585,7 +585,7 @@ def duration(self) -> float | None:
 
     @duration.expression  # type: ignore[no-redef]
     @provide_session
-    def duration(cls, *, session: Session = NEW_SESSION) -> Case:
+    def duration(cls: type[DagRun], *, session: Session = NEW_SESSION) -> ColumnElement[Any]:
         dialect_name = get_dialect_name(session)
         if dialect_name == "mysql":
             return func.timestampdiff(text("SECOND"), cls.start_date, cls.end_date)
 
@@ -197,11 +197,13 @@ connection-types:
       - Nebius
       - Ollama
       - OpenAI
+      - OpenAI Decisions
       - OpenRouter
       - OVHcloud
       - Pydantic AI Gateway
       - SambaNova
       - Snowflake Cortex
+      - System One
       - Together AI
       - TypeSafe
       - Vercel AI Gateway
 
@@ -170,11 +170,13 @@ def get_provider_info():
                     "Nebius",
                     "Ollama",
                     "OpenAI",
+                    "OpenAI Decisions",
                     "OpenRouter",
                     "OVHcloud",
                     "Pydantic AI Gateway",
                     "SambaNova",
                     "Snowflake Cortex",
+                    "System One",
                     "Together AI",
                     "TypeSafe",
                     "Vercel AI Gateway",
 
@@ -69,6 +69,9 @@ def resolve_sqlglot_dialect(dialect_name: str | None) -> str | None:
     exp.Except,
 )
 
+# Snowflake ``LIST``/``LS @stage``; sqlglot >= 30.22 only (older releases mis-parse it as exp.Alias).
+_LIST_STAGE: type[exp.Expr] | None = getattr(exp, "ListStage", None)
+
 # Read-only metadata statements that introspect the schema without touching data:
 # - Describe: DESCRIBE / DESC <table> (and EXPLAIN on some dialects)
 # - Show: SHOW TABLES / SHOW COLUMNS / SHOW DATABASES, etc.
@@ -323,10 +326,11 @@ def collect_table_references(
             continue
         # A bare aliased expression is never a valid top-level statement -- sqlglot only
         # produces one by mis-parsing input it does not model. Snowflake ``LIST @stage`` /
-        # ``LS @stage`` (list a stage's files) parse this way, as ``Column AS Parameter``.
-        # Read-only mode already rejects it via the statement-type allow-list; this also
-        # refuses it on the ``allow_writes`` path, where only this scan runs.
-        if isinstance(stmt, exp.Alias):
+        # ``LS @stage`` (list a stage's files) parse this way, as ``Column AS Parameter``,
+        # before sqlglot 30.22, which models them as ``exp.ListStage`` instead.
+        # Read-only mode already rejects both via the statement-type allow-list; this also
+        # refuses them on the ``allow_writes`` path, where only this scan runs.
+        if isinstance(stmt, exp.Alias) or (_LIST_STAGE is not None and isinstance(stmt, _LIST_STAGE)):
             unverifiable.append("a statement sqlglot could not parse as a query (e.g. Snowflake LIST @stage)")
             continue
         # A function whose string argument reaches a file, another table, or a program
 
@@ -358,7 +358,10 @@ def test_execute_raises_when_cost_cap_exceeded(self, mock_hook_cls, cap_kwargs,
             op.execute(context=usage_budget_context)
 
         pushes = {c.kwargs["key"]: c.kwargs["value"] for c in ti.xcom_push.call_args_list}
-        assert pushes["usage"]["requests"] == 1
+        # Not `requests`: pydantic-ai >= 2.54 counts only responses the agent acted on, so the
+        # capped response's request is no longer counted, though its tokens and cost still are.
+        assert pushes["usage"]["input_tokens"] == 100
+        assert pushes["usage"]["output_tokens"] == 50
         assert pushes["usage"]["cost"] == str(PRICED_COST)
 
     @patch("airflow.providers.common.ai.operators.llm.PydanticAIHook", autospec=True)
 
@@ -81,10 +81,12 @@ def _extract_toolset_block() -> dict:
     "nebius": "Nebius",
     "ollama": "Ollama",
     "openai": "OpenAI",
+    "openai_decisions": "OpenAI Decisions",
     "openrouter": "OpenRouter",
     "ovhcloud": "OVHcloud",
     "sambanova": "SambaNova",
     "snowflake": "Snowflake Cortex",
+    "system_one": "System One",
     "together": "Together AI",
     "typesafe": "TypeSafe",
     "vercel": "Vercel AI Gateway",
@@ -163,14 +165,15 @@ def _extract_toolset_block() -> dict:
 # `provider.yaml`'s vendor list is derived from a current pydantic-ai, but the provider
 # supports `pydantic-ai-slim>=2.33.0` (`pyproject.toml`) and CI's "Low dep tests" job installs
 # exactly that floor. Vendors upstream added after the floor are legitimately absent there --
-# bisected empirically: `vllm` landed in 2.38.0, `github_copilot` in 2.42.0, and `typesafe` in
-# 2.45.0 (`openai_codex` also landed after the floor, in 2.41.0, but is excluded from this
-# reasoning -- see `OPENAI_CODEX_AVAILABLE_FROM` above) -- so only the "installed but not
-# declared" direction holds on every supported version. The opposite direction ("declared but
-# not installed") is a real drift signal solely from the release that carries every vendor
-# currently declared, i.e. the latest of the three, `typesafe`'s 2.45.0: raise this when a
-# newly declared vendor lands in a later release.
-ALL_DECLARED_VENDORS_AVAILABLE_FROM = Version("2.45.0")
+# bisected empirically: `vllm` landed in 2.38.0, `github_copilot` in 2.42.0, `typesafe` in
+# 2.45.0, `system_one` in 2.53.0 and `openai_decisions` in 2.55.0 (`openai_codex` also landed
+# after the floor, in 2.41.0, but is excluded from this reasoning -- see
+# `OPENAI_CODEX_AVAILABLE_FROM` above) -- so only the "installed but not declared" direction
+# holds on every supported version. The opposite direction ("declared but not installed") is a
+# real drift signal solely from the release that carries every vendor currently declared, i.e.
+# the latest of them, `openai_decisions`'s 2.55.0: raise this when a newly declared vendor lands
+# in a later release.
+ALL_DECLARED_VENDORS_AVAILABLE_FROM = Version("2.55.0")
 
 requires_all_declared_vendors = pytest.mark.skipif(
     Version(Version(version("pydantic-ai-slim")).base_version) < ALL_DECLARED_VENDORS_AVAILABLE_FROM,
 
@@ -141,6 +141,35 @@ def test_setting_names_are_the_defaults_plus_automatic_anthropic_caching(self):
         assert {*ALL_DEFAULTS, "anthropic_cache"} == PROMPT_CACHE_SETTING_NAMES
 
 
+_ANTHROPIC_OK_STREAM = "".join(
+    f"event: {event['type']}\ndata: {json.dumps(event)}\n\n"
+    for event in [
+        {
+            "type": "message_start",
+            "message": {
+                "id": "msg_1",
+                "type": "message",
+                "role": "assistant",
+                "model": "claude-sonnet-4-5",
+                "content": [],
+                "stop_reason": None,
+                "stop_sequence": None,
+                "usage": {"input_tokens": 10, "output_tokens": 0},
+            },
+        },
+        {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "ok"}},
+        {"type": "content_block_stop", "index": 0},
+        {
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+            "usage": {"output_tokens": 1},
+        },
+        {"type": "message_stop"},
+    ]
+).encode()
+
+
 class TestPromptCachingOnTheAnthropicWire:
     """What an Anthropic model actually sends, captured at the HTTP transport."""
 
@@ -149,7 +178,14 @@ def _send(*, model_settings: Any = None, prompt: Any = "hi") -> dict[str, Any]:
         bodies: list[dict[str, Any]] = []
 
         def handler(request: httpx2.Request) -> httpx2.Response:
-            bodies.append(json.loads(request.content))
+            body = json.loads(request.content)
+            bodies.append(body)
+            # pydantic-ai >= 2.52 streams whenever its default max_tokens exceeds the SDK's
+            # non-streaming limit and the client timeout is not the SDK default, as here.
+            if body.get("stream"):
+                return httpx2.Response(
+                    200, headers={"content-type": "text/event-stream"}, content=_ANTHROPIC_OK_STREAM
+                )
             return httpx2.Response(
                 200,
                 json={
 
@@ -418,7 +418,9 @@ def test_legit_cte_reference_is_excluded(self):
     def test_flags_inline_comment_as_unverifiable(self, sql, dialect):
         """Comments hide parser-vs-engine differentials (MySQL executable comments), so reject them."""
         scan = collect_table_references(parse_sql(sql, dialect=dialect))
-        assert scan.unverifiable_sources
+        # Newer sqlglot tokenizes MySQL's `--+1` as arithmetic, as MySQL does (a comment needs
+        # whitespace after `--`); with no differential left, `secret` is then a scanned table.
+        assert scan.unverifiable_sources or ("", "", "secret") in scan.tables
 
     @pytest.mark.parametrize(
         "sql",