A Failed Airflow Task Can Still End in a Successful DAG Run

A downstream consumer that sees an Airflow DAG run marked success may be looking at a valid scheduler result and still be making the wrong operational decision. In Airflow’s documented model, the terminal DAG-run state is determined from leaf tasks, not from an assertion that every required task succeeded. A successful all_done cleanup that is the sole leaf can therefore coexist with an earlier failed task and a green run. Apache Airflow’s Dag Runs documentation explicitly warns about that case.

This playbook isolates that one defect class: Airflow DAG run success after task failure. It separates three propositions that operators too often collapse into one: a task failed; the scheduler assigned the run success; and an external owner should accept the run. The laboratory uses only manually triggered DAGs with schedule=None, catchup=False, retries=0, inert local or log-only effects, and no production connections. It compares ordinary tasks only.

The repair has two layers. First, the graph gets a dedicated all_success verdict leaf with the correct direct parents. Second, an external acceptance gate checks the complete task inventory and requires every named required task plus mandatory cleanup to be success. That second layer matters because a controlled skip can still produce a successful DAG run. Airflow 3.3.0 is the intended documentation baseline, not a claim about the newest release.

Define the run signal your downstream consumer trusts

Assume a harmless consumer named release_probe watches one manually triggered run and, if accepted, writes only a local JSON decision such as {"decision":"accept"}. It does not publish data, notify customers, refresh dashboards, or call a production service. That keeps the experiment about scheduler semantics rather than business effects.

The weak contract is: “accept when the DAG run state is success.” The stronger contract is: “accept only the expected run identity and graph revision when the complete expected task inventory is present, the DAG run is success, every required business task is success, mandatory cleanup is success, and no required state is missing, skipped, removed, unknown, or nonterminal.” Those are different predicates.

Airflow documents why. A DAG run succeeds when all leaf nodes are success or skipped; it fails when a leaf is failed or upstream_failed. A leaf is simply a task with no children. That means the green run badge is a statement about terminal leaf-state evaluation, not a universal quantifier over every task instance.

That distinction is the operational boundary of this article. Broader context on Airflow’s place in the data toolchain is useful for orchestration literacy, but it is not evidence for this leaf-state mechanism. The mechanism comes from Airflow’s own scheduler documentation.

Treat the downstream consumer’s acceptance policy as a separate, versioned contract. Airflow is allowed to report its documented scheduler result; the consumer is responsible for deciding whether that result contains enough evidence for business acceptance. “Green” is therefore an input to the gate, not the gate itself.

Freeze a disposable Airflow baseline

The experiment should run in one localhost-only, self-managed environment. Use the Airflow 3.3.0 image family and record what is actually inside the running container before trusting any trace; the 3.3.0 Python 3.13 image is published in the official image repository. Airflow’s 3.3.0 documentation shows airflow.sdk imports for DAG and trigger-rule objects, while common operators such as PythonOperator and EmptyOperator are supplied by apache-airflow-providers-standard.

A same-day documentation check exposes an important reproducibility lesson. The fixed documentation snapshot rendered 3.3.0, but a later check on September 24, 2026 found that the mutable stable selector rendered 3.3.2. The archived 3.3.0 page still carries the same leaf-state warning. This playbook therefore uses 3.3.0 as an intentionally frozen semantic baseline; the stable selector is not an environment lock or evidence of the latest release.

Create the disposable directory and start a local container:

mkdir -p leaf-audit/{dags,tests,evidence}
cd leaf-audit

docker run --name leaf-audit-airflow --rm \
  -p 127.0.0.1:8080:8080 \
  -e AIRFLOW__CORE__LOAD_EXAMPLES=False \
  -e AIRFLOW__DATABASE__SQL_ALCHEMY_CONN='sqlite:////opt/airflow/airflow.db' \
  -v "$PWD/dags:/opt/airflow/dags" \
  -v "$PWD/tests:/opt/airflow/tests" \
  -v "$PWD/evidence:/opt/airflow/evidence" \
  apache/airflow:3.3.0-python3.13 standalone

Record, rather than assume, the exact runtime:

docker exec leaf-audit-airflow airflow version
docker exec leaf-audit-airflow python --version
docker exec leaf-audit-airflow python - <<'PY'
from importlib.metadata import version
import sqlite3

print("standard_provider=", version("apache-airflow-providers-standard"))
print("sqlite=", sqlite3.sqlite_version)
PY
docker exec leaf-audit-airflow airflow config get-value core executor
docker exec leaf-audit-airflow airflow config get-value database sql_alchemy_conn
docker exec leaf-audit-airflow python -m pip freeze | sort > evidence/pip-freeze.txt

Also run airflow version, airflow info, and the relevant command --help text as part of the evidence bundle; Airflow 3.3.0 documents dags trigger, dags state, and tasks states-for-dag-run as supported CLI surfaces.

