Validate a serialized Dag and fill its unset settings from the Airflow config

airflow-core/adr/lang-sdk/0004-dag-parsing.md

@@ -178,7 +178,7 @@ Airflow Supervisor                    Bridge              Language Runtime
 
 ### DagFileParsingResult Format
 
-The language runtime must produce a `DagFileParsingResult` that matches Python Airflow's DagSerialization format exactly. The Airflow scheduler deserializes this into its internal model — any divergence causes parsing failures.
+The language runtime must produce a `DagFileParsingResult` that matches Python Airflow's DagSerialization format exactly, except that it may omit the config-backed Dag fields marked in the table below, which Airflow fills in from its config before it loads the Dag. The Airflow scheduler deserializes this into its internal model — any divergence causes parsing failures.
 
 **Envelope:**
 
@@ -216,9 +216,11 @@ The language runtime must produce a `DagFileParsingResult` that matches Python A
 | `start_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
 | `end_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
 | `tags` | list | if non-empty | Unwrapped from `__type`/`__var` |
-| `catchup` | bool | if `true` | |
-| `max_active_tasks` | int | if non-default | |
-| `max_active_runs` | int | if non-default | |
+| `catchup` | bool | if set | Airflow fills an unset field from its config |
+| `max_active_tasks` | int | if set | Airflow fills an unset field from its config |
+| `max_active_runs` | int | if set | Airflow fills an unset field from its config |
+| `max_consecutive_failed_dag_runs` | int | if set | Airflow fills an unset field from its config |
+| `disable_bundle_versioning` | bool | if set | Airflow fills an unset field from its config |
 
 **Timetable encoding:**
 
