@@ -1161,6 +1161,12 @@ repos:
         language: python
         files: ^.*/example_dags/.*\.py$
         pass_filenames: true
+      - id: check-example-dag-shell-commands
+        name: Check example Dags do not build shell commands from runtime values
+        entry: ./scripts/ci/prek/check_example_dag_shell_commands.py
+        language: python
+        files: ^.*/example_dags/.*\.py$
+        pass_filenames: true
       - id: check-apache-license-rat
         name: Check if licenses are OK for Apache
         entry: ./scripts/ci/prek/check_license.py
 
@@ -59,20 +59,24 @@ def example_dag_decorator(url: str = "https://httpbingo.org/get"):
     """
     get_ip = GetRequestOperator(task_id="get_ip", url=url)
 
-    @task(multiple_outputs=True)
-    def prepare_command(raw_json: dict[str, Any]) -> dict[str, str]:
+    @task
+    def get_external_ip(raw_json: dict[str, Any]) -> str:
         external_ip = raw_json["origin"]
         try:
-            ipaddress.ip_address(external_ip)
-            return {
-                "command": f"echo 'Seems like today your server executing Airflow is connected from IP {external_ip}'",
-            }
+            return str(ipaddress.ip_address(external_ip))
         except ValueError:
             raise ValueError(f"Invalid IP address: '{external_ip}'.")
 
-    command_info = prepare_command(get_ip.output)
+    external_ip = get_external_ip(get_ip.output)
 
-    BashOperator(task_id="echo_ip_info", bash_command=command_info["command"])
+    # Pass the value to the command through the environment rather than formatting it into the
+    # command text, so the shell never interprets it as part of the command.
+    BashOperator(
+        task_id="echo_ip_info",
+        bash_command='echo "Seems like today your server executing Airflow is connected from IP $EXTERNAL_IP"',
+        env={"EXTERNAL_IP": external_ip},
+        append_env=True,
+    )
 
 
 example_dag = example_dag_decorator()
 
@@ -139,6 +139,7 @@
 from airflow.utils.helpers import prune_dict
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.log.task_log_address import prepare_task_log_contexts, render_task_log_filename
+from airflow.utils.retries import run_with_db_retries
 from airflow.utils.session import create_session, provide_session
 
 if TYPE_CHECKING:
@@ -818,8 +819,10 @@ def on_trigger_event(self, trigger_id: int, event: TriggerEvent) -> None:
         Trigger.submit_event(trigger_id=trigger_id, event=event)
 
     def clean_unused(self) -> None:
-        """Remove triggers that are no longer needed."""
-        Trigger.clean_unused()
+        """Remove triggers that are no longer needed, retrying database errors in a new transaction."""
+        for attempt in run_with_db_retries():
+            with attempt, create_session() as session:
+                Trigger.clean_unused(session=session)
 
     def handle_failed_triggers(self):
         """
 
@@ -37,8 +37,7 @@
 from airflow.models.taskinstance import TaskInstance
 from airflow.serialization.enums import stringify_encoding_keys
 from airflow.triggers.base import BaseTaskEndEvent
-from airflow.utils.retries import run_with_db_retries
-from airflow.utils.session import NEW_SESSION, create_session, provide_session
+from airflow.utils.session import NEW_SESSION, provide_session
 from airflow.utils.sqlalchemy import UtcDateTime, with_row_locks
 from airflow.utils.state import TaskInstanceState
 