Treat the resulting manifest as evidence, not decoration. It should contain the container image reference, apache-airflow version, Python version, apache-airflow-providers-standard version, configured executor, metadata-database engine and SQLite library version, plus the SHA-256 hashes of both DAG files, the tests, the gate, and the required-task manifest. Record the exact command lines used to create each run. The reason is diagnostic: if the observed state ledger disagrees with the known-answer ledger, an owner must be able to tell whether the disagreement belongs to graph code, package resolution, scheduler configuration, evidence collection, or the documented assumption.

Do not fill an “actual” column from a Docker tag or documentation page. A tag identifies the requested artifact, but the acceptance record is the output of the running installation. Likewise, the standard-provider version must come from package metadata in that container. This distinction is especially important because provider packages can have their own release cadence. The import smoke test is a release gate: if DAG, TriggerRule, EmptyOperator, PythonOperator, or AirflowSkipException cannot import in the recorded runtime, stop before generating scheduler evidence.

Evidence status: the environment available for preparing this playbook exposed Python 3.13.5 but had no Airflow installation. An attempted constrained Airflow 3.3.0 installation could not fetch dependencies because outbound package-network access was unavailable. No scheduler execution below is therefore labeled observed. The code is complete, and the ledgers are documented or mathematically expected; a platform owner must capture the exact Airflow, Python, standard-provider, executor, and SQLite versions above before changing the decision from HOLD. This is also why workflow orchestration foundations are context, not a substitute for an environment manifest.

This table is intentionally incomplete rather than fabricated. Until the rightmost evidence exists for the same scheduler that produced the test runs, the laboratory has no “tested Airflow 3.3.0” claim.

Write the task-state acceptance ledger first

Write expected outcomes before reading runtime output. That prevents the implementation from defining its own expected result after the fact. The fixture has four operational tasks: prepare, transform, publish, and cleanup. prepare, transform, and publish are required work. cleanup is also mandatory under this article’s explicit policy: a cleanup failure makes the run unacceptable. The repaired graph adds a control task, verdict.

Airflow distinguishes failed, upstream_failed, and skipped. A task is failed when its execution errors; upstream_failed means an upstream task failed and the trigger rule required it; and skipped is its own state. Those task states are evidence distinct from the DAG-run state. Apache Airflow’s Tasks documentation defines those states, while the DAG-run page defines the separate leaf-based terminal rule.

The independent known-answer ledger is:

Before executing anything, save this ledger with a review identity such as known-answer-v1. That file is the experiment's oracle; runtime collectors are not allowed to rewrite it. The graph revision can change only through an explicit code review that also updates topology expectations and the external manifest. This keeps “what should happen” independent from “what happened,” a basic control when the very signal under investigation can be misleading.

The task-state entries are expected scheduler consequences of the documented trigger rules; the run-state entries follow mathematically from the documented leaf rule. They are not observed scheduler output in the evidence available for this playbook. Attempt numbers, timestamps, and exact scheduler transition times must be captured from the executed run rather than invented.

Read the graph from direct parents to leaves

For the flawed graph the edge set is exactly prepare -> transform, transform -> publish, and publish -> cleanup. The only leaf is cleanup. For the repaired graph those same edges remain, and four additional edges point to the verdict: prepare -> verdict, transform -> verdict, publish -> verdict, and cleanup -> verdict. The only leaf becomes verdict.

That direct-parent wording matters. Apache Airflow’s DAG trigger-rule documentation says the default all_success rule waits for all upstream direct parents to succeed, while all_done waits for upstream tasks to be done regardless of success. A task placed last in a picture is not automatically a full-workflow assertion. Its direct dependencies and its membership in the leaf set determine what evidence it contributes.

Run the failed-work, successful-cleanup case

The intentionally flawed DAG keeps the failure synthetic and effects inert. transform reads a run-conf mode and raises an ordinary RuntimeError; cleanup only logs. The sole cleanup leaf uses all_done.

# dags/leaf_audit_flawed.py
import pendulum

from airflow.exceptions import AirflowSkipException
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, TriggerRule

GRAPH_REVISION = "leaf-audit-v1"


def transform_fn(**context):
    mode = context["dag_run"].conf.get("transform_mode", "success")
    print(f"transform_mode={mode}")
    if mode == "fail":
        raise RuntimeError("synthetic transform failure")
    if mode == "skip":
        raise AirflowSkipException("synthetic required-task skip")
    print("transform completed: inert fixture")


def cleanup_fn(**context):
    print("cleanup completed: inert fixture")


