Add unified emission_policy config for OpenLineage provider

[kacpermuda]

Introduces a single [openlineage] emission_policy configuration option that replaces
the patchwork of per-feature toggles with one coherent rule language, plus a Dag-author
authoring API.

What changed

New config: emission_policy — a JSON array of rule objects. Each rule has shape
{"scope": {...}, "controls": {...}, "match_mode": ..., "locked": ...}:

[openlineage]
emission_policy = [
  {"scope": {"operator": "airflow.providers.http.operators.http.HttpOperator"},
   "controls": {"emit": false}},
  {"scope": {"dag_id": "expensive_dag"}, "controls": {"extract_operator_metadata": false}},
  {"scope": {"dag_id": "expensive_dag", "task_id": "send_report"},
   "controls": {"extract_operator_metadata": true}},
  {"scope": {"dag_id": "reporting_dag"}, "controls": {"emit_dag_events": false}},
  {"scope": {"dag_id": "^staging_.*"}, "match_mode": "regex", "controls": {"emit": false}},
  {"scope": {"dag_id": "audit_dag"}, "controls": {"include_full_task_info": true}},
  {"scope": {}, "controls": {"include_source_code": false}, "locked": true}
]

• scope (required, {} for global) accepts dag_id, task_id, operator. Scope must be exactly one of: global, operator-only, dag-only, or dag+task.
• controls (required, non-empty): emit, emit_task_events, emit_dag_events, extract_operator_metadata, include_source_code, hook_lineage, include_full_task_info.
• match_mode: "regex" — every value inside scope is treated as a re.fullmatch pattern.
• locked: true — admin floor lock; see the admin/dag-author split below.

Scope priority (most-specific wins): dag_id + task_id → dag_id → operator → global. Within a tier, last rule wins.

Admin vs Dag-author split - no Airflow restart required. The conf option is the admin lever: it lives in airflow.cfg, applies deployment-wide, and an admin can mark any rule "locked": true to make that field non-overridable. Layered on top, Dag authors call extend_global_openlineage_emission_policy(dag_or_task, ...) directly in DAG code to adjust the policy at parse time without restarting Airflow. Authoring overrides win against unlocked conf rules but cannot touch fields the admin has locked, so admins retain hard guarantees while DAG authors keep day-to-day flexibility.

Strict validation — unknown keys at any nesting level cause the rule to be skipped with a WARNING (catches typos like {"scope": {"dag_id": ...}}).

Audit logging — every non-default resolved flag is logged at INFO with the exact rule that caused it, giving a clear audit trail.

Contradiction detection — two rules in the same priority tier setting the same flag to opposite values trigger a WARNING (last value still wins).

Scope of change — which configs are affected

Superseded by emission_policy (these four control what is emitted and how much):

Legacy option	Translated to
disabled_for_operators = some.Operator	{"scope": {"operator": "some.Operator"}, "controls": {"emit": false}} (one per entry)
disable_source_code = True	{"scope": {}, "controls": {"include_source_code": false}}
include_full_task_info = True	{"scope": {}, "controls": {"include_full_task_info": true}}
selective_enable = True	{"scope": {}, "controls": {"emit": false}} global baseline + per-task opt-in rules injected at runtime

Not touched — completely unrelated to this change:

• transport / config_path — where events go; unchanged
• disabled — global kill switch; still takes full precedence over everything
• namespace — logical namespace for events; unchanged
• extractors / custom_run_facets — extension points; unchanged
• execution_timeout / debug_mode / dag_state_change_process_pool_size — operational knobs; unchanged
• spark_inject_parent_job_info / spark_inject_transport_info — Spark integration; unchanged

Backwards compatibility

The four superseded options are not removed. Any active legacy option is always translated into equivalent emission_policy rules — regardless of whether emission_policy itself is set. Translated rules are prepended before any user-provided rules, so explicit user rules win within each tier (last-wins).

• When no legacy option is set and emission_policy is empty: zero warnings, built-in defaults.
• When any legacy option is set: a DeprecationWarning is issued listing every translated option. Migrate to emission_policy exclusively to silence it.

The legacy options can be removed in a future version.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (claude-sonnet-4-5)

Generated-by: Claude Code (claude-sonnet-4-5) following the guidelines

[mobuchowski]

If we want to control dag-level behavior, it should be something - decorator, config - that is configurable per dag in dag code. Otherwise, a restart is required to change it - which is too cumbersome for many customers, especially with large setups.

[shnhdan]

Hey @kacpermuda — just updated #63499 with consistent gating across all
get_hook_lineage_collector() calls in s3.py. Wanted to flag that the
two PRs are complementary — #66992 controls the OL listener read side,
#63499 gates the calls at the source. Happy to coordinate if needed.