@@ -249,30 +248,30 @@ def fetch_trigger_ids_with_non_task_associations(cls, *, session: Session = NEW_
         return set(session.scalars(query))
 
     @classmethod
-    def clean_unused(cls) -> None:
+    @provide_session
+    def clean_unused(cls, *, session: Session = NEW_SESSION) -> None:
         """
         Delete up to one batch of triggers that have no dependent tasks, assets, or callbacks.
 
         Triggers have a one-to-many relationship to task instances, so we need to clean those up first.
         Afterward we can drop unreferenced triggers; later triggerer loops drain any remaining batches.
+
+        Database errors are not retried; to retry, run the call again in a new transaction.
         """
         batch_size = conf.getint("triggerer", "unreferenced_triggers_cleanup_batch_size", fallback=500)
         if batch_size <= 0:
             raise ValueError("[triggerer] unreferenced_triggers_cleanup_batch_size must be at least 1")
 
-        for attempt in run_with_db_retries():
-            with attempt:
-                with create_session(scoped=False) as session:
-                    # Update all task instances with trigger IDs that are not DEFERRED to remove them
-                    session.execute(
-                        update(TaskInstance)
-                        .where(
-                            TaskInstance.state != TaskInstanceState.DEFERRED,
-                            TaskInstance.trigger_id.is_not(None),
-                        )
-                        .values(trigger_id=None)
-                    )
-                    cls._delete_unused_batch(batch_size, session=session)
+        # Update all task instances with trigger IDs that are not DEFERRED to remove them
+        session.execute(
+            update(TaskInstance)
+            .where(
+                TaskInstance.state != TaskInstanceState.DEFERRED,
+                TaskInstance.trigger_id.is_not(None),
+            )
+            .values(trigger_id=None)
+        )
+        cls._delete_unused_batch(batch_size, session=session)
 
     @classmethod
     def _delete_unused_batch(cls, batch_size: int, *, session: Session) -> None:
 
@@ -47,7 +47,9 @@
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 from pydantic import TypeAdapter
-from sqlalchemy import delete, update
+from sqlalchemy import create_engine, delete, event, func, select, text, update
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.pool import NullPool
 from structlog.typing import FilteringBoundLogger
 
 from airflow._shared.timezones import timezone
@@ -685,6 +687,71 @@ def test_heartbeat_watchdog_disabled_when_threshold_is_zero(supervisor_builder,
     perform_heartbeat_mock.assert_called_once()
 
 
+def _is_trigger_delete(statement: str) -> bool:
+    return "delete from trigger" in " ".join(statement.lower().replace('"', "").replace("`", "").split())
+
+
+def test_clean_unused_retries_failed_batches_in_new_transactions(jobless_supervisor, session):
+    session.add(Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}))
+    session.commit()
+
+    delete_attempt_count = 0
+    begin_count = 0
+
+    def fail_first_two_deletes(connection, cursor, statement, parameters, context, executemany):
+        nonlocal delete_attempt_count
+        if _is_trigger_delete(statement):
+            delete_attempt_count += 1
+            if delete_attempt_count <= 2:
+                raise OperationalError(statement, parameters, RuntimeError("injected cleanup failure"))
+
+    def count_begin(cleanup_session, transaction, connection):
+        nonlocal begin_count
+        begin_count += 1
+
+    event.listen(session.bind, "before_cursor_execute", fail_first_two_deletes)
+    event.listen(session.__class__, "after_begin", count_begin)
+    try:
+        jobless_supervisor.clean_unused()
+    finally:
+        event.remove(session.bind, "before_cursor_execute", fail_first_two_deletes)
+        event.remove(session.__class__, "after_begin", count_begin)
+
+    assert (delete_attempt_count, begin_count) == (3, 3)
+    session.expire_all()
+    assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+
+
+@pytest.mark.backend("postgres")
+def test_clean_unused_recovers_after_the_database_connection_drops(jobless_supervisor, session):
+    session.add(Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}))
+    session.commit()
+
+    killer_engine = create_engine(session.bind.url, poolclass=NullPool)
+    terminated = []
+
+    def terminate_connection_before_delete(connection, cursor, statement, parameters, context, executemany):
+        if not terminated and _is_trigger_delete(statement):
+            backend_pid = connection.connection.dbapi_connection.info.backend_pid
+            with killer_engine.connect() as killer:
+                terminated.append(
+                    killer.execute(
+                        text("SELECT pg_terminate_backend(:pid, 5000)"), {"pid": backend_pid}
+                    ).scalar()
+                )
+
+    event.listen(session.bind, "before_cursor_execute", terminate_connection_before_delete)
+    try:
+        jobless_supervisor.clean_unused()
+    finally:
+        event.remove(session.bind, "before_cursor_execute", terminate_connection_before_delete)
+        killer_engine.dispose()
+
+    assert terminated == [True]
+    session.expire_all()
+    assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+
+
 def test_metric_tags_default_uses_job_hostname(supervisor_builder):
     """metric_tags() defaults to {"hostname": job.hostname} when a job is present."""
     supervisor = supervisor_builder()
 
@@ -343,24 +343,26 @@ def count_commit(committed_session):
     assert commit_count == 0
 
 
-@conf_vars({("triggerer", "unreferenced_triggers_cleanup_batch_size"): "2"})
-def test_clean_unused_retries_failed_batch_in_fresh_transactions(session):
-    session.add_all(
-        [
-            Trigger(classpath=f"airflow.triggers.testing.SuccessTrigger{index}", kwargs={})
-            for index in range(5)
-        ]
-    )
+def test_clean_unused_runs_in_the_caller_transaction(session):
+    session.add(Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}))
     session.commit()
 
-    Trigger.clean_unused()
-    session.expire_all()
-    assert session.scalar(select(func.count()).select_from(Trigger)) == 3
+    Trigger.clean_unused(session=session)
+    assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+    session.rollback()
+
+    assert session.scalar(select(func.count()).select_from(Trigger)) == 1
+
+
+def test_clean_unused_raises_database_errors_to_the_caller_without_retrying(session):
+    session.add(Trigger(classpath="airflow.triggers.testing.SuccessTrigger", kwargs={}))
+    session.commit()
+    # The caller's own pending write must survive the cleanup failing.
+    job = Job()
+    session.add(job)
+    session.flush()
 
     delete_attempt_count = 0