with DAG(
    dag_id="leaf_audit_flawed_v1",
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 0},
    tags=[f"graph_revision:{GRAPH_REVISION}"],
) as dag:
    prepare = EmptyOperator(task_id="prepare")
    transform = PythonOperator(task_id="transform", python_callable=transform_fn)
    publish = EmptyOperator(task_id="publish")
    cleanup = PythonOperator(
        task_id="cleanup",
        python_callable=cleanup_fn,
        trigger_rule=TriggerRule.ALL_DONE,
    )

    prepare >> transform >> publish >> cleanup

AirflowSkipException is present only so the same inert callable can support the later skip challenge; Airflow documents that raising it marks the current task skipped. The first control uses transform_mode=fail, not skip.

After the scheduler has parsed the file, unpause and manually trigger a unique run. Airflow 3.3.0 documents -r/--run-id, -c/--conf, JSON output, and the warning that a paused DAG leaves a triggered run queued.

docker exec leaf-audit-airflow airflow dags unpause -y leaf_audit_flawed_v1

docker exec leaf-audit-airflow airflow dags trigger \
  -r 'manual__leaf-audit-v1__transform-fail__20260924T120000Z' \
  -c '{"graph_revision":"leaf-audit-v1","transform_mode":"fail"}' \
  -o json \
  leaf_audit_flawed_v1

Poll with a deadline instead of sleeping an arbitrary amount, then capture both run and task-instance states. airflow dags state accepts a run ID, and airflow tasks states-for-dag-run can return JSON for all task instances in that run. If the deadline expires, the evidence is incomplete and the external decision is HOLD.

Follow the failed task through upstream_failed to green

The expected chain is not a scheduler malfunction. prepare succeeds. transform executes once because retries are zero, raises the controlled exception, and becomes failed. publish has the default all_success rule, so its required parent did not succeed; Airflow defines the resulting blocked state as upstream_failed. cleanup has all_done, and its direct parent publish is terminal, so cleanup is eligible to run and succeeds.

At that point cleanup is the only leaf. Airflow’s documented DAG-run rule says a run is success when all leaves are success or skipped, and its own note specifically warns that a successful all_done leaf can make the whole run successful even when something failed in the middle. That is the precise Airflow cleanup masks failure case.

The consumer must therefore retain two statements at once: “the scheduler is expected to assign this flawed run success” and “the external owner must reject it because required task transform is failed and publish is upstream_failed.” Conflating those statements would be the defect.

Add a verdict leaf with the right dependencies

The graph-level repair keeps the business-task behavior and cleanup behavior unchanged. It adds a normal EmptyOperator named verdict with all_success, and critically makes prepare, transform, publish, and cleanup direct parents. Cleanup remains all_done, so it can still run after failed work.

# dags/leaf_audit_repaired.py
import pendulum

from airflow.exceptions import AirflowSkipException
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, TriggerRule

GRAPH_REVISION = "leaf-audit-v2"


def transform_fn(**context):
    mode = context["dag_run"].conf.get("transform_mode", "success")
    print(f"transform_mode={mode}")
    if mode == "fail":
        raise RuntimeError("synthetic transform failure")
    if mode == "skip":
        raise AirflowSkipException("synthetic required-task skip")
    print("transform completed: inert fixture")


def cleanup_fn(**context):
    fail_cleanup = bool(context["dag_run"].conf.get("fail_cleanup", False))
    print(f"fail_cleanup={fail_cleanup}")
    if fail_cleanup:
        raise RuntimeError("synthetic mandatory cleanup failure")
    print("cleanup completed: inert fixture")


with DAG(
    dag_id="leaf_audit_repaired_v2",
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    catchup=False,
    default_args={"retries": 0},
    tags=[f"graph_revision:{GRAPH_REVISION}"],
) as dag:
    prepare = EmptyOperator(task_id="prepare")
    transform = PythonOperator(task_id="transform", python_callable=transform_fn)
    publish = EmptyOperator(task_id="publish")
    cleanup = PythonOperator(
        task_id="cleanup",
        python_callable=cleanup_fn,
        trigger_rule=TriggerRule.ALL_DONE,
    )
    verdict = EmptyOperator(
        task_id="verdict",
        trigger_rule=TriggerRule.ALL_SUCCESS,
    )

    prepare >> transform >> publish >> cleanup
    [prepare, transform, publish, cleanup] >> verdict

Under the no-skip failure case, cleanup can succeed, but verdict cannot: one direct parent is failed and another is upstream_failed. The expected verdict state is upstream_failed; because verdict is now the only leaf, the run is expected to be failed. This is an Airflow leaf node success rules repair, not yet the full business-acceptance repair.

Why checking only the cleanup parent repeats the bug

A tempting but wrong edge set is only cleanup -> verdict. With that graph, verdict’s all_success rule asks one question: did cleanup succeed? It does not recursively reinterpret all ancestors as direct-parent assertions. Airflow’s trigger-rule description is explicit that these checks concern direct upstream parents. A successful cleanup would therefore make that poorly wired verdict succeed and recreate the false green one hop later.

