Fix OpenLineage emitting duplicate START events for rescheduled sensors

[kaxil]

On Airflow 3, a sensor in reschedule mode emits an OpenLineage START event on every poke
instead of only the first. The run id is derived from try_number and a reschedule does not
increment it, so all of those events carry the same run id: downstream consumers see one run
entering RUNNING over and over for the life of the sensor.

The listener already has a guard meant to stop exactly this, but it tests a dict key with
hasattr:

if hasattr(context, "task_reschedule_count") and context["task_reschedule_count"] > 0:

context is airflow.sdk.definitions.context.Context, a TypedDict, so at runtime it is a
plain dict with no attribute of that name. The condition is always false and the return
below it is unreachable. The mixed access on the line gives it away: hasattr treats
context as an object, then the subscript treats it as a dict, and only the second is right.

Airflow 2 is unaffected, which is why this went unnoticed. Its hook calls
is_ti_rescheduled_already()
against the task_reschedule table and returns before the dead check is ever reached. The
Airflow 3 hook has no equivalent and cannot have one: that helper is defined under
if not AIRFLOW_V_3_0_PLUS: and needs a SQLAlchemy session against the metadata DB, which a
task runner has no access to. The hasattr line is the only reschedule protection Airflow 3
has, and it has never fired. It arrived that way in #45294, which split the hook by Airflow
version and swapped the is_ti_rescheduled_already() call inside the emission closure for the
hasattr check, adding the working guard back only on the Airflow 2 branch.

The revived guard is scoped to reschedule-mode sensors, not to the row count alone.
task_reschedule_count answers "have rows been written for this task instance", which is not
the same question as "did we already emit a START".
_maybe_reschedule_startup_failure()
writes those rows for any operator when a worker cannot see the DAG file, up to
[workers] missing_dag_retries (default 3), and it runs inside startup() before the
on_task_instance_running hook fires. So a plain PythonOperator that hit bundle-sync lag
reaches its first real attempt with a non-zero count and nothing emitted yet. A guard keyed on
the count alone would suppress that attempt's START and then emit a COMPLETE for a run that
never entered RUNNING, losing the RUNNING transition, the START eventTime that makes run
duration computable, and the inputs that extractors report at start. The Airflow 2 sibling
gates on isinstance(task, BaseSensorOperator) and task.reschedule before it ever looks at
the table; getattr(task, "reschedule", False) is the same test without the import, since
reschedule is a property on the sensor base.

The check sits in the hook rather than in the emission closure, where the broken one was.
That closure runs under _execute(..., use_fork=True), so leaving the guard there forks the
task runner once per suppressed poke to do nothing. Guarding in the hook matches where the
Airflow 2 branch guards, and costs nothing extra because
get_dag_run_dag_and_task_from_ti()
on the line above already builds and caches the template context on Airflow 3.

The regression test drives the real Airflow 3 path, a RuntimeTaskInstance with a live
TIRunContext, rather than the mocked-context helper in the same file. Nothing in the suite
had ever set the count above zero, so the guard was never exercised in its firing state. The
non-sensor row is what pins the scoping: with the count-only form of the check it fails.