diff --git a/airflow-core/adr/lang-sdk/0004-dag-parsing.md b/airflow-core/adr/lang-sdk/0004-dag-parsing.md
index d4e85acde8950..064f44767d28f 100644
--- a/airflow-core/adr/lang-sdk/0004-dag-parsing.md
+++ b/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
 
diff --git a/airflow-core/src/airflow/serialization/schema.json b/airflow-core/src/airflow/serialization/schema.json
index 086bccb34bb5b..6ecea324714a1 100644
--- a/airflow-core/src/airflow/serialization/schema.json
+++ b/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",
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py
index b58bb2505b0f4..8a6b259994458 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/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."""
diff --git a/airflow-core/tests/unit/models/test_dagrun.py b/airflow-core/tests/unit/models/test_dagrun.py
index 47343eeefae52..e85713943d6c7 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/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)
diff --git a/airflow-core/tests/unit/models/test_serialized_dag.py b/airflow-core/tests/unit/models/test_serialized_dag.py
index 7609c57ba0128..7b47f70fdf2ba 100644
--- a/airflow-core/tests/unit/models/test_serialized_dag.py
+++ b/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",
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 279a7bb02a154..ce2c2fbb8a151 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -5155,3 +5155,218 @@ def get_weight(self, ti):
         op = BaseOperator(task_id="empty_task", weight_rule=NotRegisteredPriorityWeightStrategy())
         with pytest.raises(ValueError, match="Unknown priority strategy"):
             OperatorSerialization.serialize(op)
+
+
+class TestValidateSerializedDag:
+    @staticmethod
+    def _serialize() -> dict:
+        with DAG(dag_id="checked_dag", schedule=None) as dag:
+            BaseOperator(task_id="extract") >> BaseOperator(task_id="load")
+        return DagSerialization.to_dict(dag)
+
+    def test_accepts_a_dag_that_loads(self):
+        data = self._serialize()
+        data["__version"] = 2
+        before = copy.deepcopy(data)
+
+        DagSerialization.validate_serialized_dag(data)
+
+        assert data == before
+
+    @pytest.mark.parametrize(
+        ("change", "error"),
+        [
+            pytest.param(
+                {"max_active_runs": "many"},
+                "Dag 'checked_dag' does not match the schema at $.dag.max_active_runs: "
+                "'many' is not of type 'number'",
+                id="schema",
+            ),
+            pytest.param(
+                {"timetable": {"__type": "no.such.Timetable", "__var": {}}},
+                "Dag 'checked_dag' cannot be deserialized: TimetableNotRegistered: ",
+                id="deserialize",
+            ),
+        ],
+    )
+    def test_rejects_a_dag_that_does_not_load(self, change, error):
+        data = self._serialize()
+        data["dag"].update(change)
+
+        with pytest.raises(DeserializationError, match=f"^{re.escape(error)}"):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_returns_the_deserialized_dag(self):
+        dag = DagSerialization.validate_serialized_dag(self._serialize())
+
+        assert isinstance(dag, SerializedDAG)
+        assert dag.dag_id == "checked_dag"
+        assert dag.task_dict["extract"].downstream_task_ids == {"load"}
+
+    @pytest.mark.parametrize(
+        ("entry", "json_path"),
+        [
+            pytest.param({}, "$.dag.tasks[2]", id="empty"),
+            pytest.param("x", "$.dag.tasks[2]", id="not-an-object"),
+            pytest.param(
+                {"__type": "dag", "__var": {"task_id": "t"}}, "$.dag.tasks[2].__type", id="wrong-type"
+            ),
+            pytest.param({"__type": "operator", "__var": {}}, "$.dag.tasks[2].__var", id="no-task-id"),
+            pytest.param(
+                {"__type": "operator", "__var": {"task_id": 5}},
+                "$.dag.tasks[2].__var.task_id",
+                id="task-id-not-a-string",
+            ),
+        ],
+    )
+    def test_rejects_a_task_entry_that_is_not_an_operator(self, entry, json_path):
+        data = self._serialize()
+        data["dag"]["tasks"].append(entry)
+
+        with pytest.raises(
+            DeserializationError,
+            match=rf"^Dag 'checked_dag' does not match the schema at {re.escape(json_path)}: ",
+        ):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_rejects_a_repeated_task_id(self):
+        data = self._serialize()
+        data["dag"]["tasks"].append(copy.deepcopy(data["dag"]["tasks"][0]))
+
+        with pytest.raises(
+            DeserializationError, match="^Dag 'checked_dag' has more than one task with id 'extract'$"
+        ):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_reports_the_cause_of_a_deserialization_error(self):
+        data = self._serialize()
+        data["dag"]["tasks"][0]["__var"]["downstream_task_ids"] = ["ghost"]
+
+        with pytest.raises(
+            DeserializationError, match="^Dag 'checked_dag' cannot be deserialized: KeyError: 'ghost'$"
+        ):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_rejects_a_dag_with_a_cycle(self):
+        data = self._serialize()
+        load = next(task for task in data["dag"]["tasks"] if task["__var"]["task_id"] == "load")
+        load["__var"]["downstream_task_ids"] = ["extract"]
+
+        with pytest.raises(DeserializationError, match="^Dag 'checked_dag' has a cycle through task 'load'$"):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_rejects_a_cycle_in_the_legacy_downstream_key(self):
+        data = self._serialize()
+        load = next(task for task in data["dag"]["tasks"] if task["__var"]["task_id"] == "load")
+        load["__var"]["_downstream_task_ids"] = ["extract"]
+
+        with pytest.raises(DeserializationError, match="^Dag 'checked_dag' has a cycle through task 'load'$"):
+            DagSerialization.validate_serialized_dag(data)
+
+    @pytest.mark.parametrize(
+        ("change", "error"),
+        [
+            pytest.param(
+                {},
+                "Dag 'checked_dag' sets max_active_runs 16, but ContinuousTimetable allows at most 1",
+                id="unset",
+            ),
+            pytest.param(
+                {"max_active_runs": 2},
+                "Dag 'checked_dag' sets max_active_runs 2, but ContinuousTimetable allows at most 1",
+                id="above-the-limit",
+            ),
+        ],
+    )
+    def test_rejects_max_active_runs_above_the_timetable_limit(self, change, error):
+        data = DagSerialization.to_dict(DAG(dag_id="checked_dag", schedule="@continuous", max_active_runs=1))
+        del data["dag"]["max_active_runs"]
+        data["dag"].update(change)
+
+        with pytest.raises(DeserializationError, match=f"^{re.escape(error)}$"):
+            DagSerialization.validate_serialized_dag(data)
+
+    def test_rejects_catchup_without_a_start_date(self):
+        data = DagSerialization.to_dict(DAG(dag_id="checked_dag", schedule="@daily"))
+        data["dag"]["catchup"] = True
+
+        with pytest.raises(DeserializationError, match="^Dag 'checked_dag' sets catchup but no start_date$"):
+            DagSerialization.validate_serialized_dag(data)
+
+    @pytest.mark.parametrize(
+        "kwargs",
+        [
+            pytest.param(
+                {
+                    "schedule": "@daily",
+                    "default_args": {"start_date": datetime(2024, 1, 1, tzinfo=dt_timezone.utc)},
+                },
+                id="start-date-in-default-args",
+            ),
+            pytest.param({"schedule": None}, id="unscheduled"),
+        ],
+    )
+    def test_accepts_catchup_the_sdk_dag_accepts(self, kwargs):
+        dag = DAG(dag_id="checked_dag", catchup=True, **kwargs)
+
+        assert DagSerialization.validate_serialized_dag(DagSerialization.to_dict(dag)).catchup is True
+
+
+class TestFillConfigDefaults:
+    CONFIG = {
+        ("core", "max_active_tasks_per_dag"): "7",
+        ("core", "max_active_runs_per_dag"): "3",
+        ("core", "max_consecutive_failed_dag_runs_per_dag"): "5",
+        ("scheduler", "catchup_by_default"): "True",
+        ("dag_processor", "disable_bundle_versioning"): "False",
+    }
+    FIELDS = (
+        "max_active_tasks",
+        "max_active_runs",
+        "max_consecutive_failed_dag_runs",
+        "catchup",
+        "disable_bundle_versioning",
+    )
+
+    @staticmethod
+    def _serialize() -> dict:
+        return DagSerialization.to_dict(DAG(dag_id="native_dag", schedule=None))
+
+    @conf_vars(CONFIG)
+    def test_fills_an_unset_setting_from_the_config(self):
+        data = self._serialize()
+        for field in self.FIELDS:
+            del data["dag"][field]
+
+        DagSerialization.fill_config_defaults(data)
+
+        assert {field: data["dag"][field] for field in self.FIELDS} == {
+            "max_active_tasks": 7,
+            "max_active_runs": 3,
+            "max_consecutive_failed_dag_runs": 5,
+            "catchup": True,
+            "disable_bundle_versioning": False,
+        }
+
+    @conf_vars(CONFIG)
+    def test_keeps_a_setting_the_dag_sets(self):
+        data = self._serialize()
+        data["dag"].update(
+            max_active_tasks=16,
+            max_active_runs=16,
+            max_consecutive_failed_dag_runs=0,
+            catchup=False,
+            disable_bundle_versioning=True,
+        )
+        before = copy.deepcopy(data)
+
+        DagSerialization.fill_config_defaults(data)
+
+        assert data == before
+
+    def test_leaves_a_payload_without_a_dag_to_validation(self):
+        data = {"__version": 3}
+
+        DagSerialization.fill_config_defaults(data)
+
+        assert data == {"__version": 3}
diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
index c3b027b6e3119..7f92223f604e6 100644
--- a/airflow-core/tests/unit/serialization/test_dag_version_diff.py
+++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
@@ -2896,7 +2896,6 @@ def test_diff_ignores_wrapper_types_missing_on_both_sides(include_values, proven
     base["dag"]["tasks"][0].pop("__type")
     base["dag"].pop("task_group")
     target = copy.deepcopy(base)
-    DagSerialization.validate_schema(base)
     assert DagSerialization.from_dict(copy.deepcopy(base)).task_dict == {}
     target_version = "new" if provenance_changed else "old"
 
diff --git a/go-sdk/schema/dag-schema.json b/go-sdk/schema/dag-schema.json
index 086bccb34bb5b..6ecea324714a1 100644
--- a/go-sdk/schema/dag-schema.json
+++ b/go-sdk/schema/dag-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",
diff --git a/java-sdk/sdk/schema/dag-schema.json b/java-sdk/sdk/schema/dag-schema.json
index 086bccb34bb5b..6ecea324714a1 100644
--- a/java-sdk/sdk/schema/dag-schema.json
+++ b/java-sdk/sdk/schema/dag-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",
diff --git a/scripts/ci/lang_sdk_serialization/serialize_python.py b/scripts/ci/lang_sdk_serialization/serialize_python.py
index 3bb1406cb0e3a..0647b53ac379c 100644
--- a/scripts/ci/lang_sdk_serialization/serialize_python.py
+++ b/scripts/ci/lang_sdk_serialization/serialize_python.py
@@ -19,8 +19,8 @@
 
 Each Dag is built with the Python authoring API and written as ``DagSerialization.to_dict`` returns it,
 keyed by Dag id. ``--receive`` also takes a language SDK's output as Airflow receives it: it fills in the
-Dag fields the SDK leaves to Airflow's config, writes the result, and loads every Dag through
-``DagSerialization.validate_schema`` and ``from_dict``. compare.py runs it as::
+Dag fields the SDK leaves to Airflow's config with ``DagSerialization.fill_config_defaults``, writes the
+result, and checks every Dag with ``DagSerialization.validate_serialized_dag``. compare.py runs it as::
 
     uv run --project airflow-core --no-dev python scripts/ci/lang_sdk_serialization/serialize_python.py \
         scripts/ci/lang_sdk_serialization/test_dags.yaml serialized_python.json \
@@ -30,30 +30,17 @@
 from __future__ import annotations
 
 import argparse
-import copy
 import datetime
 import json
 import sys
-from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
 import yaml
 
-from airflow.configuration import conf
 from airflow.sdk import DAG, BaseOperator, TaskGroup
 from airflow.serialization.serialized_objects import DagSerialization
 
-# The Dag fields the Python DAG reads from config when they are left unset. A language SDK cannot read
-# Airflow's config, so it leaves them out, and Airflow fills them in when it receives the SDK's Dags.
-CONFIG_BACKED_DAG_FIELDS: dict[str, Callable[[], Any]] = {
-    "max_active_tasks": lambda: conf.getint("core", "max_active_tasks_per_dag"),
-    "max_active_runs": lambda: conf.getint("core", "max_active_runs_per_dag"),
-    "max_consecutive_failed_dag_runs": lambda: conf.getint("core", "max_consecutive_failed_dag_runs_per_dag"),
-    "catchup": lambda: conf.getboolean("scheduler", "catchup_by_default"),
-    "disable_bundle_versioning": lambda: conf.getboolean("dag_processor", "disable_bundle_versioning"),
-}
-
 
 class NoopOperator(BaseOperator):
     """Stands in for a language SDK's task: a task with no Python behaviour."""
@@ -102,13 +89,11 @@ def get_node(dag: DAG, groups: dict[str, TaskGroup], node_id: str):
 def receive(sdk_output: Path, received_output: Path) -> None:
     received = json.loads(sdk_output.read_text())
     for data in received.values():
-        for key, read_config in CONFIG_BACKED_DAG_FIELDS.items():
-            data["dag"].setdefault(key, read_config())
+        DagSerialization.fill_config_defaults(data)
     received_output.write_text(json.dumps(received, indent=2) + "\n")
     for dag_id, data in received.items():
         try:
-            DagSerialization.validate_schema(data)
-            DagSerialization.from_dict(copy.deepcopy(data))
+            DagSerialization.validate_serialized_dag(data)
         except Exception:
             print(f"Airflow cannot load Dag {dag_id!r} as the SDK wrote it", file=sys.stderr)
             raise
diff --git a/ts-sdk/schema/dag-schema.json b/ts-sdk/schema/dag-schema.json
index 086bccb34bb5b..6ecea324714a1 100644
--- a/ts-sdk/schema/dag-schema.json
+++ b/ts-sdk/schema/dag-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",