@@ -404,7 +406,7 @@ uv run validation/serialization/compare.py \
     validation/serialization/serialized_java.json
 ```
 
-Both share test cases defined in `test_dags.yaml`, ensuring the Java SDK produces byte-identical output to Python's `DagSerialization.serialize_dag()` for the same inputs.
+Both share test cases defined in `test_dags.yaml`, ensuring the Java SDK produces byte-identical output to Python's `DagSerialization.serialize_dag()` for the same inputs, apart from the config-backed Dag fields it may omit, which Airflow fills in before it loads the Dag.
 
 ## Consequences
 

airflow-core/src/airflow/serialization/schema.json

@@ -257,7 +257,18 @@
     },
     "tasks": {
       "type": "array",
-      "additionalProperties": { "$ref": "#/definitions/operator" }
+      "items": {
+        "type": "object",
+        "required": [ "__type", "__var" ],
+        "properties": {
+          "__type": { "const": "operator" },
+          "__var": {
+            "type": "object",
+            "required": [ "task_id" ],
+            "properties": { "task_id": { "type": "string" } }
+          }
+        }
+      }
     },
     "params": {
       "type": "array",

airflow-core/src/airflow/serialization/serialized_objects.py

@@ -22,6 +22,7 @@
 
 import collections.abc
 import contextlib
+import copy
 import datetime
 import enum
 import itertools
@@ -39,11 +40,14 @@
 import lazy_object_proxy
 import pydantic
 from dateutil import relativedelta
+from jsonschema import ValidationError
 from pendulum.tz.timezone import FixedTimezone, Timezone
 
+from airflow._shared.dagnode.cycle import detect_cycle
 from airflow._shared.module_loading import qualname
 from airflow._shared.timezones.timezone import from_timestamp, parse_timezone, utcnow
 from airflow.callbacks.callback_requests import DagCallbackRequest, TaskCallbackRequest
+from airflow.configuration import conf
 from airflow.exceptions import AirflowException, DeserializationError, SerializationError
 from airflow.models.connection import Connection
 from airflow.models.expandinput import SchedulerMappedArgument, create_expand_input
@@ -296,6 +300,16 @@ def _decode_start_trigger_args(var: dict[str, Any]) -> StartTriggerArgs:
     )
 
 
+def _build_json_path(error: ValidationError) -> str:
+    """
+    Return where *error* is in the document, such as ``$.dag.tasks[2].__type``.
+
+    Unlike ``ValidationError.json_path``, whose quoting of a key such as ``__type`` depends on the
+    jsonschema version, this is the same under every version.
+    """
+    return "$" + "".join(f"[{part}]" if isinstance(part, int) else f".{part}" for part in error.absolute_path)
+
+
 class _XComRef(NamedTuple):
     """
     Store info needed to create XComArg.
@@ -2131,6 +2145,88 @@ def from_dict(cls, serialized_obj: dict) -> SerializedDAG:
         # Pass client_defaults directly to deserialize_dag
         return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
 
+    @classmethod
+    def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+        """
+        Fill in the Dag settings a serialized Dag leaves unset from the Airflow config, as a Python Dag does.
+
+        A Lang-SDK runtime cannot read the Airflow config, so it leaves ``max_active_tasks``,
+        ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup`` and
+        ``disable_bundle_versioning`` out unless the Dag sets them. A value the Dag sets is kept.
+        *serialized_obj* is changed in place.
+        """
+        dag = serialized_obj.get("dag")
+        if not isinstance(dag, dict):
+            # validate_serialized_dag rejects it.
+            return
+        for key, get, section, option in (
+            ("max_active_tasks", conf.getint, "core", "max_active_tasks_per_dag"),
+            ("max_active_runs", conf.getint, "core", "max_active_runs_per_dag"),
+            (
+                "max_consecutive_failed_dag_runs",
+                conf.getint,
+                "core",
+                "max_consecutive_failed_dag_runs_per_dag",
+            ),
+            ("catchup", conf.getboolean, "scheduler", "catchup_by_default"),
+            ("disable_bundle_versioning", conf.getboolean, "dag_processor", "disable_bundle_versioning"),
+        ):
+            if key not in dag:
+                dag[key] = get(section, option)
+
+    @classmethod
+    def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) -> SerializedDAG:
+        """
+        Check that a serialized Dag, such as one a Lang-SDK runtime produced, can be stored and loaded.
+
+        It must match the JSON schema, have unique task ids and deserialize. Like a Dag built with the SDK,
+        it must not set a ``max_active_runs`` its timetable forbids, nor ``catchup`` without a
+        ``start_date``, and its task graph must have no cycle. *serialized_obj* is not changed.
+
+        :return: The deserialized Dag.
+        :raises DeserializationError: if it does not.
+        """
+        dag = serialized_obj.get("dag")
+        dag_id = dag.get("dag_id") if isinstance(dag, dict) else None
+        try:
+            cls.validate_schema(serialized_obj)
+        except ValidationError as e:
+            raise DeserializationError(
+                dag_id, f"Dag {dag_id!r} does not match the schema at {_build_json_path(e)}: {e.message}"
+            ) from e
+        task_ids = collections.Counter(
+            task[Encoding.VAR]["task_id"] for task in serialized_obj["dag"]["tasks"]
+        )
+        if duplicates := sorted(task_id for task_id, count in task_ids.items() if count > 1):
+            raise DeserializationError(
+                dag_id, f"Dag {dag_id!r} has more than one task with id {', '.join(map(repr, duplicates))}"
+            )
+        try:
+            deserialized = cls.from_dict(copy.deepcopy(serialized_obj))
+        except Exception as e:
+            cause = e.__cause__ if isinstance(e, DeserializationError) and e.__cause__ else e
+            raise DeserializationError(
+                dag_id, f"Dag {dag_id!r} cannot be deserialized: {type(cause).__name__}: {cause}"
+            ) from e
+        if (
+            limit := deserialized.timetable.active_runs_limit
+        ) is not None and deserialized.max_active_runs > limit:
+            raise DeserializationError(
+                dag_id,
+                f"Dag {dag_id!r} sets max_active_runs {deserialized.max_active_runs}, "
+                f"but {type(deserialized.timetable).__name__} allows at most {limit}",
+            )
+        if (
+            deserialized.catchup
+            and deserialized.timetable.can_be_scheduled
+            and not (deserialized.start_date or "start_date" in deserialized.default_args)
+        ):
+            raise DeserializationError(dag_id, f"Dag {dag_id!r} sets catchup but no start_date")
+        downstream = {task_id: task.downstream_task_ids for task_id, task in deserialized.task_dict.items()}
+        if (task_id := detect_cycle(downstream, downstream.__getitem__)) is not None:
+            raise DeserializationError(dag_id, f"Dag {dag_id!r} has a cycle through task {task_id!r}")
+        return deserialized
+
 
 class TaskGroupSerialization(BaseSerialization):
     """JSON serializable representation of a task group."""

airflow-core/tests/unit/models/test_dagrun.py

@@ -617,7 +617,8 @@ def on_success_callable(context):
             "test_state_succeeded2": TaskInstanceState.SUCCESS,
         }
         dag.relative_fileloc = relative_fileloc
-        SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker")
+        dag_maker.dag.relative_fileloc = relative_fileloc
+        SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag_maker.dag), bundle_name="dag_maker")
         session.commit()
 
         dag_run = self.create_dag_run(dag=dag, task_states=initial_task_states, session=session)
@@ -664,7 +665,8 @@ def on_failure_callable(context):
             "test_state_failed2": TaskInstanceState.FAILED,
         }
         dag.relative_fileloc = relative_fileloc
-        SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker")
+        dag_maker.dag.relative_fileloc = relative_fileloc
+        SerializedDagModel.write_dag(LazyDeserializedDAG.from_dag(dag_maker.dag), bundle_name="dag_maker")
         session.commit()
 
         dag_run = self.create_dag_run(dag=dag, task_states=initial_task_states, session=session)

airflow-core/tests/unit/models/test_serialized_dag.py

@@ -885,11 +885,11 @@ def test_dynamic_dag_update_preserves_null_check(self, dag_maker, session):
         Test that dynamic DAG update gracefully handles case where SerializedDagModel doesn't exist.
         This preserves the null-check fix from PR #56422 and tests the direct UPDATE path.
         """
-        with dag_maker(dag_id="test_missing_serdag", serialized=True, session=session) as dag:
+        with dag_maker(dag_id="test_missing_serdag", serialized=True, session=session):
             EmptyOperator(task_id="task1")
 
         # Write the DAG first
-        lazy_dag = LazyDeserializedDAG.from_dag(dag)
+        lazy_dag = LazyDeserializedDAG.from_dag(dag_maker.dag)
         SDM.write_dag(
             dag=lazy_dag,
             bundle_name="test_bundle",