Make the topology executable evidence with a structural test:

# tests/test_graphs.py
import sys

sys.path.insert(0, "/opt/airflow")

from dags.leaf_audit_flawed import dag as flawed
from dags.leaf_audit_repaired import dag as repaired
from airflow.sdk import TriggerRule


def leaves(dag):
    return {t.task_id for t in dag.tasks if not t.downstream_task_ids}


def main():
    assert leaves(flawed) == {"cleanup"}
    assert flawed.get_task("cleanup").upstream_task_ids == {"publish"}
    assert flawed.get_task("cleanup").trigger_rule == TriggerRule.ALL_DONE

    verdict = repaired.get_task("verdict")
    assert leaves(repaired) == {"verdict"}
    assert verdict.upstream_task_ids == {
        "prepare", "transform", "publish", "cleanup"
    }
    assert verdict.trigger_rule == TriggerRule.ALL_SUCCESS
    assert repaired.get_task("cleanup").trigger_rule == TriggerRule.ALL_DONE
    assert set(t.task_id for t in repaired.tasks) == {
        "prepare", "transform", "publish", "cleanup", "verdict"
    }
    print("structural assertions: PASS")


if name == "__main__":
    main()

Airflow’s Best Practices documentation includes DAG loading and dependency-structure unit tests, so this kind of edge assertion is a supported testing pattern. It proves graph shape; it does not prove what a live scheduler actually assigned to a particular run.

Prove the happy path and the cleanup-failure policy

The repaired graph needs positive evidence, not just a failure test. Trigger three separate runs with unique run IDs: transformed work failure, happy path, and mandatory-cleanup failure. Reusing one run and manually changing its state would destroy the evidence chain.

docker exec leaf-audit-airflow airflow dags unpause -y leaf_audit_repaired_v2

# Required-work failure
docker exec leaf-audit-airflow airflow dags trigger \
  -r 'manual__leaf-audit-v2__transform-fail__20260924T121000Z' \
  -c '{"graph_revision":"leaf-audit-v2","transform_mode":"fail","fail_cleanup":false}' \
  leaf_audit_repaired_v2

# Positive control
docker exec leaf-audit-airflow airflow dags trigger \
  -r 'manual__leaf-audit-v2__happy__20260924T122000Z' \
  -c '{"graph_revision":"leaf-audit-v2","transform_mode":"success","fail_cleanup":false}' \
  leaf_audit_repaired_v2

# Mandatory cleanup failure
docker exec leaf-audit-airflow airflow dags trigger \
  -r 'manual__leaf-audit-v2__cleanup-fail__20260924T123000Z' \
  -c '{"graph_revision":"leaf-audit-v2","transform_mode":"success","fail_cleanup":true}' \
  leaf_audit_repaired_v2

For the happy path, the known-answer result is all five task instances success, verdict as the only successful leaf, and DAG run success. For cleanup failure, prepare, transform, and publish should succeed; cleanup should fail; verdict should become upstream_failed; and the run should fail because its leaf is upstream_failed. Those are expected results derived from the fixed graph and documented terminal rule, not scheduler observations from an executed run.

The cleanup policy must not drift between layers. In this playbook cleanup is mandatory. Therefore both graph and external gate reject a cleanup failure. A different organization might classify cleanup as best-effort, but then it must explicitly redesign both the graph and acceptance manifest. Quietly making cleanup optional in one layer and mandatory in another produces contradictory release evidence.

There is also a stop condition: if the scheduler-observed trace differs from the known-answer ledger, for example when a task gets a state not predicted here, do not “fix” the ledger to make the run pass. Freeze the run ID, capture versions, graph hash, task inventory, states, attempts, and logs, then investigate the mismatch. Documentation, installed software, and evidence are three independent artifacts. An unexplained disagreement means HOLD.

Challenge the repair with a skipped required task

A good repair test attacks its own assumptions. Run the same repaired DAG with the required transform task deliberately raising AirflowSkipException. This is not branching; it is a controlled state injection inside an ordinary Python task. Airflow documents that AirflowSkipException marks the current task skipped.

docker exec leaf-audit-airflow airflow dags trigger \
  -r 'manual__leaf-audit-v2__required-skip__20260924T124000Z' \
  -c '{"graph_revision":"leaf-audit-v2","transform_mode":"skip","fail_cleanup":false}' \
  leaf_audit_repaired_v2

The independent expected trace is: prepare=success; transform=skipped; publish=skipped; cleanup=success; verdict=skipped; DAG run success. Why? Airflow states that skipped tasks cascade through all_success. publish therefore skips after its required parent skips. Cleanup uses all_done, so it can still run once its parent is terminal. Verdict uses all_success and has skipped direct parents, so the skip propagates there as well.

