Status

Motivation

Dynamic task mapping (AIP-42) made cardinality dynamic: run the same task over a collection discovered at run time. What is still static is the shape, meaning which tasks run in a region and how they depend on each other.

Some workflows cannot know that until they run. A step inspects its input and only then decides what work is needed and how it connects.

1. Triage-driven response (data and ops). An incoming item is classified at run time, and the classification decides which remediation tasks run and in what order: a support ticket routed to refund, escalation, enrichment or a human; a data-quality failure routed to quarantine or backfill; an alert routed through a diagnosis chain whose later steps depend on what the first probe found. The set of actions is fixed and reviewed. The ordering and the dependencies between them are not.

2. Plan-driven pipelines (data). An upstream analysis decides which transformations this run needs and how they depend on each other. Selective builds are the clearest case: an impact analysis over changed sources says which models to build for this run, with the dependency edges between exactly those models. Today authors either regenerate the Dag file per plan, which produces a new Dag version per run and drowns author intent in version churn, or hide the whole plan inside one task and give up per-model retry and per-model observability.

3. Agent-composed workflows (agentic). A planner is an LLM handed a reviewed set of allowed actions, and it composes the response: which actions to take, how they connect, with what arguments. The set of things the model may compose is a list in the Dag file, reviewed like any other code.

The tasks themselves stay deterministic in all three.

Current workarounds and their limitations

• Branching. A branch task already selects a subset: choose_branch returns a task id or a list of task ids, so any subset of its declared successors can run. What a branch cannot do is decide the dependency edges among the tasks it selected.
• One opaque task. Loses per-task retry, logs and observability.
• Regenerating the Dag file per plan. A Dag factory that reads the plan at parse time makes the plan a parse-time input, so every distinct plan is a new Dag version. Each run still pins the structure it ran, so history is recoverable, but version history stops describing author intent and the churn makes it hard to see when the pipeline actually changed.
• A Dag per shape plus TriggerDagRunOperator. Loses the single-run view and cross-shape lineage, and the shapes still have to be enumerated.
• SubDagOperator. Deprecated in Airflow 2 and removed in Airflow 3; poor isolation and scheduler pathologies. Reviving it is a non-goal.
• Dynamic task mapping (AIP-42). Runtime cardinality over one task. It cannot say "given this result, run these different tasks, with these dependencies between them".

Terminology

The proposal reuses Airflow's existing vocabulary wherever it fits. A task, a task instance, a node and an edge mean what they already mean in Airflow. Note that "dependency" is used in this document for the scheduler's own dep mechanism, so the plan's wiring is always called edges. These are the terms this AIP adds.

Scope

A Task Group whose members and internal edges are decided at run time from a fixed, serialized catalog:

• A generator task emits a plan: which catalog entries run, how they depend on each other, and what inputs each gets.
• The group materializes exactly the selected entries as task instances. An entry the plan did not select never becomes a task instance.
• The region converges on a synthesized merge node so the group composes with downstream tasks under default trigger rules.
• The parent Dag stays static, serialized and versioned. Two runs that take different shapes share one Dag version.

The plan source is an override point: the base form takes a planner= callable and a subclass overrides plan(). That is how an LLM planner ships as a provider-level subclass.

Relationship to other AIPs

