Recover Dag processor work and expire inactive processors - #74791

Lost responses and reparses during an active import must not consume later requests. Callback replay must preserve priority, and routine cleanup must remain safe on SQLite and avoid scanning completed Job history.

Legacy Dags whose files were removed must become stale after a 2.x upgrade even when migration 0082 assigns a non-null bundle name. Existing files must remain active when bundle names or directories change.

The upstream migration sequence has advanced, so claim migrations and their
regressions need the next available number. Claim receipts inherit Job
cleanup through their foreign key and must retain replay protection while
the Job remains alive.

Bundle discovery now returns source failure information alongside files.
The upgrade regression must use the discovered files to exercise legacy
Dag deactivation after migration 0082.

Inactive bundles must stop scheduling without waiting through multiple
cleanup intervals, and expired owners must not retain parse checkpoints
or leave callbacks outside normal database cleanup.

Callback context needs a versioned response schema, while heartbeat
thresholds must agree on when a processor can lose ownership. Regression
coverage must exercise the configured database backend as well as replay,
claim limits, inactive bundles, and interrupted cleanup.

Existing heartbeat settings must not prevent inactive bundles and retired
processor metadata from being cleaned up when API expiration is unsafe.

Distinct file paths must retain separate priority requests even when the
database collation ignores case or accents, so requesting one file does
not release another file's claim or lose the requested reparse.

Airflow's structlog bridge does not render stdlib `extra=` fields, so the
warning logged when scheduler maintenance expired an API Dag processor Job
carried only its message. Operators tracing a callback that ran twice
could not tell which Job or host was expired, or which timeout applied.

Maintenance also validated `job_heartbeat_timeout` on every run. That
timeout only applies to API Dag processor Jobs, and database-mode
deployments may legitimately set `[dag_processor] health_check_threshold`
above it, so each scheduler logged a configuration warning every cleanup
interval about an option the deployment never uses. The timeout is now
checked only when a live API Dag processor Job exists, and a test keeps
Job expiry batched, since nothing failed if that limit was dropped.

Before maintenance moved to the scheduler, every Dag processor loop
removed all warnings of stale Dags. Purging one batch per cleanup
interval left warnings of a bulk-deactivated bundle listed in the Dag
warnings API for many minutes, so they now drain in committed batches
like the Dag and publication-receipt steps.

A Dag processor numbers callback and priority claims independently. If
both kinds shared one claim record, a processor that had already claimed
callbacks at a higher sequence would see every priority claim as stale,
and priority reparses would never reach it. The claim body bounds cap how
many rows one request can lock, and the route's Job check covers a Job
that closes between token admission and the row lock. Nothing failed if
any of these was removed.

Some tests did not exercise this change. The takeover test passed without
the new heartbeat-timeout validation, whose guard the registration test
already covers, so it is dropped. The replaced legacy inventory test also
passed without this change, so the original from the publication API
change is restored. Tests of DagFileProcessorManager belong with the
manager tests, SQLite-only tests should declare their backend instead of
skipping themselves, and the lock-timeout test no longer names an
inactive bundle it does not use.

The parsing_cleanup_interval description read as if every Dag processor
Job is ended once it exceeds job_heartbeat_timeout. Only Jobs registered
through the Execution API carry a session and are expired, so a
database-mode operator could expect a hung processor to be ended when it
never is.

The stale Dag warning and publication receipt cleanups each capped their
key batches at 499, but only one said why: each two-column key takes two
bind parameters, and SQLite before 3.32 allows 999. Two copies of that
bound could drift apart and break cleanup on older SQLite.

The claim and acknowledgment routes took an untyped execution token,
while the other Dag processor routes refuse any other token type before
the handler runs. Main added migrations 0152 and 0153, so the
work-claims migration is now 0155.

Main added migration 0154 and the publication-state migration below this
one is now 0155, so the work-claims migration and its test move to 0156.