This is the skip challenge that prevents the graph repair from becoming a new false guarantee. An all_success verdict is useful for the failure case, but it is not equivalent to “every named required task ended in exactly the string success.”

A skipped leaf can still allow a successful run

Airflow’s run rule counts a skipped leaf as compatible with run success. Therefore a skipped verdict leaf can still permit a successful run even though the strict business contract says a required task may not be skipped. This is the Airflow skipped task false success case and the reason the external check must remain fail-closed.

The repaired graph should stay: it converts ordinary failure into a failed leaf while preserving cleanup execution. The external gate should add the stricter invariant: every task ID in required_success must be present exactly once and have state success; cleanup must be present and success; the control verdict should be success; and the complete task inventory must match the expected graph revision. A skip, removal, missing task, unknown state, or still-running task is nonaccepting even if Airflow itself legitimately reports the run success.

That division of responsibility is deliberate. The DAG’s trigger rules encode scheduling behavior. The external manifest encodes the consumer’s acceptance semantics. Neither should be misrepresented as the other.

Build the external acceptance gate

Make the gate consume public evidence rather than importing Airflow metadata models into task code. Airflow 3’s public-interface guidance identifies supported integration surfaces including the REST API and Task SDK, while the 3.3.0 CLI reference documents run-state and all-task-state commands; direct metadata-database coupling is not the contract this playbook relies on.

Version the required-task manifest beside the DAG:

{
  "dag_id": "leaf_audit_repaired_v2",
  "graph_revision": "leaf-audit-v2",
  "expected_run_state": "success",
  "task_inventory": ["prepare", "transform", "publish", "cleanup", "verdict"],
  "required_success": ["prepare", "transform", "publish", "cleanup", "verdict"]
}

Normalize the collected scheduler evidence to this small schema. The collector may use the Airflow 3.3.0 CLI or REST API, but it must preserve the original values and retain the raw response separately:

{
  "dag_id": "leaf_audit_repaired_v2",
  "run_id": "manual__leaf-audit-v2__happy__20260924T122000Z",
  "graph_revision": "leaf-audit-v2",
  "run_state": "success",
  "complete": true,
  "tasks": [
    {
      "task_id": "prepare",
      "state": "success",
      "try_number": 1,
      "start_date": "...",
      "end_date": "..."
    }
  ]
}

Then use a client-side gate that refuses empty, partial, duplicated, or unknown evidence:

# gate.py
import argparse
import json
import sys
from pathlib import Path

TERMINAL = {"success", "failed", "skipped", "upstream_failed", "removed"}


def load(path):
    return json.loads(Path(path).read_text())


def evaluate(manifest, evidence, expected_run_id):
    hold, repair = [], []

    if evidence.get("dag_id") != manifest["dag_id"]:
        hold.append("dag_id mismatch")
    if evidence.get("run_id") != expected_run_id:
        hold.append("run_id mismatch")
    if evidence.get("graph_revision") != manifest["graph_revision"]:
        hold.append("graph_revision mismatch")
    if evidence.get("complete") is not True:
        hold.append("task inventory not proven complete")

    tasks = evidence.get("tasks")
    if not isinstance(tasks, list) or not tasks:
        hold.append("task list missing or empty")
        tasks = []

    by_id = {}
    for row in tasks:
        task_id = row.get("task_id")
        if not task_id or task_id in by_id:
            hold.append(f"duplicate or invalid task_id: {task_id!r}")
            continue
        by_id[task_id] = row

        state = row.get("state")
        if state not in TERMINAL:
            hold.append(
                f"nonterminal or unknown state: {task_id}={state!r}"
            )

    observed = set(by_id)
    expected = set(manifest["task_inventory"])
    if observed != expected:
        hold.append(
            f"inventory mismatch missing={sorted(expected - observed)} "
            f"unexpected={sorted(observed - expected)}"
        )

    if evidence.get("run_state") != manifest["expected_run_state"]:
        repair.append(
            f"run_state={evidence.get('run_state')!r}, "
            f"expected={manifest['expected_run_state']!r}"
        )

    for task_id in manifest["required_success"]:
        state = by_id.get(task_id, {}).get("state")
        if state != "success":
            repair.append(
                f"required task {task_id} is {state!r}, not 'success'"
            )

    if hold:
        return "HOLD", hold + repair
    if repair:
        return "REPAIR", repair
    return "ACCEPT", []


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--manifest", required=True)
    p.add_argument("--evidence", required=True)
    p.add_argument("--run-id", required=True)
    a = p.parse_args()

    decision, reasons = evaluate(
        load(a.manifest),
        load(a.evidence),
        a.run_id,
    )
    print(json.dumps({"decision": decision, "reasons": reasons}, indent=2))
    sys.exit(0 if decision == "ACCEPT" else 2)