• AIP-42 Dynamic Task Mapping Same expansion mechanism. Mapping makes count dynamic from a collection; this makes shape dynamic from a plan. Selected entries materialize through the existing mapped-instance path, which is why the grid needs no new concept.
• AIP-63: DAG Versioning The emitted shape is run data and never enters the serialized Dag or the Dag hash, so runs of different shapes share one version. The catalog is in the serialized Dag, so adding or removing an entry is a version bump. Changing what a planner may do is an author change; what it did on a given run is run data.
• AIP-72 Task Execution Interface aka Task SDK The authoring surface and the plan accessor live in the Task SDK. The planner runs on a worker; the scheduler runs no user code.
• AIP-103: Task State Management Cross-task state and the merge aggregate want a group-scoped store rather than XCom. This AIP depends on AIP-103 for that and does not build a new store. AIP-111 reached the same conclusion for carried loop state, and group scope is not yet proposed as its own work by either.
• AIP-111: Task Loops A loop repeats a fixed group a runtime number of times; a dynamic group reshapes a fixed catalog. They share the expansion mechanism and one dependency for nesting (see Composition). Their integration surfaces are disjoint: a serialized flag, a scheduler dependency that no-ops outside its own group type, a context accessor, and one reserved control key each, so they land independently.
• AIP-112: Task Steps Structure inside one task; this operates between tasks. Where the work does not need independent retry, its own row in the grid, or an approval part-way through the shape, Task Steps covers it instead, and the docs should point there. 

Considerations

What change do you propose?

A Task Group whose members are chosen at run time from a declared catalog, authored as one construct and rendered as a group:

DynamicTaskGroup(
    group_id="resolve",
    planner=triage,                          # OR subclass and override plan()
    validator=check_plan,                    # OR override validate_plan(); raises to reject
    catalog=[validate, issue_refund, notify, # the fixed set; this is what serializes
             enrich, route_to_human, follow_up],
    merge=summarize,                         # terminal convergence
    max_selected=8,                          # runaway guard
)

The plan source is the extension point, so an LLM planner is a subclass:

class AgentTaskGroup(DynamicTaskGroup):
    def plan(self, context) -> dict:
        # hand the model self.catalog as its tool schema plus the goal; it returns the plan
        return agent.run_sync(...).output

Providers can ship other plan sources the same way: a YAML reader, a control-table query, an impact analysis.

The group has two override points and they are symmetric: one produces a plan, one checks it. validator= takes a callable, or a subclass overrides validate_plan(). It runs inside the generator once the built-in checks have passed, and raising from it rejects the plan, so the region never materializes. Whatever rules an author needs live there as ordinary Python, which is why the catalog itself stays a flat list. Trying to express "notify at most once", "issue_refund needs validate ahead of it", "no more than one entry that writes to production" as catalog syntax means growing a rule language, and the set of rules people want does not converge.

Naming: max_selected, since max_active would collide with max_active_runs and max_active_tasks and read as a concurrency throttle. catalog= is an explicit literal because a reviewer needs to read the list of permitted actions; a decorator that inferred the set from context would hide it. Whether a .dynamic() method on @task_group, symmetric with .loop(), should exist alongside the constructor is an open question.

The catalog: allowlist and versioning anchor

The catalog is the list of tasks a planner may select and wire. A plan naming anything else is rejected before any task runs.

It is also what serializes, so the Dag stays versionable no matter what shape a run takes, and a change to what a planner may reach is an author change with an audit trail.

What the catalog bounds and what it does not:

• It bounds which callables can run. A plan cannot name a task that was not pre-declared.
• It does not bound their arguments. The plan carries per-task inputs, so a planner picks an entry from the allowed set and supplies its arguments. A capped action can be handed an uncapped amount, and it is the task's own code that has to reject it.
• It does not bound the wiring. Any acyclic arrangement of the selected entries is expressible.

So this is an allowlist of callables, not a bound on what a run can do with them. Rules beyond membership belong in the validator rather than in the catalog, which is why catalog= stays a flat list of tasks. Whether plan-supplied inputs are exempt from Jinja rendering is an open question, and it decides how much the allowlist is worth in the agentic case.

The plan

The plan is what one run's planner returns. It names the tasks that run, the edges between them, and the arguments each gets. A task is a record whose id identifies it within the plan and whose entry names the catalog entry it runs:

{
    "tasks": [
        {"id": "validate", "entry": "validate", "inputs": {"ticket_id": "T-1"}},
        {"id": "issue_refund", "entry": "issue_refund"},
        "notify",
    ],
    "edges": [["validate", "issue_refund"], ["issue_refund", "notify"]],
}

