diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 7ccab9e4dc971..ec3dd995779e7 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -430,6 +430,25 @@ repos:
           ^go-sdk/schema/dag-schema\.json$|
           ^scripts/ci/lang_sdk_serialization/.*$|
           ^scripts/ci/prek/check_go_sdk_serialization_conformance\.py$
+      - id: check-java-sdk-serialization-conformance
+        name: Check the Java SDK serializes Dags the way Airflow does
+        description: "Serialize the shared test Dags with the Java SDK and with Airflow, and compare the two"
+        entry: ./java-sdk/scripts/ci/prek/check_serialization_conformance.py
+        language: system
+        pass_filenames: false
+        require_serial: true
+        files: >
+          (?x)
+          ^airflow-core/src/airflow/serialization/serialized_objects\.py$|
+          ^airflow-core/src/airflow/serialization/schema\.json$|
+          ^java-sdk/scripts/ci/prek/check_serialization_conformance\.py$|
+          ^java-sdk/sdk/schema/dag-schema\.json$|
+          ^java-sdk/sdk/build\.gradle\.kts$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/(Bundle|DagDef|Deps|Arg|TaskGroupRef)\.kt$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde\.kt$|
+          ^java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/(Refs|Fields)\.kt$|
+          ^java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/.*$|
+          ^scripts/ci/lang_sdk_serialization/.*$
       - id: check-go-version-in-sync
         name: Check Go toolchain version is consistent across build files
         entry: ./scripts/ci/prek/check_go_version_in_sync.py
diff --git a/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md b/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
index c1f9d06fcf64a..48588b35a7743 100644
--- a/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
+++ b/airflow-core/adr/lang-sdk/0007-taskflow-across-language-boundary.md
@@ -256,6 +256,11 @@ modes. It is also why the spec is deliberately not `@task.stub`-shaped — no fi
 natively authored call. (Its docstrings still say "stub", reflecting the only producer that
 exists today; that is wording to revisit, not a constraint in the format.)
 
+A runtime that holds its own Dag may resolve call arguments from that Dag rather than from the
+delivered spec. The spec stays the recorded and displayed form, which keeps one wire format across
+both authoring modes. The Go SDK reads the delivered bindings and the Java SDK resolves locally;
+both emit the same spec.
+
 ### H. Scope: what does not cross the boundary
 
 The following raise when the Dag is serialized, rather than being silently dropped or deferred to
diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
index 446557db8ea89..12b65d85580f9 100644
--- a/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
+++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/java.rst
@@ -675,7 +675,9 @@ generated view, with a no-argument ``depends()`` method:
 
 Every ``@Builder.Task`` method must be called in the wiring class; a task the wiring missed fails at
 Dag-parse time.  ``lit(...)`` wires an inline constant where no upstream feeds a parameter.  A bare
-``double`` cannot be an ``Arg``, so a constant is wrapped.  A view method that takes no arguments
+``double`` cannot be an ``Arg``, so a constant is wrapped.  Airflow records what each task is called
+with, and that record travels as JSON, so a constant has to be a string, number, boolean, list or
+map.  A view method that takes no arguments
 returns the same handle every time, so it names one node wherever it appears; one that takes
 arguments is called once, and the wiring fails if it is called again with arguments, so hold its
 handle in a local and reuse that.
@@ -690,9 +692,9 @@ class that supplies only task bodies, for a Dag a Python file declares, carries
 
 .. note::
 
-   Runtime argument bindings win over Java-declared wiring.  When the supervisor delivers bindings
-   for a run (see :ref:`java-sdk/arg-binding`), the binding at a parameter's position is what the
-   task receives.  Wired inputs are the fallback, which is what a native Java Dag always uses.
+   A native Java Dag binds its task arguments from its own wiring, and the ``_arg_bindings`` it
+   serializes are what Airflow records and shows.  Runtime bindings (see :ref:`java-sdk/arg-binding`)
+   are what a ``@Builder.TaskHandler`` class reads, for a task whose Dag a Python file declares.
 
 Task groups
 ~~~~~~~~~~~
@@ -772,6 +774,9 @@ Durations and date-times are ISO-8601 strings in annotations (``retryDelay = "PT
 ``java.time.OffsetDateTime`` values in ``config`` calls.  An unknown key or a mismatched value type
 fails the build for an annotation, and the ``config`` call itself for an object.
 
+A Dag with a cron ``schedule`` runs in the time zone of its ``startDate``.  With no ``startDate`` it
+is scheduled in UTC, so set ``startDate`` to pin the zone.
+
 .. _java-sdk/task-state-store:
 
 Task state store
diff --git a/dev/breeze/doc/ci/04_selective_checks.md b/dev/breeze/doc/ci/04_selective_checks.md
index ec177bd87a808..9e16809dde851 100644
--- a/dev/breeze/doc/ci/04_selective_checks.md
+++ b/dev/breeze/doc/ci/04_selective_checks.md
@@ -548,6 +548,12 @@ when some files are not changed. Those are the rules implemented:
     Gradle wrapper, which downloads the Gradle distribution, and the latter additionally
     resolves the whole Java SDK dependency graph from Maven Central, so we avoid those
     downloads on PRs that do not touch `java-sdk/`)
+  * if no `Java SDK conformance files` changed - `check-java-sdk-serialization-conformance`
+    is skipped (it compiles the Java SDK with Gradle and resolves its dependencies from Maven
+    Central). The group is wider than `Java SDK files`: it also covers Airflow's serializer
+    (`serialized_objects.py`), `schema.json` and the shared harness under
+    `scripts/ci/lang_sdk_serialization/`, because the check compares the SDK against those and
+    none of them forces `full_tests_needed`
   * if no `TS SDK files` (`ts-sdk/`) changed - `check-ts-sdk-supervisor-schema` check is
     skipped (it regenerates and diffs the generated ts-sdk file; a change to the supervisor
     wire schema alone deliberately does not trigger it - regenerating the ts-sdk types is
diff --git a/dev/breeze/src/airflow_breeze/utils/selective_checks.py b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
index 0388797404ed1..39334c54135d8 100644
--- a/dev/breeze/src/airflow_breeze/utils/selective_checks.py
+++ b/dev/breeze/src/airflow_breeze/utils/selective_checks.py
@@ -140,6 +140,7 @@ class FileGroupForCi(Enum):
     AGENT_FRAMEWORK_FILES = auto()
     GO_SDK_FILES = auto()
     JAVA_SDK_FILES = auto()
+    JAVA_SDK_CONFORMANCE_FILES = auto()
     TS_SDK_FILES = auto()
     TS_SDK_DOCS_FILES = auto()
     AIRFLOW_CTL_FILES = auto()
@@ -517,6 +518,14 @@ def __hash__(self):
             # `.md` excluded — doc-only edits do not affect the Gradle build.
             r"^java-sdk/(?!.*\.md$).*",
         ],