-    cleanup_sessions = []
-    commit_count = 0
-    rollback_count = 0
 
     def fail_delete(connection, cursor, statement, parameters, context, executemany):
         nonlocal delete_attempt_count
@@ -369,42 +371,15 @@ def fail_delete(connection, cursor, statement, parameters, context, executemany)
             delete_attempt_count += 1
             raise OperationalError(statement, parameters, RuntimeError("injected cleanup failure"))
 
-    def capture_session(cleanup_session, transaction, connection):
-        cleanup_sessions.append(cleanup_session)
-
-    def count_commit(committed_session):
-        nonlocal commit_count
-        commit_count += 1
-
-    def count_rollback(rolled_back_session):
-        nonlocal rollback_count
-        rollback_count += 1
-
     event.listen(session.bind, "before_cursor_execute", fail_delete)
-    event.listen(session.__class__, "after_begin", capture_session)
-    event.listen(session.__class__, "after_commit", count_commit)
-    event.listen(session.__class__, "after_rollback", count_rollback)
     try:
         with pytest.raises(OperationalError, match="injected cleanup failure"):
-            Trigger.clean_unused()
+            Trigger.clean_unused(session=session)
     finally:
         event.remove(session.bind, "before_cursor_execute", fail_delete)
-        event.remove(session.__class__, "after_begin", capture_session)
-        event.remove(session.__class__, "after_commit", count_commit)
-        event.remove(session.__class__, "after_rollback", count_rollback)
 
-    assert len({id(cleanup_session) for cleanup_session in cleanup_sessions}) == 3
-    assert (delete_attempt_count, commit_count, rollback_count) == (3, 0, 3)
-    session.expire_all()
-    assert session.scalar(select(func.count()).select_from(Trigger)) == 3
-
-    Trigger.clean_unused()
-    session.expire_all()
-    assert session.scalar(select(func.count()).select_from(Trigger)) == 1
-
-    Trigger.clean_unused()
-    session.expire_all()
-    assert session.scalar(select(func.count()).select_from(Trigger)) == 0
+    assert delete_attempt_count == 1
+    assert job in session
 
 
 @patch.object(TriggererCallback, "handle_event")
 
@@ -853,13 +853,9 @@ def test_get_conn_with_password_none_auth(self, mock_connect):
         ],
     )
     def test_get_conn_with_wrong_connection_parameters(self, host, port, schema, message):
-        connection = Connection(
-            conn_id="test",
-            conn_type="hive",
-            host=host,
-            port=port,
-            schema=schema,
-        )
+        # Newer core Connection rejects invalid ports on assignment; a spec'd mock lets the hook's own
+        # check, which still guards older Airflow versions, see them.
+        connection = mock.Mock(spec=Connection, conn_id="test", host=host, port=port, schema=schema)
         hook = HiveCliHook()
         if message:
             with pytest.raises(Exception, match=message):
 
@@ -135,7 +135,10 @@ def connection():
     @task_group(prefix_group_id=False)
     def standard_tasks_group():
         classic_bash = BashOperator(
-            task_id="classic_bash", bash_command="echo Parameter is {{ params.mapping_count }}"
+            task_id="classic_bash",
+            bash_command='echo "Parameter is $MAPPING_COUNT"',
+            env={"MAPPING_COUNT": "{{ params.mapping_count }}"},
+            append_env=True,
         )
 
         empty = EmptyOperator(task_id="not_executed")
 
@@ -97,7 +97,13 @@ def execute(self, context):
     task_2 = PythonOperator(
         task_id="task_2",
         python_callable=do_nothing,
-        inlets=[Asset(uri="file://host1/dir2/file2.txt"), Asset(uri="unknown_scheme://host1/dir2/file3.txt")],
+        inlets=[
+            Asset(
+                uri="file://host1/dir2/file2.txt",
+                extra={"etag": '"0x8D0000000000001"', "owner": "O'Brien", "has_header": True, "note": None},
+            ),
+            Asset(uri="unknown_scheme://host1/dir2/file3.txt"),
+        ],
         max_retry_delay=42,
         doc="text doc",
         doc_md="should be skipped",
@@ -107,7 +113,12 @@ def execute(self, context):
     )
     task_3 = EmptyOperator(
         task_id="task_3",
-        outlets=[Asset(uri="file://host/dir/file.txt")],
+        outlets=[
+            Asset(
+                uri="file://host/dir/file.txt",
+                extra={"checks": [{"passed": True}], "ratio": 0.5},
+            )
+        ],
         doc_md="MD doc",
         doc_json="should be skipped",
         doc_yaml="should be skipped",