A bare string is shorthand for a record whose id and entry are both that label and which takes no inputs, so the common case stays short. Edges are pairs of ids.

Keeping id separate from entry is what makes it possible to select one entry more than once with different inputs. Today an id must be a catalog label, so the two always match and an entry appears at most once; an open question recommends lifting that, and the record form is what the lift needs in order to be expressible at all.

An empty plan is legal. {"tasks": []} selects nothing, the region stays empty, and the merge still succeeds.

Guardrails

• max_selected is a required hard cap on how many entries one plan may select. Catalog size is the absolute bound.
• The catalog is a closed list, so a plan cannot name an undeclared task.
• Plan validation is fail-closed: an unknown entry, over the cap, malformed or cyclic edges, a duplicate id, or an edge referencing a task the plan did not select all fail the generator.
• The author's validator runs after those checks and can reject a plan for any reason it likes: a count of a given entry, a required upstream, a combination the author does not want composed. It sees the normalized plan and rejects by raising, so a rejected plan fails the generator like any other invalid plan.
• The synthesized generator, region and merge ids are reserved. An entry colliding with one, or two entries sharing a label, fail at parse time.

Serializing every candidate as a node and skipping the unselected ones was considered and rejected: a catalog of 20 with 3 selected leaves 17 skipped rows per run, and skipped elsewhere signals "branch not taken" or "upstream failed", so a healthy run reads as a broken one.

Runtime model

The group serializes as three nodes: the generator, the region that holds the emitted shape, and the terminal merge.

<group>.plan  ──▶  <group>.generated{ selected entries  ──▶  merge }
   generator              the runtime-shaped region (collapsible)

The catalog is a parse-time registry; its labels serialize as metadata on the group, so the allowlist is versioned without any entry becoming a node in the Dag.

At run time:

1. The generator runs the planner, validates the plan against the catalog, writes the normalized plan somewhere the scheduler can read it (see Where the emitted plan lives), and returns the selected entries in order.
2. Only the selected entries materialize as task instances. N selected gives N instances, a plan that selects nothing leaves the region empty, and an entry the plan did not name never becomes a task instance. Each instance is labeled by the entry it runs, so the grid shows validate and issue_refund rather than positional indexes.
3. A new scheduler dependency sequences those instances by the emitted edges: an instance waits for the instances of its plan-upstream tasks, and failure and skip propagate along the emitted edges.
4. The merge runs once the region terminates.

The scheduler runs no user code. The plan is produced by a worker task, and the dependency reads only run data.

An invalid plan fails the generator, the region never materializes, and it fails closed instead of producing a half-specified graph.

How the selected entries are materialized is an implementation-time decision, and it is the one this AIP most deliberately leaves open, because the options differ in what they cost the author. Serializing every catalog entry as a node keeps each entry's own pool, queue, retries and outlets, but leaves the unselected ones visible as skipped rows. Expanding a single dispatching task shows nothing unselected in the grid, but the entries then share that task's configuration. A dispatching task per entry, or per author-declared placement lane, sits between the two. The prototype does the single-dispatching-task version, which is why its entries share a pool.

Per-entry configuration is therefore a property of the materialization, not of the feature, and the choice does not change the authoring surface. Per-entry pool and queue are worth calling out because they have already come up in early feedback: a catalog of mixed actions plausibly wants a warehouse pool for one entry and an API-rate-limit pool for another. pool, queue, pool_slots, priority_weight, executor, executor_config and the retry limit are all columns on the task-instance row, so a materialization can vary them per entry. outlets is the exception: it is resolved from the serialized Dag for asset scheduling before any task instance exists, so per-entry asset events need more than a materialization choice.

The merge node

The feature synthesizes the merge and fixes its success semantics instead of exposing a trigger_rule:

• It succeeds when the region terminates normally, including when the plan selects nothing.
• It propagates failure when a selected entry failed.
• Downstream edges attach to it, so dynamic_group >> next composes under all_success with no trigger_rule on the author's side.
• It carries the aggregate the downstream reads.

The obvious choice does not work. With none_failed_min_one_success on an ordinary task, an empty plan leaves the region's body with nothing to run, so the merge's only upstream is skipped and the trigger-rule dependency resolves the merge to skipped (the skipped == upstream branch, which is evaluated before the zero-success branch), so everything downstream skips silently. AIP-111 has the same problem with its skipped tail, and reached the same answer: the convergence node cannot be an ordinary task carrying a trigger rule.

Worth noting that today's branch >> join idiom has the identical hole. A branch that selects nothing leaves a join under none_failed_min_one_success skipped by the same condition. Convergence after a region that may select nothing is a general unsolved case that predates this feature.

Proposed mechanism. The scheduler dependency this AIP already adds should own the merge as well. It reads the plan to sequence the selected instances, so gating the merge on region termination and marking it upstream_failed when a selected entry failed is the same data on the same code path, and the marginal cost over what is already proposed is small. Two alternatives and why not: a new TriggerRule member becomes public the moment anyone reads a serialized Dag and is then permanent; all_done plus logic in the merge's own body needs no scheduler change but moves failure propagation to the worker, which then needs authoritative sibling instance states it cannot infer from missing outputs, and it also runs the merge when the generator failed, which contradicts failing the region closed. One consequence: the merge then gets its semantics from the scheduler, so this is a scheduler change.

Where determinism still holds

Airflow does not become non-deterministic. The parent Dag's edges into and out of the group stay static and versioned, and what a run may vary is bounded by the catalog. The deterministic boundary moves from the task to the group.

Consequences:

• Reproducibility means the emitted plan is recorded, which it is. Replay from a recorded plan is an optional mode for regulated cases; re-planning is the default.
• Reshape on rerun follows the existing model for a mapped task whose cardinality changes: tasks absent from the new shape resolve to REMOVED, which the trigger-rule dependency already accounts for when it counts upstream states.
• max_selected and catalog size are governance dials, not correctness controls.

Side-effect safety. Containment bounds control flow and which tasks run. It does not bound side-effect safety across shapes: an approved task wired into an unexpected shape still fires its real side effect. Two levers exist and they cut differently. The catalog bounds membership, so one discipline is to catalog only tasks whose every possible wiring is safe. What the catalog cannot do is say anything about the wiring itself, and for a mixed catalog that is where the danger is.

This is what validator is for. An author who needs issue_credit to have request_approval upstream of it writes that check, and a plan that violates it fails the generator before any task instance exists. That matters more than where the check lives: an entry can test its own wiring at run time, but by then the region has materialized and its siblings may already have fired. Rejecting the plan is the only point at which nothing has happened yet. Where a planner is a model, the same rules should also go in its prompt, so the model usually produces a valid plan and the validator is the thing that makes it true rather than likely. The validator runs once a plan exists, so it is a run-time check. How much wiring safety could instead be caught at parse time, before any run, is still open.

Interaction with Dag-level settings

A runtime-chosen shape runs into several settings that are fixed when the Dag is written.

• dagrun_timeout. The author sets one budget without knowing how many entries a run will select. A plan of two short entries and a plan of eight long ones share it. On timeout the scheduler fails the DagRun and sets every unfinished task instance to skipped, so a run that selects a large shape under a budget written for a small one loses the entries still in flight, and the synthesized merge is skipped along with them regardless of the semantics in The merge node. "Terminates normally" there therefore excludes DagRun timeout, and the state-transition table has to cover it. Set the timeout for the worst plan max_selected allows and it stops being a useful signal for small plans. Whether the feature should expose a per-run budget, or let a planner propose one that the author's ceiling caps, is unresolved.
• Deadlines and SLA. The same problem in the other direction: a deadline on the region cannot be derived from a shape that does not exist yet.
• max_active_tasks and pool capacity. A plan selecting max_selected entries at once contends against everything else in the Dag. The cap bounds how many instances run. It says nothing about how much resource they draw.

