Paginate Airflow task instances before building run graphs

[kstonekuan]

Current behavior

AirflowClient.task_instances() says it returns every task instance for one run, including every dynamically mapped process_batch instance. It makes one GET request with no limit, offset, or cursor.

Airflow's Get Task Instances endpoint is paginated. The generated API client documents limit=50 and offset=0 as its defaults and returns total_entries: https://github.com/apache/airflow-client-python/blob/main/docs/TaskInstanceApi.md#get_task_instances

The workspace run graph consumes that one returned list and calculates mapped totals and per-state counts from it at packages/hflow-server/src/hflow_server/_graph.py:513-540. A fan-out larger than the API page reports only the first page, so mapped_summary.total and by_state omit later batches.

The neighboring dag_runs() API makes its truncation explicit in both its signature and docstring at src/hflow/runtime/_client.py:281-293. task_instances() has no equivalent caller-selected cap because its contract is complete run detail.

Verified against main at fbfd553.

Definition of done

1. AirflowClient.task_instances() returns all task instances for the selected run across every API page.
2. Dynamically mapped instances retain every map_index exactly once.
3. A failure on any later page raises AirflowClientError. The client doesn't return a partial list as complete.
4. Pagination terminates safely from the documented response metadata and can't loop forever on an inconsistent response.
5. Existing run-id URL encoding and token refresh behavior remain unchanged on every page.
6. The workspace run graph reports complete mapped totals and state counts for a fan-out larger than one Airflow page.
7. Add a local HTTP-boundary regression with more than one page. Assert the returned business result and requested pagination values, not Airflow internals.
8. Keep dag_runs() caller-capped behavior unchanged.
9. Update the task_instances() docstring with the pagination contract.

Non-goals

• A generic pagination framework for every Airflow endpoint
• Removing the deliberate limits on recent DAG-run lists
• Changing run-graph payloads or topology
• Changing Airflow's own page size or ordering defaults

Validation

uv run ruff check --fix
uv run ruff format
uv run ty check
uv run pytest -q tests/test_runtime_client.py \
  packages/hflow-server/tests/test_server_run_graph.py
HFLOW_DOCKER_TESTS=1 uv run pytest -q tests/test_runtime_integration.py
uv run pytest -q

[chiruu12]

Confirmed on main at e6009fc. task_instances makes one GET to .../taskInstances with no limit or offset and returns response["task_instances"] as if it were complete. dag_runs, right above it, takes a caller limit and says so in its docstring.

Plan: page on limit/offset, stop from total_entries, and refuse to continue on a page that returns nothing or repeats a map_index already seen, so an inconsistent response cannot spin. A failure on any page raises AirflowClientError instead of returning a short list as complete. The regression drives a two-page fan-out at the HTTP boundary and asserts the merged map_index set and the offsets actually requested. dag_runs stays as it is.

Could I take this one?