AIP-96: wire CHECKPOINTED through supervisor → API server → DB

[1fanwang]

Stacked on #66402 (CHECKPOINTED state + AirflowTaskCheckpointed exception). The foundation PR reached the message boundary — the worker constructs TaskState(state=CHECKPOINTED) and the supervisor accepts it — but the DB row never transitioned because:

1. The supervisor's STATES_SENT_DIRECTLY list didn't include CHECKPOINTED, so it fell through to client.task_instances.finish() which is constrained to terminal states.
2. The API server's state-update endpoint had explicit branches for terminal / retry / success / deferred / reschedule payloads but no branch for TITargetStatePayload (the discriminated-union catch-all that CHECKPOINTED routes through).

This PR closes both gaps:

• Supervisor: CHECKPOINTED is added to STATES_SENT_DIRECTLY. The TaskState handler calls a new client.task_instances.checkpoint() method when the worker reports CHECKPOINTED.
• Client: new checkpoint() method PATCHes /task-instances/{id}/state with TITargetStatePayload(state=CHECKPOINTED) — the same shape DEFERRED and UP_FOR_RESCHEDULE already use.
• API server: a new branch in the state-update handler accepts TITargetStatePayload and updates the row to checkpointed without additional bookkeeping.

checkpoint_data persistence is intentionally not added in this PR. The right shape (XCom-backed, new TaskInstance JSON column, separate metadata table) is the open AIP-96 question this PR deliberately leaves for the AIP discussion. Opening this draft so the discussion has a concrete state-transition wiring to react to — reviewers can keep this shape, suggest a different one, or reject the entire approach without me having silently picked it.

Real e2e validation

Verified end-to-end on airflow standalone (scheduler + API server + LocalExecutor + sqlite):

worker  | Task checkpointed; reporting CHECKPOINTED state.
\` curl PATCH /task-instances/{id}/state \`
api     | (new branch updates state=checkpointed)
sqlite> SELECT state FROM task_instance WHERE dag_id='e2e_checkpointed';
        state = checkpointed

Before this PR (only #66402 applied), the same DAG ran the worker through the catch block but the DB row landed at failed because the supervisor → API path rejected the state. After this PR, the row lands at checkpointed.

Testing

• Unit test test_task_instance_checkpoint in task-sdk/tests/task_sdk/api/test_client.py verifies the client's PATCH carries state="checkpointed".
• The existing TestTaskCheckpointed.test_run_returns_checkpointed_state continues to cover the worker → state-return contract from AIP-96 (draft): add CHECKPOINTED state and AirflowTaskCheckpointed exception #66402.
• Real-services e2e captured above.

Open questions for the AIP discussion (deferred)

1. checkpoint_data persistence. This PR drops the operator-supplied payload. Where should it land — XCom, a new TaskInstance JSON column, a separate metadata table?
2. Auto-resume vs manual-resume-only. Once a row is at checkpointed, should the scheduler auto-transition it back to queued after a delay (like up_for_retry), or only resume on explicit user action?
3. Trigger-rule treatment. Downstream tasks of a checkpointed upstream wait per existing State.unfinished semantics. Should all_done treat checkpointed as "done" for fan-out scenarios?

I deliberately did not pick on any of these — happy to follow up once the AIP discussion converges.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.