if name == "__main__":
    main()

This code is a client implementation of a local acceptance specification, not an Airflow scheduler specification. Its fail-closed rule is intentionally stricter than DAG-run success. REVERT is not inferred from one run; it is a release-owner action when a newly deployed graph/gate revision regresses against a preserved known-good revision.

Keep three source roles separate. Airflow's core-concept pages are explanatory project documentation for scheduler behavior. The supported CLI or REST interface is the integration contract used to retrieve scheduler-owned evidence. The JSON manifest above is the consumer's normative acceptance specification for this workflow revision. Finally, gate.py is merely one client implementation of that local specification. A defect in the client must not be “resolved” by weakening the manifest, and a change in Airflow documentation must not silently alter an already approved business contract.

The gate also needs a provenance wrapper around normalized evidence. Store the raw API or CLI payload, its content hash, collector version, collection timestamp, and normalization result. complete=true may be set only by the collector after it has proved full inventory coverage; a caller must not be able to hand-edit that flag to make a partial response acceptable. In a REST implementation, follow every page until the server-reported total has been collected. In a CLI implementation, reject malformed JSON or a task list that does not exactly reconcile with the manifest. Those rules turn absence of evidence into HOLD rather than accidental success.

Collect scheduler evidence without rewriting history

An evidence bundle should make the scheduler decision reconstructable without touching task states. For each run retain: DAG ID, run ID, graph revision, run state, full task inventory, each task state, try number, start/end timestamps, trigger rule, direct-parent IDs, leaf set, environment versions, code hash, command invocation, and raw logs for executed tasks. That is the minimum useful scheduler state reconciliation packet for this defect class.

Use the CLI only from outside task code. A bounded shell harness can poll the run, then save the all-task state response:

DAG='leaf_audit_repaired_v2'
RUN='manual__leaf-audit-v2__happy__20260924T122000Z'
DEADLINE=$((SECONDS + 120))

while (( SECONDS < DEADLINE )); do
  STATE=$(docker exec leaf-audit-airflow \
    airflow dags state "$DAG" "$RUN" | tail -n 1)

  printf '%s run_state=%s\n' "$(date -u +%FT%TZ)" "$STATE" \
    >> "evidence/${RUN}.poll.log"

  if [[ "$STATE" == "success" || "$STATE" == "failed" ]]; then
    break
  fi

  sleep 2
done

if [[ "$STATE" != "success" && "$STATE" != "failed" ]]; then
  echo 'HOLD: deadline expired before terminal run state' >&2
  exit 2
fi

docker exec leaf-audit-airflow airflow tasks states-for-dag-run \
  -o json "$DAG" "$RUN" \
  > "evidence/${RUN}.task-states.json"

docker exec leaf-audit-airflow airflow dags state "$DAG" "$RUN" \
  > "evidence/${RUN}.run-state.txt"

The exact JSON fields emitted by the installed CLI must be checked with --help and inspected rather than assumed; a normalization step should preserve the raw file and fail if required fields are absent. Airflow 3.3.0 documents the commands and JSON option, but the acceptance code remains responsible for schema validation.

Graph evidence should come from the checked-in DAG plus structural test output, not from a screenshot alone. Save a source hash and an explicit topology manifest. Logs belong in the evidence packet too, but logs do not override scheduler state. Broader monitoring and logging responsibilities provide operational context; for this acceptance decision the authoritative inputs are the preserved Airflow run/task records and the versioned graph contract.

A practical evidence directory should be immutable after the decision is signed. Name it by DAG ID and run ID, and include runtime.txt, trigger.json, run-state.txt, task-states.json, topology.json, source.sha256, poll.log, and task logs where attempts actually executed. The normalized evidence.json used by the gate should point back to those raw artifacts by hash. This prevents a later operator from accidentally comparing a run-state file from one manual run with task states from another. Identity reconciliation is part of correctness, not clerical cleanup.

Attempts also need careful interpretation. With retries=0, an executed task should not enter a retry cycle, but blocked tasks such as upstream_failed may not have a worker execution attempt comparable to a failed Python task. Capture try_number exactly as Airflow reports it; do not manufacture a uniform value of one for every row. The acceptance contract here uses final state, identity, and completeness; attempt numbers are retained for diagnosis and audit rather than treated as a substitute for task state.

Static tests and executed runs prove different things

Run the loader and structure checks before scheduler tests:

docker exec leaf-audit-airflow \
  python /opt/airflow/dags/leaf_audit_flawed.py

docker exec leaf-audit-airflow \
  python /opt/airflow/dags/leaf_audit_repaired.py

