[fas89]

Stacked on #736 (feat/airflow3-dags): it rewrites the same generators. Draft until #736 merges; this branch then gets rebased onto main and retargeted. Do not merge before.

What and why

Two defects the #736 review turned up, plus the non-shipped example.

1. fluid init's DAG helper never ran the generator it called. generate_dag_for_project ran fluid generate-airflow --output-dir ..., a flag that command never had. argparse exited 2 and every project got the basic fallback DAG. When fluid was not on PATH, it got no DAG at all. The helper's tests mocked subprocess.run, so nothing parsed that argv. With valid flags, generate-airflow (deprecated) writes only echo placeholder tasks for the shipped templates. It now does the following, per the decisions made during review:

• Runs fluid generate schedule --scheduler airflow --output <abs dags> --contract-path contract.fluid.yaml --overwrite. This is the command generate-airflow's own deprecation notice names. It runs in-process through the CLI's own build_parser, as stage 3 (artifact_fanout) runs it.
• Writes no DAG, and says why, when the contract schedules nothing: no build with a cron trigger and no orchestration.tasks. For example, customer-360's build is trigger.type: manual, and the command would synthesize a nightly DAG that only logs.
• Writes no DAG when the command refuses the contract, for example an invalid cron. The old fallback wrote a basic DAG on a schedule the contract never declared, which hid the error.
• Judges both conditions on contract.fluid.yaml as the command loads it, including $ref builds and the legacy singular build:.
• Writes a dags/README.md that lists each DAG this run wrote and says to edit the contract, not the generated DAG.

create_basic_dag stays a Python helper. Its fluid apply --auto-approve (rejected, exit 2) is now --yes, and its project path is shell-quoted and then emitted as a Python string literal. No CLI command reaches the helper with DAGs on today: fluid demo passes no_dag=True.

2. Snowflake orchestration python tasks lost the whole DAG. providers/snowflake/orchestration/airflow_generator.py emitted python_callable=<name> for a function the module never defined. Airflow raised NameError at parse and dropped the DAG; this reproduces in a real DagBag on 3.2.2 and 2.10.5. The generator now defines a stub of the callable above the task. The stub raises AirflowFailException naming the task, so the DAG loads and that one task fails at once instead of burning retries. A callable name that would rebind something the DAG uses (dag, an operator class, an earlier task, ImportError, a dunder) is refused with a ValueError. sanitize_identifier and py_str_literal escaping are unchanged. This generator is library-only (fluid export uses providers/snowflake/codegen).

3. examples/generated_dags/financial_analytics_dag.py is regenerated through AwsProvider.export and black at width 100. It now uses schedule= and Airflow-3-first imports with the Airflow 2 fallback.

Borrow before build

• In repo: ship.py::_resolve_fluid_bin; stage 3's in-process generate_schedule.run; fluid_apply.has_scheduled_builds/has_explicit_tasks; codegen_utils escaping; the tests/cli/_schedule_dag_fixtures.load_dag stub-Airflow harness (adapted).
• Outside: dag-factory resolves callables at parse time from files or dotted paths. Not adopted: contract text must never become code executed at parse time.
• Airflow @task.stub is for externally executed tasks and needs a recent standard provider, so it is not used.
• DagBag import_errors integrity checks follow Astronomer's testing guide.

Validation

• Real Airflow 3.2.2 and 2.10.5 DagBag (uv venvs with the snowflake, amazon and common-sql providers; 3.2.2 also has the standard provider): the Snowflake python-task DAG, first-dag and pipeline-orchestration fluid apply DAGs, and the example all load with 0 import errors. The base code gives NameError: name 'score_customers' is not defined on both versions.
• airflow tasks test on the stub task: 3.2.2 new_state=failed, 2.10.5 Marking task as FAILED. With NotImplementedError both went UP_FOR_RETRY.
• Journeys through the real helper:
  • first-dag, pipeline-orchestration and incremental-processing get their DAG plus README.
  • customer-360 gets none, with the reason.
  • An invalid cron gets none; the command's own error and the exit code are shown.
  • A re-run regenerates.
  • A pre-existing dags/helpers.py stays out of the README.
  • A relative project dir and a singular build: contract both work.
  • The README's regenerate command works.
• The basic DAG's validate, plan and apply --yes ran for real in a project path containing a space and ' (all rc 0).
• New tests fail against the base code. Targeted suites pass: 509. Lint matches CI: ruff, black 24.10.0 on the whole tree, license headers, lint-imports (6/6).
• Full suite on duckdb 1.5.6: the only failures are known environment baselines (masking needs numpy, no dbt, macOS temp-root sandbox, live claude/MCP tests), none in touched areas. DuckDB 2.0.0 (released 2026-10-21) breaks 48 unrelated tests repo-wide and is why main CI is red; the DuckDB 2.0 port session owns that.

Reviews

• /security-review: no findings in round 1, round 2 or the round-3 delta. 69 hostile callable values and 15 hostile project paths were checked: no parse-time code, no shell injection, $ref confined, nothing written outside dags/.
• /world-class-inspection: round 1 returned CONCERNS (fallback semantics; subprocess design; README contradicting generated DAGs). All are resolved, per the round-2 re-check. Remaining P3, latent: an orchestration.engine: none opt-out is not honoured, and should_generate_dag can miss $ref builds (Trello).
• /defect-scan: round 1 had 4 hunters plus a referee. Its confirmed defects were fixed and re-verified: fallback on a benign refusal, README listing files the run did not write, relative paths, decoding, the task_vars false positive, retries of the stub, and harness blind spots. Round 2 left 5 Low latent items: engine opt-out; the outer gate on $ref; the message when the command fails after writing; the reason printed twice; the command's Tip missing --overwrite (Trello).

Follow-ups (on the Trello board)

• Land this PR after fix(airflow): every generated DAG loads on Airflow 3 and on Airflow 2.7+ #736: rebase, retarget, merge on green.
• Init helper latent items from the last round.
• Scheduler: synthesis gives manual builds a nightly no-op DAG; the generic scheduler drops bash/python orchestration.tasks.
• Init DAG path: unreachable from the CLI, dead --no-dag, flat dags/ vs schedule-sync, stale DAGs on regenerate, Jinja in the project path.
• Snowflake orchestration generator: decide whether to delete it or align it with the schema.
• AWS generator: Glue key mismatch visible in the example.

[github-actions]

📄 Documentation Reminder

This PR appears to be missing a documentation reference. Our docs live in a separate repo.

Please update the PR description with one of:

• Link a docs PR — check the "Docs PR linked" box and paste the URL
• Mark as no docs needed — check "No docs needed" with a justification
• Acknowledge docs TODO — check "Docs TODO" and create the docs PR before merge

See the Contributing Guide for details.