You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "shivaam (via GitHub)" <gi...@apache.org> on 2026/06/27 07:34:09 UTC
[PR] Add TypeScript coordinator runtime [airflow]
shivaam opened a new pull request, #69079:
URL: https://github.com/apache/airflow/pull/69079
Adds TypeScript coordinator-mode execution on top of #67908.
This PR lets Airflow execute TypeScript task handlers through a bundled Node.js entrypoint:
- Adds `airflow.sdk.coordinators.typescript.TypescriptCoordinator`, modeled on the subprocess coordinator path.
- Adds the TypeScript coordinator runtime that speaks the supervisor protocol over `--comm` and `--logs` sockets.
- Adds TaskClient-backed runtime support for Variables, XCom get/set, and Connections.
- Adds `airflow-ts-pack`, which bundles a TypeScript entrypoint into `bundle.mjs` and writes sibling `airflow-metadata.yaml` with the SDK supervisor schema version.
- Documents Python stub Dag + TypeScript handler usage, bundle layout, and Airflow coordinator configuration.
This is intentionally still Python-stub-Dag mode. TypeScript Dag declaration is not introduced here; TypeScript registers handlers for Dag/task IDs declared by Python stub tasks.
Stacking note: this PR is stacked on #67908. Once #67908 lands, I will rebase this branch onto `main` so the diff only shows the coordinator/runtime changes.
Testing performed:
- TypeScript SDK package checks: format, lint, typecheck, test, build, format check.
- TypeScript SDK root prek hook: `ts-sdk-lint`.
- Python coordinator unit tests: `task-sdk/tests/task_sdk/coordinators/typescript/test_coordinator.py`.
- Built-package import smoke test for root, coordinator, and bundle subpath exports.
- Full daemon-mode E2E from this worktree without Breeze/Docker:
- Built SDK and `airflow-ts-pack` bundle.
- Bootstrapped isolated `AIRFLOW_HOME`.
- Ran Airflow `api-server`, `dag-processor`, and `scheduler`.
- Verified `hello_typescript`, `polyglot`, `connection_test`, and `api_showcase` task runs reached success.
- Verified signal handling Dags: one SIGTERM task attempt reached failed, and one retryable SIGTERM task retried then reached success with `ctx.signal.aborted` markers.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Codex (GPT-5)
Generated-by: Codex (GPT-5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3489981675
##########
ts-sdk/src/coordinator/comm-channel.ts:
##########
@@ -0,0 +1,265 @@
+/*!
+ * 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.
+ */
+
+// Comm socket client — length-prefixed msgpack frames over TCP.
+// Mirrors the Airflow supervisor's comm socket protocol.
+//
+// The channel is the sole reader on the socket. The task sends
+// requests and awaits id-correlated replies. The supervisor's only
+// unprompted frame is the greeting (StartupDetails /
+// DagFileParseRequest), pre-caught into the `greeting` promise that
+// `connect()` awaits — the protocol sends nothing else
+// supervisor-initiated (comms.py: "No messages are sent to task
+// process except in response to a request").
+
+import type { Socket } from "node:net";
+import { encodeRequest, encodeResponse, FrameReader, type Frame } from "./frames.js";
+import { connectTcp } from "./tcp-connect.js";
+import { Deferred } from "./deferred.js";
+import type { LogChannel } from "./log-channel.js";
+
+/** What `CommChannel.connect` resolves to: the live channel plus the
+ * supervisor's first frame (StartupDetails / DagFileParseRequest),
+ * already in hand so the caller never has to manage a "frame arrived
+ * with no consumer" window. */
+export interface CommConnection {
+ channel: CommChannel;
+ firstFrame: Frame;
+}
+
+export interface SendResponseOptions {
+ timeoutMs?: number;
+}
+
+export class CommChannel {
+ private readonly sock: Socket;
+ private readonly reader = new FrameReader();
+ private readonly logs: LogChannel | null;
+ private nextId = 0;
+ private pendingReplies = new Map<number, (frame: Frame) => void>();
+ private closed = false;
+ private closeError: Error | null = null;
+
+ // The greeting (first supervisor-initiated frame). The promise is
+ // its own buffer: arriving before `connect()` awaits is fine — it
+ // stays settled with the value, so there is no race to handle.
+ private readonly greeting = new Deferred<Frame>();
+
+ private constructor(sock: Socket, logs: LogChannel | null) {
+ this.sock = sock;
+ this.logs = logs;
+ // A `new Socket()` (from `connectTcp`) starts paused: it buffers
+ // inbound bytes and emits no `data` until a listener attaches
+ // and flips it to flowing. Attaching synchronously here — same
+ // tick as construction, before the event loop can deliver a
+ // read, and as the only reader — loses nothing, double-reads
+ // nothing.
+ sock.on("data", (chunk) => this.handleData(chunk));
+ sock.on("close", () => this.handleClose(null));
+ sock.on("error", (err) => this.handleClose(err));
+ }
+
+ /** Connect and wait for the supervisor's greeting; rejects if the
+ * socket dies before it arrives. */
+ static async connect(addr: string, logs: LogChannel | null = null): Promise<CommConnection> {
+ const sock = await connectTcp(addr);
+ const channel = new CommChannel(sock, logs);
+ const firstFrame = await channel.greeting.promise;
+ return { channel, firstFrame };
+ }
+
+ /** Send a request to the supervisor and await its matching response. */
+ async request(body: unknown): Promise<Frame> {
+ const id = this.nextId++;
+ const type = describeFrameType(body);
+ this.logs?.debug("Sending request", { id, type });
+ return new Promise<Frame>((resolve, reject) => {
+ if (this.closed) {
+ reject(this.closeError ?? new Error("Comm channel closed"));
+ return;
+ }
+ this.pendingReplies.set(id, (frame) => {
+ this.logs?.debug("Response received", {
+ id,
+ request_type: type,
+ response_type: describeFrameType(frame.body),
+ error: frame.error ?? null,
+ });
+ resolve(frame);
+ });
Review Comment:
Not sure if I missed anything. This has the potential to be stuck forever (until the process is killed) if the frame it expects got dropped or corrupt for whatever reason. This should probably add a timeout or something.
##########
ts-sdk/src/coordinator/comm-channel.ts:
##########
@@ -0,0 +1,265 @@
+/*!
+ * 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.
+ */
+
+// Comm socket client — length-prefixed msgpack frames over TCP.
+// Mirrors the Airflow supervisor's comm socket protocol.
+//
+// The channel is the sole reader on the socket. The task sends
+// requests and awaits id-correlated replies. The supervisor's only
+// unprompted frame is the greeting (StartupDetails /
+// DagFileParseRequest), pre-caught into the `greeting` promise that
+// `connect()` awaits — the protocol sends nothing else
+// supervisor-initiated (comms.py: "No messages are sent to task
+// process except in response to a request").
+
+import type { Socket } from "node:net";
+import { encodeRequest, encodeResponse, FrameReader, type Frame } from "./frames.js";
+import { connectTcp } from "./tcp-connect.js";
+import { Deferred } from "./deferred.js";
+import type { LogChannel } from "./log-channel.js";
+
+/** What `CommChannel.connect` resolves to: the live channel plus the
+ * supervisor's first frame (StartupDetails / DagFileParseRequest),
+ * already in hand so the caller never has to manage a "frame arrived
+ * with no consumer" window. */
+export interface CommConnection {
+ channel: CommChannel;
+ firstFrame: Frame;
+}
+
+export interface SendResponseOptions {
+ timeoutMs?: number;
+}
+
+export class CommChannel {
+ private readonly sock: Socket;
+ private readonly reader = new FrameReader();
+ private readonly logs: LogChannel | null;
+ private nextId = 0;
+ private pendingReplies = new Map<number, (frame: Frame) => void>();
+ private closed = false;
+ private closeError: Error | null = null;
+
+ // The greeting (first supervisor-initiated frame). The promise is
+ // its own buffer: arriving before `connect()` awaits is fine — it
+ // stays settled with the value, so there is no race to handle.
+ private readonly greeting = new Deferred<Frame>();
+
+ private constructor(sock: Socket, logs: LogChannel | null) {
+ this.sock = sock;
+ this.logs = logs;
+ // A `new Socket()` (from `connectTcp`) starts paused: it buffers
+ // inbound bytes and emits no `data` until a listener attaches
+ // and flips it to flowing. Attaching synchronously here — same
+ // tick as construction, before the event loop can deliver a
+ // read, and as the only reader — loses nothing, double-reads
+ // nothing.
+ sock.on("data", (chunk) => this.handleData(chunk));
+ sock.on("close", () => this.handleClose(null));
+ sock.on("error", (err) => this.handleClose(err));
+ }
+
+ /** Connect and wait for the supervisor's greeting; rejects if the
+ * socket dies before it arrives. */
+ static async connect(addr: string, logs: LogChannel | null = null): Promise<CommConnection> {
+ const sock = await connectTcp(addr);
+ const channel = new CommChannel(sock, logs);
+ const firstFrame = await channel.greeting.promise;
+ return { channel, firstFrame };
+ }
+
+ /** Send a request to the supervisor and await its matching response. */
+ async request(body: unknown): Promise<Frame> {
+ const id = this.nextId++;
+ const type = describeFrameType(body);
+ this.logs?.debug("Sending request", { id, type });
+ return new Promise<Frame>((resolve, reject) => {
+ if (this.closed) {
+ reject(this.closeError ?? new Error("Comm channel closed"));
+ return;
+ }
+ this.pendingReplies.set(id, (frame) => {
+ this.logs?.debug("Response received", {
+ id,
+ request_type: type,
+ response_type: describeFrameType(frame.body),
+ error: frame.error ?? null,
+ });
+ resolve(frame);
+ });
Review Comment:
Not sure if I missed anything. This seems to have the potential to be stuck forever (until the process is killed) if the frame it expects got dropped or corrupt for whatever reason. This should probably add a timeout or something.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3511031038
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -4515,6 +4515,18 @@
"format": "date-time",
"title": "Timestamp",
"type": "string"
+ },
+ "partition_key": {
+ "anyOf": [
+ {
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Partition Key"
Review Comment:
`partition_key` is added to `AssetEventDagRunReference` in this generated snapshot. Checked the actual Pydantic models (`_generated.py:713`, `taskinstance.py:309`, and the `AssetEventDagRunReferenceResult`
subclass in `comms.py:450`) - none of them declare this field, and I regenerated the snapshot locally to confirm: the field disappears entirely on a clean regen.
It's harmless at runtime (optional, never populated from the wire), but
`generate-supervisor-schemas-snapshot` (it keeps task-`sdk/src/airflow/sdk/execution_time/schema/schema.json` in sync with the actual Pydantic models) will go red on this PR since the diff also touches `comms.py/processor.py`, which trigger it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3511031038
##########
task-sdk/src/airflow/sdk/execution_time/schema/schema.json:
##########
@@ -4515,6 +4515,18 @@
"format": "date-time",
"title": "Timestamp",
"type": "string"
+ },
+ "partition_key": {
+ "anyOf": [
+ {
+ "type": "string"
+ },
+ {
+ "type": "null"
+ }
+ ],
+ "default": null,
+ "title": "Partition Key"
Review Comment:
`partition_key` is added to `AssetEventDagRunReference` in this generated snapshot. Checked the actual Pydantic models (`_generated.py:713`, `taskinstance.py:309`, and the `AssetEventDagRunReferenceResult`
subclass in `comms.py:450`) - none of them declare this field, and I regenerated the snapshot locally to confirm: the field disappears entirely on a clean regen.
It's harmless at runtime (optional, never populated from the wire), but
`generate-supervisor-schemas-snapshot` (it keeps task-`sdk/src/airflow/sdk/execution_time/schema/schema.json` in sync with the actual Pydantic models) will go red on this PR since the diff also touches `comms.py/processor.py`, which trigger it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3511172778
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -0,0 +1,120 @@
+/*!
+ * 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.
+ */
+
+// Log channel — newline-delimited JSON log records over TCP.
+//
+// The Airflow coordinator's `_bridge` reads lines from this socket,
+// parses each as JSON, and re-emits through structlog using the same
+// handler used for ordinary Python task logs
+// (`process_log_messages_from_subprocess`). Required fields are
+// `event`, `level`, `logger`, `timestamp`. Extra fields pass through
+// as structured log keys.
+//
+// The `logger` field becomes the bracketed name column in structlog's
+// ConsoleRenderer output (e.g. `[ts-sdk.runtime] Coordinator runtime
+// started`). Use hierarchical names — `ts-sdk.runtime`, `ts-sdk.comm`,
+// `ts-sdk.client` — so SDK-emitted lines are visibly distinct from
+// user task logs (which typically use the task's module name).
+
+import type { Socket } from "node:net";
+import { connectTcp } from "./tcp-connect.js";
+
+export type LogLevel = "debug" | "info" | "warning" | "error";
+
+export interface LogRecord {
+ event: string;
+ level: LogLevel;
+ logger: string;
+ timestamp: string;
+ [key: string]: unknown;
+}
+
+const DEFAULT_LOGGER_NAME = "ts-sdk";
+
+export class LogChannel {
+ private readonly sock: Socket;
+ private readonly name: string;
+ private readonly isRoot: boolean;
+
+ private constructor(sock: Socket, name: string, isRoot: boolean) {
+ this.sock = sock;
+ this.name = name;
+ this.isRoot = isRoot;
+ }
Review Comment:
```suggestion
private constructor(sock: Socket, name: string, isRoot: boolean) {
this.sock = sock;
this.name = name;
this.isRoot = isRoot;
sock.on("error", (err) => {
process.stderr.write(`[${this.name}] log socket error: ${err.message}\n`);
});
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3490050681
##########
ts-sdk/src/coordinator/runtime.ts:
##########
@@ -0,0 +1,399 @@
+/*!
+ * 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.
+ */
+
+// Coordinator runtime entrypoint.
+//
+// Invoked by Airflow's coordinator subprocess path:
+//
+// node my-bundle.mjs --comm=host:port --logs=host:port
+//
+// where `my-bundle.mjs` is a user-bundled Node script that imports
+// the SDK, calls `registerTask(...)` for each handler, then calls
+// `startCoordinator()`.
+//
+// Lifecycle:
+// 1. Parse --comm / --logs from argv
+// 2. Connect both TCP sockets
+// 3. Read the first frame from comm:
+// - DagFileParseRequest → respond with DagFileParsingResult, exit
+// - StartupDetails → run task, respond Succeed or Fail, exit
+//
+import { createCoordinatorClient } from "./client.js";
+import { CommChannel } from "./comm-channel.js";
+import { LogChannel } from "./log-channel.js";
+import {
+ asMsgFromSupervisor,
+ SUPERVISOR_API_VERSION,
+ type RuntimeDagFileParsingResult,
+ type RuntimeRetryTask,
+ type RuntimeSucceedTask,
+ type RuntimeTaskState,
+ type StartupDetails,
+} from "./protocol.js";
+import { getRegisteredTask, listRegisteredTasks } from "../sdk/registry.js";
+import type { TaskContext, TaskHandlerArgs } from "../sdk/task.js";
+import type { JsonValue } from "../sdk/client-types.js";
+
+export const COORDINATOR_SIGNAL_GRACE_PERIOD_MS = 30_000;
+export const COORDINATOR_RESPONSE_TIMEOUT_MS = 30_000;
+
+/** Options for `startCoordinator()`. */
+export interface StartCoordinatorOptions {
+ /** Comm socket address (host:port). Must be supplied together with `logsAddr`; otherwise parsed from argv. */
+ commAddr?: string;
+ /** Logs socket address (host:port). Must be supplied together with `commAddr`; otherwise parsed from argv. */
+ logsAddr?: string;
+ /** Source argv. Defaults to `process.argv`. */
+ argv?: readonly string[];
+}
+
+interface ParsedArgs {
+ commAddr: string;
+ logsAddr: string;
+}
+
+type SignalListener = NodeJS.SignalsListener;
+
+interface SignalEventSource {
+ on(signal: NodeJS.Signals, listener: SignalListener): SignalEventSource;
+ off(signal: NodeJS.Signals, listener: SignalListener): SignalEventSource;
+}
+
+interface CoordinatorCancellationOptions {
+ processEvents?: SignalEventSource;
+ forceExit?: (code: number) => never;
+}
+
+export interface CoordinatorCancellation {
+ readonly signal: AbortSignal;
+ isCancellationRequested(): boolean;
+ dispose(): void;
+}
+
+const COORDINATOR_TERMINATION_SIGNALS: readonly NodeJS.Signals[] = ["SIGTERM", "SIGINT"];
+const COORDINATOR_FORCE_EXIT_CODE = 1;
+
+export function parseArgs(argv: readonly string[]): ParsedArgs {
+ let commAddr: string | null = null;
+ let logsAddr: string | null = null;
+ for (const arg of argv) {
+ if (arg.startsWith("--comm=")) {
+ commAddr = arg.slice("--comm=".length);
+ } else if (arg.startsWith("--logs=")) {
+ logsAddr = arg.slice("--logs=".length);
+ }
+ }
+ if (!commAddr) throw new Error("Missing --comm=host:port");
+ if (!logsAddr) throw new Error("Missing --logs=host:port");
+ return { commAddr, logsAddr };
+}
+
+/** Start the coordinator runtime. Resolves when the subprocess has
+ * delivered its terminal frame and closed both sockets. */
+export async function startCoordinator(opts: StartCoordinatorOptions = {}): Promise<void> {
+ const argv = opts.argv ?? process.argv;
+ const parsed =
+ opts.commAddr && opts.logsAddr
+ ? { commAddr: opts.commAddr, logsAddr: opts.logsAddr }
+ : parseArgs(argv);
+
+ let logs: LogChannel | null = null;
+ let comm: CommChannel | null = null;
+ let cancellation: CoordinatorCancellation | null = null;
+
+ try {
+ // Connect log channel first so early failures are captured.
+ // Root logger is `ts-sdk`; subsystems use child names (`ts-sdk.runtime`,
+ // `ts-sdk.comm`, `ts-sdk.client`) so structlog's ConsoleRenderer prints
+ // them as a distinct `[name]` column on the supervisor side.
+ logs = await LogChannel.connect(parsed.logsAddr);
+ const runtimeLogs = logs.child("runtime");
+ const tasks = listRegisteredTasks();
+ runtimeLogs.info("Coordinator runtime started", {
+ registered_tasks: tasks,
+ count: tasks.length,
+ // Cadwyn schema version this SDK was generated against. Logged
+ // for operator visibility; not sent on the wire.
+ supervisor_api_version: SUPERVISOR_API_VERSION,
+ });
+
+ const connection = await CommChannel.connect(parsed.commAddr, logs.child("comm"));
+ comm = connection.channel;
+ const firstFrame = connection.firstFrame;
+ runtimeLogs.debug("Connected comm socket", { commAddr: parsed.commAddr });
+ cancellation = createCoordinatorCancellation(runtimeLogs);
+
+ const body = asMsgFromSupervisor(firstFrame.body);
+
+ if (body.type === "DagFileParseRequest") {
+ runtimeLogs.info("Received Dag parse request", {
+ file: body.file,
+ bundle_path: body.bundle_path,
+ });
+ const response = handleParse(body, runtimeLogs);
+ await sendSupervisorResponse(firstFrame.id, response, comm, runtimeLogs);
+ } else if (body.type === "StartupDetails") {
+ runtimeLogs.info("Received task startup details", {
+ dag_id: body.ti.dag_id,
+ task_id: body.ti.task_id,
+ run_id: body.ti.run_id,
+ try_number: body.ti.try_number,
+ map_index: body.ti.map_index ?? -1,
+ bundle: body.bundle_info.name,
+ });
+ const response = await handleTask(
+ body,
+ comm,
+ runtimeLogs,
+ logs.child("client"),
+ cancellation.signal,
+ cancellation.isCancellationRequested,
+ );
+ await sendSupervisorResponse(firstFrame.id, response, comm, runtimeLogs);
+ if (response.type === "SucceedTask") {
+ runtimeLogs.info("Task succeeded", { task_id: body.ti.task_id });
+ }
+ } else {
+ const errMsg = `First frame must be DagFileParseRequest or StartupDetails, got ${body.type}`;
+ runtimeLogs.error("Unexpected first frame", { type: body.type });
+ await sendSupervisorResponse(firstFrame.id, null, comm, runtimeLogs, {
+ error: "protocol_error",
+ detail: errMsg,
+ });
+ }
+ } finally {
+ cancellation?.dispose();
+ await comm?.close();
+ await logs?.close();
+ }
+}
+
+export function createCoordinatorCancellation(
+ logs: LogChannel | null,
+ opts: CoordinatorCancellationOptions = {},
+): CoordinatorCancellation {
+ const controller = new AbortController();
+ const processEvents = opts.processEvents ?? process;
+ const forceExit = opts.forceExit ?? ((code: number): never => process.exit(code));
+ let disposed = false;
+ let forceExitTimer: ReturnType<typeof setTimeout> | null = null;
+ const signalListeners = new Map<NodeJS.Signals, SignalListener>();
+
+ const handleSignal = (signal: NodeJS.Signals): void => {
+ if (disposed) return;
+ if (controller.signal.aborted) {
+ logs?.warning("Additional cancellation signal received", { signal });
+ return;
+ }
+
+ logs?.warning("Cancellation signal received", {
+ signal,
+ grace_period_ms: COORDINATOR_SIGNAL_GRACE_PERIOD_MS,
+ });
+ controller.abort(new Error(`Task cancelled by ${signal}`));
+ forceExitTimer = setTimeout(() => {
+ logs?.error("Cancellation grace period expired; forcing process exit", {
+ signal,
+ grace_period_ms: COORDINATOR_SIGNAL_GRACE_PERIOD_MS,
+ });
+ forceExit(COORDINATOR_FORCE_EXIT_CODE);
+ }, COORDINATOR_SIGNAL_GRACE_PERIOD_MS);
+ };
+
+ for (const signal of COORDINATOR_TERMINATION_SIGNALS) {
+ const listener: SignalListener = () => handleSignal(signal);
+ signalListeners.set(signal, listener);
+ processEvents.on(signal, listener);
+ }
+
+ return {
+ signal: controller.signal,
+ isCancellationRequested: () => controller.signal.aborted,
+ dispose: () => {
+ disposed = true;
+ for (const [signal, listener] of signalListeners) {
+ processEvents.off(signal, listener);
+ }
+ signalListeners.clear();
+ if (forceExitTimer !== null) {
+ clearTimeout(forceExitTimer);
+ forceExitTimer = null;
+ }
+ },
+ };
+}
+
+function handleParse(
+ request: { file: string; bundle_path: string },
+ logs: LogChannel,
+): RuntimeDagFileParsingResult {
+ // TypeScript-native Dag parsing is not yet supported.
+ // Respond with an empty result so the Python-stub-Dag workflow works.
+ logs.info("Parse-mode response (TS Dag parsing not yet supported)", {
+ registered_tasks: listRegisteredTasks(),
+ });
+ const response: RuntimeDagFileParsingResult = {
+ type: "DagFileParsingResult",
+ fileloc: request.file,
+ serialized_dags: [],
+ };
+ return response;
+}
+
+async function handleTask(
+ details: StartupDetails,
+ comm: CommChannel,
+ logs: LogChannel,
+ clientLogs: LogChannel,
+ signal: AbortSignal,
+ isCancellationRequested: () => boolean,
+): Promise<RuntimeSucceedTask | RuntimeRetryTask | RuntimeTaskState> {
+ const ti = details.ti;
+ const handler = getRegisteredTask(ti.dag_id, ti.task_id);
+
+ if (!handler) {
+ logs.warning("No handler registered for task", {
+ dag_id: ti.dag_id,
+ task_id: ti.task_id,
+ available: listRegisteredTasks(),
+ });
+ // A missing handler means this bundle cannot run the task, so retrying the
+ // same bundle/configuration mismatch would not help.
+ const response: RuntimeTaskState = {
+ type: "TaskState",
+ state: "removed",
+ end_date: new Date().toISOString(),
+ };
+ return response;
+ }
+
+ const ctx = buildContext(details, signal);
+ const client = createCoordinatorClient(comm, ctx, clientLogs);
+ const args: TaskHandlerArgs = { ctx, client };
+ // Startup-details fields already logged above (`Received task
+ // startup details`); this line just marks the handler-call boundary.
+ logs.debug("Dispatching to handler", { task_id: ctx.taskId });
+
+ try {
+ const result = await handler(args);
Review Comment:
If a cancel signal arrives after the result is actually sent, the handler’s result would have already been collected, but discard into the ether and never received by anyone. I would scope the abort signal to only apply if cancellable happened during the communication. Something like
```ts
try {
const result = await handler(args);
// Don't check isCancellationRequested here, just process.
} catch (err) {
// Check here instead.
if (isCancellationRequested()) {
return buildCancellationResponse(details, logs, ctx);
}
// Same...
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3511189546
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -0,0 +1,120 @@
+/*!
+ * 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.
+ */
+
+// Log channel — newline-delimited JSON log records over TCP.
+//
+// The Airflow coordinator's `_bridge` reads lines from this socket,
+// parses each as JSON, and re-emits through structlog using the same
+// handler used for ordinary Python task logs
+// (`process_log_messages_from_subprocess`). Required fields are
+// `event`, `level`, `logger`, `timestamp`. Extra fields pass through
+// as structured log keys.
+//
+// The `logger` field becomes the bracketed name column in structlog's
+// ConsoleRenderer output (e.g. `[ts-sdk.runtime] Coordinator runtime
+// started`). Use hierarchical names — `ts-sdk.runtime`, `ts-sdk.comm`,
+// `ts-sdk.client` — so SDK-emitted lines are visibly distinct from
+// user task logs (which typically use the task's module name).
+
+import type { Socket } from "node:net";
+import { connectTcp } from "./tcp-connect.js";
+
+export type LogLevel = "debug" | "info" | "warning" | "error";
+
+export interface LogRecord {
+ event: string;
+ level: LogLevel;
+ logger: string;
+ timestamp: string;
+ [key: string]: unknown;
+}
+
+const DEFAULT_LOGGER_NAME = "ts-sdk";
+
+export class LogChannel {
+ private readonly sock: Socket;
+ private readonly name: string;
+ private readonly isRoot: boolean;
+
+ private constructor(sock: Socket, name: string, isRoot: boolean) {
+ this.sock = sock;
+ this.name = name;
+ this.isRoot = isRoot;
+ }
Review Comment:
Nice catch by jason inline on the test file — CommChannel wires sock.on("close"/"error") in its constructor, LogChannel doesn't.
The only thing catching errors right now is a one-shot listener left over from
connectTcp()'s promise, so the first post-connect error gets silently swallowed and the second has nothing to catch it - silently losing task logs, or a hard crash.
Should be a quick fix, just mirror the comm channel's
handler here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #69079:
URL: https://github.com/apache/airflow/pull/69079#issuecomment-4928387350
> Nice! Here're all the follow-up issues that I saw during the review so far. I prefer to have real issues to keep track of them instead of having follow-up statements in the review comments that might be missed in the long run.
>
> * #69286 - TypeScript SDK: add a `ts pack` bundle build tool
> * #69287 - TypeScript SDK: Dag/task-aware multi-bundle selection in NodeCoordinator
> * #69288 - TypeScript SDK: native TypeScript Dag declaration (remove Python stub requirement)
> * #69289 - TypeScript SDK: coordinator schema-mapping tests (camelCase to snake_case)
> * #69290 - TypeScript SDK: Airflow-level integration tests for coordinator-mode tasks
> * #69291 - TypeScript SDK: refactor shared logic between NodeCoordinator and ExecutableCoordinator
> * #69292 - TypeScript SDK: log socket buffering and reconnect hardening
Thank you @jason810496 . I will get to working on these tasks one by one. Curious, is there a task that you think is higher priority out of these?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3490130454
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -0,0 +1,120 @@
+/*!
+ * 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.
+ */
+
+// Log channel — newline-delimited JSON log records over TCP.
+//
+// The Airflow coordinator's `_bridge` reads lines from this socket,
+// parses each as JSON, and re-emits through structlog using the same
+// handler used for ordinary Python task logs
+// (`process_log_messages_from_subprocess`). Required fields are
+// `event`, `level`, `logger`, `timestamp`. Extra fields pass through
+// as structured log keys.
+//
+// The `logger` field becomes the bracketed name column in structlog's
+// ConsoleRenderer output (e.g. `[ts-sdk.runtime] Coordinator runtime
+// started`). Use hierarchical names — `ts-sdk.runtime`, `ts-sdk.comm`,
+// `ts-sdk.client` — so SDK-emitted lines are visibly distinct from
+// user task logs (which typically use the task's module name).
+
+import type { Socket } from "node:net";
+import { connectTcp } from "./tcp-connect.js";
+
+export type LogLevel = "debug" | "info" | "warning" | "error";
+
+export interface LogRecord {
+ event: string;
+ level: LogLevel;
+ logger: string;
+ timestamp: string;
+ [key: string]: unknown;
+}
+
+const DEFAULT_LOGGER_NAME = "ts-sdk";
+
+export class LogChannel {
+ private readonly sock: Socket;
+ private readonly name: string;
+ private readonly isRoot: boolean;
+
+ private constructor(sock: Socket, name: string, isRoot: boolean) {
+ this.sock = sock;
+ this.name = name;
+ this.isRoot = isRoot;
+ }
+
+ static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME): Promise<LogChannel> {
+ return new LogChannel(await connectTcp(addr), name, true);
+ }
+
+ /** Create a sibling logger that shares the underlying socket but
+ * carries a hierarchical name (`parent.suffix`). Only the root
+ * owns the socket — children's `close()` is a no-op. */
+ child(suffix: string): LogChannel {
+ return new LogChannel(this.sock, `${this.name}.${suffix}`, false);
+ }
+
+ /** Name reported in the `logger` field of every record this
+ * instance emits. Useful for tests. */
+ get loggerName(): string {
+ return this.name;
+ }
+
+ send(
+ record: Omit<LogRecord, "timestamp" | "logger"> & {
+ timestamp?: string;
+ logger?: string;
+ },
+ ): void {
+ // Prepend the logger name to the event message so it surfaces in
+ // the Airflow UI task log view, which renders the message text but
+ // hides the `logger` JSON field. The field is still emitted for
+ // JSON consumers (grep/jq). Remove the prefix here if the
+ // supervisor-side renderer ever starts showing the logger column.
+ const line = JSON.stringify({
+ logger: this.name,
+ ...record,
+ event: `[${this.name}] ${record.event}`,
+ timestamp: record.timestamp ?? new Date().toISOString(),
+ });
+ this.sock.write(Buffer.from(line + "\n", "utf8"));
Review Comment:
Both Java and Go implement a buffer here to temporarily hold logs if `sock` is not connected. This is important because we might need to log before all sockets are fully connected, and those logs should still be sent after connection. (This can be hardened later.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "phanikumv (via GitHub)" <gi...@apache.org>.
phanikumv commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3511189546
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -0,0 +1,120 @@
+/*!
+ * 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.
+ */
+
+// Log channel — newline-delimited JSON log records over TCP.
+//
+// The Airflow coordinator's `_bridge` reads lines from this socket,
+// parses each as JSON, and re-emits through structlog using the same
+// handler used for ordinary Python task logs
+// (`process_log_messages_from_subprocess`). Required fields are
+// `event`, `level`, `logger`, `timestamp`. Extra fields pass through
+// as structured log keys.
+//
+// The `logger` field becomes the bracketed name column in structlog's
+// ConsoleRenderer output (e.g. `[ts-sdk.runtime] Coordinator runtime
+// started`). Use hierarchical names — `ts-sdk.runtime`, `ts-sdk.comm`,
+// `ts-sdk.client` — so SDK-emitted lines are visibly distinct from
+// user task logs (which typically use the task's module name).
+
+import type { Socket } from "node:net";
+import { connectTcp } from "./tcp-connect.js";
+
+export type LogLevel = "debug" | "info" | "warning" | "error";
+
+export interface LogRecord {
+ event: string;
+ level: LogLevel;
+ logger: string;
+ timestamp: string;
+ [key: string]: unknown;
+}
+
+const DEFAULT_LOGGER_NAME = "ts-sdk";
+
+export class LogChannel {
+ private readonly sock: Socket;
+ private readonly name: string;
+ private readonly isRoot: boolean;
+
+ private constructor(sock: Socket, name: string, isRoot: boolean) {
+ this.sock = sock;
+ this.name = name;
+ this.isRoot = isRoot;
+ }
Review Comment:
Nice catch by jason inline on the test file — CommChannel wires sock.on("close"/"error") in its constructor, LogChannel doesn't.
Right now the only thing catching errors is a one-shot listener left over from connectTcp()'s promise — so the first post-connect error slips by silently, and the second one has nothing left to catch it. Could mean quietly losing task logs, or a hard crash
Should be a quick fix, just mirror the comm channel's
handler here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #69079:
URL: https://github.com/apache/airflow/pull/69079#issuecomment-4849469329
> Need to fix the static check failure obviously but otherwise very nice work
Thank you. I have been spending the last couple of months understanding the Task SDK execution path, edge worker work, and coordinator model, so it’s really encouraging to see this coming together.
I’ve pushed a follow-up for the static check failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #69079:
URL: https://github.com/apache/airflow/pull/69079#issuecomment-4827863085
@uranusjr @jason810496 PR for TS sdk runtime.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3490056832
##########
ts-sdk/src/generated/supervisor.ts:
##########
Review Comment:
We should have a pre-commit hook to ensure this file is up-to-date.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3503185522
##########
ts-sdk/src/generated/supervisor.ts:
##########
@@ -0,0 +1,1871 @@
+/*!
+ * 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.
+ */
+
+// AUTO-GENERATED by scripts/generate-supervisor.mjs — do not edit by hand.
+// Source: ../task-sdk/src/airflow/sdk/execution_time/schema/schema.json
+//
+// Re-run with: pnpm run generate:supervisor
+
+export type Name = string;
+export type Id = number;
+export type Timestamp = string;
+export type Extra = {
+ [k: string]: JsonValue;
+} | null;
+/**
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "JsonValue".
+ */
+export type JsonValue = unknown;
+export type Name1 = string;
+export type Uri = string;
+export type Group = string;
+export type Extra1 = {
+ [k: string]: JsonValue;
+} | null;
+export type RunId = string;
+export type DagId = string;
+export type LogicalDate = string | null;
+export type StartDate = string;
+export type EndDate = string | null;
+export type State = string;
+export type DataIntervalStart = string | null;
+export type DataIntervalEnd = string | null;
+export type PartitionKey = string | null;
+export type CreatedDagruns = DagRunAssetReference[];
+export type SourceTaskId = string | null;
+export type SourceDagId = string | null;
+export type SourceRunId = string | null;
+export type SourceMapIndex = number | null;
+export type PartitionKey1 = string | null;
+export type AssetEvents = AssetEventResponse[];
+export type Type = "AssetEventsResult";
+export type Name2 = string | null;
+export type Uri1 = string | null;
+export type Type1 = string;
+export type Name3 = string;
+export type Uri2 = string;
+export type Name4 = string;
+export type Uri3 = string;
+export type Group1 = string;
+export type Extra3 = {
+ [k: string]: JsonValue;
+} | null;
+export type Type2 = "AssetResult";
+export type Type3 = "AssetStateStoreResult";
+export type Assets = AssetResult[];
+export type Type4 = "AssetsByAliasResult";
+export type State1 = "awaiting_input" | null;
+export type Timeout = string | null;
+export type NextMethod = string;
+export type NextKwargs = {
+ [k: string]: JsonValue;
+} | null;
+export type RenderedMapIndex = string | null;
+export type Type5 = "AwaitInputTask";
+export type Name5 = string;
+export type Version = string | null;
+export type VersionData = {
+ [k: string]: unknown;
+} | null;
+export type Name6 = string;
+export type Type6 = "ClearAssetStateStoreByName";
+export type Uri4 = string;
+export type Type7 = "ClearAssetStateStoreByUri";
+export type TiId = string;
+export type Type8 = "ClearTaskStateStore";
+export type ConnId = string;
+export type ConnType = string;
+export type Host = string | null;
+export type Schema = string | null;
+export type Login = string | null;
+export type Password = string | null;
+export type Port = number | null;
+export type Extra4 = string | null;
+export type Type9 = "ConnectionResult";
+export type TiId1 = string;
+/**
+ * @minItems 1
+ */
+export type Options = [string, ...string[]];
+export type Subject = string;
+export type Body = string | null;
+export type Defaults = string[] | null;
+export type Multiple = boolean | null;
+export type Params = {
+ [k: string]: unknown;
+} | null;
+export type AssignedUsers = HITLUser[] | null;
+export type Id1 = string;
+export type Name7 = string;
+export type Type10 = "CreateHITLDetailPayload";
+export type Count = number;
+export type Type11 = "DRCount";
+export type Filepath = string;
+export type BundleName = string;
+export type BundleVersion = string | null;
+export type Msg = string | null;
+export type DagId1 = string;
+export type RunId1 = string;
+export type DagId2 = string;
+export type RunId2 = string;
+export type LogicalDate1 = string | null;
+export type DataIntervalStart1 = string | null;
+export type DataIntervalEnd1 = string | null;
+export type RunAfter = string;
+export type StartDate1 = string | null;
+export type EndDate1 = string | null;
+export type ClearNumber = number;
+/**
+ * Class with DagRun types.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "DagRunType".
+ */
+export type DagRunType =
+ | "backfill"
+ | "scheduled"
+ | "manual"
+ | "operator_triggered"
+ | "asset_triggered"
+ | "asset_materialization";
+/**
+ * All possible states that a DagRun can be in.
+ *
+ * These are "shared" with TaskInstanceState in some parts of the code,
+ * so please ensure that their values always match the ones with the
+ * same name in TaskInstanceState.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "DagRunState".
+ */
+export type DagRunState = "queued" | "running" | "success" | "failed";
+export type Conf = {
+ [k: string]: unknown;
+} | null;
+export type TriggeringUserName = string | null;
+export type SourceTaskId1 = string | null;
+export type SourceDagId1 = string | null;
+export type SourceRunId1 = string | null;
+export type SourceMapIndex1 = number | null;
+export type SourceAliases = AssetAliasReferenceAssetEventDagRun[];
+export type Timestamp1 = string;
+export type PartitionKey2 = string | null;
+export type ConsumedAssetEvents = AssetEventDagRunReference[];
+export type PartitionKey3 = string | null;
+export type PartitionDate = string | null;
+export type Note = string | null;
+export type TeamName = string | null;
+export type Id2 = string;
+export type TaskId = string;
+export type DagId3 = string;
+export type RunId3 = string;
+export type TryNumber = number;
+export type DagVersionId = string;
+export type MapIndex = number;
+export type Hostname = string | null;
+export type ContextCarrier = {
+ [k: string]: unknown;
+} | null;
+export type Queue = string;
+export type IsFailureCallback = boolean | null;
+export type Type12 = "DagCallbackRequest";
+export type File = string;
+export type BundlePath = string;
+export type BundleName1 = string;
+export type Filepath1 = string;
+export type BundleName2 = string;
+export type BundleVersion1 = string | null;
+export type Msg1 = string | null;
+/**
+ * All possible states that a Task Instance can be in.
+ *
+ * Note that None is also allowed, so always use this in a type hint with Optional.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "TaskInstanceState".
+ */
+export type TaskInstanceState =
+ | "removed"
+ | "scheduled"
+ | "queued"
+ | "running"
+ | "success"
+ | "restarting"
+ | "failed"
+ | "up_for_retry"
+ | "up_for_reschedule"
+ | "upstream_failed"
+ | "skipped"
+ | "deferred"
+ | "awaiting_input";
+export type TaskRescheduleCount = number;
+export type MaxTries = number;
+export type Key = string;
+export type Value = string | null;
+export type Variables = VariableResponse[];
+export type ConnId1 = string;
+export type ConnType1 = string;
+export type Host1 = string | null;
+export type Schema1 = string | null;
+export type Login1 = string | null;
+export type Password1 = string | null;
+export type Port1 = number | null;
+export type Extra6 = string | null;
+export type Connections = ConnectionResponse[];
+export type NextMethod1 = string | null;
+export type NextKwargs1 =
+ | {
+ [k: string]: unknown;
+ }
+ | string
+ | null;
+export type XcomKeysToClear = string[];
+export type ShouldRetry = boolean;
+export type StartDate2 = string | null;
+export type Type13 = "TaskCallbackRequest";
+export type Filepath2 = string;
+export type BundleName3 = string;
+export type BundleVersion2 = string | null;
+export type Msg2 = string | null;
+export type EmailType = "failure" | "retry";
+export type Type14 = "EmailRequest";
+export type CallbackRequests = (DagCallbackRequest | TaskCallbackRequest | EmailRequest)[];
+export type Type15 = "DagFileParseRequest";
+export type Fileloc = string;
+export type LastLoaded = string | null;
+export type SerializedDags = LazyDeserializedDAG[];
+export type Warnings = unknown[] | null;
+export type ImportErrors = {
+ [k: string]: string;
+} | null;
+export type Type16 = "DagFileParsingResult";
+export type DagId4 = string;
+export type IsPaused = boolean;
+export type BundleName4 = string | null;
+export type BundleVersion3 = string | null;
+export type RelativeFileloc = string | null;
+export type Owners = string | null;
+export type Tags = string[];
+export type NextDagrun = string | null;
+export type Type17 = "DagResult";
+export type DagId5 = string;
+export type RunId4 = string;
+export type LogicalDate2 = string | null;
+export type DataIntervalStart2 = string | null;
+export type DataIntervalEnd2 = string | null;
+export type RunAfter1 = string;
+export type StartDate3 = string | null;
+export type EndDate2 = string | null;
+export type ClearNumber1 = number | null;
+export type Conf1 = {
+ [k: string]: unknown;
+} | null;
+export type TriggeringUserName1 = string | null;
+export type ConsumedAssetEvents1 = AssetEventDagRunReference[];
+export type PartitionKey4 = string | null;
+export type PartitionDate1 = string | null;
+export type Note1 = string | null;
+export type TeamName1 = string | null;
+export type Type18 = "DagRunResult";
+export type Type19 = "DagRunStateResult";
+export type State2 = "deferred" | null;
+export type Classpath = string;
+export type TriggerKwargs =
+ | {
+ [k: string]: JsonValue;
+ }
+ | string
+ | null;
+export type TriggerTimeout = string | null;
+export type Queue1 = string | null;
+export type NextMethod2 = string;
+export type NextKwargs2 = {
+ [k: string]: JsonValue;
+} | null;
+export type RenderedMapIndex1 = string | null;
+export type Type20 = "DeferTask";
+export type Name8 = string;
+export type Key1 = string;
+export type Type21 = "DeleteAssetStateStoreByName";
+export type Uri5 = string;
+export type Key2 = string;
+export type Type22 = "DeleteAssetStateStoreByUri";
+export type TiId2 = string;
+export type Key3 = string;
+export type Type23 = "DeleteTaskStateStore";
+export type Key4 = string;
+export type Type24 = "DeleteVariable";
+export type Key5 = string;
+export type DagId6 = string;
+export type RunId5 = string;
+export type TaskId1 = string;
+export type MapIndex1 = number | null;
+export type Type25 = "DeleteXCom";
+/**
+ * Error types used in the API client.
+ */
+export type ErrorType =
+ | "CONNECTION_NOT_FOUND"
+ | "VARIABLE_NOT_FOUND"
+ | "XCOM_NOT_FOUND"
+ | "ASSET_NOT_FOUND"
+ | "TASK_STORE_NOT_FOUND"
+ | "ASSET_STORE_NOT_FOUND"
+ | "DAGRUN_ALREADY_EXISTS"
+ | "PERMISSION_DENIED"
+ | "GENERIC_ERROR"
+ | "API_SERVER_ERROR";
+export type Detail = {
+ [k: string]: unknown;
+} | null;
+export type Type26 = "ErrorResponse";
+/**
+ * Error types used in the API client.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "ErrorType".
+ */
+export type ErrorType1 =
+ | "CONNECTION_NOT_FOUND"
+ | "VARIABLE_NOT_FOUND"
+ | "XCOM_NOT_FOUND"
+ | "ASSET_NOT_FOUND"
+ | "TASK_STORE_NOT_FOUND"
+ | "ASSET_STORE_NOT_FOUND"
+ | "DAGRUN_ALREADY_EXISTS"
+ | "PERMISSION_DENIED"
+ | "GENERIC_ERROR"
+ | "API_SERVER_ERROR";
+export type Name9 = string;
+export type Type27 = "GetAssetByName";
+export type Uri6 = string;
+export type Type28 = "GetAssetByUri";
+export type Name10 = string | null;
+export type Uri7 = string | null;
+export type After = string | null;
+export type Before = string | null;
+export type Limit = number | null;
+export type Ascending = boolean;
+export type Type29 = "GetAssetEventByAsset";
+export type AliasName = string;
+export type After1 = string | null;
+export type Before1 = string | null;
+export type Limit1 = number | null;
+export type Ascending1 = boolean;
+export type Type30 = "GetAssetEventByAssetAlias";
+export type Name11 = string;
+export type Key6 = string;
+export type Type31 = "GetAssetStateStoreByName";
+export type Uri8 = string;
+export type Key7 = string;
+export type Type32 = "GetAssetStateStoreByUri";
+export type AliasName1 = string;
+export type Type33 = "GetAssetsByAlias";
+export type ConnId2 = string;
+export type Type34 = "GetConnection";
+export type DagId7 = string;
+export type LogicalDates = string[] | null;
+export type RunIds = string[] | null;
+export type States = string[] | null;
+export type Type35 = "GetDRCount";
+export type DagId8 = string;
+export type Type36 = "GetDag";
+export type DagId9 = string;
+export type RunId6 = string;
+export type Type37 = "GetDagRun";
+export type DagId10 = string;
+export type RunId7 = string;
+export type Type38 = "GetDagRunState";
+export type TiId3 = string;
+export type Type39 = "GetHITLDetailResponse";
+export type TiId4 = string;
+export type Type40 = "GetPrevSuccessfulDagRun";
+export type DagId11 = string;
+export type LogicalDate3 = string;
+export type State3 = string | null;
+export type Type41 = "GetPreviousDagRun";
+export type DagId12 = string;
+export type TaskId2 = string;
+export type LogicalDate4 = string | null;
+export type MapIndex2 = number;
+export type Type42 = "GetPreviousTI";
+export type DagId13 = string;
+export type MapIndex3 = number | null;
+export type TaskIds = string[] | null;
+export type TaskGroupId = string | null;
+export type LogicalDates1 = string[] | null;
+export type RunIds1 = string[] | null;
+export type States1 = string[] | null;
+export type Type43 = "GetTICount";
+export type DagId14 = string;
+export type RunId8 = string;
+export type Type44 = "GetTaskBreadcrumbs";
+export type TiId5 = string;
+export type TryNumber1 = number;
+export type Type45 = "GetTaskRescheduleStartDate";
+export type TiId6 = string;
+export type Key8 = string;
+export type Type46 = "GetTaskStateStore";
+export type DagId15 = string;
+export type MapIndex4 = number | null;
+export type TaskIds1 = string[] | null;
+export type TaskGroupId1 = string | null;
+export type LogicalDates2 = string[] | null;
+export type RunIds2 = string[] | null;
+export type Type47 = "GetTaskStates";
+export type Key9 = string;
+export type Type48 = "GetVariable";
+export type Prefix = string | null;
+export type Limit2 = number;
+export type Offset = number;
+export type Type49 = "GetVariableKeys";
+export type Key10 = string;
+export type DagId16 = string;
+export type RunId9 = string;
+export type TaskId3 = string;
+export type MapIndex5 = number | null;
+export type IncludePriorDates = boolean;
+export type Type50 = "GetXCom";
+export type Key11 = string;
+export type DagId17 = string;
+export type RunId10 = string;
+export type TaskId4 = string;
+export type Type51 = "GetXComCount";
+export type Key12 = string;
+export type DagId18 = string;
+export type RunId11 = string;
+export type TaskId5 = string;
+export type Offset1 = number;
+export type Type52 = "GetXComSequenceItem";
+export type Key13 = string;
+export type DagId19 = string;
+export type RunId12 = string;
+export type TaskId6 = string;
+export type Start = number | null;
+export type Stop = number | null;
+export type Step = number | null;
+export type IncludePriorDates1 = boolean;
+export type Type53 = "GetXComSequenceSlice";
+export type TiId7 = string;
+/**
+ * @minItems 1
+ */
+export type Options1 = [string, ...string[]];
+export type Subject1 = string;
+export type Body1 = string | null;
+export type Defaults1 = string[] | null;
+export type Multiple1 = boolean | null;
+export type Params1 = {
+ [k: string]: unknown;
+} | null;
+export type AssignedUsers1 = HITLUser[] | null;
+export type Type54 = "HITLDetailRequestResult";
+export type InactiveAssets = AssetProfile[] | null;
+export type Type55 = "InactiveAssetsResult";
+export type Name12 = string | null;
+export type Type56 = "MaskSecret";
+export type Ok = boolean;
+export type Type57 = "OKResponse";
+export type DataIntervalStart3 = string | null;
+export type DataIntervalEnd3 = string | null;
+export type StartDate4 = string | null;
+export type EndDate3 = string | null;
+export type Type58 = "PrevSuccessfulDagRunResult";
+export type Type59 = "PreviousDagRunResult";
+export type TaskId7 = string;
+export type DagId20 = string;
+export type RunId13 = string;
+export type LogicalDate5 = string | null;
+export type StartDate5 = string | null;
+export type EndDate4 = string | null;
+export type State4 = string | null;
+export type TryNumber2 = number;
+export type MapIndex6 = number | null;
+export type Duration = number | null;
+export type Type60 = "PreviousTIResult";
+export type Key14 = string;
+export type Value1 = string | null;
+export type Description = string | null;
+export type Type61 = "PutVariable";
+export type State5 = "up_for_reschedule" | null;
+export type RescheduleDate = string;
+export type EndDate5 = string;
+export type Type62 = "RescheduleTask";
+export type Type63 = "ResendLoggingFD";
+export type State6 = "up_for_retry" | null;
+export type EndDate6 = string;
+export type RenderedMapIndex2 = string | null;
+export type RetryDelaySeconds = number | null;
+export type RetryReason = string | null;
+export type Type64 = "RetryTask";
+export type Type65 = "SentFDs";
+export type Fds = number[];
+export type Name13 = string;
+export type Key15 = string;
+export type Type66 = "SetAssetStateStoreByName";
+export type Uri9 = string;
+export type Key16 = string;
+export type Type67 = "SetAssetStateStoreByUri";
+export type Type68 = "SetRenderedFields";
+export type RenderedMapIndex3 = string;
+export type Type69 = "SetRenderedMapIndex";
+export type TiId8 = string;
+export type Key17 = string;
+export type ExpiresAt = string | null;
+export type Type70 = "SetTaskStateStore";
+export type Key18 = string;
+export type DagId21 = string;
+export type RunId14 = string;
+export type TaskId8 = string;
+export type MapIndex7 = number | null;
+export type DagResult1 = boolean;
+export type MappedLength = number | null;
+export type Type71 = "SetXCom";
+export type Tasks = (string | [unknown, unknown])[];
+export type Type72 = "SkipDownstreamTasks";
+export type DagRelPath = string;
+export type StartDate6 = string;
+export type SentryIntegration = string;
+export type Type73 = "StartupDetails";
+export type State7 = "success" | null;
+export type EndDate7 = string;
+export type TaskOutlets = AssetProfile[] | null;
+export type OutletEvents =
+ | {
+ [k: string]: unknown;
+ }[]
+ | null;
+export type RenderedMapIndex4 = string | null;
+export type Type74 = "SucceedTask";
+export type Count1 = number;
+export type Type75 = "TICount";
+export type Breadcrumbs = {
+ [k: string]: unknown;
+}[];
+export type Type76 = "TaskBreadcrumbsResult";
+export type StartDate7 = string | null;
+export type Type77 = "TaskRescheduleStartDate";
+export type State8 = "failed" | "skipped" | "removed";
+export type EndDate8 = string | null;
+export type Type78 = "TaskState";
+export type RenderedMapIndex5 = string | null;
+export type Type79 = "TaskStateStoreResult";
+export type Type80 = "TaskStatesResult";
+export type LogicalDate6 = string | null;
+export type RunAfter2 = string | null;
+export type Conf2 = {
+ [k: string]: unknown;
+} | null;
+export type ResetDagRun = boolean | null;
+export type PartitionKey5 = string | null;
+export type Note2 = string | null;
+export type DagId22 = string;
+export type DagRunId = string;
+export type Type81 = "TriggerDagRun";
+export type TiId9 = string;
+/**
+ * @minItems 1
+ */
+export type ChosenOptions = [string, ...string[]];
+export type ParamsInput = {
+ [k: string]: unknown;
+} | null;
+export type Type82 = "UpdateHITLDetail";
+export type TiId10 = string;
+export type Type83 = "ValidateInletsAndOutlets";
+export type Keys = string[];
+export type TotalEntries = number;
+export type Type84 = "VariableKeysResult";
+export type Key19 = string;
+export type Value2 = string | null;
+export type Type85 = "VariableResult";
+export type Len = number;
+export type Type86 = "XComCountResponse";
+export type Key20 = string;
+export type Type87 = "XComResult";
+export type Type88 = "XComSequenceIndexResult";
+export type Root = JsonValue[];
+export type Type89 = "XComSequenceSliceResult";
+
+export interface SupervisorWireSchema {}
+/**
+ * Schema for AssetAliasModel used in AssetEventDagRunReference.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetAliasReferenceAssetEventDagRun".
+ */
+export interface AssetAliasReferenceAssetEventDagRun {
+ name: Name;
+}
+/**
+ * Asset event schema with fields that are needed for Runtime.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetEventResponse".
+ */
+export interface AssetEventResponse {
+ id: Id;
+ timestamp: Timestamp;
+ extra?: Extra;
+ asset: AssetResponse;
+ created_dagruns: CreatedDagruns;
+ source_task_id?: SourceTaskId;
+ source_dag_id?: SourceDagId;
+ source_run_id?: SourceRunId;
+ source_map_index?: SourceMapIndex;
+ partition_key?: PartitionKey1;
+}
+/**
+ * Asset schema for responses with fields that are needed for Runtime.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetResponse".
+ */
+export interface AssetResponse {
+ name: Name1;
+ uri: Uri;
+ group: Group;
+ extra?: Extra1;
+}
+/**
+ * DagRun serializer for asset responses.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "DagRunAssetReference".
+ */
+export interface DagRunAssetReference {
+ run_id: RunId;
+ dag_id: DagId;
+ logical_date?: LogicalDate;
+ start_date: StartDate;
+ end_date?: EndDate;
+ state: State;
+ data_interval_start?: DataIntervalStart;
+ data_interval_end?: DataIntervalEnd;
+ partition_key?: PartitionKey;
+}
+/**
+ * Response to GetAssetEvent request.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetEventsResult".
+ */
+export interface AssetEventsResult {
+ asset_events: AssetEvents;
+ type?: Type;
+}
+/**
+ * Profile of an asset-like object.
+ *
+ * Asset will have name, uri defined, with type set to 'Asset'.
+ * AssetNameRef will have name defined, type set to 'AssetNameRef'.
+ * AssetUriRef will have uri defined, type set to 'AssetUriRef'.
+ * AssetAlias will have name defined, type set to 'AssetAlias'.
+ *
+ * Note that 'type' here is distinct from 'asset_type' the user declares on an
+ * Asset (or subclass). This field is for distinguishing between different
+ * asset-related types (Asset, AssetRef, or AssetAlias).
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetProfile".
+ */
+export interface AssetProfile {
+ name?: Name2;
+ uri?: Uri1;
+ type: Type1;
+}
+/**
+ * Schema for AssetModel used in AssetEventDagRunReference.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetReferenceAssetEventDagRun".
+ */
+export interface AssetReferenceAssetEventDagRun {
+ name: Name3;
+ uri: Uri2;
+ extra: Extra2;
+}
+export interface Extra2 {
+ [k: string]: JsonValue;
+}
+/**
+ * Response to ReadXCom request.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetResult".
+ */
+export interface AssetResult {
+ name: Name4;
+ uri: Uri3;
+ group: Group1;
+ extra?: Extra3;
+ type?: Type2;
+}
+/**
+ * Response to GetAssetStateStore; wraps the generated API response for supervisor to worker comms.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetStateStoreResult".
+ */
+export interface AssetStateStoreResult {
+ value: JsonValue;
+ type?: Type3;
+}
+/**
+ * Response to GetAssetsByAlias; list of concrete assets resolved from an alias.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AssetsByAliasResult".
+ */
+export interface AssetsByAliasResult {
+ assets: Assets;
+ type?: Type4;
+}
+/**
+ * Park a task instance awaiting human input (Human-in-the-loop), without a trigger.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "AwaitInputTask".
+ */
+export interface AwaitInputTask {
+ state?: State1;
+ timeout?: Timeout;
+ next_method: NextMethod;
+ next_kwargs?: NextKwargs;
+ rendered_map_index?: RenderedMapIndex;
+ type?: Type5;
+}
+/**
+ * Schema for telling task which bundle to run with.
+ *
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "BundleInfo".
+ */
+export interface BundleInfo {
+ name: Name5;
+ version?: Version;
+ version_data?: VersionData;
+}
+/**
+ * This interface was referenced by `SupervisorWireSchema`'s JSON-Schema
+ * via the `definition` "ClearAssetStateStoreByName".
+ */
+export interface ClearAssetStateStoreByName {
+ name: Name6;
+ type?: Type6;
+}
Review Comment:
Would it be possible to introduce a post step to normalize the type naming? Since there're quite a low of type annotated with number suffix, just wonder could we remove them (Though it should not impact the comms itself).
Not sure will this be too complicated or it's not possible to make it?
##########
ts-sdk/src/coordinator/client.ts:
##########
@@ -0,0 +1,184 @@
+/*!
+ * 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.
+ */
+
+import type { CommChannel } from "./comm-channel.js";
+import type { LogChannel } from "./log-channel.js";
+import type { TaskContext } from "../sdk/task.js";
+import type { TaskClient } from "../sdk/client.js";
+import type { ConnectionResult, GetXComOpts, SetXComOpts } from "../sdk/client-types.js";
+import { VariableNotFoundError } from "../sdk/client.js";
+import type {
+ GetVariable,
+ GetXCom,
+ SetXCom,
+ GetConnection,
+ ConnectionResult as WireConnectionResult,
+} from "./protocol.js";
+
+function toWireMapIndex(m: number | null | undefined): number | null {
+ return m == null || m < 0 ? null : m;
+}
+
+function fromWireConnection(body: WireConnectionResult): ConnectionResult {
+ return {
+ id: body.conn_id,
+ type: body.conn_type,
+ host: body.host ?? null,
+ schema: body.schema ?? null,
+ login: body.login ?? null,
+ password: body.password ?? null,
+ port: body.port ?? null,
+ extra: body.extra ?? null,
+ };
+}
+
+export function createCoordinatorClient(
+ comm: CommChannel,
+ ctx: TaskContext,
+ logs: LogChannel | null = null,
+): TaskClient {
+ async function rpc<T>(
+ op: string,
+ expectedType: string | null,
+ request: unknown,
+ extract: (body: Record<string, unknown> | null) => T,
+ ): Promise<T | null> {
+ logs?.debug(`${op} request`);
+ const frame = await comm.request(request);
+ const err = parseFrameError(frame);
+ if (err) {
+ if (isNotFound(err)) {
+ logs?.debug(`${op} not found`, { error: err.code });
+ return null;
+ }
+ logs?.warning(`${op} failed`, { error: err.code });
+ throw new Error(`${op} failed: ${err.code}`);
+ }
+ const body = frame.body as Record<string, unknown> | null;
+ if (expectedType !== null && body?.type !== expectedType) {
+ logs?.error(`${op} unexpected response type`, {
+ expected: expectedType,
+ got: body?.type ?? null,
+ });
+ throw new Error(`${op}: unexpected response type ${JSON.stringify(body?.type)}`);
+ }
+ logs?.debug(`${op} ok`);
+ return extract(body);
+ }
+
+ const client: TaskClient = {
+ // ---- Variables ----
+
+ async getVariable(key: string): Promise<string | null> {
+ const msg: GetVariable = { type: "GetVariable", key };
+ return rpc("GetVariable", "VariableResult", msg, (body) => (body!.value as string) ?? null);
+ },
+
+ async getVariableOrThrow(key: string): Promise<string> {
+ const value = await client.getVariable(key);
+ if (value == null) throw new VariableNotFoundError(key);
+ return value;
+ },
+
+ // ---- XCom ----
+
+ async getXCom<T = unknown>(opts: GetXComOpts): Promise<T | null> {
+ const msg: GetXCom = {
+ type: "GetXCom",
+ key: opts.key,
+ dag_id: opts.dagId ?? ctx.dagId,
+ task_id: opts.taskId ?? ctx.taskId,
+ run_id: opts.runId ?? ctx.runId,
+ map_index: toWireMapIndex(opts.mapIndex ?? ctx.mapIndex),
+ include_prior_dates: opts.includePriorDates ?? false,
+ };
+ return rpc("GetXCom", "XComResult", msg, (body) => (body!.value as T) ?? null);
+ },
+
+ async setXCom(opts: SetXComOpts): Promise<void> {
+ const msg: SetXCom = {
+ type: "SetXCom",
+ key: opts.key,
+ value: opts.value,
+ dag_id: opts.dagId ?? ctx.dagId,
+ task_id: opts.taskId ?? ctx.taskId,
+ run_id: opts.runId ?? ctx.runId,
+ map_index: toWireMapIndex(opts.mapIndex ?? ctx.mapIndex),
+ };
Review Comment:
It seems we will drop the explicit null or -1 from user.
##########
ts-sdk/src/coordinator/client.ts:
##########
@@ -0,0 +1,184 @@
+/*!
+ * 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.
+ */
+
+import type { CommChannel } from "./comm-channel.js";
+import type { LogChannel } from "./log-channel.js";
+import type { TaskContext } from "../sdk/task.js";
+import type { TaskClient } from "../sdk/client.js";
+import type { ConnectionResult, GetXComOpts, SetXComOpts } from "../sdk/client-types.js";
+import { VariableNotFoundError } from "../sdk/client.js";
+import type {
+ GetVariable,
+ GetXCom,
+ SetXCom,
+ GetConnection,
+ ConnectionResult as WireConnectionResult,
+} from "./protocol.js";
+
+function toWireMapIndex(m: number | null | undefined): number | null {
+ return m == null || m < 0 ? null : m;
+}
+
+function fromWireConnection(body: WireConnectionResult): ConnectionResult {
+ return {
+ id: body.conn_id,
+ type: body.conn_type,
+ host: body.host ?? null,
+ schema: body.schema ?? null,
+ login: body.login ?? null,
+ password: body.password ?? null,
+ port: body.port ?? null,
+ extra: body.extra ?? null,
+ };
+}
+
+export function createCoordinatorClient(
+ comm: CommChannel,
+ ctx: TaskContext,
+ logs: LogChannel | null = null,
+): TaskClient {
+ async function rpc<T>(
+ op: string,
+ expectedType: string | null,
+ request: unknown,
+ extract: (body: Record<string, unknown> | null) => T,
+ ): Promise<T | null> {
+ logs?.debug(`${op} request`);
+ const frame = await comm.request(request);
+ const err = parseFrameError(frame);
+ if (err) {
+ if (isNotFound(err)) {
+ logs?.debug(`${op} not found`, { error: err.code });
+ return null;
+ }
+ logs?.warning(`${op} failed`, { error: err.code });
+ throw new Error(`${op} failed: ${err.code}`);
+ }
+ const body = frame.body as Record<string, unknown> | null;
+ if (expectedType !== null && body?.type !== expectedType) {
+ logs?.error(`${op} unexpected response type`, {
+ expected: expectedType,
+ got: body?.type ?? null,
+ });
+ throw new Error(`${op}: unexpected response type ${JSON.stringify(body?.type)}`);
+ }
+ logs?.debug(`${op} ok`);
+ return extract(body);
+ }
+
+ const client: TaskClient = {
+ // ---- Variables ----
+
+ async getVariable(key: string): Promise<string | null> {
+ const msg: GetVariable = { type: "GetVariable", key };
+ return rpc("GetVariable", "VariableResult", msg, (body) => (body!.value as string) ?? null);
+ },
+
+ async getVariableOrThrow(key: string): Promise<string> {
+ const value = await client.getVariable(key);
+ if (value == null) throw new VariableNotFoundError(key);
+ return value;
+ },
+
+ // ---- XCom ----
+
+ async getXCom<T = unknown>(opts: GetXComOpts): Promise<T | null> {
+ const msg: GetXCom = {
+ type: "GetXCom",
+ key: opts.key,
+ dag_id: opts.dagId ?? ctx.dagId,
+ task_id: opts.taskId ?? ctx.taskId,
+ run_id: opts.runId ?? ctx.runId,
+ map_index: toWireMapIndex(opts.mapIndex ?? ctx.mapIndex),
+ include_prior_dates: opts.includePriorDates ?? false,
+ };
+ return rpc("GetXCom", "XComResult", msg, (body) => (body!.value as T) ?? null);
+ },
+
+ async setXCom(opts: SetXComOpts): Promise<void> {
+ const msg: SetXCom = {
+ type: "SetXCom",
+ key: opts.key,
+ value: opts.value,
+ dag_id: opts.dagId ?? ctx.dagId,
+ task_id: opts.taskId ?? ctx.taskId,
+ run_id: opts.runId ?? ctx.runId,
+ map_index: toWireMapIndex(opts.mapIndex ?? ctx.mapIndex),
+ };
+ await rpc("SetXCom", null, msg, () => undefined);
+ },
+
+ // ---- Connections ----
+
+ async getConnection(connId: string): Promise<ConnectionResult | null> {
+ const msg: GetConnection = { type: "GetConnection", conn_id: connId };
+ return rpc("GetConnection", "ConnectionResult", msg, (body) =>
+ fromWireConnection(body as unknown as WireConnectionResult),
+ );
+ },
+ };
+ return client;
+}
+
+// -------- Error handling (two functions) --------
+//
+// parseFrameError: extract a structured error from the frame (once).
+// isNotFound: decide if the error means "absent" (return null to caller)
+// or "failed" (throw).
+
+interface FrameError {
+ code: string;
+ statusCode?: number;
+}
+
+/** Extract the error code and optional HTTP status from a response frame.
+ * Returns `null` for non-error frames. */
+function parseFrameError(frame: { body: unknown; error?: unknown }): FrameError | null {
+ // Case 1: error field on the frame itself
+ if (frame.error != null) {
+ if (typeof frame.error === "string") return { code: frame.error };
+ if (typeof frame.error === "object") {
+ const e = frame.error as Record<string, unknown>;
+ if (typeof e.error === "string") {
+ const detail = e.detail as { status_code?: number } | undefined;
+ return { code: e.error, statusCode: detail?.status_code };
+ }
+ }
+ }
+ // Case 2: ErrorResponse in body
+ const body = frame.body as Record<string, unknown> | null;
+ if (body?.type === "ErrorResponse" && typeof body.error === "string") {
+ const detail = body.detail as { status_code?: number } | undefined;
+ return { code: body.error, statusCode: detail?.status_code };
+ }
+ return null;
+}
+
+// Exact supervisor ErrorType codes that mean "absent", not "failed".
+const NOT_FOUND_CODES = new Set(["VARIABLE_NOT_FOUND", "XCOM_NOT_FOUND", "CONNECTION_NOT_FOUND"]);
+
+/** Is this error a "not found" (caller should get null, not a throw)?
+ * Covers both dedicated NOT_FOUND codes and API_SERVER_ERROR with 404. */
+function isNotFound(err: FrameError): boolean {
+ if (NOT_FOUND_CODES.has(err.code)) return true;
+ // The supervisor wraps API server 404s as API_SERVER_ERROR with
+ // detail.status_code=404 (supervisor.py: WatchedSubprocess.handle_requests).
+ // Dag / Dag run lookups hit this path.
+ return err.code === "API_SERVER_ERROR" && err.statusCode === 404;
Review Comment:
Worthwhile to add TODO for this part as folding all the not found errors as null is not accurate (it's fine for now if only support XCom, Variable, Connection, but it doesn't seem fine in the future).
##########
ts-sdk/tests/coordinator/log-channel.test.ts:
##########
@@ -0,0 +1,133 @@
+/*!
+ * 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.
+ */
+
+import * as net from "node:net";
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import { LogChannel } from "../../src/coordinator/log-channel.js";
+
+interface Fixture {
+ server: net.Server;
+ port: number;
+ received: Buffer[];
+ sockClosed: Promise<void>;
+}
+
+async function makeServer(): Promise<Fixture> {
+ const received: Buffer[] = [];
+ let resolveSockClosed: () => void;
+ const sockClosed = new Promise<void>((r) => {
+ resolveSockClosed = r;
+ });
+ const server = net.createServer((sock) => {
+ sock.on("data", (chunk) => received.push(chunk));
+ sock.on("close", () => resolveSockClosed());
+ });
Review Comment:
We did protect the socket for comms with `sock.on("error"` but we miss for the log socket here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3490130454
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -0,0 +1,120 @@
+/*!
+ * 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.
+ */
+
+// Log channel — newline-delimited JSON log records over TCP.
+//
+// The Airflow coordinator's `_bridge` reads lines from this socket,
+// parses each as JSON, and re-emits through structlog using the same
+// handler used for ordinary Python task logs
+// (`process_log_messages_from_subprocess`). Required fields are
+// `event`, `level`, `logger`, `timestamp`. Extra fields pass through
+// as structured log keys.
+//
+// The `logger` field becomes the bracketed name column in structlog's
+// ConsoleRenderer output (e.g. `[ts-sdk.runtime] Coordinator runtime
+// started`). Use hierarchical names — `ts-sdk.runtime`, `ts-sdk.comm`,
+// `ts-sdk.client` — so SDK-emitted lines are visibly distinct from
+// user task logs (which typically use the task's module name).
+
+import type { Socket } from "node:net";
+import { connectTcp } from "./tcp-connect.js";
+
+export type LogLevel = "debug" | "info" | "warning" | "error";
+
+export interface LogRecord {
+ event: string;
+ level: LogLevel;
+ logger: string;
+ timestamp: string;
+ [key: string]: unknown;
+}
+
+const DEFAULT_LOGGER_NAME = "ts-sdk";
+
+export class LogChannel {
+ private readonly sock: Socket;
+ private readonly name: string;
+ private readonly isRoot: boolean;
+
+ private constructor(sock: Socket, name: string, isRoot: boolean) {
+ this.sock = sock;
+ this.name = name;
+ this.isRoot = isRoot;
+ }
+
+ static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME): Promise<LogChannel> {
+ return new LogChannel(await connectTcp(addr), name, true);
+ }
+
+ /** Create a sibling logger that shares the underlying socket but
+ * carries a hierarchical name (`parent.suffix`). Only the root
+ * owns the socket — children's `close()` is a no-op. */
+ child(suffix: string): LogChannel {
+ return new LogChannel(this.sock, `${this.name}.${suffix}`, false);
+ }
+
+ /** Name reported in the `logger` field of every record this
+ * instance emits. Useful for tests. */
+ get loggerName(): string {
+ return this.name;
+ }
+
+ send(
+ record: Omit<LogRecord, "timestamp" | "logger"> & {
+ timestamp?: string;
+ logger?: string;
+ },
+ ): void {
+ // Prepend the logger name to the event message so it surfaces in
+ // the Airflow UI task log view, which renders the message text but
+ // hides the `logger` JSON field. The field is still emitted for
+ // JSON consumers (grep/jq). Remove the prefix here if the
+ // supervisor-side renderer ever starts showing the logger column.
+ const line = JSON.stringify({
+ logger: this.name,
+ ...record,
+ event: `[${this.name}] ${record.event}`,
+ timestamp: record.timestamp ?? new Date().toISOString(),
+ });
+ this.sock.write(Buffer.from(line + "\n", "utf8"));
Review Comment:
Both Java and Go implement a buffer here so if `sock` is not connected. This is important because we might need to log before all sockets are fully connected, and those logs should still be sent after connection. (This can be hardened later.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add TypeScript coordinator runtime [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69079:
URL: https://github.com/apache/airflow/pull/69079#discussion_r3490056832
##########
ts-sdk/src/generated/supervisor.ts:
##########
Review Comment:
We should have a pre-commit hook to ensure this file is up-to-date. See Go SDK's approach and discussions in #68629.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org