This AIP solves none of them. A fixed author budget and a planner that decides how much work to do are in tension. The limits the author set have to win: a planner cannot raise dagrun_timeout or max_selected, only work inside them.

The plan accessor

A catalog entry reads its wiring from the task context, so it never hard-codes task ids and can be reused across shapes:

• plan.id -- this instance's id in the plan, which is what the emitted edges reference. Equal to plan.entry today.
• plan.entry -- which catalog entry this instance is running.
• plan.inputs -- the arguments the planner supplied for this task.
• plan.upstream(name) and plan.upstreams() -- outputs of this task's plan-upstream tasks, resolved by their instance index.
• plan.all_outputs() -- every selected task's output, which is what the merge reads.
• plan.edges, plan.selected, plan.raw -- the emitted shape, so a task can check its own wiring.

These are the running task's own view. The validator is handed the whole plan instead, so it reads plan.tasks (the records, each with id, entry and inputs) and asks about wiring with plan.has_edge(src, dst). Two views of one object, and which one you get depends on where you are.

name is a plan id. plan.upstream(name) raises when the planner did not wire name upstream of this task, rather than returning None. A task that wants to tolerate a missing upstream tests plan.edges first. This is the main thing that trips people up when writing a catalog entry, since the same entry runs under different wirings on different runs.

Examples

A deterministic planner, showing the parts that trip people up: how an entry reads its wiring instead of hard-coding task ids, where inputs come from, and downstream wiring with no trigger_rule on the author's side.

from airflow.sdk import DynamicTaskGroup, dag, get_current_context, task

@task
def validate() -> dict:
    plan = get_current_context()["plan"]
    return {"valid": True, "ticket_id": plan.inputs["ticket_id"]}

@task
def issue_refund() -> dict:
    plan = get_current_context()["plan"]
    checked = plan.upstream("validate")      # raises if the planner did not wire this edge
    return {"refunded": True, "ticket_id": checked["ticket_id"]}

@task
def notify() -> dict:
    plan = get_current_context()["plan"]
    return {"notified": sorted(plan.upstreams())}

def triage(context) -> dict:                 # a plain, unit-testable callable
    kind = context["params"]["ticket_kind"]
    ticket_id = context["params"]["ticket_id"]
    if kind == "refund":
        return {
            "tasks": [{"id": "validate", "entry": "validate",
                       "inputs": {"ticket_id": ticket_id}}, "issue_refund", "notify"],
            "edges": [["validate", "issue_refund"], ["issue_refund", "notify"]],
        }
    return {                                 # a fan-out instead of a chain
        "tasks": [{"id": t, "entry": t, "inputs": {"ticket_id": ticket_id}}
                  for t in ("enrich_policy", "enrich_claims")] + ["notify"],
        "edges": [["enrich_policy", "notify"], ["enrich_claims", "notify"]],
    }

def check_plan(plan) -> None:                # runs in the generator, before anything materializes
    if sum(t.entry == "notify" for t in plan.tasks) > 1:
        raise ValueError("notify is not idempotent; select it at most once")

@dag(params={"ticket_kind": "refund", "ticket_id": "T-1"})
def handle_ticket():
    resolve = DynamicTaskGroup(
        group_id="resolve",
        planner=triage,
        validator=check_plan,
        catalog=[validate, issue_refund, notify, enrich_policy, enrich_claims],
        merge=summarize,
        max_selected=5,
    )
    resolve >> publish()                     # attaches to the merge; no trigger_rule needed

Both run on one Dag version. A planner that returns {"tasks": []} selects nothing, the merge succeeds, and publish runs.