docker exec leaf-audit-airflow \
  python /opt/airflow/tests/test_graphs.py

Airflow’s Best Practices page says a loader test catches import and dependency errors and shows unit tests that assert DAG structure. It also distinguishes a whole-DAG dag.test() integration exercise from those unit tests. Neither parsing nor a local test should be mislabeled scheduler-observed deployment evidence.

The complete acceptance campaign is one flawed control plus four repaired-graph runs: required-work failure, happy path, cleanup failure, and required-task skip. For every run, compare observed scheduler state to the ledger written earlier. Scheduler observations are still pending in the evidence available for this playbook, so the correct decision is HOLD, not an invented “pass.” Once the exact environment executes the campaign, the platform owner should attach observed evidence and preserve any mismatch rather than editing history.

Reconcile missing, renamed and nonterminal tasks

The gate must be robust against evidence shape, not only state values. A stale manifest is a normal operational hazard. Suppose graph revision leaf-audit-v3 renames publish to publish_local, but the consumer still requires publish. An implementation that filters observed tasks to “whatever required names happen to exist” could produce an empty or shortened all-success list and accidentally pass. The correct result is HOLD because identity and inventory do not reconcile.

Test three negative cases directly against gate.py. First, remove one expected task row from the evidence and leave complete=false: HOLD. Second, present publish_local where publish is expected: HOLD with both a missing and unexpected ID. Third, set a required task to running, queued, none, or an unknown future string: HOLD because the state is not an accepted terminal observation. Airflow’s task-state documentation shows that queued, running, retry-related, and other lifecycle states are distinct from terminal success.

A removed task deserves special suspicion. Airflow documents removed as a task that vanished from the DAG after the run started, and trigger-rule logic can count it as “done” for rules such as all_done; it does not count as success. The external contract therefore rejects removed for every required task and treats an inventory discrepancy as nonaccepting even if the scheduler can finish the run.

For REST-based collectors, pagination is part of correctness. Continue until the reported total is fully collected, reject duplicate IDs, and mark complete=true only after proving the page sequence covers the entire response. A timeout, truncated page, missing total, decode failure, or authorization error is HOLD. Unknown evidence must never collapse into an empty all-success set.

Repair downstream decisions already made from a false green

A graph change repairs future scheduling semantics; it cannot undo an external action already taken from a false green. If a consumer previously trusted only DAG-run success, start by identifying the affected acceptance window using immutable run IDs and the graph revision that produced each run. Do not bulk-mark historical task instances successful to make the history look consistent.

The DAG owner should enumerate candidate runs whose graph had an all_done sole leaf or equivalent acceptance risk. The platform owner should collect each run’s complete task inventory and reconcile required states against the versioned contract. The downstream consumer owner should place automated consumption on HOLD while that review is incomplete. If an already accepted run contains failed, upstream_failed, skipped, removed, or missing required work, classify the prior acceptance as invalid even if the recorded DAG-run state is legitimately success. Airflow’s documented leaf rule is precisely why those facts can coexist.

Assessment of real downstream outputs is system-specific and outside this inert laboratory. The safe general rule is that changing a trigger rule does not reverse files written, tables replaced, messages emitted, or external services called before the repair. Owners must decide whether each downstream artifact should be quarantined, recomputed, invalidated, or left in place with an exception record. That is an operational remediation decision, not something Airflow can infer from graph state.

The same discipline used in data lake and warehouse operational practices applies at the boundary: preserve evidence before mutation and keep ownership explicit. The link provides broader platform context; the specialist acceptance test remains the task-state reconciliation described here.

Roll out the graph and gate as one versioned change

Do not deploy the new verdict edge set while leaving the old consumer predicate untouched. The graph and the required-task manifest form one acceptance contract and should share a release identifier. In this fixture that pair is leaf-audit-v2: the DAG carries it as a tag and each manual run conf carries it as graph_revision; the external manifest requires the same string. The gate also pins the expected run ID supplied by the caller.

Before rollout, preserve the previous DAG file, manifest, environment manifest, and known-answer evidence. Then run, in the disposable environment, the structural tests and the full campaign: flawed control, repaired required-work failure, repaired happy path, repaired cleanup failure, repaired skip. The first demonstrates the hazard; the next four demonstrate the new graph and the stricter external contract. None of those results should be promoted from “expected” to “observed” until the scheduler evidence exists.

For a real release, compare code hashes, edge lists, leaf sets, task inventories, task states, and run states between candidate and known-good revisions. A green happy path is necessary but not enough: the negative controls must fail closed in the intended layer. The skip case is especially valuable because it proves the consumer is not outsourcing strict acceptance to all_success semantics.

This sequencing complements broader cloud-native pipeline delivery practices without changing executors or infrastructure to manufacture a result. The point is to hold the platform constant while testing one graph semantic.

