[FrankYang0529]

Implement Go SDK: Serialize a native Dag to Dag JSON.

• DagRef.serialize writes {"__version": 3, "dag": {...}} for a registered Dag. The dag object holds the timetable, the tasks, the edge labels and a dag_dependencies entry for each TriggerDagRun task. The task groups form a tree, written as children[label] = [kind, value], and each group records its group edges.
• A TriggerDagRun task gets the fields that Python writes for a TriggerDagRunOperator. The TriggerDagRun task also carries the other parameters, such as poke_interval. An unset LogicalDate is written as NOTSET.
• genspec generates spec_fields.gen.go, which maps each field of DagSpec, TaskSpec and TaskGroupSpec to its schema property and schema default, as the TS SDK generates its field table. A field that is unset or at its schema default is left out.
• When a Dag leaves max_active_runs and four other fields unset, Python takes their values from the Airflow config. A bundle cannot read that config, so the serializer leaves the five fields out. Airflow core fills them in from the config when it receives the serialized Dag.
• DagSpec has a new Queue field, because the Go SDK has no default_args to give every task a queue. A Go task runs on the queue that DagSpec.Queue names, unless its TaskSpec sets a Queue. A TriggerDagRun task does not take the Dag's Queue, because a Python worker runs the task.
• Each number in TriggerDagRunSpec.Conf used to be copied as a json.Number, and is now copied as an int64, a uint64 or a float64. A bundle sends the serialized Dag to Airflow as msgpack, which writes a json.Number as a string. An integer that fits in neither 64-bit integer type is rejected.

Review follow-ups

• Literal inputs. airflow.Inputs takes a task or airflow.Literal(value). A literal is serialized as the TS SDK writes it, {"name": "argN", "kind": "literal", "value": ...}, and adds no edge.

  extracted := dag.Task(extract)
  dag.Task(load, airflow.Inputs(extracted, airflow.Literal("uk")))

• Switch. A dag.Switch task is written with _can_skip_downstream, like a dag.If task.

• Edge labels. A labelled edge from a task that no group holds to a task group also labels the edges to the group's first tasks, as Python's extract >> Label("rows") >> transform does.

• Empty nested group. An edge out of an empty nested group falls back to its parent group, as Python's find_leaves does.

• Validation. @continuous requires MaxActiveRuns to be 1. EmailOnFailure and EmailOnRetry are documented as having no effect yet.

• Shared conformance cases. The recorded Go payload moved into scripts/ci/lang_sdk_serialization/test_dags.yaml as cases for branches, switches, edge labels, group options, TriggerDagRun, literal inputs and a six-field cron.

  • Each case lists the features it requires, and compare.py --supports lets an SDK run only the cases it can build. The TS SDK runs the cases that need no feature.
  • go_native.json is generated from these cases.

• Public API in the conformance test. The conformance test builds every Dag through the public API (airflow.Dag, dag.Task, the spec structs) from an external test package.

---

Was generative AI tooling used to co-author this PR?

• Yes - Claude Code

Generated-by: Claude Code (Opus 5.5) following the guidelines

[jason810496]

Thanks for the PR, LGTM overall.
The first 5 comments are my manual review, rest of the comments are reviews by claude.