A precondition that spans tasks belongs in the validator, not in the task that would be harmed by breaking it:

def check_credit_plan(plan) -> None:        # the validator for a different group
    for t in plan.tasks:
        if t.entry == "issue_credit" and t.inputs.get("amount", 0) > 200:
            if not plan.has_edge("request_approval", t.id):
                raise ValueError("credit over the 200 cap requires request_approval upstream")

The same check can be written inside issue_credit against plan.edges, and an entry that wants to behave differently under different wirings has to read plan.edges anyway. But by the time the entry runs, the region has materialized and its siblings may already have fired. The validator is the last point at which rejecting costs nothing. Either way the framework enforces no cross-task precondition of its own: both the validator and the in-task check are author code.

Rendering

The grid needs no new concept. Because only selected entries materialize, it lists the instances that ran under the region, labeled by entry.

The graph draws the emitted shape as an overlay: the static topology comes from the serialized Dag as it does today, and the shape a particular run took is layered on top from that run's data. A read-only endpoint returns, per group, the instances that ran and the run-scoped wiring, and the client replaces the region's collapsed placeholder with one node per instance, wired by the emitted edges. Before the run the region renders as one collapsed placeholder next to the generator, the same model dynamic task mapping already established.

The graph today renders a mapped task as one node and does not lay out per-index nodes, so this is a new UI capability carrying a second source of truth for edges. It is more than an endpoint. And the overlay must be gated on the generator having succeeded in this run, or a cleared and re-run Dag draws the previous run's plan.

Where the emitted plan lives, and for how long

The planner is user code, so it runs on a worker. The scheduler is what needs the result, because it does the sequencing. The prototype writes the plan to a reserved XCom key. That works, and it has a problem: XCom is writable by any task in the Dag. The plan is validated once, inside the generator, at the moment it is written. Nothing stops a later task in the same Dag from overwriting that key afterwards, and the scheduler would then sequence against edges that were never validated, including a cycle. The validation is in the wrong place to defend the read.

So the plan needs somewhere only the generator can write and the scheduler can read. TaskMap, which already carries mapped-task expansion data from worker to scheduler, is the precedent for what that looks like: a dedicated table rather than a key in a user-writable namespace or Task State Store (AIP-103) most likely; we will pick what we use for AIP-111. Worth noting the row is written by the Execution API on the task's push rather than by the task itself, which is the property being borrowed.

Every consumer of the plan needs the same run-and-attempt gate: the scheduler dependency, the merge, the graph endpoint, and plan.upstream() inside a running entry. XCom keys carry no attempt number, so a retried generator overwrites its own plan while instances from the earlier attempt may still exist. Clearing does not help: clear_task_instances contains no XCom deletion, so a cleared generator leaves the previous plan in place for anything that reads it. Getting that gate consistently right across the SDK, the scheduler and the API server is where this feature is most likely to break, and it needs a specified data-access contract.

Retention is the second half of the same problem. If the emitted edges live in XCom, then db clean and XCom expiry destroy the record of what a planner composed, and the run stops being explainable at whatever the retention window is. If they live in a queryable run-scoped row so the graph endpoint can read them cheaply, that is a migration. This AIP does not get to claim both a durable audit record and no new storage, and which one it picks is an open question.

Composition with mapping and loops

Mapping, loops and dynamic groups all materialize a static region into runtime instances addressed by a single-dimension instance identity, which is why they reuse each other's machinery and why nesting one inside another has nowhere to put a second dimension. The fix needs a multi-dimension identity; this AIP calls it the composite instance key.

The closest existing case, task mapping inside a mapped task group, was omitted deliberately rather than because it was hard: "While the technical aspect of this feature is not particularly difficult, we have decided to intentionally omit this feature since it adds considerable UI complexities, and may not be necessary for general use cases. This restriction may be revisited in the future depending on user feedback" (airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst). The one-dimension limit affects mapping, loops and this equally, so it should be designed once.