+        FileGroupForCi.JAVA_SDK_CONFORMANCE_FILES: [
+            # The Java SDK, plus what its serialization conformance check compares it against:
+            # Airflow's serializer and schema, and the shared harness.
+            r"^java-sdk/(?!.*\.md$).*",
+            r"^airflow-core/src/airflow/serialization/serialized_objects\.py$",
+            r"^airflow-core/src/airflow/serialization/schema\.json$",
+            r"^scripts/ci/lang_sdk_serialization/.*",
+        ],
         FileGroupForCi.TS_SDK_DOCS_FILES: [
             # TypeDoc renders the reference from the SDK sources and category entry points,
             # and the landing page is authored in ts-sdk/docs — unlike TS_SDK_FILES, `.md`
@@ -1942,6 +1951,11 @@ def skip_prek_hooks(self) -> str:
             # from Maven Central. Skip it when no java-sdk files changed so unrelated PRs do not
             # depend on that resolution.
             prek_hooks_to_skip.add("regenerate-java-sdk-verification-metadata")
+        if not self._matching_files(FileGroupForCi.JAVA_SDK_CONFORMANCE_FILES, CI_FILE_GROUP_MATCHES):
+            # This hook compiles the Java SDK with Gradle and resolves its dependencies from Maven
+            # Central. Skip it unless a java-sdk file, Airflow's serializer or schema, or the shared
+            # harness changed. Those last do not force full_tests_needed, so they need their own group.
+            prek_hooks_to_skip.add("check-java-sdk-serialization-conformance")
         if not self._matching_files(FileGroupForCi.TS_SDK_FILES, CI_FILE_GROUP_MATCHES):
             # This hook regenerates ts-sdk/src/generated/supervisor.ts from the wire schema and
             # diffs it. Schema-only changes deliberately do not trigger it: regenerating the
diff --git a/dev/breeze/tests/test_selective_checks.py b/dev/breeze/tests/test_selective_checks.py
index 7e8d6e8d71243..28664c92413a3 100644
--- a/dev/breeze/tests/test_selective_checks.py
+++ b/dev/breeze/tests/test_selective_checks.py
@@ -113,7 +113,7 @@
 
 
 ALL_SKIPPED_COMMITS_ON_NO_CI_IMAGE = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -127,7 +127,7 @@
 ALL_SKIPPED_COMMITS_BY_DEFAULT_ON_ALL_TESTS_NEEDED = "identity,update-uv-lock"
 
 ALL_SKIPPED_COMMITS_IF_ONLY_UI_OPENAPI_CHANGED = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,"
     "lint-helm-chart,mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,"
     "mypy-airflow-e2e-tests,mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,"
     "mypy-kubernetes-tests,mypy-scripts,mypy-shared-configuration,mypy-shared-dagnode,"
@@ -139,7 +139,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NO_UI = (
-    "check-ts-sdk-supervisor-schema,identity,ktlint,mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
+    "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
     "mypy-shared-configuration,mypy-shared-dagnode,mypy-shared-listeners,mypy-shared-logging,"
@@ -149,7 +149,7 @@
     "regenerate-java-sdk-verification-metadata,ts-compile-lint-simple-auth-manager-ui,ts-compile-lint-ui,update-uv-lock"
 )
 ALL_SKIPPED_COMMITS_IF_NO_HELM_TESTS = (
-    "check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -160,7 +160,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NO_UI_AND_HELM_TESTS = (
-    "check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -175,7 +175,7 @@
 # forced. airflow-core Python changed (so mypy-airflow-core + flynt run); no
 # provider.yaml, helm, or UI files changed, so those checks stay skipped.
 ALL_SKIPPED_COMMITS_IF_ONLY_API_SOURCE_CHANGED = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -188,7 +188,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NO_PROVIDERS_AND_UI = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -200,7 +200,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NO_PROVIDERS = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -213,7 +213,7 @@
 
 
 ALL_SKIPPED_COMMITS_IF_NO_PROVIDERS_UI_AND_HELM_TESTS = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -225,7 +225,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NO_CODE_PROVIDERS_AND_HELM_TESTS = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -236,7 +236,7 @@
 )
 
 ALL_SKIPPED_COMMITS_IF_NOT_IMPORTANT_FILES_CHANGED = (
-    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
+    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,lint-helm-chart,"
     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
     "mypy-scripts,"
@@ -481,7 +481,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "run-amazon-tests": "false",
                     "docs-build": "true",
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -527,7 +527,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "run-api-tests": "true",
                     "docs-build": "true",
                     "skip-prek-hooks": (
-                        "check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -782,7 +782,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "docs-build": "true",
                     "full-tests-needed": "false",
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -820,7 +820,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "docs-build": "false",
                     "full-tests-needed": "false",
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -883,7 +883,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "docs-build": "true",
                     "full-tests-needed": "false",
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-core,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -919,7 +919,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                     "docs-build": "false",
                     "full-tests-needed": "false",
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-e2e-tests,"
                         "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                         "mypy-scripts,"
@@ -1268,7 +1268,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                 "docs-build": "false",
                 "run-kubernetes-tests": "false",
                 "skip-prek-hooks": (
-                    "check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                    "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                     "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                     "mypy-scripts,"
@@ -1475,7 +1475,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                 "run-amazon-tests": "false",
                 "docs-build": "true",
                 "skip-prek-hooks": (
-                    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,"
+                    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,flynt,identity,ktlint,"
                     "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                     "mypy-scripts,"
@@ -1879,7 +1879,7 @@ def assert_outputs_are_printed(expected_outputs: dict[str, str], stderr: str):
                 ("shared/logging/src/airflow_shared/logging/remote.py",),
                 {
                     "skip-prek-hooks": (
-                        "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                        "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                         "mypy-airflow-core,mypy-airflow-ctl,mypy-airflow-ctl-tests,"
                         "mypy-airflow-e2e-tests,mypy-dev,mypy-devel-common,mypy-docker-tests,"
                         "mypy-helm-tests,mypy-kubernetes-tests,mypy-scripts,"
@@ -1911,7 +1911,10 @@ def test_expected_output_pull_request_main(
     assert_outputs_are_printed(expected_outputs, str(stderr))
 
 
-@pytest.mark.parametrize("hook", ["ktlint", "regenerate-java-sdk-verification-metadata"])
+@pytest.mark.parametrize(
+    "hook",
+    ["ktlint", "regenerate-java-sdk-verification-metadata", "check-java-sdk-serialization-conformance"],
+)
 @pytest.mark.parametrize(
     ("files", "hook_skipped"),
     [
@@ -1948,6 +1951,43 @@ def test_java_sdk_gradle_hooks_only_run_for_java_sdk_changes(
     assert (hook in skipped_hooks) is hook_skipped
 
 
+@pytest.mark.parametrize(
+    ("files", "hook_skipped"),
+    [
+        pytest.param(
+            ("airflow-core/src/airflow/serialization/serialized_objects.py",),
+            False,
+            id="runs when Airflow's serializer changes",
+        ),
+        pytest.param(
+            ("airflow-core/src/airflow/serialization/schema.json",),
+            False,
+            id="runs when Airflow's Dag schema changes",
+        ),
+        pytest.param(
+            ("scripts/ci/lang_sdk_serialization/compare.py",),
+            False,
+            id="runs when the shared conformance harness changes",
+        ),
+        pytest.param(
+            ("SECURITY.md",),
+            True,
+            id="skipped when none of its inputs change",
+        ),
+    ],
+)
+def test_java_sdk_conformance_hook_runs_for_serializer_changes(files: tuple[str, ...], hook_skipped: bool):
+    stderr = SelectiveChecks(
+        files=files,
+        commit_ref=NEUTRAL_COMMIT,
+        github_event=GithubEvents.PULL_REQUEST,
+        pr_labels=tuple(),
+        default_branch="main",
+    )
+    skipped_hooks = get_outputs_from_stderr(str(stderr))["skip-prek-hooks"].split(",")
+    assert ("check-java-sdk-serialization-conformance" in skipped_hooks) is hook_skipped
+
+
 @pytest.mark.parametrize(
     ("files", "hook_skipped"),
     [
@@ -2821,7 +2861,7 @@ def test_expected_output_push(
                 "docs-build": "true",
                 "docs-list-as-string": ALL_DOCS_SELECTED_FOR_BUILD,
                 "skip-prek-hooks": (
-                    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                     "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                     "mypy-scripts,"
@@ -2863,7 +2903,7 @@ def test_expected_output_push(
                 "microsoft.mssql mongo mysql openlineage oracle postgres "
                 "presto salesforce samba sftp ssh standard trino",
                 "skip-prek-hooks": (
-                    "check-ts-sdk-supervisor-schema,identity,ktlint,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
+                    "check-java-sdk-serialization-conformance,check-ts-sdk-supervisor-schema,identity,ktlint,mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                     "mypy-scripts,"
                     "mypy-shared-configuration,mypy-shared-dagnode,mypy-shared-listeners,mypy-shared-logging,"
@@ -2908,7 +2948,7 @@ def test_expected_output_push(
                 "docs-build": "true",
                 "docs-list-as-string": "apache-airflow",
                 "skip-prek-hooks": (
-                    "check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
+                    "check-java-sdk-serialization-conformance,check-provider-yaml-valid,check-ts-sdk-supervisor-schema,identity,ktlint,lint-helm-chart,"
                     "mypy-airflow-ctl,mypy-airflow-ctl-tests,mypy-airflow-e2e-tests,"
                     "mypy-dev,mypy-devel-common,mypy-docker-tests,mypy-helm-tests,mypy-kubernetes-tests,"
                     "mypy-scripts,"
diff --git a/java-sdk/README.md b/java-sdk/README.md
index 33a3c560f7b98..6b4ba3522fd44 100644
--- a/java-sdk/README.md
+++ b/java-sdk/README.md
@@ -635,17 +635,17 @@ prek hook regenerate it.
 | capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet |
 | capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet |
 | **Native-Dag authoring** |  |  |  |  |
-| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet |
-| capability: `task-args` | MUST † | n/a | – |  |
-| capability: `dag-params` | MUST † | n/a | – |  |
-| capability: `taskflow-dependencies` | MUST † | n/a | – |  |
-| capability: `branching` | SHOULD † | n/a | – |  |
-| capability: `dag-test` | SHOULD † | n/a | – |  |
-| capability: `task-group` | MAY † | n/a | – |  |
-| capability: `dynamic-task-mapping` | MAY † | n/a | – |  |
-| capability: `asset-inlets-outlets` | MAY † | n/a | – |  |
-| capability: `asset-scheduling` | MAY † | n/a | – |  |
-| capability: `object-store` | MAY † | n/a | – |  |
+| capability: `native-dag-authoring` | SHOULD | ✓ | 3.4 |  |
+| capability: `task-args` | MUST † | ✓ | 3.4 |  |
+| capability: `dag-params` | MUST † | ✗ | – |  |
+| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4 |  |
+| capability: `branching` | SHOULD † | ✗ | – |  |
+| capability: `dag-test` | SHOULD † | ✗ | – |  |
+| capability: `task-group` | MAY † | ✓ | 3.4 |  |
+| capability: `dynamic-task-mapping` | MAY † | ✗ | – |  |
+| capability: `asset-inlets-outlets` | MAY † | ✗ | – |  |
+| capability: `asset-scheduling` | MAY † | ✗ | – |  |
+| capability: `object-store` | MAY † | ✗ | – |  |
 
 *Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.*
 
diff --git a/java-sdk/capabilities.yaml b/java-sdk/capabilities.yaml
index 6f61f999240fd..75c6053fc892e 100644
--- a/java-sdk/capabilities.yaml
+++ b/java-sdk/capabilities.yaml
@@ -52,8 +52,8 @@ states:
     supported: true
     since: "3.3"
 
-# Runtime capabilities reflect the task-facing Client surface; native-Dag authoring is not
-# implemented yet, so every native capability is unsupported.
+# Runtime capabilities reflect the task-facing Client surface. Native-Dag capabilities describe
+# what a Java-authored Dag can declare and deliver through Dag serialization.
 capabilities:
   mixed-lang-stub-target:
     supported: true
@@ -96,20 +96,23 @@ capabilities:
     supported: false
     note: "no task-facing asset-event API yet"
   native-dag-authoring:
-    supported: false
-    note: "native Dag authoring not implemented yet"
+    supported: true
+    since: "3.4"
   task-args:
-    supported: false
+    supported: true
+    since: "3.4"
   dag-params:
     supported: false
   taskflow-dependencies:
-    supported: false
+    supported: true
+    since: "3.4"
   branching:
     supported: false
   dag-test:
     supported: false
   task-group:
-    supported: false
+    supported: true
+    since: "3.4"
   dynamic-task-mapping:
     supported: false
   asset-inlets-outlets:
diff --git a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
index 1101dbb9084f3..cc0d9c594f89e 100644
--- a/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
+++ b/java-sdk/processor/src/main/kotlin/org/apache/airflow/sdk/BuilderProcessor.kt
@@ -362,11 +362,15 @@ class BuilderProcessor : AbstractProcessor() {
     if (decl.dataParams.isEmpty()) {
       method.addStatement($$"return $T.node($L, $L)", REFS_TYPE, group, def)
     } else {
+      // The parameter names ride along so the serialized Dag can name each
+      // argument, as the binding spec ADR-0007 defines requires.
       method.addStatement(
-        $$"return $T.call($L, $L, $L)",
+        $$"return $T.call($L, $L, $T.of($L), $L)",
         REFS_TYPE,
         group,
         def,
+        LIST_TYPE,
+        CodeBlock.join(decl.dataParams.map { CodeBlock.of($$"$S", it.name) }, ", "),
         decl.dataParams.joinToString { it.name },
       )
     }
diff --git a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
index d75c33aa45d9e..c3759fa19985d 100644
--- a/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
+++ b/java-sdk/processor/src/test/kotlin/org/apache/airflow/sdk/BuilderTest.kt
@@ -136,6 +136,7 @@ class BuilderTest {
 
          import java.lang.Integer;
          import java.lang.Void;
+         import java.util.List;
          import org.apache.airflow.sdk.Arg;
          import org.apache.airflow.sdk.Deps;
          import org.apache.airflow.sdk.TaskDef;
@@ -159,7 +160,7 @@ class BuilderTest {
            }
 
            default TaskRef<Void> t3(Arg<? extends Integer> value) {
-             return Refs.call("", new TaskDef("t3", TestExampleBuilder.T3.class), value);
+             return Refs.call("", new TaskDef("t3", TestExampleBuilder.T3.class), List.of("value"), value);
            }
          }
         """,
@@ -381,7 +382,7 @@ class BuilderTest {
 
            default TaskRef<Void> t(Arg<? extends String> text, Arg<?> anything,
                Arg<? extends List<String>> items, Arg<? extends Long> boxed) {
-             return Refs.call("", new TaskDef("t", TestExampleBuilder.T.class), text, anything, items, boxed);
+             return Refs.call("", new TaskDef("t", TestExampleBuilder.T.class), List.of("text", "anything", "items", "boxed"), text, anything, items, boxed);
            }
          }
         """,
diff --git a/java-sdk/scripts/ci/prek/check_serialization_conformance.py b/java-sdk/scripts/ci/prek/check_serialization_conformance.py
new file mode 100755
index 0000000000000..9869f4ccc5ca0
--- /dev/null
+++ b/java-sdk/scripts/ci/prek/check_serialization_conformance.py
@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Check the Java SDK serializes Dags as Airflow does, with scripts/ci/lang_sdk_serialization/compare.py."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[4] / "scripts" / "ci" / "prek"))
+
+from common_prek_utils import AIRFLOW_ROOT_PATH
+
+if __name__ not in ("__main__", "__mp_main__"):
+    raise SystemExit(
+        "This file is intended to be executed as an executable program. You cannot use it as a module."
+        f"To run this script, run the ./{__file__} command"
+    )
+
+if __name__ == "__main__":
+    java_sdk = AIRFLOW_ROOT_PATH / "java-sdk"
+    # Gradle prints its own progress on stdout, so the classpath is the last line.
+    printed = subprocess.run(
+        [str(java_sdk / "gradlew"), "-p", str(java_sdk), "-q", ":sdk:printConformanceClasspath"],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    lines = printed.stdout.strip().splitlines()
+    if printed.returncode or not lines:
+        sys.stderr.write(printed.stdout)
+        sys.stderr.write(printed.stderr)
+        raise SystemExit("Could not build the Java SDK conformance classpath; see the Gradle output above")
+    classpath = lines[-1]
+    compare = AIRFLOW_ROOT_PATH / "scripts" / "ci" / "lang_sdk_serialization" / "compare.py"
+    serializer = ["java", "-cp", classpath, "org.apache.airflow.sdk.conformance.SerializeJavaKt"]
+    command = [
+        sys.executable,
+        str(compare),
+        "--sdk",
+        "java",
+        "--supports",
+        "literal_inputs",
+        "--",
+        *serializer,
+    ]
+    sys.exit(subprocess.run(command, check=False).returncode)
diff --git a/java-sdk/sdk/build.gradle.kts b/java-sdk/sdk/build.gradle.kts
index 9ef581634bd7f..0159e1d2803d6 100644
--- a/java-sdk/sdk/build.gradle.kts
+++ b/java-sdk/sdk/build.gradle.kts
@@ -865,3 +865,12 @@ publishing {
         }
     }
 }
+
+// Prints the classpath that runs the conformance serializer, for
+// java-sdk/scripts/ci/prek/check_serialization_conformance.py.
+tasks.register("printConformanceClasspath") {
+    dependsOn("testClasses")
+    // Capture early to keep compatibility to the Gradle configuration cache.
+    val classpath = sourceSets.test.get().runtimeClasspath
+    doLast { println(classpath.asPath) }
+}
diff --git a/java-sdk/sdk/module.md b/java-sdk/sdk/module.md
index 07e272d14315f..3d11a42da5ea5 100644
--- a/java-sdk/sdk/module.md
+++ b/java-sdk/sdk/module.md
@@ -56,17 +56,17 @@ meaning of each dimension is defined in the
 | capability: `asset-event-emit` | MAY | ✗ | – | runtime does not emit asset events yet |
 | capability: `asset-event-read` | MAY | ✗ | – | no task-facing asset-event API yet |
 | **Native-Dag authoring** |  |  |  |  |
-| capability: `native-dag-authoring` | SHOULD | ✗ | – | native Dag authoring not implemented yet |
-| capability: `task-args` | MUST † | n/a | – |  |
-| capability: `dag-params` | MUST † | n/a | – |  |
-| capability: `taskflow-dependencies` | MUST † | n/a | – |  |
-| capability: `branching` | SHOULD † | n/a | – |  |
-| capability: `dag-test` | SHOULD † | n/a | – |  |
-| capability: `task-group` | MAY † | n/a | – |  |
-| capability: `dynamic-task-mapping` | MAY † | n/a | – |  |
-| capability: `asset-inlets-outlets` | MAY † | n/a | – |  |
-| capability: `asset-scheduling` | MAY † | n/a | – |  |
-| capability: `object-store` | MAY † | n/a | – |  |
+| capability: `native-dag-authoring` | SHOULD | ✓ | 3.4 |  |
+| capability: `task-args` | MUST † | ✓ | 3.4 |  |
+| capability: `dag-params` | MUST † | ✗ | – |  |
+| capability: `taskflow-dependencies` | MUST † | ✓ | 3.4 |  |
+| capability: `branching` | SHOULD † | ✗ | – |  |
+| capability: `dag-test` | SHOULD † | ✗ | – |  |
+| capability: `task-group` | MAY † | ✓ | 3.4 |  |
+| capability: `dynamic-task-mapping` | MAY † | ✗ | – |  |
+| capability: `asset-inlets-outlets` | MAY † | ✗ | – |  |
+| capability: `asset-scheduling` | MAY † | ✗ | – |  |
+| capability: `object-store` | MAY † | ✗ | – |  |
 
 *Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when `native-dag-authoring` is supported.*
 
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
index d5ba757beab45..ff6f740af180e 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/DagDef.kt
@@ -316,6 +316,9 @@ class TaskDef(
 
   internal val configValues = linkedMapOf<String, Any>()
   internal val inputs = mutableListOf<Arg<*>>()
+
+  /** Name of the task parameter each of [inputs] feeds, in the same order. */
+  internal val inputNames = mutableListOf<String>()
   internal val upstreams = linkedSetOf<TaskDef>()
   internal var owner: DagDef? = null
 
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
index 0dab9c978f701..01b26177cd6dd 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Deps.kt
@@ -125,6 +125,10 @@ interface Deps {
    * `transform(extract(), lit(0.9))`. It is passed to the task as a constant
    * and creates no dependency edge.
    *
+   * The Dag's call arguments travel to Airflow as JSON, so the value has to be
+   * a string, number, boolean, list, or map; anything else fails when the Dag
+   * is parsed.
+   *
    * @param value Constant to bind; may be null for a nullable parameter.
    */
   fun <T> lit(value: T?): Arg<T> = Arg.lit(value)
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
index 4b42eb68d4c39..47b44662b9969 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/Server.kt
@@ -33,8 +33,10 @@ import kotlinx.coroutines.runBlocking
 import org.apache.airflow.sdk.execution.CoordinatorComm
 import org.apache.airflow.sdk.execution.LogSender
 import org.apache.airflow.sdk.execution.Logger
+import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
 import org.apache.airflow.sdk.execution.comm.ErrorResponse
 import org.apache.airflow.sdk.execution.comm.StartupDetails
+import org.apache.airflow.sdk.execution.parseDags
 import org.apache.airflow.sdk.execution.runTask
 import kotlin.text.substringAfterLast
 import kotlin.text.substringBeforeLast
@@ -181,6 +183,7 @@ class Server(
     val frame = coordinator.readMessage()
     when (val body = frame.body) {
       is StartupDetails -> runTaskAndReport(bundle, body, coordinator)
+      is DagFileParseRequest -> parseDagsAndReport(bundle, body, coordinator)
       is ErrorResponse -> throw ApiError("[${body.error}] ${body.detail}")
       else -> throw ApiError("Unexpected initial frame (id=${frame.id})")
     }
@@ -194,4 +197,12 @@ class Server(
     val result = runTask(bundle, startup, coordinator)
     coordinator.communicate<Unit>(result)
   }
+
+  private suspend fun parseDagsAndReport(
+    bundle: Bundle,
+    request: DagFileParseRequest,
+    coordinator: CoordinatorComm,
+  ) {
+    coordinator.communicate<Unit>(parseDags(bundle, request))
+  }
 }
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
new file mode 100644
index 0000000000000..89245cf6946ad
--- /dev/null
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/execution/Serde.kt
@@ -0,0 +1,490 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.execution
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import org.apache.airflow.sdk.Bundle
+import org.apache.airflow.sdk.DagDef
+import org.apache.airflow.sdk.GroupEdges
+import org.apache.airflow.sdk.GroupExpansion
+import org.apache.airflow.sdk.LiteralArg
+import org.apache.airflow.sdk.TaskDef
+import org.apache.airflow.sdk.TaskGroupRef
+import org.apache.airflow.sdk.TaskRef
+import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
+import org.apache.airflow.sdk.internal.Field
+import org.apache.airflow.sdk.internal.SchemaFields
+import java.nio.file.InvalidPathException
+import java.nio.file.Paths
+import java.time.Duration
+import java.time.Instant
+import java.time.OffsetDateTime
+
+// Serializes Dags to Airflow DagSerialization v3 JSON, mirroring the TypeScript
+// SDK's serde (ts-sdk/src/coordinator/serde.ts), which in turn matches Python's
+// DagSerialization output.
+
+private val defaultsMapper = ObjectMapper()
+
+// Python drops both unless the operator names an email recipient. A Java task
+// has no email field to name one, so they are never written.
+private val OMITTED_TASK_KEYS = setOf("email_on_failure", "email_on_retry")
+
+/**
+ * Processes a [DagFileParseRequest] by serialising every Dag registered on
+ * [bundle] to DagSerialization v3 and returning the result as a
+ * DagFileParsingResult body.
+ *
+ * A Dag that cannot be serialized becomes an import error rather than taking
+ * the rest of the bundle with it. Airflow keys an import error by the
+ * bundle-relative path and holds one row per file, so every failure here is
+ * reported under that one key with its Dag named in the message.
+ */
+internal fun parseDags(
+  bundle: Bundle,
+  request: DagFileParseRequest,
+): Map<String, Any?> {
+  val fileloc = request.file ?: ""
+  val relativeFileloc = computeRelativeFileloc(fileloc, request.bundlePath)
+  val serializedDags = mutableListOf<Map<String, Any?>>()
+  val failures = mutableListOf<String>()
+  bundle.dags.values.forEach { dag ->
+    runCatching { serializeDag(dag, fileloc, relativeFileloc) }
+      .onSuccess { serializedDags += mapOf("data" to mapOf("__version" to 3, "dag" to it)) }
+      .onFailure { failures += "Dag \"${dag.id}\": ${it.message ?: it.javaClass.name}" }
+  }
+  return linkedMapOf<String, Any?>(
+    "type" to "DagFileParsingResult",
+    "fileloc" to fileloc,
+    "serialized_dags" to serializedDags,
+  ).apply {
+    if (failures.isNotEmpty()) this["import_errors"] = mapOf(relativeFileloc to failures.joinToString("\n"))
+  }
+}
+
+/**
+ * Converts a [DagDef] to Airflow DagSerialization v3 format. Required fields are
+ * always present; config-driven fields follow the rules in [applyDagConfig]
+ * (some always emitted, some only when set).
+ */
+internal fun serializeDag(
+  dag: DagDef,
+  fileloc: String,
+  relativeFileloc: String,
+): Map<String, Any?> {
+  // Group edges mean tasks only once the Dag is complete, so they are worked
+  // out here rather than carried on the groups themselves.
+  val expansion = dag.expandGroupEdges()
+  val downstream = linkedMapOf<String, MutableList<String>>()
+  dag.tasks.forEach { (taskId, def) ->
+    expansion.upstreamsOf(def).forEach { upstream ->
+      downstream.getOrPut(upstream) { mutableListOf() } += taskId
+    }
+  }
+
+  val result =
+    linkedMapOf<String, Any?>(
+      "dag_id" to dag.id,
+      "fileloc" to fileloc,
+      "relative_fileloc" to relativeFileloc,
+      "timezone" to dagTimezone(dag.dagConfig),
+      "timetable" to serializeTimetable(dag.id, dag.dagConfig),
+      "tasks" to dag.tasks.map { (taskId, def) -> serializeTask(taskId, def, downstream[taskId]) },
+      "dag_dependencies" to emptyList<Any?>(),
+      "task_group" to serializeTaskGroups(dag, expansion),
+      "edge_info" to emptyMap<String, Any?>(),
+      "params" to emptyList<Any?>(),
+      "deadline" to null,
+      "allowed_run_types" to null,
+    )
+  applyDagConfig(result, dag.dagConfig)
+  return result
+}
+
+/**
+ * Converts one task to the Airflow serialization format. `downstream` is the
+ * inverted view of the Dag's upstream edges, sorted for stable JSON.
+ */
+private fun serializeTask(
+  taskId: String,
+  def: TaskDef,
+  downstream: List<String>?,
+): Map<String, Any?> {
+  val data =
+    linkedMapOf<String, Any?>(
+      "task_id" to taskId,
+      "task_type" to def.definition.simpleName,
+      "_task_module" to def.definition.packageName,
+      "language" to "java",
+      // Python's operator serializer always emits template_fields (its list
+      // value never matches the tuple default it is compared against), so it
+      // is unconditional here too. Java tasks have no template fields.
+      "template_fields" to emptyList<Any?>(),
+      // What marks a task whose arguments Airflow resolves per instance for a
+      // runtime outside Python, as `@task.stub` does on the Python side.
+      // `get_arg_bindings` reads nothing without it.
+      "is_stub" to true,
+    )
+  argBindings(taskId, def)?.let { data["_arg_bindings"] = it }
+  // Emit only config entries that differ from their schema default, mirroring
+  // Python BaseSerialization's "omit hard-coded default" behavior, which the Go
+  // and TypeScript SDKs mirror too. Operator fields are stored unwrapped, so the
+  // __type encoding is stripped. If core grows a task-level fill_config_defaults,
+  // every SDK has to keep explicitly set values instead, or an explicit retries=0
+  // reads as unset and picks up the configured default.
+  def.configValues.forEach { (key, value) ->
+    if (key !in OMITTED_TASK_KEYS && !matchesSchemaDefault(SchemaFields.TASK[key], value)) {
+      data[key] = unwrapTypeEncoding(serializeValue(value))
+    }
+  }
+  if (!downstream.isNullOrEmpty()) {
+    data["downstream_task_ids"] = downstream.sorted()
+  }
+  return mapOf(
+    "__type" to "operator",
+    "__var" to data,
+  )
+}
+
+/**
+ * The task's arguments as the binding spec Airflow records, one entry per
+ * argument in the order the Dag's call passed them, as
+ * [ADR-0007](../../adr/lang-sdk/0007-taskflow-across-language-boundary.md)
+ * defines it.
+ *
+ * An upstream's handle becomes an `xcom` binding naming that task, and
+ * anything else a `literal` carrying the value. `value_schema` is left out: it
+ * constrains the decode side, and a Java task decodes into the type its own
+ * parameter declares.
+ *
+ * A Java task reads its arguments from the Dag in its own bundle rather than
+ * from this spec, so what it carries is what Airflow shows and what a change
+ * to an argument is seen in.
+ *
+ * Null for a task the Dag called with no arguments, which needs no spec.
+ */
+private fun argBindings(
+  taskId: String,
+  def: TaskDef,
+): List<Map<String, Any?>>? {
+  // A Dag wired by hand through Refs names no argument, and a spec without
+  // names binds nothing, so it is left out rather than written half-filled.
+  if (def.inputNames.size != def.inputs.size || def.inputs.isEmpty()) return null
+  return def.inputNames.zip(def.inputs) { name, input ->
+    when (input) {
+      is TaskRef<*> -> mapOf("name" to name, "kind" to "xcom", "task_id" to input.def.id)
+      is LiteralArg<*> ->
+        mapOf("name" to name, "kind" to "literal", "value" to plainJson(input.value, name, taskId))
+    }
+  }
+}
+
+/**
+ * [value] as the JSON the binding spec travels as, rejecting anything that has
+ * no JSON form.
+ *
+ * The spec is part of the serialized Dag, so a literal Airflow cannot store is
+ * refused where the Dag is written rather than where the task reads it.
+ */
+private fun plainJson(
+  value: Any?,
+  name: String,
+  taskId: String,
+): Any? =
+  when (value) {
+    null, is String, is Boolean -> value
+    is Double ->
+      value.takeIf { it.isFinite() }
+        ?: throw IllegalArgumentException(
+          "Argument '$name' of task '$taskId' is $value, which JSON has no form for; pass it as a string",
+        )
+    is Float -> plainJson(value.toDouble(), name, taskId)
+    is Int, is Long, is Short, is Byte -> value
+    is Collection<*> -> value.map { plainJson(it, name, taskId) }
+    is Array<*> -> value.map { plainJson(it, name, taskId) }
+    is Map<*, *> ->
+      value.entries.associate { (key, entry) ->
+        require(key is String) { "Argument '$name' of task '$taskId' has a map key that is not a string" }
+        key to plainJson(entry, name, taskId)
+      }
+    else ->
+      throw IllegalArgumentException(
+        "Argument '$name' of task '$taskId' is a ${value.javaClass.name}, which has no JSON form; the " +
+          "Dag's call arguments travel as JSON, so pass a string, number, boolean, list, or map",
+      )
+  }
+
+/**
+ * Writes Dag-level config onto [data], leaving out every field the Dag did
+ * not set. That includes the fields Python reads from Airflow's config
+ * (max_active_tasks, max_active_runs, max_consecutive_failed_dag_runs,
+ * catchup, disable_bundle_versioning): Airflow fills those in from its own
+ * config when it receives the Dag.
+ */
+private fun applyDagConfig(
+  data: MutableMap<String, Any?>,
+  config: Map<String, Any>,
+) {
+  listOf("description", "dag_display_name", "doc_md", "start_date", "end_date", "dagrun_timeout").forEach { key ->
+    config[key]?.let { data[key] = unwrapTypeEncoding(serializeValue(it)) }
+  }
+  (config["tags"] as? List<*>)?.let { tags ->
+    // Python stores tags in a set and serializes them sorted (for a stable
+    // dag_hash); mirror that regardless of registration order.
+    data["tags"] = tags.map { it.toString() }.distinct().sorted()
+  }
+  listOf(
+    "max_active_tasks",
+    "max_active_runs",
+    "max_consecutive_failed_dag_runs",
+    "catchup",
+    "disable_bundle_versioning",
+  ).forEach { key -> config[key]?.let { data[key] = it } }
+  // fail_fast and render_template_as_native_obj have schema default false, so
+  // Python omits them when false; keep that behavior.
+  if (config["fail_fast"] == true) data["fail_fast"] = true
+  if (config["render_template_as_native_obj"] == true) data["render_template_as_native_obj"] = true
+  config["is_paused_upon_creation"]?.let { data["is_paused_upon_creation"] = it }
+}
+
+// TODO: respect [scheduler] create_cron_data_intervals like Python's
+// _create_timetable; the JVM bundle cannot read airflow.cfg, so the
+// supervisor must send those flags over the coordinator protocol first.
+// The same gap applies to [core] default_timezone, which Python's
+// _extract_tz uses for a Dag with no start date; this uses UTC.
+// The TypeScript SDK waits on the same flag; tracked at
+// https://github.com/apache/airflow/issues/67938
+private fun serializeTimetable(
+  dagId: String,
+  config: Map<String, Any>,
+): Map<String, Any?> =
+  when (val schedule = config["schedule"] as String?) {
+    null -> mapOf("__type" to "airflow.timetables.simple.NullTimetable", "__var" to emptyMap<String, Any?>())
+    "@once" -> mapOf("__type" to "airflow.timetables.simple.OnceTimetable", "__var" to emptyMap<String, Any?>())
+    "@continuous" ->
+      mapOf("__type" to "airflow.timetables.simple.ContinuousTimetable", "__var" to emptyMap<String, Any?>())
+    else -> {
+      val expression = CRON_PRESETS[schedule] ?: schedule
+      require(isCronExpression(expression)) {
+        "Schedule '$schedule' of Dag '$dagId' is not a cron expression or a preset " +
+          "(${(CRON_PRESETS.keys + CRON_ALIASES).joinToString()}, @once, @continuous); a schedule the " +
+          "scheduler cannot parse would leave the Dag unschedulable"
+      }
+      mapOf(
+        "__type" to "airflow.timetables.trigger.CronTriggerTimetable",
+        "__var" to
+          mapOf(
+            "expression" to expression,
+            "timezone" to dagTimezone(config),
+            "interval" to 0.0,
+            "run_immediately" to false,
+          ),
+      )
+    }
+  }
+
+/**
+ * The Dag's timezone, as Python's `encode_timezone` writes it: `"UTC"` for a
+ * zero offset, otherwise the offset in seconds.
+ *
+ * Python takes it from `start_date`, so a cron schedule runs in the zone the
+ * Dag's start date was written in. A Dag with no start date runs in UTC.
+ */
+private fun dagTimezone(config: Map<String, Any>): Any =
+  (config["start_date"] as? OffsetDateTime)
+    ?.offset
+    ?.totalSeconds
+    ?.takeIf { it != 0 }
+    ?: "UTC"
+
+/**
+ * Presets expanded the way `CronMixin.__init__` expands them, so the
+ * serialized expression is the one Python records, which the Dag's summary and
+ * its hash are both taken from. Mirrors `airflow.utils.dates.cron_presets`.
+ */
+private val CRON_PRESETS =
+  mapOf(
+    "@hourly" to "0 * * * *",
+    "@daily" to "0 0 * * *",
+    "@weekly" to "0 0 * * 0",
+    "@monthly" to "0 0 1 * *",
+    "@quarterly" to "0 0 1 */3 *",
+    "@yearly" to "0 0 1 1 *",
+  )
+
+/** One value in a cron field: a number, `*`, `?`, or a three-letter month or weekday name. */
+private const val CRON_VALUE = "(\\d+|\\*|\\?|[A-Z]{3})"
+
+/**
+ * One comma-separated element of a cron field: a value or a range, either
+ * stepped, with croniter's `L`, `W` and `#` qualifiers.
+ */
+private val CRON_ELEMENT =
+  Regex("$CRON_VALUE(-$CRON_VALUE)?([/#]\\d+)?[LW]*|L(-\\d+)?|LW", RegexOption.IGNORE_CASE)
+
+/**
+ * Aliases croniter accepts that `cron_presets` does not expand, so Python
+ * stores them unexpanded and so does this.
+ */
+private val CRON_ALIASES = setOf("@midnight", "@annually")
+
+/**
+ * Whether [expression] has the shape croniter accepts: five or six
+ * space-separated fields of cron characters, or an `@` alias.
+ *
+ * A shape check, not a parse: croniter validates the ranges, and repeating
+ * that here would be a second implementation to keep in step. What it catches
+ * is prose, such as `"every tuesday"`, which would otherwise be written into a
+ * Dag the scheduler then fails to build a timetable for.
+ */
+private fun isCronExpression(expression: String): Boolean {
+  val trimmed = expression.trim()
+  if (trimmed.startsWith("@")) return trimmed in CRON_PRESETS || trimmed in CRON_ALIASES
+  val fields = trimmed.split(Regex("\\s+"))
+  return fields.size in 5..6 &&
+    fields.all { field -> field.split(',').all { CRON_ELEMENT.matches(it) } }
+}
+
+/**
+ * Serializes the Dag's task groups as Python's `TaskGroupSerialization` does:
+ * a root group holding the tasks in no group and the top-level groups, each
+ * group nesting its own tasks and groups.
+ */
+private fun serializeTaskGroups(
+  dag: DagDef,
+  expansion: GroupExpansion,
+): Map<String, Any?> {
+  val grouped = dag.groups.values.flatMapTo(mutableSetOf()) { it.taskIds }
+  return taskGroupObject(
+    null,
+    dag.tasks.keys.filterNot { it in grouped },
+    dag.groups.values.filterNot { '.' in it.id },
+    expansion,
+  )
+}
+
+/** One group object: the root when [group] is null, otherwise a nested one. */
+private fun taskGroupObject(
+  group: TaskGroupRef?,
+  taskIds: List<String>,
+  children: List<TaskGroupRef>,
+  expansion: GroupExpansion,
+): Map<String, Any?> =
+  mapOf(
+    // The local segment: Python's TaskGroup stores the ID it was given, and
+    // rebuilds the full one from where the group sits in the tree.
+    "_group_id" to group?.id?.substringAfterLast('.'),
+    "group_display_name" to "",
+    "prefix_group_id" to true,
+    "tooltip" to "",
+    "ui_color" to "CornflowerBlue",
+    "ui_fgcolor" to "#000",
+    "children" to
+      taskIds.associateWith { listOf("operator", it) } +
+      children.associate {
+        it.id to listOf("taskgroup", taskGroupObject(it, it.taskIds, it.children, expansion))
+      },
+    "upstream_group_ids" to group.edges(expansion).upstreamGroupIds.sorted(),
+    "downstream_group_ids" to group.edges(expansion).downstreamGroupIds.sorted(),
+    "upstream_task_ids" to group.edges(expansion).upstreamTaskIds.sorted(),
+    "downstream_task_ids" to group.edges(expansion).downstreamTaskIds.sorted(),
+  )
+
+/** The edges this group records for itself; the root group records none. */
+private fun TaskGroupRef?.edges(expansion: GroupExpansion): GroupEdges = this?.let { expansion.edgesOf(it.id) } ?: GroupEdges()
+
+/**
+ * Recursively serializes a value with Airflow's type/var encoding, matching
+ * Python's `BaseSerialization.serialize()` output: primitives pass through,
+ * date-times become `{"__type": "datetime", "__var": epoch_seconds}`,
+ * durations `{"__type": "timedelta", "__var": total_seconds}`, and maps
+ * `{"__type": "dict", "__var": {...}}`.
+ */
+internal fun serializeValue(value: Any?): Any? =
+  when (value) {
+    null -> null
+    is String, is Boolean, is Int, is Long, is Double -> value
+    is Byte, is Short -> (value as Number).toInt()
+    is Float -> value.toDouble()
+    is OffsetDateTime -> serializeValue(value.toInstant())
+    is Instant ->
+      mapOf(
+        "__type" to "datetime",
+        "__var" to value.epochSecond + value.nano / 1e9,
+      )
+    is Duration ->
+      mapOf(
+        "__type" to "timedelta",
+        "__var" to value.toNanos() / 1e9,
+      )
+    is Map<*, *> ->
+      mapOf(
+        "__type" to "dict",
+        "__var" to value.entries.associate { (k, v) -> k.toString() to serializeValue(v) },
+      )
+    is List<*> -> value.map(::serializeValue)
+    is Array<*> -> value.map(::serializeValue)
+    else -> value
+  }
+
+/**
+ * Extracts the `__var` part from a type-encoded value: in Python's
+ * `serialize_to_json`, non-decorated fields are serialized then unwrapped.
+ */
+internal fun unwrapTypeEncoding(value: Any?): Any? {
+  val map = value as? Map<*, *> ?: return value
+  if ("__type" !in map) return value
+  return if ("__var" in map) map["__var"] else value
+}
+
+/** Whether a config value equals the schema default and can be omitted. */
+private fun matchesSchemaDefault(
+  field: Field?,
+  value: Any,
+): Boolean {
+  val defaultJson = field?.defaultJson ?: return false
+  val node = defaultsMapper.readTree(defaultJson)
+  return when (value) {
+    is String -> node.isTextual && node.asText() == value
+    is Boolean -> node.isBoolean && node.asBoolean() == value
+    is Number -> node.isNumber && node.asDouble() == value.toDouble()
+    is Duration -> node.isNumber && node.asDouble() == value.toNanos() / 1e9
+    else -> false
+  }
+}
+
+private fun computeRelativeFileloc(
+  fileloc: String,
+  bundlePath: String?,
+): String {
+  if (fileloc.isEmpty()) return ""
+  if (bundlePath.isNullOrEmpty()) return "."
+  return try {
+    Paths
+      .get(bundlePath)
+      .relativize(Paths.get(fileloc))
+      .toString()
+      .ifEmpty { "." }
+  } catch (e: InvalidPathException) {
+    "."
+  } catch (e: IllegalArgumentException) {
+    "."
+  }
+}
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
index 559e975645be5..e3920f2b67f76 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/ArgValues.kt
@@ -51,10 +51,9 @@ import java.lang.reflect.Type
  * binding at their position (through [TaskArgs]); [TaskInput] fields resolve
  * bindings by name.
  *
- * A natively authored Dag has no stub call site, so the supervisor sends no
- * bindings for it and the inputs the Dag itself wired stand in. When the
- * supervisor sends bindings they are used for every parameter; the Dag's own
- * inputs are read only when it sends none.
+ * A natively authored Dag resolves its own arguments instead: the bundle that
+ * holds the task also holds the call that wired it, so a task that wired any
+ * input reads those and never the bindings.
  *
  * A count that does not match is fatal for flat parameters and a warning for a
  * [TaskInput]: a position has no name to fall back on, while a field does, so
@@ -92,7 +91,7 @@ object ArgValues {
     // Runtime bindings carry argument names to match fields against. A wired
     // input carries none, so it decodes into the whole input at once -- which
     // is well defined because a TaskInput is a task's only data parameter.
-    wiredInputs(context, client)?.let { wired ->
+    wiredInputs(context)?.let { wired ->
       warnWiredArity(client, type, wired.size)
       return type.cast(decode(resolveWiredAll(wired.take(1), client).single(), type))
         ?: throw missingInput(wired[0], type.simpleName)
@@ -232,14 +231,15 @@ object ArgValues {
   ): Any? = decode(value, type)
 
   /**
-   * The inputs the Dag wired for this task, or null when the run's arguments
-   * come from the stub call site. A task with no wired inputs reads the
-   * bindings, so a stub call that bound nothing keeps its own diagnostics.
+   * The inputs the Dag wired for this task, or null when it wired none and the
+   * arguments come from the stub call site instead.
+   *
+   * A natively authored Dag always answers here, so its serialized binding
+   * spec is what Airflow records and shows rather than something read back at
+   * run time. A task handler wires nothing, so it reads the bindings and keeps
+   * its own diagnostics.
    */
-  internal fun wiredInputs(
-    context: Context,
-    client: Client,
-  ): List<Arg<*>>? = if (client.argBindings.isEmpty()) context.taskDef?.inputs?.takeIf { it.isNotEmpty() } else null
+  internal fun wiredInputs(context: Context): List<Arg<*>>? = context.taskDef?.inputs?.takeIf { it.isNotEmpty() }
 
   /**
    * The failure for a wired argument that resolved to nothing where a value is
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
index 387e6fec3658e..761d5b8f28261 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/Refs.kt
@@ -92,14 +92,16 @@ object Refs {
   fun <T> node(
     groupId: String,
     def: TaskDef,
-  ): TaskRef<T> = call(groupId, def)
+  ): TaskRef<T> = call(groupId, def, emptyList())
 
   /**
-   * Records a task and the data edge for every [TaskRef] among [args]; a
-   * literal argument records a baked value and no edge.
+   * Records a task called with named arguments, as a generated wiring view
+   * does. [names] is the task parameter each argument feeds, in the same
+   * order, and is what the serialized Dag carries as the task's binding spec.
    *
    * @param groupId Full ID of the task group holding it, empty when it sits in
-   *    none.
+   *    none. The generated wiring view knows which it is.
+   *
    * @return The handle representing this task, memoized by [TaskDef.id] so a result
    *    held in a local and reused refers to one node.
    * @throws IllegalArgumentException if an argument is a raw Java `null`
@@ -110,8 +112,12 @@ object Refs {
   fun <T> call(
     groupId: String,
     def: TaskDef,
+    names: List<String>,
     vararg args: Arg<*>?,
   ): TaskRef<T> {
+    require(names.isEmpty() || names.size == args.size) {
+      "Task '${def.id}' was wired with ${args.size} argument(s) under ${names.size} name(s)"
+    }
     val inputs =
       args.mapIndexed { i, arg ->
         requireNotNull(arg) {
@@ -131,6 +137,7 @@ object Refs {
     }
     inputs.filterIsInstance<TaskRef<*>>().forEach { def.dependsOn(it.def) }
     def.inputs += inputs
+    def.inputNames += names
     if (groupId.isEmpty()) {
       active.dag.addTask(def)
     } else {
diff --git a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/TaskArgs.kt b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/TaskArgs.kt
index 14c6404758b06..c4bd53a5e148b 100644
--- a/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/TaskArgs.kt
+++ b/java-sdk/sdk/src/main/kotlin/org/apache/airflow/sdk/internal/TaskArgs.kt
@@ -75,7 +75,7 @@ class TaskArgs private constructor(
       client: Client,
       declared: Int,
     ): TaskArgs {
-      ArgValues.wiredInputs(context, client)?.let { wired ->
+      ArgValues.wiredInputs(context)?.let { wired ->
         // Positional binding is strict in both directions: a parameter has no
         // name to fall back on, so a count that does not match cannot be
         // resolved and the run fails here rather than mid-task.
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgTestSupport.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgTestSupport.kt
index 50b0a8ef3ae19..0929dfe4e4ec8 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgTestSupport.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgTestSupport.kt
@@ -129,6 +129,6 @@ internal class NoopTask : Task {
 internal fun contextWiredWith(inputs: List<Arg<*>>): Context {
   val dag = DagDef("d")
   val def = TaskDef("t", NoopTask::class.java)
-  Refs.record(dag, listOf("t"), emptyList()) { Refs.call<Unit>("", def, *inputs.toTypedArray()) }
+  Refs.record(dag, listOf("t"), emptyList()) { Refs.call<Unit>("", def, emptyList(), *inputs.toTypedArray()) }
   return taskContext().also { it.taskDef = def }
 }
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgValuesTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgValuesTest.kt
index b6b5347ba4ec4..3c0a6fed27913 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgValuesTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ArgValuesTest.kt
@@ -121,6 +121,17 @@ internal class ArgValuesTest {
     assertEquals(0.5, input.threshold)
   }
 
+  @Test
+  @DisplayName("Should bind the Dag's wired input over the runtime bindings")
+  fun shouldPreferWiredInputOverRuntimeBindings() {
+    val (client, _) = clientWith(listOf(literal("threshold", 9.0)))
+    val context = contextWiredWith(listOf(Arg.lit(mapOf("threshold" to 0.5))))
+
+    val input = ArgValues.bindInput(context, client, ThresholdInput::class.java)
+
+    assertEquals(0.5, input.threshold)
+  }
+
   @Test
   @DisplayName("Should match a camelCase field to a snake_case argument through the fold")
   fun shouldFoldSnakeCaseArgument() {
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ServerTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ServerTest.kt
index e436dd2fc6160..d6e93fe211421 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ServerTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/ServerTest.kt
@@ -27,16 +27,25 @@ import kotlinx.coroutines.runBlocking
 import org.apache.airflow.sdk.execution.CoordinatorComm
 import org.apache.airflow.sdk.execution.Frame
 import org.apache.airflow.sdk.execution.IncomingFrame
+import org.apache.airflow.sdk.execution.RawFrame
 import org.apache.airflow.sdk.execution.comm.TaskState
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.Timeout
 import org.msgpack.core.MessagePack
+import org.msgpack.core.buffer.ArrayBufferInput
 import java.io.ByteArrayOutputStream
 import java.util.concurrent.ArrayBlockingQueue
 import java.util.concurrent.TimeUnit
 
+private class ServerNoopTask : Task {
+  override fun execute(
+    context: Context,
+    client: Client,
+  ) = Unit
+}
+
 class ServerTest {
   private fun hexToBytes(hex: String): ByteArray =
     hex
@@ -112,6 +121,60 @@ class ServerTest {
     comm.close()
   }
 
+  @Test
+  @DisplayName("Should serialize the bundle's dags when the initial frame is a parse request")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun parsesDagsAndReportsResult() {
+    val toServer = ByteChannel(autoFlush = true)
+    val fromServer = ByteChannel(autoFlush = true)
+    val comm = CoordinatorComm(toServer, fromServer)
+    val server = Server(InetSocketAddress("localhost", 0), InetSocketAddress("localhost", 0))
+    val bundle = Bundle(listOf(DagDef("parsed_dag").addTask(TaskDef("t", ServerNoopTask::class.java))))
+
+    val reported = ArrayBlockingQueue<RawFrame>(1)
+    val supervisor =
+      Thread {
+        runBlocking {
+          toServer.writeFrame(parseRequestFrame(3, "/bundle/dags/java.py", "/bundle"))
+          val prefix = fromServer.readByteArray(4)
+          val payload = fromServer.readByteArray(Frame.parseLengthPrefix(prefix).toInt())
+          val raw = Frame.decodeRaw(ArrayBufferInput(payload))
+          reported.put(raw)
+          toServer.writeFrame(ackFrame(raw.id))
+        }
+      }
+    supervisor.start()
+
+    runBlocking { server.dispatchTask(bundle, comm) }
+    supervisor.join()
+
+    val body = reported.take().rawBody as Map<*, *>
+    Assertions.assertEquals("DagFileParsingResult", body["type"])
+    Assertions.assertEquals("/bundle/dags/java.py", body["fileloc"])
+    val dags = body["serialized_dags"] as List<*>
+    Assertions.assertEquals(1, dags.size)
+    val dag = ((dags[0] as Map<*, *>)["data"] as Map<*, *>)["dag"] as Map<*, *>
+    Assertions.assertEquals("parsed_dag", dag["dag_id"])
+    comm.close()
+  }
+
+  private fun parseRequestFrame(
+    id: Int,
+    file: String,
+    bundlePath: String,
+  ): ByteArray {
+    val out = ByteArrayOutputStream()
+    MessagePack.newDefaultPacker(out).use { packer ->
+      packer.packArrayHeader(2)
+      packer.packInt(id)
+      packer.packMapHeader(3)
+      packer.packString("type").packString("DagFileParseRequest")
+      packer.packString("file").packString(file)
+      packer.packString("bundle_path").packString(bundlePath)
+    }
+    return out.toByteArray()
+  }
+
   private companion object {
     // [2, msg, null] with msg coming from
     // https://github.com/astronomer/airflow/blob/f39c8da8/task-sdk/tests/task_sdk/execution_time/test_comms.py#L73-L108
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/SerializeJava.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/SerializeJava.kt
new file mode 100644
index 0000000000000..7d9aa9370103c
--- /dev/null
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/conformance/SerializeJava.kt
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Serializes the shared test Dags with this SDK, for
+// scripts/ci/lang_sdk_serialization/compare.py:
+//
+//   java -cp <sdk test runtime classpath> org.apache.airflow.sdk.conformance.SerializeJavaKt \
+//       scripts/ci/lang_sdk_serialization/test_dags.yaml serialized_java.json
+//
+// Each Dag is built with the interface API and written as the runtime answers
+// a parse request, keyed by Dag ID. A task's `upstream` becomes an ordering
+// edge, which a Java task serializes the same way as a data edge.
+package org.apache.airflow.sdk.conformance
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLMapper
+import org.apache.airflow.sdk.Arg
+import org.apache.airflow.sdk.Bundle
+import org.apache.airflow.sdk.Client
+import org.apache.airflow.sdk.Context
+import org.apache.airflow.sdk.DagDef
+import org.apache.airflow.sdk.Deps
+import org.apache.airflow.sdk.Task
+import org.apache.airflow.sdk.TaskGroupRef
+import org.apache.airflow.sdk.TaskRef
+import org.apache.airflow.sdk.execution.serializeDag
+import org.apache.airflow.sdk.internal.Field
+import org.apache.airflow.sdk.internal.FieldType
+import org.apache.airflow.sdk.internal.SchemaFields
+import java.io.File
+import java.time.Duration
+import java.time.OffsetDateTime
+
+class ConformanceTask : Task {
+  override fun execute(
+    context: Context,
+    client: Client,
+  ) = Unit
+}
+
+fun main(args: Array<String>) {
+  require(args.size == 2) { "usage: SerializeJava <test_dags.yaml> <output.json>" }
+  val cases = YAMLMapper().readTree(File(args[0])).path("dags")
+  val bundle = Bundle()
+  cases.forEach { bundle.register(buildDag(it)) }
+  val serialized =
+    bundle.dags.values.associate { dag ->
+      dag.id to mapOf("__version" to 3, "dag" to serializeDag(dag, "", "."))
+    }
+  ObjectMapper().writerWithDefaultPrettyPrinter().writeValue(File(args[1]), serialized)
+}
+
+private fun buildDag(case: JsonNode): DagDef {
+  val dag = DagDef(case.path("dag_id").asText())
+  case.path("spec").fields().forEach { (key, value) -> dag.config(key, toValue(SchemaFields.DAG, key, value)) }
+
+  // A group ID is fully qualified, so its parent is whatever comes before the last dot.
+  val groups = linkedMapOf<String, TaskGroupRef>()
+  case.path("groups").forEach { node ->
+    val groupId = node.asText()
+    val parentId = groupId.substringBeforeLast('.', "")
+    val localId = groupId.substringAfterLast('.')
+    groups[groupId] = if (parentId.isEmpty()) dag.taskGroup(localId) else groups.getValue(parentId).taskGroup(localId)
+  }
+
+  val tasks = linkedMapOf<String, TaskRef<Unit>>()
+  case.path("tasks").forEach { task ->
+    val groupId = task.path("group").asText("")
+    val localId = task.path("task_id").asText()
+    val ref =
+      if (groupId.isEmpty()) {
+        dag.task<Unit>(localId, ConformanceTask::class.java)
+      } else {
+        groups.getValue(groupId).task(localId, ConformanceTask::class.java)
+      }
+    task.path("spec").fields().forEach { (key, value) -> ref.config(key, toValue(SchemaFields.TASK, key, value)) }
+    // A task's `upstream` handles and its `literals` are its call arguments, in that order, so the
+    // Dag carries the binding spec a stub call would. Names are positional, as the Go SDK names
+    // them: the interface API has no signature to read parameter names from.
+    val inputs: List<Arg<*>> =
+      task.path("upstream").map { tasks.getValue(it.asText()) } +
+        task.path("literals").map { Arg.lit(toJsonValue(it)) }
+    inputs.filterIsInstance<TaskRef<*>>().forEach { ref.after(it) }
+    ref.def.inputs += inputs
+    ref.def.inputNames += inputs.indices.map { "arg$it" }
+    tasks[ref.def.id] = ref
+  }
+
+  case.path("order_edges").forEach { edge ->
+    val node = { id: String -> groups[id] as Deps.Flow? ?: tasks.getValue(id) }
+    node(edge[0].asText()).before(node(edge[1].asText()))
+  }
+  return dag
+}
+
+/** Reads a YAML value as the Java type the config key takes. */
+private fun toValue(
+  table: Map<String, Field>,
+  key: String,
+  node: JsonNode,
+): Any =
+  when (requireNotNull(table[key]) { "Unknown config key: '$key'" }.type) {
+    FieldType.STRING -> node.asText()
+    FieldType.BOOLEAN -> node.asBoolean()
+    FieldType.INTEGER -> node.asInt()
+    FieldType.NUMBER -> node.numberValue()
+    // `!datetime` is an ISO 8601 timestamp, and `!timedelta` a number of seconds.
+    FieldType.DATETIME -> OffsetDateTime.parse(node.asText())
+    FieldType.TIMEDELTA -> Duration.ofNanos((node.asText().toDouble() * 1e9).toLong())
+    FieldType.STRING_ARRAY -> node.map { it.asText() }
+  }
+
+/** Reads a YAML literal as the plain value `Serde` writes out. */
+private fun toJsonValue(node: JsonNode): Any? =
+  when {
+    node.isNull -> null
+    node.isTextual -> node.asText()
+    node.isBoolean -> node.asBoolean()
+    node.isIntegralNumber -> node.numberValue()
+    node.isNumber -> node.asDouble()
+    node.isArray -> node.map { toJsonValue(it) }
+    else -> node.fields().asSequence().associate { (key, value) -> key to toJsonValue(value) }
+  }
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt
new file mode 100644
index 0000000000000..e025d9a5b154f
--- /dev/null
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/execution/SerdeTest.kt
@@ -0,0 +1,428 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.airflow.sdk.execution
+
+import org.apache.airflow.sdk.Arg
+import org.apache.airflow.sdk.Bundle
+import org.apache.airflow.sdk.Client
+import org.apache.airflow.sdk.Context
+import org.apache.airflow.sdk.DagDef
+import org.apache.airflow.sdk.Task
+import org.apache.airflow.sdk.TaskDef
+import org.apache.airflow.sdk.execution.comm.DagFileParseRequest
+import org.apache.airflow.sdk.internal.Refs
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import java.time.Duration
+import java.time.OffsetDateTime
+
+private class SerdeNoopTask : Task {
+  override fun execute(
+    context: Context,
+    client: Client,
+  ) = Unit
+}
+
+@Suppress("UNCHECKED_CAST")
+private fun taskData(
+  serialized: Map<String, Any?>,
+  index: Int,
+): Map<String, Any?> {
+  val tasks = serialized["tasks"] as List<Map<String, Any?>>
+  assertEquals("operator", tasks[index]["__type"])
+  return tasks[index]["__var"] as Map<String, Any?>
+}
+
+internal class SerdeTest {
+  @Test
+  @DisplayName("Should emit required dag fields and leave unset config-backed fields out")
+  fun shouldEmitRequiredDagFields() {
+    val serialized = serializeDag(DagDef("d"), "/bundles/app/dags.jar", "app/dags.jar")
+
+    assertEquals("d", serialized["dag_id"])
+    assertEquals("/bundles/app/dags.jar", serialized["fileloc"])
+    assertEquals("app/dags.jar", serialized["relative_fileloc"])
+    assertEquals("UTC", serialized["timezone"])
+    assertEquals(
+      mapOf("__type" to "airflow.timetables.simple.NullTimetable", "__var" to emptyMap<String, Any?>()),
+      serialized["timetable"],
+    )
+    assertEquals(emptyList<Any?>(), serialized["tasks"])
+    assertEquals(emptyList<Any?>(), serialized["dag_dependencies"])
+    assertEquals(emptyMap<String, Any?>(), serialized["edge_info"])
+    assertEquals(emptyList<Any?>(), serialized["params"])
+    assertNull(serialized["deadline"])
+    assertNull(serialized["allowed_run_types"])
+    assertFalse("max_active_tasks" in serialized)
+    assertFalse("max_active_runs" in serialized)
+    assertFalse("max_consecutive_failed_dag_runs" in serialized)
+    assertFalse("catchup" in serialized)
+    assertFalse("disable_bundle_versioning" in serialized)
+    assertFalse("description" in serialized)
+    assertFalse("fail_fast" in serialized)
+    assertFalse("tags" in serialized)
+  }
+
+  @Test
+  @DisplayName("Should map schedule strings to the matching timetable")
+  fun shouldMapScheduleToTimetable() {
+    val cron = serializeDag(DagDef("d").config("schedule", "@daily"), "", ".")
+    assertEquals(
+      mapOf(
+        "__type" to "airflow.timetables.trigger.CronTriggerTimetable",
+        "__var" to
+          mapOf(
+            "expression" to "0 0 * * *",
+            "timezone" to "UTC",
+            "interval" to 0.0,
+            "run_immediately" to false,
+          ),
+      ),
+      cron["timetable"],
+    )
+
+    val once = serializeDag(DagDef("d").config("schedule", "@once"), "", ".")
+    assertEquals(
+      mapOf("__type" to "airflow.timetables.simple.OnceTimetable", "__var" to emptyMap<String, Any?>()),
+      once["timetable"],
+    )
+
+    val continuous = serializeDag(DagDef("d").config("schedule", "@continuous"), "", ".")
+    assertEquals(
+      mapOf("__type" to "airflow.timetables.simple.ContinuousTimetable", "__var" to emptyMap<String, Any?>()),
+      continuous["timetable"],
+    )
+  }
+
+  @Test
+  @DisplayName("Should apply dag config values with Python's emit rules")
+  fun shouldApplyDagConfig() {
+    val dag =
+      DagDef("d")
+        .config("description", "demo")
+        .config("tags", listOf("b", "a", "b"))
+        .config("catchup", true)
+        .config("fail_fast", true)
+        .config("max_active_runs", 3)
+        .config("dagrun_timeout", Duration.ofMinutes(5))
+        .config("start_date", OffsetDateTime.parse("2026-01-01T00:00:00Z"))
+
+    val serialized = serializeDag(dag, "", ".")
+
+    assertEquals("demo", serialized["description"])
+    assertEquals(listOf("a", "b"), serialized["tags"])
+    assertEquals(true, serialized["catchup"])
+    assertEquals(true, serialized["fail_fast"])
+    assertEquals(3, serialized["max_active_runs"])
+    assertEquals(300.0, serialized["dagrun_timeout"])
+    assertEquals(1.7672256E9, serialized["start_date"])
+  }
+
+  @Test
+  @DisplayName("Should serialize tasks with identity fields, config, and sorted downstream ids")
+  fun shouldSerializeTasks() {
+    val extractDef =
+      TaskDef("extract", SerdeNoopTask::class.java)
+        .config("retries", 2)
+        .config("queue", "q")
+        .config("retry_delay", Duration.ofMinutes(10))
+        .config("email_on_failure", false)
+        .config("email_on_retry", false)
+    val transformDef =
+      TaskDef("transform", SerdeNoopTask::class.java)
+        .dependsOn(extractDef)
+        // Explicitly at schema defaults: omitted from the serialized form.
+        .config("retries", 0)
+        .config("queue", "default")
+        .config("retry_delay", Duration.ofMinutes(5))
+    val dag = DagDef("d").addTask(extractDef).addTask(transformDef)
+
+    val serialized = serializeDag(dag, "", ".")
+
+    val extract = taskData(serialized, 0)
+    assertEquals("extract", extract["task_id"])
+    assertEquals("SerdeNoopTask", extract["task_type"])
+    assertEquals("org.apache.airflow.sdk.execution", extract["_task_module"])
+    assertEquals("java", extract["language"])
+    assertEquals(emptyList<Any?>(), extract["template_fields"])
+    assertEquals(2, extract["retries"])
+    assertEquals("q", extract["queue"])
+    assertEquals(600.0, extract["retry_delay"])
+    assertEquals(listOf("transform"), extract["downstream_task_ids"])
+    assertFalse("email_on_failure" in extract)
+    assertFalse("email_on_retry" in extract)
+
+    val transform = taskData(serialized, 1)
+    assertEquals("transform", transform["task_id"])
+    assertFalse("retries" in transform)
+    assertFalse("queue" in transform)
+    assertFalse("retry_delay" in transform)
+    assertFalse("downstream_task_ids" in transform)
+
+    assertEquals(
+      mapOf("extract" to listOf("operator", "extract"), "transform" to listOf("operator", "transform")),
+      (serialized["task_group"] as Map<*, *>)["children"],
+    )
+  }
+
+  @Test
+  @DisplayName("Should serialize nested task groups with their own edges")
+  fun shouldSerializeTaskGroups() {
+    val dag = DagDef("d")
+    val extract = dag.task<Unit>("extract", SerdeNoopTask::class.java)
+    val staging = dag.taskGroup("staging")
+    staging.task<Unit>("stage", SerdeNoopTask::class.java)
+    staging.taskGroup("checks").task<Unit>("nulls", SerdeNoopTask::class.java)
+    extract.before(staging)
+
+    val root = serializeDag(dag, "", ".")["task_group"] as Map<*, *>
+
+    val group = { name: String, children: Map<String, Any?>, upstreamTasks: List<String> ->
+      mapOf(
+        "_group_id" to name,
+        "group_display_name" to "",
+        "prefix_group_id" to true,
+        "tooltip" to "",
+        "ui_color" to "CornflowerBlue",
+        "ui_fgcolor" to "#000",
+        "children" to children,
+        "upstream_group_ids" to emptyList<String>(),
+        "downstream_group_ids" to emptyList<String>(),
+        "upstream_task_ids" to upstreamTasks,
+        "downstream_task_ids" to emptyList<String>(),
+      )
+    }
+    val checks = group("checks", mapOf("staging.checks.nulls" to listOf("operator", "staging.checks.nulls")), emptyList())
+    assertEquals(
+      mapOf(
+        "extract" to listOf("operator", "extract"),
+        "staging" to
+          listOf(
+            "taskgroup",
+            group(
+              "staging",
+              mapOf(
+                "staging.stage" to listOf("operator", "staging.stage"),
+                "staging.checks" to listOf("taskgroup", checks),
+              ),
+              listOf("extract"),
+            ),
+          ),
+      ),
+      root["children"],
+    )
+    assertEquals(null, root["_group_id"])
+  }
+
+  @Test
+  @DisplayName("Should serialize wiring-registered dags with their data-flow edges")
+  fun shouldSerializeWiredDag() {
+    val dag = DagDef("d")
+    Refs.record(dag, listOf("extract", "transform"), emptyList()) {
+      val extracted = Refs.node<Long>("", TaskDef("extract", SerdeNoopTask::class.java))
+      Refs.call<Unit>("", TaskDef("transform", SerdeNoopTask::class.java), listOf("rows"), extracted)
+    }
+
+    val serialized = serializeDag(dag, "", ".")
+
+    assertEquals(listOf("transform"), taskData(serialized, 0)["downstream_task_ids"])
+    assertEquals(
+      listOf(mapOf("name" to "rows", "kind" to "xcom", "task_id" to "extract")),
+      taskData(serialized, 1)["_arg_bindings"],
+    )
+    assertEquals(true, taskData(serialized, 1)["is_stub"])
+  }
+
+  @Test
+  @DisplayName("Should bind a wired literal as a literal argument binding")
+  fun shouldSerializeLiteralArgBinding() {
+    val dag = DagDef("d")
+    Refs.record(dag, listOf("transform"), emptyList()) {
+      Refs.call<Unit>("", TaskDef("transform", SerdeNoopTask::class.java), listOf("region"), Arg.lit("uk"))
+    }
+
+    val serialized = serializeDag(dag, "", ".")
+
+    assertEquals(
+      listOf(mapOf("name" to "region", "kind" to "literal", "value" to "uk")),
+      taskData(serialized, 0)["_arg_bindings"],
+    )
+  }
+
+  @Test
+  @DisplayName("Should leave out the binding spec of a task called with no arguments")
+  fun shouldOmitArgBindingsWithoutArguments() {
+    val serialized =
+      serializeDag(DagDef("d").addTask(TaskDef("t", SerdeNoopTask::class.java)), "", ".")
+
+    assertFalse("_arg_bindings" in taskData(serialized, 0))
+    assertEquals(true, taskData(serialized, 0)["is_stub"])
+  }
+
+  @Test
+  @DisplayName("Should reject a wired literal that has no JSON form")
+  fun shouldRejectNonJsonLiteral() {
+    val dag = DagDef("d")
+    Refs.record(dag, listOf("t"), emptyList()) {
+      Refs.call<Unit>("", TaskDef("t", SerdeNoopTask::class.java), listOf("at"), Arg.lit(Duration.ofSeconds(5)))
+    }
+
+    val error = assertThrows(IllegalArgumentException::class.java) { serializeDag(dag, "", ".") }
+
+    assertEquals(
+      "Argument 'at' of task 't' is a java.time.Duration, which has no JSON form; the Dag's call " +
+        "arguments travel as JSON, so pass a string, number, boolean, list, or map",
+      error.message,
+    )
+  }
+
+  @Test
+  @DisplayName("Should take the dag timezone from the start date, as Python does")
+  fun shouldTakeTimezoneFromStartDate() {
+    val dag =
+      DagDef("d")
+        .config("schedule", "0 3 * * *")
+        .config("start_date", OffsetDateTime.parse("2026-01-01T00:00:00+05:30"))
+
+    val serialized = serializeDag(dag, "", ".")
+
+    assertEquals(19800, serialized["timezone"])
+    assertEquals(19800, (serialized["timetable"] as Map<*, *>)["__var"].let { (it as Map<*, *>)["timezone"] })
+  }
+
+  private fun cronExpression(schedule: String): Any? {
+    val timetable = serializeDag(DagDef("d").config("schedule", schedule), "", ".")["timetable"] as Map<*, *>
+    return (timetable["__var"] as Map<*, *>)["expression"]
+  }
+
+  @Test
+  @DisplayName("Should accept every cron schedule croniter does")
+  fun shouldAcceptCroniterSchedules() {
+    listOf(
+      "0 9 * * MON,WED,FRI",
+      "0 0 * JAN,JUL *",
+      "0 0 * * MON#2",
+      "0 0 15W * *",
+      "*/5 1-5/2 * * MON-FRI",
+    ).forEach { assertEquals(it, cronExpression(it)) }
+  }
+
+  @Test
+  @DisplayName("Should serialize croniter's @midnight and @annually aliases unexpanded")
+  fun shouldKeepCronAliasesUnexpanded() {
+    assertEquals("@midnight", cronExpression("@midnight"))
+    assertEquals("@annually", cronExpression("@annually"))
+    assertEquals("0 0 * * *", cronExpression("@daily"))
+  }
+
+  @Test
+  @DisplayName("Should reject schedules that are not cron expressions")
+  fun shouldRejectProseAndUnknownAliases() {
+    listOf("every tuesday", "@bogus", "0 0 * * tuesday").forEach {
+      assertThrows(IllegalArgumentException::class.java) { cronExpression(it) }
+    }
+  }
+
+  @Test
+  @DisplayName("Should reject a schedule the scheduler cannot build a timetable from")
+  fun shouldRejectNonCronSchedule() {
+    val dag = DagDef("d").config("schedule", "every monday")
+
+    val error = assertThrows(IllegalArgumentException::class.java) { serializeDag(dag, "", ".") }
+
+    assertEquals(
+      "Schedule 'every monday' of Dag 'd' is not a cron expression or a preset " +
+        "(@hourly, @daily, @weekly, @monthly, @quarterly, @yearly, @midnight, @annually, @once, @continuous); a schedule the " +
+        "scheduler cannot parse would leave the Dag unschedulable",
+      error.message,
+    )
+  }
+
+  @Test
+  @DisplayName("Should report a dag that cannot be serialized as an import error")
+  fun shouldReportUnserializableDagAsImportError() {
+    val broken = DagDef("broken").config("schedule", "every monday")
+    val healthy = DagDef("healthy").addTask(TaskDef("t", SerdeNoopTask::class.java))
+    val request =
+      DagFileParseRequest().also {
+        it.file = "/bundles/app/dags.jar"
+        it.bundlePath = "/bundles"
+      }
+
+    val result = parseDags(Bundle(listOf(broken, healthy)), request)
+
+    assertEquals(1, (result["serialized_dags"] as List<*>).size)
+    assertEquals(
+      mapOf(
+        "app/dags.jar" to
+          "Dag \"broken\": Schedule 'every monday' of Dag 'broken' is not a cron expression or a preset " +
+          "(@hourly, @daily, @weekly, @monthly, @quarterly, @yearly, @midnight, @annually, @once, @continuous); a schedule the " +
+          "scheduler cannot parse would leave the Dag unschedulable",
+      ),
+      result["import_errors"],
+    )
+  }
+
+  @Test
+  @DisplayName("Should wrap parsed dags in a DagFileParsingResult body")
+  fun shouldBuildParsingResult() {
+    val bundle = Bundle(listOf(DagDef("d").addTask(TaskDef("t", SerdeNoopTask::class.java))))
+    val request =
+      DagFileParseRequest().also {
+        it.file = "/bundles/app/dags.jar"
+        it.bundlePath = "/bundles"
+      }
+
+    val result = parseDags(bundle, request)
+
+    assertEquals("DagFileParsingResult", result["type"])
+    assertEquals("/bundles/app/dags.jar", result["fileloc"])
+    val dags = result["serialized_dags"] as List<*>
+    assertEquals(1, dags.size)
+    val data = (dags[0] as Map<*, *>)["data"] as Map<*, *>
+    assertEquals(3, data["__version"])
+    val dag = data["dag"] as Map<*, *>
+    assertEquals("d", dag["dag_id"])
+    assertEquals("app/dags.jar", dag["relative_fileloc"])
+  }
+
+  @Test
+  @DisplayName("Should encode temporals and nested maps with the type/var envelope")
+  fun shouldEncodeValuesWithTypeEnvelope() {
+    assertEquals(
+      mapOf("__type" to "timedelta", "__var" to 90.0),
+      serializeValue(Duration.ofSeconds(90)),
+    )
+    assertEquals(
+      mapOf("__type" to "datetime", "__var" to 1.7672256E9),
+      serializeValue(OffsetDateTime.parse("2026-01-01T00:00:00Z")),
+    )
+    assertEquals(
+      mapOf("__type" to "dict", "__var" to mapOf("k" to listOf(1, 2))),
+      serializeValue(mapOf("k" to listOf(1, 2))),
+    )
+    assertEquals(42, unwrapTypeEncoding(mapOf("__type" to "timedelta", "__var" to 42)))
+    assertEquals(mapOf("plain" to 1), unwrapTypeEncoding(mapOf("plain" to 1)))
+  }
+}
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
index fad223f605fd2..b9fd480e10c3d 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/ArgValuesTest.kt
@@ -141,7 +141,7 @@ internal class ArgValuesTest {
       .distinct()
       .forEach { dag.addTask(it) }
     val def = TaskDef("consumer", NoopArgTask::class.java)
-    Refs.record(dag, listOf("consumer"), emptyList()) { Refs.call<Unit>("", def, *inputs.toTypedArray()) }
+    Refs.record(dag, listOf("consumer"), emptyList()) { Refs.call<Unit>("", def, emptyList(), *inputs.toTypedArray()) }
     return contextWithoutTaskDef().also { it.taskDef = def }
   }
 
diff --git a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
index 975dd8e449c68..3a06f7c472bd0 100644
--- a/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
+++ b/java-sdk/sdk/src/test/kotlin/org/apache/airflow/sdk/internal/RefsTest.kt
@@ -54,7 +54,7 @@ internal class RefsTest {
     val dag = DagDef("d")
     Refs.record(dag, listOf("p", "c"), emptyList()) {
       val producer = Refs.node<Long>("", TaskDef("p", NoopRefTask::class.java))
-      Refs.call<Unit>("", TaskDef("c", NoopRefTask::class.java), producer, Arg.lit(5))
+      Refs.call<Unit>("", TaskDef("c", NoopRefTask::class.java), emptyList(), producer, Arg.lit(5))
     }
 
     val consumerDef = dag.tasks.getValue("c")
@@ -126,7 +126,7 @@ internal class RefsTest {
     val error =
       assertThrows(IllegalArgumentException::class.java) {
         Refs.record(DagDef("d"), listOf("t"), emptyList()) {
-          Refs.call<Unit>("", TaskDef("t", NoopRefTask::class.java), Arg.lit(1), null)
+          Refs.call<Unit>("", TaskDef("t", NoopRefTask::class.java), emptyList(), Arg.lit(1), null)
         }
       }
 
@@ -140,7 +140,7 @@ internal class RefsTest {
       assertThrows(IllegalArgumentException::class.java) {
         Refs.record(DagDef("d"), listOf("t"), emptyList()) {
           Refs.node<Unit>("", TaskDef("t", NoopRefTask::class.java))
-          Refs.call<Unit>("", TaskDef("t", NoopRefTask::class.java), Arg.lit(1))
+          Refs.call<Unit>("", TaskDef("t", NoopRefTask::class.java), emptyList(), Arg.lit(1))
         }
       }
 
diff --git a/scripts/ci/lang_sdk_serialization/compare.py b/scripts/ci/lang_sdk_serialization/compare.py
index 0d1dc19bf8fb5..e6fbab441c88d 100644
--- a/scripts/ci/lang_sdk_serialization/compare.py
+++ b/scripts/ci/lang_sdk_serialization/compare.py
@@ -35,6 +35,9 @@
 the Dags whose features are all among them. A Dag that requires nothing is always in it, and the SDK
 has to write exactly the Dags of the copy.
 
+Each SDK's own prek hook runs it this way, such as
+java-sdk/scripts/ci/prek/check_serialization_conformance.py for the Java SDK.
+
 On a failure the files are kept, and their directory is printed.
 """
 
@@ -147,21 +150,25 @@ def get_task_defaults() -> dict[str, Any]:
     return {key: field["default"] for key, field in fields.items() if field.get("default") is not None}
 
 
-def normalize_as_javascript(value: Any) -> Any:
-    """Read a JSON value as JavaScript does: one number type, and a bool that is not a number."""
+def normalize_numbers(value: Any) -> Any:
+    """
+    Read a JSON value with one number type, and a bool that is not a number.
+
+    An SDK may write ``2`` where Python writes ``2.0``, as JavaScript does, which has one number type.
+    """
     if isinstance(value, bool):
         return ("bool", value)
     if isinstance(value, (int, float)):
         return float(value)
     if isinstance(value, list):
-        return [normalize_as_javascript(item) for item in value]
+        return [normalize_numbers(item) for item in value]
     if isinstance(value, dict):
-        return {key: normalize_as_javascript(item) for key, item in value.items()}
+        return {key: normalize_numbers(item) for key, item in value.items()}
     return value
 
 
 def is_same_json(python: Any, sdk: Any) -> bool:
-    return normalize_as_javascript(python) == normalize_as_javascript(sdk)
+    return normalize_numbers(python) == normalize_numbers(sdk)
 
 
 def find_differences(path: str, python: Any, sdk: Any) -> list[str]:
diff --git a/scripts/ci/lang_sdk_serialization/test_dags.yaml b/scripts/ci/lang_sdk_serialization/test_dags.yaml
index a9591031bb8a9..ae5a756426d92 100644
--- a/scripts/ci/lang_sdk_serialization/test_dags.yaml
+++ b/scripts/ci/lang_sdk_serialization/test_dags.yaml
@@ -79,7 +79,7 @@ dags:
           wait_for_past_depends_before_skipping: true
           wait_for_downstream: true
           retries: 2
-          queue: typescript
+          queue: workers
           pool: tiny
           pool_slots: 2
           execution_timeout: !timedelta 120
@@ -143,7 +143,7 @@ dags:
 
   # Fan-out and fan-in, so downstream_task_ids has to be sorted rather than kept in wiring
   # order. The schedule is not written "@daily" because Python expands a preset before
-  # serializing it, which serde.test.ts covers separately.
+  # serializing it, which each SDK's own serializer tests cover separately.
   - dag_id: conformance_diamond
     spec:
       schedule: "0 0 * * *"
diff --git a/scripts/tests/ci/lang_sdk_serialization/test_compare.py b/scripts/tests/ci/lang_sdk_serialization/test_compare.py
index a5efc191657b9..6bba7d0abbc3f 100644
--- a/scripts/tests/ci/lang_sdk_serialization/test_compare.py
+++ b/scripts/tests/ci/lang_sdk_serialization/test_compare.py
@@ -48,6 +48,7 @@ def build_serialized(fileloc: str, tasks: list[dict]) -> dict:
                 "timezone": "UTC",
                 "catchup": False,
                 "tags": ["a"],
+                "params": [],
                 "task_group": {"prefix_group_id": True, "children": {"extract": ["operator", "extract"]}},
                 "tasks": [{"__type": "operator", "__var": task} for task in tasks],
             },
@@ -121,6 +122,16 @@ def test_accepts_the_differences_an_sdk_is_allowed():
             ["d: tags[0] is 'b', Python writes 'a'"],
             id="nested-list",
         ),
+        pytest.param(
+            lambda sdk: sdk["d"]["dag"].update(
+                params=[["limit", {"__class": "airflow.sdk.definitions.param.Param", "default": 5}]]
+            ),
+            [
+                "d: params is [['limit', {'__class': 'airflow.sdk.definitions.param.Param', 'default': 5}]], "
+                "Python writes []"
+            ],
+            id="dag-params",
+        ),
         pytest.param(
             lambda sdk: sdk["d"]["dag"]["task_group"].update(prefix_group_id=False),
             ["d: task_group.prefix_group_id is False, Python writes True"],