kaxil commented on code in PR #74484: URL: https://github.com/apache/airflow/pull/74484#discussion_r4234430410
########## ts-sdk/src/sdk/operator.ts: ########## @@ -0,0 +1,139 @@ +/*! + * 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. + */ + +// Internal: what an operator the SDK runs itself implements, so the Dag, the bundle, the +// serializer and the runtime treat every operator alike and a new one is a single file. +// +// The coordinator types are imported as types only: the authoring layer must not load the +// coordinator at run time. + +import type { CoordinatorClient } from "../coordinator/client.js"; +import type { LogChannel } from "../coordinator/log-channel.js"; +import type { + RuntimeAwaitInputTask, + RuntimeDeferTask, + RuntimeRetryTask, + RuntimeSucceedTask, + RuntimeTaskState, + StartupDetails, +} from "../coordinator/protocol.js"; +import { brand, hasBrand } from "./brand.js"; +import type { JsonValue } from "./client-types.js"; +import type { Dag } from "./dag.js"; +import type { TaskContext } from "./task.js"; + +/** What an operator may end the task with. */ +export type OperatorOutcome = + | RuntimeSucceedTask + | RuntimeRetryTask + | RuntimeTaskState + | RuntimeDeferTask + | RuntimeAwaitInputTask; + +/** What the SDK hands an operator at run time. Operators only read it. */ +export interface OperatorContext { + readonly details: StartupDetails; + readonly dag: Dag; + readonly ctx: TaskContext; + readonly client: CoordinatorClient; + readonly logs: LogChannel; + /** Bind the task's declared inputs, applying the `withArgNames` renames of `argNames`. */ + resolveArgs(argNames?: ReadonlyMap<string, string>): Promise<object>; + /** Push `value` under `return_value` when it is given, then succeed. */ + succeed(value?: JsonValue): Promise<RuntimeSucceedTask>; + /** Fail, or hand the task back for retry when it has retries left. */ + fail(message: string): RuntimeRetryTask | RuntimeTaskState; + /** Skip `taskIds`, as `SkipMixin.skip` does. */ + skip(taskIds: readonly string[]): Promise<void>; + /** Park in `awaiting_input`. The next run calls `executeComplete`. */ + awaitInput(opts?: { + timeoutSeconds?: number; + kwargs?: Record<string, JsonValue>; + }): RuntimeAwaitInputTask; + /** + * Defer to a Python trigger. The next run calls `executeComplete`. Without a `queue`, the + * trigger gets the task's queue when triggerer queues are enabled. + */ + defer(opts: { + classpath: string; + kwargs: Record<string, JsonValue>; + timeoutSeconds?: number; + queue?: string | null; + }): RuntimeDeferTask; +} + +/** A Dag a task depends on, which the UI draws as an edge out of the task's own Dag. */ +export interface OperatorDagDependency { + readonly target: string; + /** Airflow's `dependency_type`, such as `"trigger"`. */ + readonly dependencyType: string; +} + +// Carries an operator's argument and result types without carrying a value, as `TaskRef` does +// for its return type. The argument is a parameter so that an operator of any argument type is +// an `Operator<never, unknown>`. +declare const OPERATOR_TYPES: unique symbol; + +/** + * An operator the SDK runs itself, passed to `dag.task` in place of a handler. + * + * `TArgs` is what the task's factory is called with, and `TResult` what it returns. + */ +export interface Operator<TArgs extends object | void = void, TResult = unknown> { + /** @internal Never set; see {@link OPERATOR_TYPES}. */ + readonly [OPERATOR_TYPES]?: (args: TArgs) => TResult; + /** `_operator_name` the UI shows: the Python operator this mirrors. */ + readonly operatorName: string; + /** Whether it may skip its downstream, as `SkipMixin` marks a Python operator. */ + readonly canSkipDownstream?: boolean; + /** An operator has no handler name to take a task id from, so `dag.task` requires one. */ + readonly requiresTaskId?: boolean; + /** How `dag.task` names the operator's tasks in its errors. Defaults to `operatorName`. */ + readonly label?: string; + /** A call that names the task, which `dag.task` shows when `requiresTaskId` is not met. */ + readonly taskIdExample?: string; + /** Whether the task's factory must be called without inputs. */ + readonly takesNoInputs?: boolean; + /** + * The option that bounds how long it waits, which `dag.task` names when it rejects an + * `executionTimeout`: the operator does not enforce that one. + */ + readonly executionTimeoutAlternative?: string; + /** + * Extra fields merged into the task's serialized record. `label` names the task, for an error + * about a value it holds. + */ + serialize?(label: string): Record<string, JsonValue>; + /** The Dags this task depends on. */ + getDagDependencies?(): readonly OperatorDagDependency[]; + execute(op: OperatorContext): Promise<OperatorOutcome>; Review Comment: `HITLTask` and `TriggerDagRunTask` are public types that extend `Operator`, so `execute`, `executeComplete`, `serialize` and the rest land in the published `.d.ts` and on the TypeDoc HITL page. Through `OperatorContext`, the public types now depend on `CoordinatorClient`, `LogChannel` and `StartupDetails`. Only the phantom member is tagged `@internal`. Tagging these members too (the TypeDoc config already sets `excludeInternal`) keeps the Operator contract free to change. `stripInternal` isn't the fix, since it would also drop the phantom member that inference needs. Related: TypeDoc warnings go from 12 to 18 with this PR, all dangling types (`TriggerDagRunSpec`, `TextArgs`, the `Operator*` types). ########## ts-sdk/src/hitl/operator.ts: ########## @@ -0,0 +1,279 @@ +/*! + * 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. + */ + +// Mirrors `HITLOperator.execute` / `execute_complete` and `ApprovalOperator.execute_complete` +// in the standard provider, on the Airflow 3.3+ path that parks the task in `awaiting_input`. +// +// The task runs twice, in two processes. The first run writes the request and parks; the +// second starts with `next_method` set and the response in `next_kwargs.event`. +// +// As `HITLOperator` does, the second run reads the options, the params and the reject policy from +// the task it is started for, and works out what a rejection skips from that task's Dag; nothing +// is carried over from the first run. A versioned Dag bundle starts both runs from the same +// version. With an unversioned bundle, a bundle deployed while the task waits is the one the +// response is read against, as it is for any task that runs after a deployment. + +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord, type Dag } from "../sdk/dag.js"; +import { getDagDownstreamTaskIds } from "../sdk/dag-graph.js"; +import { brandOperator, type OperatorContext, type OperatorOutcome } from "../sdk/operator.js"; +import { runInTaskScope } from "../sdk/task.js"; +import type { HITLResult, HITLTask, HITLTaskFields, HITLText, HITLUser } from "./spec.js"; + +export const APPROVE = "Approve"; +export const REJECT = "Reject"; + +/** Build the operator for a task whose options `spec.ts` has checked. */ +export function createHITLTask<TArgs extends object | void>( + fields: HITLTaskFields, +): HITLTask<TArgs> { + const isApproval = fields.kind === "approval"; + const task: HITLTask<never> = { + ...fields, + operatorName: isApproval ? "ApprovalOperator" : "HITLOperator", + // An approval skips what follows it on "Reject". + canSkipDownstream: isApproval, + requiresTaskId: true, + label: "human-in-the-loop", + taskIdExample: isApproval + ? 'dag.task("sign_off", approval({ subject: "..." }))' + : 'dag.task("choose", hitl({ subject: "...", options: ["..."] }))', + executionTimeoutAlternative: "responseTimeout", + execute: (op) => executeHITL(task, op), + executeComplete: (op, event) => resumeHITL(task, op, event), + }; + return brandOperator(task) as HITLTask<TArgs>; +} + +/** The first run: write the request, then park the task until a response arrives. */ +async function executeHITL(task: HITLTask<never>, op: OperatorContext): Promise<OperatorOutcome> { + const { details, ctx, client, logs } = op; + // The inputs are read here only, for the text: the resumed run must not depend on upstream + // XComs still being there. + const args = await op.resolveArgs(task.argNames); + const render = (text: HITLText<never, string | null>) => + runInTaskScope({ ctx, client }, () => renderText(text, args)); + const subject = await render(task.subject); + const body = task.body === undefined ? null : await render(task.body); + if (typeof subject !== "string" || subject.length === 0) { + return op.fail(`The subject of task "${ctx.taskId}" has to be a non-empty string`); + } + if (body !== null && body !== undefined && typeof body !== "string") { + return op.fail(`The body of task "${ctx.taskId}" has to be a string or null`); + } + + const options = task.options as [string, ...string[]]; + const defaults = task.defaults === undefined ? null : [...task.defaults]; + const assignedUsers = task.assignedUsers.map((user) => ({ id: user.id, name: user.name })); + await client.createHITLDetail({ + ti_id: details.ti.id, + options, + subject, + body: body ?? null, + defaults, + multiple: task.multiple, + params: serializeParams(task.params), + assigned_users: assignedUsers, + }); + + logs.info("Waiting for response", { task_id: ctx.taskId }); + return op.awaitInput({ timeoutSeconds: task.responseTimeout }); +} + +/** + * The params as `Param.serialize()` writes them. The source is always "task": + * `HITLOperator` drops the params that come from the Dag before it sends them. + */ +function serializeParams(params: HITLTask<never>["params"]): Record<string, JsonValue> { + return Object.fromEntries( + Object.entries(params).map(([name, { value, description, schema }]) => [ + name, + { value, description: description ?? null, schema: schema ?? {}, source: "task" }, + ]), + ); +} + +/** The second run: read the response against the task's options, then finish. */ +async function resumeHITL( + task: HITLTask<never>, + op: OperatorContext, + event: unknown, +): Promise<OperatorOutcome> { + const { dag, ctx, logs } = op; + // Throws, failing the task, when the answer cannot be read or is not allowed. + const result = readAnswer(event, task); + const responder = result.responded_by_user?.name ?? "the response timeout default"; + logs.info("Received response", { chosen_options: result.chosen_options, responder }); + + // `ApprovalOperator.execute_complete`: on "Reject", fail, or skip what follows. + if (task.kind === "approval" && result.chosen_options[0] === REJECT) { + if (task.failOnReject) return op.fail(`Rejected by ${responder}`); + const skipped = getSkipTargets(dag, ctx.taskId, task.ignoreDownstreamTriggerRules); + if (skipped.length > 0) { + // `SkipMixin.skip` ends the task as soon as it skips, so Python pushes no response then. + logs.info("Skipping downstream tasks", { task_ids: skipped }); + await op.skip(skipped); + return op.succeed(); + } + logs.info("No downstream tasks; nothing to do."); + } else if (task.kind === "approval") { + logs.info("Approved. Proceeding with downstream tasks..."); + } + return op.succeed(result as unknown as JsonValue); Review Comment: The `as unknown as JsonValue` is only needed because `HITLResult` and `HITLUser` are interfaces, which TypeScript won't match against `JsonValue`'s index signature. Declaring them as `type` aliases lets `op.succeed(result)` typecheck without the cast. ########## ts-sdk/src/hitl/operator.ts: ########## @@ -0,0 +1,279 @@ +/*! + * 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. + */ + +// Mirrors `HITLOperator.execute` / `execute_complete` and `ApprovalOperator.execute_complete` +// in the standard provider, on the Airflow 3.3+ path that parks the task in `awaiting_input`. +// +// The task runs twice, in two processes. The first run writes the request and parks; the +// second starts with `next_method` set and the response in `next_kwargs.event`. +// +// As `HITLOperator` does, the second run reads the options, the params and the reject policy from +// the task it is started for, and works out what a rejection skips from that task's Dag; nothing +// is carried over from the first run. A versioned Dag bundle starts both runs from the same +// version. With an unversioned bundle, a bundle deployed while the task waits is the one the +// response is read against, as it is for any task that runs after a deployment. + +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord, type Dag } from "../sdk/dag.js"; +import { getDagDownstreamTaskIds } from "../sdk/dag-graph.js"; +import { brandOperator, type OperatorContext, type OperatorOutcome } from "../sdk/operator.js"; +import { runInTaskScope } from "../sdk/task.js"; +import type { HITLResult, HITLTask, HITLTaskFields, HITLText, HITLUser } from "./spec.js"; + +export const APPROVE = "Approve"; +export const REJECT = "Reject"; + +/** Build the operator for a task whose options `spec.ts` has checked. */ +export function createHITLTask<TArgs extends object | void>( + fields: HITLTaskFields, +): HITLTask<TArgs> { + const isApproval = fields.kind === "approval"; + const task: HITLTask<never> = { + ...fields, + operatorName: isApproval ? "ApprovalOperator" : "HITLOperator", + // An approval skips what follows it on "Reject". + canSkipDownstream: isApproval, + requiresTaskId: true, + label: "human-in-the-loop", + taskIdExample: isApproval + ? 'dag.task("sign_off", approval({ subject: "..." }))' + : 'dag.task("choose", hitl({ subject: "...", options: ["..."] }))', + executionTimeoutAlternative: "responseTimeout", + execute: (op) => executeHITL(task, op), + executeComplete: (op, event) => resumeHITL(task, op, event), + }; + return brandOperator(task) as HITLTask<TArgs>; +} + +/** The first run: write the request, then park the task until a response arrives. */ +async function executeHITL(task: HITLTask<never>, op: OperatorContext): Promise<OperatorOutcome> { + const { details, ctx, client, logs } = op; + // The inputs are read here only, for the text: the resumed run must not depend on upstream + // XComs still being there. + const args = await op.resolveArgs(task.argNames); + const render = (text: HITLText<never, string | null>) => + runInTaskScope({ ctx, client }, () => renderText(text, args)); + const subject = await render(task.subject); + const body = task.body === undefined ? null : await render(task.body); + if (typeof subject !== "string" || subject.length === 0) { + return op.fail(`The subject of task "${ctx.taskId}" has to be a non-empty string`); + } + if (body !== null && body !== undefined && typeof body !== "string") { Review Comment: Nothing tests this guard: replacing it with `if (false)` keeps the suite green. The same holds for the non-timeout branch of `failureReason` (only `error_type: "timeout"` is exercised), the extra-keys check in `assignedUsers` (`spec.ts:405`) and the non-object spec check (`spec.ts:262`). A body-returns-a-number row next to the subject cases, a `{ error: "boom", error_type: "unknown" }` resume row, and two `rejects` rows in `spec.test.ts` would cover them. ########## ts-sdk/src/hitl/operator.ts: ########## @@ -0,0 +1,279 @@ +/*! + * 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. + */ + +// Mirrors `HITLOperator.execute` / `execute_complete` and `ApprovalOperator.execute_complete` +// in the standard provider, on the Airflow 3.3+ path that parks the task in `awaiting_input`. +// +// The task runs twice, in two processes. The first run writes the request and parks; the +// second starts with `next_method` set and the response in `next_kwargs.event`. +// +// As `HITLOperator` does, the second run reads the options, the params and the reject policy from +// the task it is started for, and works out what a rejection skips from that task's Dag; nothing +// is carried over from the first run. A versioned Dag bundle starts both runs from the same +// version. With an unversioned bundle, a bundle deployed while the task waits is the one the +// response is read against, as it is for any task that runs after a deployment. + +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord, type Dag } from "../sdk/dag.js"; +import { getDagDownstreamTaskIds } from "../sdk/dag-graph.js"; +import { brandOperator, type OperatorContext, type OperatorOutcome } from "../sdk/operator.js"; +import { runInTaskScope } from "../sdk/task.js"; +import type { HITLResult, HITLTask, HITLTaskFields, HITLText, HITLUser } from "./spec.js"; + +export const APPROVE = "Approve"; +export const REJECT = "Reject"; + +/** Build the operator for a task whose options `spec.ts` has checked. */ +export function createHITLTask<TArgs extends object | void>( + fields: HITLTaskFields, +): HITLTask<TArgs> { + const isApproval = fields.kind === "approval"; + const task: HITLTask<never> = { + ...fields, + operatorName: isApproval ? "ApprovalOperator" : "HITLOperator", + // An approval skips what follows it on "Reject". + canSkipDownstream: isApproval, + requiresTaskId: true, + label: "human-in-the-loop", + taskIdExample: isApproval + ? 'dag.task("sign_off", approval({ subject: "..." }))' + : 'dag.task("choose", hitl({ subject: "...", options: ["..."] }))', + executionTimeoutAlternative: "responseTimeout", + execute: (op) => executeHITL(task, op), + executeComplete: (op, event) => resumeHITL(task, op, event), + }; + return brandOperator(task) as HITLTask<TArgs>; +} + +/** The first run: write the request, then park the task until a response arrives. */ +async function executeHITL(task: HITLTask<never>, op: OperatorContext): Promise<OperatorOutcome> { + const { details, ctx, client, logs } = op; + // The inputs are read here only, for the text: the resumed run must not depend on upstream + // XComs still being there. + const args = await op.resolveArgs(task.argNames); + const render = (text: HITLText<never, string | null>) => + runInTaskScope({ ctx, client }, () => renderText(text, args)); + const subject = await render(task.subject); + const body = task.body === undefined ? null : await render(task.body); + if (typeof subject !== "string" || subject.length === 0) { + return op.fail(`The subject of task "${ctx.taskId}" has to be a non-empty string`); + } + if (body !== null && body !== undefined && typeof body !== "string") { + return op.fail(`The body of task "${ctx.taskId}" has to be a string or null`); + } + + const options = task.options as [string, ...string[]]; + const defaults = task.defaults === undefined ? null : [...task.defaults]; + const assignedUsers = task.assignedUsers.map((user) => ({ id: user.id, name: user.name })); + await client.createHITLDetail({ + ti_id: details.ti.id, + options, + subject, + body: body ?? null, + defaults, + multiple: task.multiple, + params: serializeParams(task.params), + assigned_users: assignedUsers, + }); + + logs.info("Waiting for response", { task_id: ctx.taskId }); + return op.awaitInput({ timeoutSeconds: task.responseTimeout }); +} + +/** + * The params as `Param.serialize()` writes them. The source is always "task": + * `HITLOperator` drops the params that come from the Dag before it sends them. + */ +function serializeParams(params: HITLTask<never>["params"]): Record<string, JsonValue> { + return Object.fromEntries( + Object.entries(params).map(([name, { value, description, schema }]) => [ + name, + { value, description: description ?? null, schema: schema ?? {}, source: "task" }, + ]), + ); +} + +/** The second run: read the response against the task's options, then finish. */ +async function resumeHITL( + task: HITLTask<never>, + op: OperatorContext, + event: unknown, +): Promise<OperatorOutcome> { + const { dag, ctx, logs } = op; + // Throws, failing the task, when the answer cannot be read or is not allowed. + const result = readAnswer(event, task); + const responder = result.responded_by_user?.name ?? "the response timeout default"; + logs.info("Received response", { chosen_options: result.chosen_options, responder }); + + // `ApprovalOperator.execute_complete`: on "Reject", fail, or skip what follows. + if (task.kind === "approval" && result.chosen_options[0] === REJECT) { + if (task.failOnReject) return op.fail(`Rejected by ${responder}`); + const skipped = getSkipTargets(dag, ctx.taskId, task.ignoreDownstreamTriggerRules); + if (skipped.length > 0) { + // `SkipMixin.skip` ends the task as soon as it skips, so Python pushes no response then. + logs.info("Skipping downstream tasks", { task_ids: skipped }); + await op.skip(skipped); + return op.succeed(); + } + logs.info("No downstream tasks; nothing to do."); + } else if (task.kind === "approval") { + logs.info("Approved. Proceeding with downstream tasks..."); + } + return op.succeed(result as unknown as JsonValue); +} + +/** + * Turn the answer Airflow resumed the task with into the result downstream tasks get, or throw + * the reason the task fails. It only reads and checks; `resumeHITL` acts on what it returns. + * + * For an approval, for example: + * - admin picked "Approve": the result `["Approve"]`, answered by admin. + * - the timeout passed, with `defaults: "Approve"`: the result `["Approve"]`, answered by no + * one, `timedout: true`. + * - the timeout passed, with no defaults: throws "Response timed out: ...". + * - `["Maybe"]`, an option it does not offer: throws, as `HITLOperator.validate_chosen_options` does. + * - a params input that does not match the task's params: throws, as + * `HITLOperator.validate_params_input` does. + */ +function readAnswer(event: unknown, task: HITLTask<never>): HITLResult { + if (!isPlainRecord(event)) { + throw new Error(`Task resumed with an event it cannot read: ${JSON.stringify(event)}`); + } + + // 1. Airflow resumed the task with a failure: a timeout with no defaults, or another error. + if ("error" in event) throw new Error(failureReason(event)); + + // 2. Read the answer. A field Airflow sent in a shape it cannot be read in is undefined. + const chosen = isStringArray(event["chosen_options"]) ? event["chosen_options"] : undefined; + const respondedAt = decodeDatetime(event["responded_at"]); + const respondedBy = readUser(event["responded_by_user"]); + if (chosen === undefined || respondedAt === undefined || respondedBy === undefined) { + throw new Error(`Task resumed with a response it cannot read: ${JSON.stringify(event)}`); + } + + // 3. Check the choice is one the task allows. + const notAllowed = checkChoice(chosen, task); + if (notAllowed !== undefined) throw new Error(notAllowed); + + // 4. Check the params input is the form the task asked for. + const paramsInput = (isPlainRecord(event["params_input"]) ? event["params_input"] : {}) as Record< + string, + JsonValue + >; + const mismatch = checkParamsInput(paramsInput, Object.keys(task.params)); + if (mismatch !== undefined) throw new Error(mismatch); + + // 5. The result downstream tasks receive. + return { + chosen_options: chosen, + params_input: paramsInput, + responded_at: respondedAt.toISOString(), + responded_by_user: respondedBy, + timedout: event["timedout"] === true, + }; +} + +/** `HITLOperator.process_trigger_event_error`: why a failed resume fails the task. */ +function failureReason(event: Record<string, unknown>): string { + const reason = String(event["error"]); + return event["error_type"] === "timeout" ? `Response timed out: ${reason}` : reason; +} + +/** Who answered: `null` when the timeout defaults did, undefined when it cannot be read. */ +function readUser(value: unknown): HITLUser | null | undefined { + if (value === null || value === undefined) return null; + return isUser(value) ? { id: value.id, name: value.name } : undefined; +} + +/** + * `HITLOperator.validate_chosen_options`: why the choice is not allowed, or undefined. How many + * options were chosen is not checked again: Airflow checked that against the request when the + * answer was given. + */ +function checkChoice(chosen: string[], task: HITLTask<never>): string | undefined { + const unknown = chosen.filter((option) => !task.options.includes(option)); + if (unknown.length > 0) { + return `Responses ${JSON.stringify(unknown)} not in ${JSON.stringify(task.options)}`; + } + return undefined; +} + +/** + * `HITLOperator.validate_params_input`: why the params input is not the form the task asked for, + * or undefined. An input for no param, or a task with no params, is not checked. + */ +function checkParamsInput( + given: Record<string, JsonValue>, + declared: readonly string[], +): string | undefined { + const received = Object.keys(given); + if (declared.length === 0 || received.length === 0) return undefined; + if (received.length === declared.length && declared.every((key) => Object.hasOwn(given, key))) { + return undefined; + } + return `params_input ${JSON.stringify(received)} does not match params ${JSON.stringify(declared)}`; +} + +/** The tasks a rejection skips: the direct downstream, or every task downstream. */ +function getSkipTargets(dag: Dag, taskId: string, ignoreDownstreamTriggerRules: boolean): string[] { + const downstream = getDagDownstreamTaskIds(dag); Review Comment: If the `approval()` comes from a second copy of the SDK (a shared package with its own nested copy), `isOperator` accepts it through the `Symbol.for` brand and Approve works. On Reject, though, this calls that copy's `getDagDownstreamTaskIds`, whose static accessors read `dag.#tasks` on the other copy's `Dag` and throw `TypeError`. `Dag`, `TaskHandler` and `Bundle` already reject a foreign copy up front with `DUPLICATE_COPY_HINT`. Could `dag.task` do the same for operators, so this fails at parse time rather than on the first Reject? ########## ts-sdk/src/hitl/operator.ts: ########## @@ -0,0 +1,279 @@ +/*! + * 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. + */ + +// Mirrors `HITLOperator.execute` / `execute_complete` and `ApprovalOperator.execute_complete` +// in the standard provider, on the Airflow 3.3+ path that parks the task in `awaiting_input`. +// +// The task runs twice, in two processes. The first run writes the request and parks; the +// second starts with `next_method` set and the response in `next_kwargs.event`. +// +// As `HITLOperator` does, the second run reads the options, the params and the reject policy from +// the task it is started for, and works out what a rejection skips from that task's Dag; nothing +// is carried over from the first run. A versioned Dag bundle starts both runs from the same +// version. With an unversioned bundle, a bundle deployed while the task waits is the one the +// response is read against, as it is for any task that runs after a deployment. + +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord, type Dag } from "../sdk/dag.js"; +import { getDagDownstreamTaskIds } from "../sdk/dag-graph.js"; +import { brandOperator, type OperatorContext, type OperatorOutcome } from "../sdk/operator.js"; +import { runInTaskScope } from "../sdk/task.js"; +import type { HITLResult, HITLTask, HITLTaskFields, HITLText, HITLUser } from "./spec.js"; + +export const APPROVE = "Approve"; +export const REJECT = "Reject"; + +/** Build the operator for a task whose options `spec.ts` has checked. */ +export function createHITLTask<TArgs extends object | void>( + fields: HITLTaskFields, +): HITLTask<TArgs> { + const isApproval = fields.kind === "approval"; + const task: HITLTask<never> = { + ...fields, + operatorName: isApproval ? "ApprovalOperator" : "HITLOperator", + // An approval skips what follows it on "Reject". + canSkipDownstream: isApproval, + requiresTaskId: true, + label: "human-in-the-loop", + taskIdExample: isApproval + ? 'dag.task("sign_off", approval({ subject: "..." }))' + : 'dag.task("choose", hitl({ subject: "...", options: ["..."] }))', + executionTimeoutAlternative: "responseTimeout", + execute: (op) => executeHITL(task, op), + executeComplete: (op, event) => resumeHITL(task, op, event), + }; + return brandOperator(task) as HITLTask<TArgs>; +} + +/** The first run: write the request, then park the task until a response arrives. */ +async function executeHITL(task: HITLTask<never>, op: OperatorContext): Promise<OperatorOutcome> { + const { details, ctx, client, logs } = op; + // The inputs are read here only, for the text: the resumed run must not depend on upstream + // XComs still being there. + const args = await op.resolveArgs(task.argNames); + const render = (text: HITLText<never, string | null>) => + runInTaskScope({ ctx, client }, () => renderText(text, args)); + const subject = await render(task.subject); + const body = task.body === undefined ? null : await render(task.body); + if (typeof subject !== "string" || subject.length === 0) { + return op.fail(`The subject of task "${ctx.taskId}" has to be a non-empty string`); + } + if (body !== null && body !== undefined && typeof body !== "string") { + return op.fail(`The body of task "${ctx.taskId}" has to be a string or null`); + } + + const options = task.options as [string, ...string[]]; + const defaults = task.defaults === undefined ? null : [...task.defaults]; + const assignedUsers = task.assignedUsers.map((user) => ({ id: user.id, name: user.name })); + await client.createHITLDetail({ + ti_id: details.ti.id, + options, + subject, + body: body ?? null, + defaults, + multiple: task.multiple, + params: serializeParams(task.params), + assigned_users: assignedUsers, + }); + + logs.info("Waiting for response", { task_id: ctx.taskId }); + return op.awaitInput({ timeoutSeconds: task.responseTimeout }); +} + +/** + * The params as `Param.serialize()` writes them. The source is always "task": + * `HITLOperator` drops the params that come from the Dag before it sends them. + */ +function serializeParams(params: HITLTask<never>["params"]): Record<string, JsonValue> { + return Object.fromEntries( + Object.entries(params).map(([name, { value, description, schema }]) => [ + name, + { value, description: description ?? null, schema: schema ?? {}, source: "task" }, + ]), + ); +} + +/** The second run: read the response against the task's options, then finish. */ +async function resumeHITL( + task: HITLTask<never>, + op: OperatorContext, + event: unknown, +): Promise<OperatorOutcome> { + const { dag, ctx, logs } = op; + // Throws, failing the task, when the answer cannot be read or is not allowed. + const result = readAnswer(event, task); + const responder = result.responded_by_user?.name ?? "the response timeout default"; + logs.info("Received response", { chosen_options: result.chosen_options, responder }); + + // `ApprovalOperator.execute_complete`: on "Reject", fail, or skip what follows. + if (task.kind === "approval" && result.chosen_options[0] === REJECT) { + if (task.failOnReject) return op.fail(`Rejected by ${responder}`); + const skipped = getSkipTargets(dag, ctx.taskId, task.ignoreDownstreamTriggerRules); + if (skipped.length > 0) { + // `SkipMixin.skip` ends the task as soon as it skips, so Python pushes no response then. + logs.info("Skipping downstream tasks", { task_ids: skipped }); + await op.skip(skipped); + return op.succeed(); + } + logs.info("No downstream tasks; nothing to do."); + } else if (task.kind === "approval") { + logs.info("Approved. Proceeding with downstream tasks..."); + } + return op.succeed(result as unknown as JsonValue); +} + +/** + * Turn the answer Airflow resumed the task with into the result downstream tasks get, or throw + * the reason the task fails. It only reads and checks; `resumeHITL` acts on what it returns. + * + * For an approval, for example: + * - admin picked "Approve": the result `["Approve"]`, answered by admin. + * - the timeout passed, with `defaults: "Approve"`: the result `["Approve"]`, answered by no + * one, `timedout: true`. + * - the timeout passed, with no defaults: throws "Response timed out: ...". + * - `["Maybe"]`, an option it does not offer: throws, as `HITLOperator.validate_chosen_options` does. + * - a params input that does not match the task's params: throws, as + * `HITLOperator.validate_params_input` does. + */ +function readAnswer(event: unknown, task: HITLTask<never>): HITLResult { + if (!isPlainRecord(event)) { + throw new Error(`Task resumed with an event it cannot read: ${JSON.stringify(event)}`); + } + + // 1. Airflow resumed the task with a failure: a timeout with no defaults, or another error. + if ("error" in event) throw new Error(failureReason(event)); + + // 2. Read the answer. A field Airflow sent in a shape it cannot be read in is undefined. + const chosen = isStringArray(event["chosen_options"]) ? event["chosen_options"] : undefined; + const respondedAt = decodeDatetime(event["responded_at"]); + const respondedBy = readUser(event["responded_by_user"]); + if (chosen === undefined || respondedAt === undefined || respondedBy === undefined) { + throw new Error(`Task resumed with a response it cannot read: ${JSON.stringify(event)}`); + } + + // 3. Check the choice is one the task allows. + const notAllowed = checkChoice(chosen, task); + if (notAllowed !== undefined) throw new Error(notAllowed); + + // 4. Check the params input is the form the task asked for. + const paramsInput = (isPlainRecord(event["params_input"]) ? event["params_input"] : {}) as Record< + string, + JsonValue + >; + const mismatch = checkParamsInput(paramsInput, Object.keys(task.params)); Review Comment: This only checks that the keys match. Python's `validate_params_input` also assigns each value into `self.params`, and `ParamsDict.__setitem__` runs `Param.resolve`, so a value outside the schema fails the task there. Nothing else checks the values: the PATCH route stores `params_input` as given. With the README's `replicas` example (maximum 10), a REST answer of `{"replicas": 50}` passes here and goes downstream, while the same Dag in Python fails. Could this validate each value against its `schema`? If a validator dependency isn't wanted, the README and this docstring should say values aren't checked, and stop calling it a port of `validate_params_input`. ########## ts-sdk/README.md: ########## @@ -175,6 +175,94 @@ bundle the coordinator's `task_handler_bundle_name` names. See the guide for all See [`example/`](https://github.com/apache/airflow/tree/main/ts-sdk/example) for a working project that serves both a Dag declared in TypeScript and the tasks of two Python Dags from one bundle. +## Human-in-the-loop tasks + +A Dag declared in TypeScript can wait for a person, as `HITLOperator` and `ApprovalOperator` do in +Python. `approval(...)` and `hitl(...)` come from the `apache-airflow-ts-sdk/hitl` subpath and are +passed to `dag.task` in place of a handler. Each is named after its Python operator without +"Operator", and each option is the Python keyword argument in camelCase: + +```ts +import { Bundle, Dag } from "apache-airflow-ts-sdk"; +import { approval } from "apache-airflow-ts-sdk/hitl"; + +const dag = new Dag("release", { queue: "typescript" }); +const report = dag.task("build_report", async () => ({ version: "1.4" }))(); + +const signOff = dag.task( + "sign_off", + approval({ + subject: ({ report }: { report: { version: string } }) => `Ship ${report.version}?`, + responseTimeout: 4 * 60 * 60, // seconds + defaults: "Reject", // the answer given when the timeout passes + }), +)({ report }); +signOff.before(dag.task("publish", async () => {})()); + +await new Bundle(dag).serve(); +``` + +The task's first run writes the request and parks the task in `awaiting_input`; its Node process +exits, so no worker slot is held while it waits. Once someone responds on the Required Actions page, +or the timeout passes, the task resumes in a new process. It does not write the request again or run +`subject` or `body` again. + +- `subject` and `body` (Markdown) are a string, or a function of the task's inputs, run once. They + may read different inputs, and the task takes both. +- `responseTimeout` is in whole **seconds**. With `defaults`, the defaults are applied when it passes; + without, the task fails, and its retries apply. A HITL task does not enforce `executionTimeout`, so + `dag.task` rejects it and asks for `responseTimeout` instead. +- `assignedUsers: [{ id, name }]` limits who can respond, by the auth manager's user id. +- The task returns the response, which downstream tasks receive and XCom stores under + `return_value`: `{ chosen_options, params_input, responded_at, responded_by_user, timedout }`. + These are the keys of Python's `HITLOperator` result plus `timedout`; `responded_at` is an + ISO-8601 string, and `responded_by_user` is `null` when the timeout defaults were applied. + +**On "Reject", an approval succeeds and skips the tasks directly downstream by default**, as +`ApprovalOperator` does. The run then shows as successful, with the guarded tasks skipped. Set +`failOnReject: true` to fail the approval instead, or `ignoreDownstreamTriggerRules: true` to skip +every task downstream. `hitl(...)` offers any `options`, with `multiple` for several, and never +skips or fails on what is chosen, even an option named "Reject". + +A failed resume retries, and the retry writes a fresh request, so `approval({ failOnReject: true })` +with `retries` asks the reviewer again on each attempt. + +When the task resumes, it reads its options, its `params` and its reject options from the Dag it +runs from, as `HITLOperator` does. A versioned Dag bundle resumes it from the same version it +started on; with an unversioned bundle, a bundle deployed while the request waits is what the +response is read against. Clearing the task starts a new attempt, which writes a fresh request. + +### Forms + +`params` adds form fields to the request, as `HITLOperator`'s `params` does. Each field is a `value` +that pre-fills it, an optional `description`, and an optional JSON Schema that the form renders: + +```ts +import { hitl } from "apache-airflow-ts-sdk/hitl"; + +const rollout = dag.task( + "size_rollout", + hitl({ + subject: "Approve the rollout?", + options: ["Approve", "Reject"], Review Comment: With exactly `["Approve", "Reject"]` as options, the UI renders approve/reject buttons (`isApprovalTask` in `ui/src/utils/hitl.ts`), but `hitl()` doesn't act on Reject. Someone copying this example gets a rollout that proceeds after a reviewer clicks Reject. `example/src/hitl.ts:123` has the same setup (its `rollout` task runs either way). Should both use `approval({ ..., params })` instead? ########## ts-sdk/src/hitl/spec.ts: ########## @@ -0,0 +1,446 @@ +/*! + * 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. + */ + +// Human-in-the-loop tasks: `hitl(...)` and `approval(...)`, which mirror `HITLOperator` and +// `ApprovalOperator` in the standard provider. A factory is the Python operator's name without +// "Operator", and an option is the Python keyword argument in camelCase. + +import { getArgNames } from "../sdk/arg-names.js"; +import type { JsonValue } from "../sdk/client-types.js"; +import { isPlainRecord } from "../sdk/dag.js"; +import type { Operator } from "../sdk/operator.js"; +import { toPlainJson } from "../sdk/plain-json.js"; +import { APPROVE, createHITLTask, REJECT } from "./operator.js"; + +/** A user allowed to respond, as Airflow's auth manager identifies them. */ +export interface HITLUser { + readonly id: string; + readonly name: string; +} + +/** Text shown to the responder: fixed, or computed from the task's inputs on its first run. */ +export type HITLText<TArgs, TText extends string | null = string> = + TText | ((args: TArgs) => TText | Promise<TText>); + +/** One form field on the Required Actions page, as Python's `Param` serializes. */ +export interface HITLParam { + /** + * Pre-filled value. When `responseTimeout` passes and `defaults` is set, the scheduler records + * each param's `value` as the response, so it is also the value used on timeout. + */ + readonly value: JsonValue; + readonly description?: string; + /** + * JSON Schema. The form renders `type`, `enum`, `format`, `items`, `minimum`, `maximum`, + * `minLength`, `maxLength` and `values_display`. + */ + readonly schema?: Readonly<Record<string, JsonValue>>; +} + +/** + * The options both HITL tasks take. + * + * `TSubjectArgs` and `TBodyArgs` are the inputs `subject` and `body` read when they are + * functions; the task takes both. + */ +interface HITLBaseSpec<TSubjectArgs extends object | void, TBodyArgs extends object | void> { + /** Headline of the request. */ + readonly subject: HITLText<TSubjectArgs>; + /** Markdown shown below the headline. */ + readonly body?: HITLText<TBodyArgs, string | null>; + /** Users allowed to respond. Anyone who can act on the task may respond when unset. */ + readonly assignedUsers?: readonly HITLUser[]; + /** Form fields the responder fills in, by name. Their answers are the response's `params_input`. */ + readonly params?: Readonly<Record<string, HITLParam>>; + /** + * Whole seconds to wait for a response, since TypeScript has no timedelta. Waits indefinitely + * when unset. Use this rather than `executionTimeout`, which a HITL task does not enforce. + */ + readonly responseTimeout?: number; +} + +/** The `HITLOperator` options, named as TypeScript spells them. */ +export interface HITLSpec< + TSubjectArgs extends object | void = void, + TBodyArgs extends object | void = TSubjectArgs, +> extends HITLBaseSpec<TSubjectArgs, TBodyArgs> { + /** What the responder chooses from. */ + readonly options: readonly string[]; + /** Options chosen for the responder when `responseTimeout` passes without a response. */ + readonly defaults?: readonly string[]; + /** Whether more than one option may be chosen. */ + readonly multiple?: boolean; +} + +/** The `ApprovalOperator` options, named as TypeScript spells them. */ +export interface ApprovalSpec< + TSubjectArgs extends object | void = void, + TBodyArgs extends object | void = TSubjectArgs, +> extends HITLBaseSpec<TSubjectArgs, TBodyArgs> { + /** The answer given for the responder when `responseTimeout` passes without a response. */ + readonly defaults?: "Approve" | "Reject"; + /** + * On "Reject", skip every task downstream rather than only the tasks directly downstream. + * Defaults to false. + */ + readonly ignoreDownstreamTriggerRules?: boolean; + /** + * On "Reject", fail this task rather than skip what follows it. Defaults to false, so a + * rejection succeeds. A retry asks the reviewer again, so `retries` with this option repeats + * the request on each attempt. + */ + readonly failOnReject?: boolean; +} + +/** + * The response a HITL task pushes to XCom and passes downstream. + * + * The keys are those of Python's `HITLTriggerEventSuccessPayload` plus `timedout`, and Review Comment: `timedout` is already a key of `HITLTriggerEventSuccessPayload` (`providers/standard/.../triggers/hitl.py:55`), so "plus `timedout`" isn't right. The keys are exactly the payload's. ########## ts-sdk/tests/coordinator/integration.test.ts: ########## @@ -1305,8 +1307,757 @@ describe("coordinator runtime integration", () => { retry_reason: "Trigger timeout", }); expect( - result.logRecords.some((r) => String(r["event"]).includes("Trigger failed:\nTraceback")), + result.logRecords.some((r) => + String(r["event"]).includes("Task could not be resumed:\nTraceback"), + ), + ).toBe(true); + }); + }); + describe("human-in-the-loop", () => { + const CREATED = { body: { type: "HITLDetailRequestResult", ti_id: "ti-1" } }; + // What `handle_event_submit` stores: what the task parked with, plus the response as `event`, + // with `responded_at` in serde's datetime encoding. + const RESPONDED_AT = { + __classname__: "pendulum.datetime.DateTime", + __version__: 2, + __data__: { + timestamp: 1791462615.123456, + tz: { + __classname__: "builtins.tuple", + __version__: 1, + __data__: ["UTC", "pendulum.tz.timezone.Timezone", 1, true], + }, + }, + }; + + function parkedWith(taskId: string, tiContext: Record<string, unknown> = {}) { + return driveSupervisor( + makeStartupDetails(taskId, "test_dag", "r1", tiContext), + (msgType, body) => { + if (msgType === "CreateHITLDetailPayload") return CREATED; + if (msgType === "GetXCom") { + return { body: { type: "XComResult", key: body["key"], value: { version: "1.4" } } }; + } + return null; + }, + ); + } + + /** The task parks with empty `next_kwargs`, so a resume carries only the response. */ + function resumedWith(taskId: string, event: unknown, tiContext: Record<string, unknown> = {}) { + return driveSupervisor( + makeStartupDetails(taskId, "test_dag", "r1", { + next_method: "execute_complete", + next_kwargs: { event }, + ...tiContext, + }), + ); + } + + function answer( + chosen: string[], + user: unknown = { id: "1", name: "Ada" }, + paramsInput: Record<string, unknown> = {}, + ) { + return { + chosen_options: chosen, + params_input: paramsInput, + responded_at: RESPONDED_AT, + responded_by_user: user, + timedout: user === null, + }; + } + + function requestTypes(result: MockResult) { + return result.runtimeRequests.map((r) => r.type); + } + + it("writes the request with rendered text and parks in awaiting_input", async () => { + const build = testDag.task("build", async () => ({ version: "1.4" })); + testDag.task( + "sign_off", + approval({ + subject: ({ report }: { report: { version: string } }) => `Ship ${report.version}?`, + body: async ({ report }: { report: { version: string } }) => `**${report.version}**`, + assignedUsers: [{ id: "ada", name: "Ada" }], + responseTimeout: 3600, + }), + )({ report: build() }); + + const result = await parkedWith("sign_off", { + arg_bindings: [{ name: "report", kind: "xcom", task_id: "build" }], + }); + + const [created] = requestsOf(result, "CreateHITLDetailPayload"); + expect(created).toEqual({ + type: "CreateHITLDetailPayload", + ti_id: "ti-1", + options: ["Approve", "Reject"], + subject: "Ship 1.4?", + body: "**1.4**", + defaults: null, + multiple: false, + params: {}, + assigned_users: [{ id: "ada", name: "Ada" }], + }); + expect(result.firstResponse!.body).toEqual({ + type: "AwaitInputTask", + state: "awaiting_input", + timeout: "PT3600S", + next_method: "execute_complete", + next_kwargs: {}, + }); + expect(requestsOf(result, "SetXCom")).toEqual([]); + }); + + it("sends its defaults and multiple, with no body and no timeout as null", async () => { + testDag.task( + "choose", + hitl({ + subject: "Pick", + options: ["us", "eu"], + defaults: ["us", "eu"], + multiple: true, + }), + )(); + + const result = await parkedWith("choose"); + + expect(requestsOf(result, "CreateHITLDetailPayload")[0]).toMatchObject({ + body: null, + defaults: ["us", "eu"], + multiple: true, + }); + expect(result.firstResponse!.body).toMatchObject({ type: "AwaitInputTask", timeout: null }); + }); + + it("writes its params as Param.serialize does, with the source always task", async () => { + testDag.task( + "choose", + hitl({ + subject: "Pick", + options: ["go"], + params: { + region: { + value: "us", + description: "Where to ship", + schema: { type: "string", enum: ["us", "eu"] }, + }, + retries: { value: 3 }, + }, + }), + )(); + + const result = await parkedWith("choose"); + + expect(requestsOf(result, "CreateHITLDetailPayload")[0]).toMatchObject({ + params: { + region: { + value: "us", + description: "Where to ship", + schema: { type: "string", enum: ["us", "eu"] }, + source: "task", + }, + retries: { value: 3, description: null, schema: {}, source: "task" }, + }, + }); + }); + + it("renders the subject and the body with the renames of both", async () => { + testDag.task( + "choose", + hitl({ + subject: withArgNames( + { version: "release_version" }, + ({ version }: { version: string }) => `Ship ${version}?`, + ), + body: withArgNames( + { notes: "release_notes" }, + ({ notes }: { notes: string }) => `Notes: ${notes}`, + ), + options: ["go"], + }), + )({ version: "1.4", notes: "Faster" }); + + const result = await parkedWith("choose", { + arg_bindings: [ + { name: "release_version", kind: "literal", value: "1.4" }, + { name: "release_notes", kind: "literal", value: "Faster" }, + ], + }); + + expect(requestsOf(result, "CreateHITLDetailPayload")[0]).toMatchObject({ + subject: "Ship 1.4?", + body: "Notes: Faster", + }); + }); + + it.each<[string, () => unknown, string]>([ + [ + "throws", + () => { + throw new Error("no report"); + }, + "no report", + ], + [ + "returns an empty string", + () => "", + 'The subject of task "sign_off" has to be a non-empty string', + ], + ["returns a number", () => 7, 'The subject of task "sign_off" has to be a non-empty string'], + ])( + "fails without writing the request when its subject function %s", + async (_label, subject, reason) => { + testDag.task("sign_off", approval({ subject: subject as () => string }))(); + + const result = await parkedWith("sign_off", { should_retry: true }); + + expect(requestsOf(result, "CreateHITLDetailPayload")).toEqual([]); + expect(result.firstResponse!.body).toMatchObject({ + type: "RetryTask", + retry_reason: reason, + }); + }, + ); + + it("resumes with the response without writing the request again", async () => { + let rendered = 0; + testDag.task( + "choose", + hitl({ + subject: () => `Pick ${++rendered}`, + options: ["us", "eu", "apac"], + multiple: true, + }), + )(); + const parked = await parkedWith("choose"); + const { next_kwargs } = parked.firstResponse!.body as { + next_kwargs: Record<string, unknown>; + }; + + const result = await driveSupervisor( + makeStartupDetails("choose", "test_dag", "r1", { + next_method: "execute_complete", + next_kwargs: { ...next_kwargs, event: answer(["eu", "apac"]) }, + }), + ); + + expect(rendered).toBe(1); + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestTypes(result)).toEqual(["SetXCom"]); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ + key: "return_value", + task_id: "choose", + value: { + chosen_options: ["eu", "apac"], + params_input: {}, + responded_at: "2026-10-08T12:30:15.123Z", + responded_by_user: { id: "1", name: "Ada" }, + timedout: false, + }, + }); + }); + + it.each([ + ["an approval answered Approve", approval({ subject: "Go?" }), "Approve"], + [ + "a generic choice answered Reject", + hitl({ subject: "?", options: ["Approve", "Reject"] }), + "Reject", + ], + ])("skips nothing for %s and pushes the response", async (_label, operator, chosen) => { + testDag + .task("decide", operator)() + .before(testDag.task("publish", async () => {})()); + + const result = await resumedWith("decide", answer([chosen])); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestTypes(result)).toEqual(["SetXCom"]); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ key: "return_value" }); + }); + + it("on Reject skips the tasks directly downstream and, as SkipMixin does, pushes no response", async () => { + const decide = testDag.task("decide", approval({ subject: "Go?" }))(); + const publish = testDag.task("publish", async () => {})(); + decide.before(publish); + publish.before(testDag.task("announce", async () => {})()); + + const result = await resumedWith("decide", answer(["Reject"])); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestTypes(result)).toEqual(["SetXCom", "SkipDownstreamTasks"]); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ + key: "skipmixin_key", + value: { skipped: ["publish"] }, + }); + expect(requestsOf(result, "SkipDownstreamTasks")[0]).toMatchObject({ tasks: ["publish"] }); + }); + + it("on Reject skips a task group directly downstream by its first tasks", async () => { + const decide = testDag.task("decide", approval({ subject: "Go?" }))(); + const stage = testDag.taskGroup("stage"); + const publish = stage.task("publish", async () => {})(); + publish.before(stage.task("announce", async () => {})()); + decide.before(stage); + + const result = await resumedWith("decide", answer(["Reject"])); + + expect(requestsOf(result, "SkipDownstreamTasks")[0]).toMatchObject({ + tasks: ["stage.publish"], + }); + }); + + it("on Reject with ignoreDownstreamTriggerRules skips every task downstream", async () => { + const decide = testDag.task( + "decide", + approval({ subject: "Go?", ignoreDownstreamTriggerRules: true }), + )(); + const publish = testDag.task("publish", async () => {})(); + decide.before(publish); + publish.before(testDag.task("announce", async () => {})()); + + const result = await resumedWith("decide", answer(["Reject"])); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestTypes(result)).toEqual(["SetXCom", "SkipDownstreamTasks"]); + expect(requestsOf(result, "SkipDownstreamTasks")[0]).toMatchObject({ + tasks: ["announce", "publish"], + }); + }); + + it("on Reject with nothing downstream pushes the response", async () => { + testDag.task("decide", approval({ subject: "Go?" }))(); + + const result = await resumedWith("decide", answer(["Reject"])); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestsOf(result, "SetXCom")).toEqual([ + expect.objectContaining({ + key: "return_value", + value: expect.objectContaining({ chosen_options: ["Reject"] }), + }), + ]); + }); + + it("on Reject fails with failOnReject, before pushing anything", async () => { + testDag.task("decide", approval({ subject: "Go?", failOnReject: true }))(); + + const result = await resumedWith("decide", answer(["Reject"])); + + expect(result.firstResponse!.body).toMatchObject({ type: "TaskState", state: "failed" }); + expect(result.runtimeRequests).toEqual([]); + }); + + it("succeeds with the defaults applied on timeout", async () => { + testDag.task( + "decide", + approval({ subject: "Go?", defaults: "Approve", responseTimeout: 60 }), + )(); + + const result = await resumedWith("decide", answer(["Approve"], null)); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ + value: { responded_by_user: null, timedout: true }, + }); + }); + + it.each([ + ["retries", true, { type: "RetryTask", retry_reason: "Response timed out: no response" }], + ["fails", false, { type: "TaskState", state: "failed" }], + ])("%s when the timeout passes without defaults", async (_label, shouldRetry, outcome) => { + testDag.task("decide", approval({ subject: "Go?", responseTimeout: 60 }))(); + + const result = await resumedWith( + "decide", + { error: "no response", error_type: "timeout" }, + { should_retry: shouldRetry }, + ); + + expect(result.firstResponse!.body).toMatchObject(outcome); + }); + + it.each<[string, Record<string, unknown>]>([ + ["an option it never offered", { event: answer(["Maybe"]) }], + ["an unreadable response time", { event: { ...answer(["Approve"]), responded_at: "soon" } }], + ["an unreadable responder", { event: answer(["Approve"], { id: 1 }) }], + ["no event", {}], + ])("fails on resume with %s", async (_label, nextKwargs) => { + testDag.task("decide", approval({ subject: "Go?" }))(); + + const result = await driveSupervisor( + makeStartupDetails("decide", "test_dag", "r1", { + next_method: "execute_complete", + next_kwargs: nextKwargs, + }), + ); + + expect(result.firstResponse!.body).toMatchObject({ type: "TaskState", state: "failed" }); + expect(result.runtimeRequests).toEqual([]); + }); + + describe("params input", () => { + const withParams = () => + hitl({ + subject: "Pick", + options: ["go"], + params: { region: { value: "us" }, retries: { value: 3 } }, + }); + + it("is pushed with the response when it answers every param", async () => { + testDag.task("choose", withParams())(); + + const result = await resumedWith( + "choose", + answer(["go"], undefined, { region: "eu", retries: 5 }), + ); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ + key: "return_value", + value: { chosen_options: ["go"], params_input: { region: "eu", retries: 5 } }, + }); + }); + + it.each<[string, Record<string, unknown>]>([ + ["only the timeout defaults", { region: "us", retries: 3 }], + ["no answers when the form was not filled in", {}], + ])("is accepted with %s", async (_label, paramsInput) => { + testDag.task("choose", withParams())(); + + const result = await resumedWith("choose", answer(["go"], undefined, paramsInput)); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + }); + + it("is not checked for a task that declares no params", async () => { + testDag.task("choose", hitl({ subject: "Pick", options: ["go"] }))(); + + const result = await resumedWith("choose", answer(["go"], undefined, { region: "eu" })); + + expect(result.firstResponse!.body).toMatchObject({ type: "SucceedTask" }); + expect(requestsOf(result, "SetXCom")[0]).toMatchObject({ + value: { params_input: { region: "eu" } }, + }); + }); + + it.each<[string, Record<string, unknown>, string]>([ + ["a param the task never declared", { region: "eu", other: 1 }, "region,other"], + ["only some of the params", { region: "eu" }, "region"], + ["other names than the params", { zone: "eu", retries: 5 }, "zone,retries"], + ])("fails the task with %s", async (_label, paramsInput, received) => { + testDag.task("choose", withParams())(); + + const result = await resumedWith("choose", answer(["go"], undefined, paramsInput), { + should_retry: true, + }); + + expect(result.firstResponse!.body).toMatchObject({ + type: "RetryTask", + retry_reason: + `params_input ${JSON.stringify(received.split(","))} does not match params ` + + '["region","retries"]', + }); + expect(result.runtimeRequests).toEqual([]); + }); + }); + + it("fails on resume with a next_method it does not know", async () => { + testDag.task("decide", approval({ subject: "Go?" }))(); + + const result = await driveSupervisor( + makeStartupDetails("decide", "test_dag", "r1", { + next_method: "something_else", + next_kwargs: { event: answer(["Approve"]) }, + }), + ); + + expect(result.firstResponse!.body).toMatchObject({ type: "TaskState", state: "failed" }); + expect(result.runtimeRequests).toEqual([]); + }); + + it("fails on resume through __fail__", async () => { + testDag.task("decide", approval({ subject: "Go?" }))(); + + const result = await driveSupervisor( + makeStartupDetails("decide", "test_dag", "r1", { + next_method: "__fail__", + next_kwargs: { error: "Could not resume the task", traceback: ["Traceback", "Boom"] }, + }), + ); + + expect(result.firstResponse!.body).toMatchObject({ type: "TaskState", state: "failed" }); + expect( + result.logRecords.some((r) => + String(r["event"]).includes("Task could not be resumed:\nTraceback\nBoom"), + ), ).toBe(true); + expect(result.logRecords.some((r) => String(r["event"]).includes("Trigger failed"))).toBe( + false, + ); + }); + + // The agent-approval Dag of `example/src/ai-approval.ts`, run one task per process. + describe("agent approval", () => { + const DRAFTED_AMOUNT = 120; + const EDITED_AMOUNT = 80; + + interface ToolCall { + type: "tool-call"; + toolCallId: string; + toolName: string; + input: { orderId: string; amount: number }; + } + interface ApprovalRequest { + type: "tool-approval-request"; + approvalId: string; + toolCall: ToolCall; + } + interface ApprovalResponse { + type: "tool-approval-response"; + approvalId: string; + approved: boolean; + reason?: string; + } + type ModelMessage = + | { role: "user"; content: string } + | { role: "assistant"; content: ApprovalRequest[] } + | { role: "tool"; content: ApprovalResponse[] }; + interface AgentStep { + messages: ModelMessage[]; + pending: { id: string; toolName: string; input: ToolCall["input"] }[]; + } + + let modelCalls: number; + + async function generateText(opts: { messages: ModelMessage[] }) { + modelCalls += 1; + const response = opts.messages + .flatMap((message) => (message.role === "tool" ? message.content : [])) + .at(-1); + if (response === undefined) { + const request: ApprovalRequest = { + type: "tool-approval-request", + approvalId: "approval-1", + toolCall: { + type: "tool-call", + toolCallId: "call-1", + toolName: "issue_refund", + input: { orderId: "ord_1042", amount: DRAFTED_AMOUNT }, + }, + }; + const messages: ModelMessage[] = [{ role: "assistant", content: [request] }]; + return { text: "", content: [request], response: { messages } }; + } + const { toolCall } = opts.messages + .flatMap((message) => (message.role === "assistant" ? message.content : [])) + .find((request) => request.approvalId === response.approvalId)!; + const text = `Refunded $${toolCall.input.amount} on order ${toolCall.input.orderId}.`; + return { text, content: [], response: { messages: [] } }; + } + + beforeEach(() => { + modelCalls = 0; + const agentStep = testDag.task("agent_step", async (): Promise<AgentStep> => { + const messages: ModelMessage[] = [{ role: "user", content: "Refund order ord_1042." }]; + const result = await generateText({ messages }); + return { + messages: [...messages, ...result.response.messages], + pending: result.content.map((request) => ({ + id: request.approvalId, + toolName: request.toolCall.toolName, + input: request.toolCall.input, + })), + }; + })(); + const review = testDag.task( + "review", + hitl({ + subject: ({ draft }: { draft: AgentStep }) => `Approve ${draft.pending[0]!.toolName}?`, + body: ({ draft }: { draft: AgentStep }) => + [ + `The agent wants to call \`${draft.pending[0]!.toolName}\` with:`, + "", + "```json", + JSON.stringify(draft.pending[0]!.input, null, 2), + "```", + ].join("\n"), + options: ["Approve", "Reject"], + params: { + amount: { + value: DRAFTED_AMOUNT, + description: "Amount to refund", + schema: { type: "number" }, + }, + }, + }), + )({ draft: agentStep }); + testDag.task( + "agent_continue", + async ({ draft, decision }: { draft: AgentStep; decision: HITLResult }) => { + const edited = decision.params_input["amount"]; + const amount = typeof edited === "number" ? edited : draft.pending[0]!.input.amount; + const approvalId = draft.pending[0]!.id; + const messages: ModelMessage[] = [ + ...draft.messages.map((message): ModelMessage => { + if (message.role !== "assistant") return message; + return { + role: "assistant", + content: message.content.map((request) => + request.approvalId === approvalId + ? { + ...request, + toolCall: { + ...request.toolCall, + input: { ...request.toolCall.input, amount }, + }, + } + : request, + ), + }; + }), + { + role: "tool", + content: [ + { + type: "tool-approval-response", + approvalId, + approved: decision.chosen_options[0] === "Approve", + reason: `Approved at $${amount}`, + }, + ], + }, + ]; + const result = await generateText({ messages }); + return { text: result.text, modelCalls: 1, approvedAmount: amount }; Review Comment: The comment above says this is the Dag from `example/src/ai-approval.ts`, but it's a hand copy and has already drifted: the example returns `approvedAmount: approved ? amount : null` with a rejection reason, while this hard-codes the approved path. The one thing only this test asserts is that a resume with `arg_bindings` sends no `GetXCom`. Would a small purpose-built Dag for that be easier to keep honest than a copy of the example? ########## airflow-e2e-tests/tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_hitl.py: ########## @@ -0,0 +1,208 @@ +# 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. +"""E2E tests for human-in-the-loop tasks in a native TypeScript Dag. + +Run with:: + + E2E_TEST_MODE=ts_sdk uv run --project airflow-e2e-tests pytest \\ + tests/airflow_e2e_tests/ts_sdk_tests/test_ts_sdk_hitl.py -xvs + +What is verified +---------------- +``conftest._setup_ts_sdk_integration`` packs ``ts-sdk/example/src/ai-approval.ts`` and drops the bundle where the +Dag processor and the worker both see it. ``ts_hitl_ai_approval`` is a native TypeScript Dag with no Python Dag +file, so the Dag processor parses the ``.min.mjs`` itself. It is the agent-approval pattern, where a stand-in +model drafts a refund and a person decides before it is issued:: + + agent_step >> review >> agent_continue + +* ``agent_step`` calls the model, which drafts an ``issue_refund`` tool call and stops for approval. +* ``review`` is a ``hitl`` task. It writes the request, with the drafted call in the body and an editable + ``amount`` field, and parks. +* ``agent_continue`` appends the decision to the history and calls the model a second time. + +The Dag is triggered once and the module-scoped fixtures walk it in order: wait for ``review`` to park, then +answer it with a different amount and wait for the run to finish. Together the tests confirm: + +1. A task of a native Dag parks in ``awaiting_input`` and the request is readable through the REST API, with the + drafted call in the body and the ``amount`` param. +2. Parking needs no triggerer. ``airflow-triggerer`` is part of the base compose stack and is left running, so a + passing run alone proves nothing. The evidence is in the data: a deferral creates a Trigger row and sets the + task instance's ``trigger_id``, while a task parked in ``awaiting_input`` has none. The REST API exposes the + row as ``trigger``, which is null exactly when ``trigger_id`` is. +3. The answer resumes the run, and the edited amount reaches ``agent_continue``'s model call and not the drafted + one. ``agent_step``, which runs the model first, is not run again. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from datetime import UTC, datetime + +import pytest + +from airflow_e2e_tests.e2e_test_utils.clients import AirflowClient + +_DAG_ID = "ts_hitl_ai_approval" + +# A native Dag is parsed by running node in the Dag processor, so allow room for the first parse. +_DAG_PARSE_TIMEOUT = 300 +# Scheduling and coordinator startup for `agent_step`, then the first run of `review`. +_PARK_TIMEOUT = 600 +# Scheduling the resumed `review`, then `agent_continue`. +_RESUME_TIMEOUT = 600 + +# Match `DRAFTED_AMOUNT` in ts-sdk/example/src/ai-approval.ts. +_DRAFTED_AMOUNT = 120 +_EDITED_AMOUNT = 80 + + +@dataclass +class _ParkedReview: + """The triggered run, as it was when ``review`` parked.""" + + client: AirflowClient + run_id: str + task_instance: dict + hitl_detail: dict + + +def _states(task_instances: dict[str, dict]) -> dict[str, str | None]: + return {task_id: ti.get("state") for task_id, ti in task_instances.items()} + + +def _wait_for_review_to_park(client: AirflowClient, run_id: str) -> dict: + """Poll until ``review`` is ``awaiting_input`` and return its task instance.""" + deadline = time.monotonic() + _PARK_TIMEOUT + task_instances: dict[str, dict] = {} + while time.monotonic() < deadline: + resp = client.get_task_instances(dag_id=_DAG_ID, run_id=run_id) + task_instances = {ti["task_id"]: ti for ti in resp.get("task_instances", [])} + review = task_instances.get("review") + if review is not None and review.get("state") == "awaiting_input": + return review + states = _states(task_instances) + assert "failed" not in states.values(), f"a task failed before 'review' parked: {states}" + time.sleep(5) + raise TimeoutError( + f"'review' did not reach awaiting_input within {_PARK_TIMEOUT}s. task states: {_states(task_instances)}" + ) + + [email protected](scope="module") +def parked_review() -> _ParkedReview: + """Trigger ``ts_hitl_ai_approval`` once and wait for ``review`` to park.""" + client = AirflowClient() + # Triggering waits only 120s for the Dag, which is short for a first parse by node. + client.wait_for_dag(_DAG_ID, timeout=_DAG_PARSE_TIMEOUT) + resp = client.trigger_dag(_DAG_ID, json={"logical_date": datetime.now(UTC).isoformat()}) + run_id = resp["dag_run_id"] + task_instance = _wait_for_review_to_park(client, run_id) + hitl_detail = client.get_hitl_detail(dag_id=_DAG_ID, run_id=run_id, task_id="review") + return _ParkedReview(client=client, run_id=run_id, task_instance=task_instance, hitl_detail=hitl_detail) + + +@dataclass +class _AnsweredRun: + client: AirflowClient + run_id: str + state: str + ti_attrs: dict[str, dict] + + def xcom(self, task_id: str, key: str = "return_value"): + return self.client.get_xcom_value(dag_id=_DAG_ID, task_id=task_id, run_id=self.run_id, key=key).get( + "value" + ) + + [email protected](scope="module") +def answered_run(parked_review: _ParkedReview) -> _AnsweredRun: + """Approve ``review`` with an edited amount and wait for the run to finish.""" + client = parked_review.client + run_id = parked_review.run_id + client.respond_to_hitl( + dag_id=_DAG_ID, + run_id=run_id, + task_id="review", + chosen_options=["Approve"], + params_input={"amount": _EDITED_AMOUNT}, + ) + state = client.wait_for_dag_run(dag_id=_DAG_ID, run_id=run_id, timeout=_RESUME_TIMEOUT) + ti_resp = client.get_task_instances(dag_id=_DAG_ID, run_id=run_id) + ti_attrs = {ti["task_id"]: ti for ti in ti_resp.get("task_instances", [])} + return _AnsweredRun(client=client, run_id=run_id, state=state, ti_attrs=ti_attrs) + + +def test_review_parks_without_a_trigger(parked_review: _ParkedReview): + """``review`` is ``awaiting_input`` with no Trigger, though a triggerer is running. + + A deferral sets ``trigger_id``; the REST API shows that row as ``trigger``. A parked HITL task has none, + so nothing for the triggerer to pick up. + """ + ti = parked_review.task_instance + assert ti["state"] == "awaiting_input", ti + assert ti["trigger"] is None, f"'review' parked with a Trigger, so it deferred: {ti['trigger']!r}" + + +def test_hitl_detail_carries_the_draft(parked_review: _ParkedReview): + """The request names the tool, shows the drafted input, and offers the amount to edit.""" + detail = parked_review.hitl_detail + assert detail["subject"] == "Approve issue_refund?" + assert detail["options"] == ["Approve", "Reject"] + assert detail["response_received"] is False + + body = detail["body"] + assert "issue_refund" in body, body + assert f'"amount": {_DRAFTED_AMOUNT}' in body, body + assert '"orderId": "ord_1042"' in body, body + + amount = detail["params"]["amount"] + assert amount["value"] == _DRAFTED_AMOUNT, amount + assert amount["schema"] == {"type": "number"}, amount + assert amount["description"] == "Amount to refund", amount + + +def test_answer_resumes_the_run(answered_run: _AnsweredRun): + assert answered_run.state == "success", ( + f"expected the run to succeed; got {answered_run.state!r}. " + f"task states: {_states(answered_run.ti_attrs)}" + ) + + +def test_review_xcom_holds_the_edited_answer(answered_run: _AnsweredRun): + value = answered_run.xcom("review") + assert value["chosen_options"] == ["Approve"], value + assert value["params_input"] == {"amount": _EDITED_AMOUNT}, value + assert value["timedout"] is False, value + + +def test_agent_continue_quotes_the_edited_amount(answered_run: _AnsweredRun): + """The second model call saw the reviewer's figure, not the drafted one, and ran once.""" + value = answered_run.xcom("agent_continue") + assert value == { + "text": f"Refunded ${_EDITED_AMOUNT} on order ord_1042.", + "modelCalls": 1, Review Comment: `modelCalls` is the literal `1` in `ai-approval.ts`, so this assertion can't fail. To show the model ran once across the wait, the number has to come from something the run actually counts. Otherwise the key can go. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
