Deadline callback migration 0094 shallow-encodes nested callback kwargs → scheduler CrashLoopBackOff (KeyError('__var'))

### Apache Airflow version

3.2.0+ (root cause is migration `0094`, `airflow_version = "3.2.0"`; buggy code is still present on `main`)

### What happened

On a deployment using **Deadline Alerts** with an `AsyncCallback` whose `kwargs` contain a **nested dict**, the **scheduler enters CrashLoopBackOff**. Every scheduler loop crashes while deserializing an associated `callback` row in the deadline-processing query:

```
File ".../airflow/jobs/scheduler_job_runner.py", line 1808, in _run_scheduler_loop
for deadline in session.scalars( ... selectinload(Deadline.callback) ... )
File ".../airflow/utils/sqlalchemy.py", line 221, in process_result_value
return BaseSerialization.deserialize(value)
File ".../airflow/serialization/serialized_objects.py", line 632, in deserialize
return {k: cls.deserialize(v) for k, v in var.items()}
File ".../airflow/serialization/serialized_objects.py", line 629, in deserialize
var = encoded_var[Encoding.VAR]
KeyError:
```

The crash happens before the scheduler heartbeats, so it presents as failing liveness probes / restarts with no OOM and no other logged exception — easy to misdiagnose as a probe or DB problem.

### Root cause

Migration **`0094_3_2_0_replace_deadline_inline_callback_with_fkey.py`** (revision `e812941398f4`) moves the old inline deadline callback into the `callback` table. `callback.data` is an `ExtendedJSON` column, whose read path runs `BaseSerialization.deserialize` (`utils/sqlalchemy.py`), which requires **every nested dict to be wrapped** as `{"__type": "dict", "__var": {...}}`.

But the migration hand-builds `callback.data` and only wraps the **top level**, embedding the old callback's `kwargs` **raw**:

```sql
-- _upgrade_postgresql(): raw nested kwargs pulled from the old inline (SDK-serde) callback
COALESCE(NULLIF(d.callback::jsonb->'__data__'->'kwargs', 'null'::jsonb), '{}'::jsonb) AS cb_kwargs
...
INSERT INTO callback (... data ...)
SELECT ...
json_build_object(
'__var', json_build_object(
'path', b.cb_path,
'kwargs', b.cb_kwargs, -- <-- raw nested dict, NOT extended-serialized
'prefix', :prefix,
'dag_id', b.dag_id),
'__type', 'dict')::json
```

(`_upgrade_mysql_sqlite()` builds the same shape in Python.) The insert is done via Core (`callback_table.insert()` / `sa.text`), which also bypasses `ExtendedJSON.process_bind_param`, so `BaseSerialization.serialize` never runs.

Result: `callback.data` = `{"__type":"dict","__var":{"path":..., "kwargs": {}, ...}}`. On read, `deserialize` recurses into the outer `DAT.DICT`, reaches the bare `kwargs` (or a nested dict inside it such as metric `tags`), executes `var = encoded_var[Encoding.VAR]`, and throws `KeyError('__var')`.

**Trigger:** callback `kwargs` containing a nested dict. Callbacks with only flat/primitive kwargs survive because the single top-level wrap is enough.

### What you think should happen instead

The migration should produce the same encoding `BaseSerialization.serialize` would — i.e. **recursively** wrap nested dicts (`kwargs`, and any dict within it) as `{"__type":"dict","__var":{...}}` — so the resulting `callback.data` round-trips through `BaseSerialization.deserialize`.

Because `0094` has already shipped (3.2.0) and run on live deployments, a **forward repair migration** is also needed to re-encode already-corrupted `callback.data` rows (deployments that ran `0094` with any nested-dict callback kwargs currently have a latent scheduler crash that fires when such a deadline becomes due).

### How to reproduce

Minimal, self-contained demonstration of the serialize/deserialize invariant (no Airflow import needed): a correct `BaseSerialization.serialize({"path": "...", "kwargs": {"tags": {"a": "b"}}})` wraps `kwargs`/`tags` and round-trips, whereas the migration's shallow-wrapped shape `{"__type":"dict","__var":{"kwargs":{"tags":{...}}}}` raises `KeyError('__var')` in `deserialize`.

End-to-end: on 3.1.x create a deadline whose `AsyncCallback` kwargs contain a nested dict; upgrade across `0094`; then run the scheduler (or `SELECT` the `callback.data`) — it is stored with unwrapped nested kwargs and the scheduler crashes.

### Operating System

Linux (containerized)

### Versions of Apache Airflow Providers

N/A (core deadline alerts + serialization)

### Deployment

Other

### Anything else?

Workaround to unblock a crashing scheduler (neutralize the corrupt rows so the `WHERE ~missed` deadline query skips them):

```sql
UPDATE deadline d SET missed = true
FROM callback c
WHERE c.id = d.callback_id
AND d.missed = false
AND jsonb_path_exists(c.data, '$.**.__var.* ? (@.type() == "object" && !exists(@.__type))');
```

### Are you willing to submit PR?

Yes.

---
🤖 Filed with [Claude Code](https://claude.com/claude-code) (model: Claude Opus 4.8, `claude-opus-4-8[1m]`).

Start with migration 0094_3_2_0_replace_deadline_inline_callback_with_fkey.py, including _upgrade_postgresql() and _upgrade_mysql_sqlite(), then read BaseSerialization.deserialize in airflow/utils/sqlalchemy.py and airflow/serialization/serialized_objects.py. Verify that callback.data round-trips for nested kwargs, and add a forward repair migration for rows already written with shallow encoding.

Written by the indexing model from the issue text.