TS SDK: replace registerTask with Dag and registerDags

[jason810496]

• related: TypeScript SDK: native TypeScript Dag declaration #69288

Why

registerTask({ dagId, taskId }, handler) binds one handler to a Python stub Dag/task pair at a time and leaves no object that can carry native TypeScript Dag declaration later (#69288). This PR changes only the user-facing authoring interface so it can grow into TaskFlow and native Dag support without another breaking change.

How

Please noted that the spec at both Dag and Task level arguments (DagSpec, TaskSpec) and the task inputs are no-op, I intentionally placeholder them in this PR to prevent further breaking change.

• new Dag(dagId, spec?) takes a positional id plus an optional trailing spec object.
• dag.task(taskId, handler, options?) returns a TaskRef handle, with placeholder inputs and spec as named options (the further native Dag: { inputs: { extracted }, spec: { retries: 2 } }) so the future options need no new parameter breaking change. Unknown option keys are rejected, so a typo fails at import time instead of being silently ignored.
• DagSpec and TaskSpec are Record<string, never> while they are placeholders, so a field that would be silently dropped does not compile.
• dag.taskIds lists the attached task IDs, and registry.listDags() lists the Dags a bundle serves with their tasks, so a user can assert their handlers match the Python @task.stub names.
• A Dag instance retains its spec and each task's (taskId, handler, spec, inputs), so a future serialize() can produce the serialized Dag JSON for DagFileParsingResult.serialized_dags.
• registerTask, listRegisteredTasks, and TaskRegistration are replaced with a public DagRegistry — plus serveDags(registry) as the entrypoint. startCoordinator stays out of the public API.
• airflow-ts-pack warns instead of failing when a registered Dag has no tasks, matching airflow-go-pack. A Dag left out of the registry is simply absent from the bundle and its tasks are marked removed at runtime.

---

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

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

[uranusjr]

What’s the plan to do the external vs pure TS dags after this?

[jason810496]

What’s the plan to do the external vs pure TS dags after this?

Along with #71213, the user should set the is_mixed_language_dag for the mixed language Dag case, it will look somehow like new Dag("my_ts_mixed_lang", spec: {is_mixed_language_dag: True, schedule: "..."}).

Additionally, we can introduce a new ExternalDag or new MixedLangDag if we feel having explicitly different user interface is better. IMO, having the is_mixed_language_dag on DagSpec level plus user education should be enough for the first stage.

[jason810496]

Thanks kaxil for the review, I just resolved all the comments.

[uranusjr]

Maybe we can do something like this instead

constdag=newDag(...);// Add tasks...constregistry=newDagRegistry(dag);registry.serve();

Or, even more similar to Java SDK, new Server(registry).serve() so we can separate dag registration logic and network logic into different types.

[jason810496]

Thanks TP and Kaxil for the review!

I went with the following interface.

constdag1=newDag(...);// Add tasks...// Add Dags at constructor levelconstregistry=newDagRegistry(dag1,dag2);// Register Dag incrementally registry.register(dag3,dag4);awaitserveDags(registry);// should only be called once

The reason I used registerDags (I agreed serveDags is a better naming) in the first place is just to remove the "coordinator" term from the user interface and keep the entrypoint as simple as possible for the user.

I prefer to let serveDags(registry) do the final socket works.

For new Server(registry).serve() interface, IMO, we don't need to expose that much details to users. In the case of Python Dag world, we don't even have the user interface to register or serve the Dags at all.

[shivaam]

Just curious about the longer-term direction. Do we expect tasks to become callable, similar to Python TaskFlow?

const result = load(transform(extract()));

These calls would build the Dag and return typed task outputs rather than execute the handlers.

Is something like this planned for future native TypeScript Dags, or is the current TaskRef plus inputs design intended to be the final authoring interface?

One option would be for dag.task() to return a callable TaskDefinition<TInputs, TOutput>. Calling that definition would record the input bindings and return a TaskOutput.

[jason810496]

Just curious about the longer-term direction. Do we expect tasks to become callable, similar to Python TaskFlow?

Yes, we will support the TaskFlow-style syntax to define the task dependencies as mentioned in #69288 (comment).

One option would be for dag.task() to return a callable TaskDefinition<TInputs, TOutput>. Calling that definition would record the input bindings and return a TaskOutput.

This is what the TaskRef does today, though I haven't refine it into the "accurate" type like you mentioned, the TaskRef is more like a placeholder type right now.

[shivaam]

Thanks, looks good overall. Two potential nits.

[Copilot]

Pull request overview

Copilot reviewed 23 out of 23 changed files in this pull request and generated no new comments.

Suppressed comments (2)

ts-sdk/src/sdk/dag.ts:205

• #validateOptions claims unknown option keys are rejected, but it uses Object.keys(...) which ignores symbol keys. Callers can pass symbol-keyed options that will be silently accepted/ignored, contradicting the intended “typo fails at import time” behavior.

 #validateOptions(taskId: string, options: TaskOptions): void {
const value: unknown = options;
if (!isPlainRecord(value)) {
throw new Error(`options for Dag "${this.dagId}" task "${taskId}" must be an object`);
}

ts-sdk/src/cli/pack.ts:166

• readBundleManifest says it validates the untrusted metadata line “down to the task-id element”, but it does not validate the Dag ID key at all, and it only checks task IDs are non-empty strings. A bundle that prints the sentinel line itself (malicious or buggy) can slip invalid Dag/task IDs through pack-time validation and generate bad metadata.

 // The line is whatever the bundle printed and nothing downstream re-validates
// it, so check each Dag entry down to the task-id element.
for (const [dagId, dag] of Object.entries(manifest.dags)) {
if (dag == null || !isTaskIdList(dag.tasks)) {
throw new Error(

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.