Rollback must preserve a conservative consumer gate

If the candidate graph causes an unexplained scheduler regression, the platform release owner may REVERT the DAG package to the last known-good graph revision. That does not justify reverting the consumer to “DAG run success means accept.” The external gate protects against exactly the graph defects that rollback can reintroduce.

Rollback therefore has an asymmetry: code may revert, evidence standards may not become weaker. If the restored graph cannot satisfy the current manifest, automated consumption stops at HOLD until the contract is reconciled. Preserve both the failed rollout runs and the restored revision; do not delete inconvenient evidence.

A conservative gate also gives rollback a clean stop condition. If run ID, revision, inventory, or task states cannot be verified after restoration, there is no ACCEPT. This is safer than treating absence of contrary evidence as success.

Use the accept, repair, hold and revert decision matrix

The final decision should be mechanical enough for automation but explicit enough for accountable humans. “Accept” means the evidence satisfies the declared contract; it does not mean Airflow has certified business correctness. “Repair” means the evidence is complete enough to prove a contract violation. “Hold” means the evidence itself is incomplete, mismatched, nonterminal, or untrusted. “Revert” is a release action when a newly introduced graph/gate revision regresses and a preserved prior revision is the approved recovery target.

For the canonical flawed run, the scheduler-level expected result is success, but the external result is REPAIR because transform and publish violate required-success policy. For the repaired work-failure and cleanup-failure runs, both scheduler outcome and external gate should be nonaccepting. For the happy path, both should accept. For the skip challenge, the scheduler may again be success, yet the external gate rejects because exact required-task success is absent. These comparisons are grounded in Airflow’s documented leaf rules, trigger-rule behavior, and task-state definitions.

The gate should never choose REVERT merely because one task failed. REVERT needs release context: the failing evidence must belong to a newly introduced revision and a prior artifact must be explicitly approved for restoration. Likewise, REPAIR should not be used when evidence is incomplete; that is HOLD, because diagnosing task logic from a partial task list is speculation.

The acceptance invariant can be stated compactly:

ACCEPT = identity_match AND revision_match AND complete_inventory AND dag_run_success AND all(required_task_state == success) AND cleanup_success AND verdict_success.

Every term must be proved from preserved evidence. No term is implied merely by the green DAG-run state. That is the practical meaning of Airflow DAG status validation for a consumer that needs stricter semantics than the scheduler’s leaf rule.

Ownership should be encoded in the release record as names or team identifiers rather than inferred during an incident. The DAG owner owns dependency repair and task semantics. The platform owner owns scheduler/runtime evidence and collector integrity. The downstream consumer owner owns the decision to act on accepted evidence. The platform release owner owns rollback authorization. One person may fill multiple roles in a small team, but the responsibilities should remain distinct because “the DAG was green” is not an acceptable substitute for any of them.

For automation, return a nonzero process status for HOLD and REPAIR, write the structured decision record atomically, and let the caller perform only inert test effects in this laboratory. In a production adoption, the same pattern should place any real side effect strictly after ACCEPT. That extension is an operational design recommendation, not something demonstrated by the localhost fixture, which deliberately excludes business side effects.

Residual uncertainty remains bounded but real. This laboratory excludes scheduled/backfill semantics, dynamic mapping, sensors, branching as workflow design, setup/teardown tasks, callbacks as verdicts, production connections, and real business side effects. It does not establish that every DAG in an estate has the same defect or that the same acceptance manifest fits every workflow. It establishes a repeatable test for one question: whether cleanup can hide failed required work at the DAG-run level and what evidence a strict consumer must check instead.

Build reliable pipeline judgment through Data Engineering

The core lesson is architectural rather than cosmetic: orchestration state and business acceptance are related signals, not synonyms. Airflow can correctly report a successful run under its documented leaf rules while a consumer correctly rejects the same run under a stricter required-task contract. A verdict leaf improves failure propagation when it has the right direct parents; a complete external task-state gate closes the skip, removal, identity, and missing-evidence gaps that graph shape alone does not close.

That kind of judgment depends on fundamentals: understanding task dependencies, transformation stages, storage and governance boundaries, and the operational evidence carried between them. Refonte Learning’s Data Engineering page lists a three-month format at 12–14 hours per week and covers areas including batch and streaming ingestion, transformation, storage/governance, data warehousing and ETL, and big-data pipeline design. Those broader foundations are relevant to reasoning about pipeline contracts, but the page does not establish that this exact Airflow 3.3.0 leaf-state acceptance laboratory is part of the syllabus.

For this playbook, the operational endpoint is simpler: do not let a downstream automation accept a run because one status string is green. Preserve the graph revision, reconcile the whole expected inventory, require exact success for required work and mandatory cleanup, and hold whenever evidence is missing or contradictory.