Rejection is at parse time so an unsupported composition fails loudly instead of mis-sequencing.

The last row, a task group as a catalog entry, is a limitation for the selective-build case: "run these three transformations for model X" is not expressible when an entry is a single task. Selecting one entry more than once is cheaper than it looks. As The plan describes, the normalized plan already separates a task's id from the entry it runs; today an id must be a catalog label, which is what prevents naming the same entry twice. Lifting that gives runtime fan-out over one entry inside the existing single dimension.

The composite instance key is a separate proposal, not addressed here and expected to be derivative work. It is not contained or pre-approved by this AIP. This AIP does not add a fourth private expansion mechanism.

Dag serialization and version skew

A dynamic group must fail closed for any reader that predates the feature. An older reader treating the region as an ordinary mapped group would run entries the plan never selected. The practical consequence is a version floor: a Dag using dynamic groups will not run on an Airflow older than the release that adds them.

AIP-111 defers the same mechanism in the same words, which means nobody has specified it. Serialized Dags are read by the scheduler, the API server, the triggerer and workers, and mixed-version reads are normal during a rolling upgrade, so this should be specified once, jointly, instead of deferred twice.

Which users are affected?

Dag authors gain an optional construct. No existing Dag is affected.

Platform teams. A Dag using a dynamic group will not run on a scheduler or worker older than the release that adds it. That is a version floor during a rolling upgrade. Whether there is a DB migration depends on where the emitted plan lives (see Where the emitted plan lives).

Operator authors. No changes. A catalog entry is an ordinary task, subject to the shared-configuration limit above.

Migration Effort

None for existing Dags. Adoption is opt-in per region. The only platform consideration is the version floor.

Downsides

• Per-entry asset events are not available. outlets is resolved from the serialized Dag before task instances exist, so a run emits the region's outlets rather than one event per selected entry. Other task-level configuration depends on the materialization chosen; see Runtime model.
• New concepts to learn: the author-facing ones in Terminology, which are catalog, entry, planner, validator, plan, merge and max_selected. Whether that weight is justified depends on whether the work genuinely needs to be tasks; the docs must point at AIP-112 for the cases where they do not.
• No nesting. No mapped task or loop inside a dynamic group, and no dynamic group inside a loop, until the composite instance key lands.
• Side-effect safety across shapes stays the author's responsibility, with validator as the place to discharge it.
• Fail-closed downgrade cost. Rolling back to an Airflow without the feature wedges Dags that adopted it until they are removed or the version is restored.

Out of scope

• Inventing tasks at run time, meaning a plan naming a task not in any catalog. It has no serialized anchor, so it would either churn Dag versions per run or need a shadow per-run node store outside serialization, which is a second Dag versioning implementation.
• Per-index task configuration (see Runtime model).
• Nesting, and the composite instance key it needs.
• Replay of a recorded plan as core behavior.
• A UI for authoring dynamic groups.

What defines this AIP as "done"?

