Align the artifact sweep with the other cleanup checks

The sweep's interval check now reads like the bundle-version cleanup next to it, and its deleted-row count no longer hides a missing rowcount behind a default that can never apply. The parsing_cleanup_interval description said the artifact deletion in a clause that did not parse cleanly.

Author: jason810496

airflow-core/src/airflow/config_templates/config.yml

@@ -2727,8 +2727,8 @@ scheduler:
       description: |
         How often (in seconds) to check for stale DAGs (DAGs which are no longer present in
         the expected files) which should be deactivated, as well as assets that are no longer
-        referenced and should be marked as orphaned, and Lang-SDK task handler artifacts that no
-        stub task uses any more, which are deleted.
+        referenced and should be marked as orphaned. The same check deletes the Lang-SDK task
+        handler artifacts that no stub task uses any more.
       version_added: 2.5.0
       type: integer
       example: ~

airflow-core/src/airflow/dag_processing/manager.py

@@ -97,6 +97,7 @@
     from collections.abc import Callable, Collection, Iterable, Iterator, Sequence
     from socket import socket
 
+    from sqlalchemy.engine import CursorResult
     from sqlalchemy.orm import Session
     from sqlalchemy.sql import Select
 
@@ -462,7 +463,7 @@ def _scan_stale_dags(self):
 
     def _sweep_task_handler_artifacts(self) -> None:
         now = time.monotonic()
-        if now - self._last_task_handler_artifact_sweep_time <= self.parsing_cleanup_interval:
+        if now - self._last_task_handler_artifact_sweep_time < self.parsing_cleanup_interval:
             return
         try:
             self.delete_unreferenced_task_handler_artifacts()
@@ -508,7 +509,7 @@ def delete_unreferenced_task_handler_artifacts(self) -> int:
                         .where(LangSDKTaskHandlerArtifact.id.in_(ids), *sweepable)
                         .execution_options(synchronize_session=False)
                     )
-                    deleted += getattr(result, "rowcount", 0)
+                    deleted += cast("CursorResult", result).rowcount
             if len(ids) < _TASK_HANDLER_ARTIFACT_SWEEP_BATCH_SIZE:
                 break
         if deleted: