Bind TaskFlow stub-task call arguments in the Go SDK runtime

[jason810496]

• related: split out of Support TaskFlow call syntax on stub tasks for the Lang SDK #69757 (now merged) to keep that PR Core + Task SDK only.
• Native temporal and UUID binding is split into follow-up: Bind Python temporal and UUID arguments to their native Go types #71809

Why

#69757 ships the Python-side contract: a @task.stub TaskFlow call is serialized as an ordered arg-binding spec and returned by ti_run as TIRunContext.arg_bindings. This PR makes the Go SDK actually consume it -- Go task functions receive the Dag file's literals and upstream XComs as typed parameters instead of hand-writing GetXCom calls with hard-coded upstream task ids:

@task.stub(queue="golang")
def transform(country: str, extracted: dict): ...


with DAG(...):
    transform("uk", extract())  # extract() is a normal Python @task

// The runtime binds "uk" onto country and pulls extract's XCom into extracted.
func transform(ctx sdk.TIRunContext, log *slog.Logger, country string, extracted map[string]any) error

Supported TaskFlow syntax

The example bundle's taskflow_binding_dag (go-sdk/dags/go_examples.py + go-sdk/example/bundle/taskflowbinding/) exercises the full surface end to end:

@task.stub(queue="golang")
def via_flat_args(
    name: str,
    count: int,
    ratio: float,
    enabled: bool,
    tags: list,
    config: dict,
    numbers: list,
    note: str | None = None,
): ...


@dag(dag_id="taskflow_binding_dag")
def taskflow_binding_dag():
    via_flat_args(
        "summary",
        3,
        2.5,
        True,  # positional scalar literals (str/int/float/bool)
        ["metrics", "hourly"],  # array literal
        config=make_config(),  # keyword arg: XCom from another @task.stub
        numbers=make_numbers(),  # XCom binding onto a typed array parameter
    )  # `note` unpassed: its None default is captured as from_default
    region = make_region()
    via_struct_no_tags(RegionCode=region, Threshold=0.75)  # one XCom fanned into several calls
    via_struct_arg_tag(region_code=region, threshold=0.75)  # literal + XCom mixed as kwargs
    via_struct_unmatched_arg(region_code=region)  # defaulted param left unpassed
    via_flat_map(config={"region": "eu-west-1", "count": 3})  # dict decoded whole into a struct
    via_struct_map(payload={"region": "eu-west-1", "count": 3})  # dict onto a struct's map field
    via_plain_map(labels={"team": "data", "tier": "gold"})  # dict into a plain Go map

On the Go side a task declares either flat positional data parameters or a single struct whose fields bind by name (mixing the two shapes is rejected at registration as too ambiguous):

// Flat parameters: bound in declaration order after the injectables
// (sdk.TIRunContext, *slog.Logger, context.Context, client interfaces).
// Arity or declared-type mismatches fail the task before its body runs.
func ViaFlatArgs(ctx sdk.TIRunContext, log *slog.Logger,
    name string, count int, ratio float64, enabled bool,
    tags []string, config Config, numbers []int, note *string) (any, error)

// Sole struct parameter: exported fields bind per call-argument name.
type ViaStructArgTagInput struct {
    Region    string  `arg:"region_code"` // explicit argument name via tag
    Threshold float64 `arg:"threshold"`   // untagged fields bind their verbatim Go field name
}
func ViaStructArgTag(ctx sdk.TIRunContext, log *slog.Logger, input ViaStructArgTagInput) (any, error)

Binding semantics, mirroring positional vs keyword calls:

• Flat parameters bind by position after injectables. Literals, XComs, and captured defaults
  fill every data parameter, with None decoded into nil-capable types.
• Sole structs bind by field name or arg: tag. Unmatched fields keep their zero value,
  from_default entries may go unclaimed, and explicitly passed unclaimed arguments fail.
• Whole-value structs decode one unclaimed argument into an untagged struct. Tagged structs do
  not use this fallback.
• Invalid bindings fail before task code runs. Missing or malformed specs, unsupported kinds,
  schema mismatches, and strict struct decode errors are reported instead of silently zero-filling.

How

New pkg/binding package:

• Analyze classifies parameters once at registration (injectables vs JSON-decodable data parameters, or the single name-bound struct), recursing through nested types so an undecodable one fails when the bundle is built rather than on every execution.
• Resolve binds the wire spec onto them. The wire union surfaces as a sealed sum type (binding.XComArg/binding.LiteralArg) whose variants and DataType vocabulary are defined in terms of the generated genmodels schema types, so the runtime types cannot drift from the wire model.

Rejected Alternative

• Ad hoc xcom:/xcom-key: struct tags were considered and dropped. A task that needs an extra XCom still asks the injected client explicitly.

---

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

• Yes, with help of Claude Code (Opus 5) following the guidelines

[jason810496]

Hi @ashb , I will merge this PR first to unblock the further works. Please feel free to raise any comment afterward and I will follow-up for them.