1. The authoring surface in the Task SDK (catalog=, planner= or an overridable plan(), validator= or an overridable validate_plan(), merge=, max_selected=), with parse-time validation of the catalog, reserved ids and duplicate labels.
2. The validator hook: what it is handed, that raising rejects the plan and fails the generator closed, and that it runs after the built-in checks and before anything materializes.
3. Runtime materialization of exactly the selected entries, an empty plan skipping the region, and an invalid plan failing it closed.
4. Sequencing by the emitted edges with failure and skip propagation, and no scheduling regression for Dags that do not use the feature.
5. A published state-transition table covering: empty plan, single-entry plan, an entry failing mid-shape, skip propagation along emitted edges, cleared generator, retried generator, reshape to a smaller plan with REMOVED upstreams at the merge, and what the task downstream of the merge sees in each case.
6. The merge node's success semantics implemented as specified, including the empty plan, and verified against all_success downstream.
7. A specified run-and-attempt data-access contract for the plan, honored identically by the scheduler dependency, the merge, the graph endpoint and the plan accessor.
8. Whether plan-supplied inputs are rendered, decided and documented.
9. Cross-task and merge state on the AIP-103 group scope, with the authorization model specified for the hand-off.
10. Grid renders only the tasks that ran; graph renders the emitted shape for a selected run from run-scoped data, with the Dag version unchanged across shapes, and a named owner for the UI work.
11. Retention and audit lifetime for the emitted plan stated, including db clean and XCom expiry.
12. Serialization fails closed for older readers, specified jointly with AIP-111, with the version floor documented.
13. Rerun, clear and reshape semantics defined, including REMOVED for tasks absent from a new shape.
14. Scheduler cost measured for a Dag that uses the feature, including query count per scheduling loop and the combined cost with AIP-111's dependency.
15. Nesting rejected at parse time with an error naming the composite-key limitation.
16. dags test, backfill, sensors, deferrable tasks and an approval task inside a selected shape all work, or the ones that cannot are documented as unsupported.
17. Docs and examples cover a deterministic planner and a model-based planner, with the side-effect-safety constraint in the introduction.
18. Tests cover plan validation, expansion, edge sequencing, failure and skip propagation, the empty plan, a retried and a cleared generator, serialization round-trip and version skew, and the render overlay.

Depends on, and not delivered here: AIP-103 group-scoped state; the composite instance key, for nesting.

Open questions

1. Are plan-supplied inputs Jinja-rendered? As prototyped they land in op_kwargs, which is a template field on decorated operators, so a planner-authored string would be rendered against the task context and could read connections and variables. That would defeat the allowlist for the agentic case. Should plan inputs be exempt from templating, or should the governance claims be scoped to match? A third option raised in review is to keep templating but scope it, per-entry or against a restricted context, so the feature is not lost where the planner is trusted. Views welcome; this is the question I most want settled on the thread.

2. Per-entry pools, and per-entry task configuration generally. Early feedback asked for this before the design was reviewed: a catalog of mixed actions wants different pools, and arguably different retry counts and queues. Under the prototype's materialization, every entry in a run shares one dispatching task's configuration. Is a shared pool acceptable for a first version, or is per-entry configuration a precondition for the feature being useful? If it is a precondition, this AIP is substantially larger than described, and I would rather find that out now.

3. How should a runtime-chosen shape interact with dagrun_timeout and deadlines? See Interaction with Dag-level settings.

4. Selecting one entry more than once. Lift the "a task's id must be a catalog label" rule now, so a plan can select the same entry several times with different inputs, or defer it? Recommendation: lift it. The plan model already separates a task's id from the entry it runs, so this is a validation change. It is also the cheapest step toward runtime fan-out.

5. Where the emitted plan lives, given that a durable audit record and "no new storage" are not both available.

6. Is a new Task Group type the right shape? The novel parts are a mapped task whose expand source is a plan and a dependency that sequences map indices by runtime edges. A smaller framing exists, closer to AIP-42 plus edge sequencing plus a dispatch operator, and it would not need the composite key to be useful. I have not written that alternative up and would rather hear whether the list wants it before I do.

7. Should a catalog be a first-class shareable object that a platform team defines and Dag authors reference, or is a per-Dag literal enough? Recommendation: per-Dag literal now.

8. Should @task_group gain a .dynamic() method, symmetric with AIP-111's .loop(), or is the DynamicTaskGroup constructor the whole authoring surface?

9. Can a deferrable task or a human-in-the-loop approval defer from inside a dispatched entry? TaskDeferred unwinds the stack and resumption needs the right operator class to receive the resume call. I do not yet know whether an entry dispatched by the region can defer and resume correctly. If it cannot, the mid-shape approval case in the AIP-112 comparison under Relationship to other AIPs has to be struck.