fix(runtime): page task instances so a wide fan-out is complete

[chiruu12]

Closes #311.

task_instances made one GET to /taskInstances with no limit or offset. Airflow's endpoint defaults to 50 per page, so any run whose fan-out is wider than 50 returned the first page and the caller treated it as the whole run. ingest_dag_topology then built a run graph off a truncated list, and the mapped-task summary undercounted without any sign that it had.

The method now pages until the run is exhausted. dag_runs stays caller-capped as it already documents; this one's contract is complete run detail, so a short read is not an acceptable answer.

Three independent stops, so a server that disagrees with its own metadata cannot spin: a page that added nothing new, a page shorter than the one requested, and the reported total_entries being reached. Entries are keyed by (task_id, map_index) so overlapping pages do not double count. A failure on any page raises AirflowClientError rather than returning a short list as though it were complete.

Page size is 100 rather than the default 50. Asking for more per request costs nothing and halves the round trips.

Tests

tests/test_runtime_client.py

• every page of a 251-instance fan-out is returned, and the mapped indices are complete
• pages are requested in turn with the expected limit and offset
• a failure on a later page raises instead of returning the pages already collected

packages/hflow-server/tests/test_server_run_graph.py

• the run-graph endpoint totals a fan-out wider than one Airflow page. task_instances is left real here and the HTTP layer under it is stubbed instead, so this fails if the client ever stops paging.

All four fail on main and pass here. Reverting only the paging loop and keeping the tests gives 4 failed, 53 passed.

Local run: 269 passed across tests/test_runtime_client.py and packages/hflow-server/tests, plus 1167 passed on the root suite. ruff check and ruff format --check are clean.

Rebase note

This was written against e6009fc, before task instances became typed. It is rebased onto current main and now returns list[AirflowTaskInstance] through _parse_task_instance, so the paging loop composes with the typed model rather than bypassing it.

[Copilot]

🟡 Changes recommended

The new paging loop can still silently return partial results on malformed/empty pages and can terminate early when pages overlap due to an offset >= total_entries check instead of using the unique collected count.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR fixes incomplete Airflow run-graph ingestion by paginating AirflowClient.task_instances() so it returns complete task-instance lists for wide mapped-task fan-outs (e.g., >50 instances), and adds regression tests to ensure the paging behavior is preserved.

Changes:

• Implement pagination in AirflowClient.task_instances() with de-duplication by (task_id, map_index) and multiple termination guards.
• Add client-level stub-server tests covering multi-page fan-outs, request sequencing (limit/offset), and error propagation on later-page failures.
• Add an HTTP-boundary server test ensuring run-graph totals include fan-outs wider than one Airflow page.

File summaries

File	Description
tests/test_runtime_client.py	Extends the stub Airflow server and adds paging-focused client regression tests.
src/hflow/runtime/_client.py	Updates task_instances() to page through /taskInstances and documents the complete-results contract.
packages/hflow-server/tests/test_server_run_graph.py	Adds an integration-style test that fails if the client stops paging and mapped totals are truncated.

Review details

• Files reviewed: 3/3 changed files
• Comments generated: 2
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[kstonekuan]

The paging itself is right, and the three-stop reasoning is the part I would have asked for: added == 0 is a real loop guard, keying on (task_id, map_index) handles overlap, counting total_entries against what was kept rather than the offset is a subtlety most people miss, and raising instead of returning a short list is DoD 3 exactly. One thing to fix before this lands.

The short-page stop truncates on any deployment whose api.maximum_page_limit is below 100. Airflow clamps the limit query parameter to that setting and does not error, so a server configured lower serves shorter pages than you asked for, and len(page) < _TASK_INSTANCE_PAGE_SIZE reads that as "run exhausted". A 251-instance fan-out, server clamping at the value shown:

maximum_page_limit=100   3 requests   returned 251 of 251   COMPLETE
maximum_page_limit= 50   1 request    returned  50 of 251   TRUNCATED
maximum_page_limit= 25   1 request    returned  25 of 251   TRUNCATED

One request, then a silent partial list reported as complete. That is the defect #311 exists to remove, reintroduced under a config change rather than a fan-out size, and it fails DoD 1 and DoD 3 on that deployment. The default for maximum_page_limit is 100, which is exactly what you request, so it passes on a stock Airflow and breaks the moment an operator lowers it. Your stub honours limit verbatim, which is why no test sees this.

The information to distinguish the two cases is already in the response you read. When total_entries is present and has not been reached, a short page means the server clamped, not that the run ended, so the offset should advance by what was served and the loop continue. The short-page rule is only a safe terminator when total_entries is absent. Your added == 0 stop is what keeps that from spinning, so the loop guard does not weaken.

Worth a test with a stub that clamps below the requested limit, since that is the shape no current test can produce.

Two things I checked that are fine. An empty run still returns [] rather than raising, because {"task_instances": [], "total_entries": 0} hits the empty-page break before the missing-list refusal, so the run graph does not start answering 502 for a run with no tasks. And dag_runs is untouched and still caller-capped, with a test saying so.

Gate is clean on the merged result otherwise: ruff check, ruff format --check, ty check, 1417 passed / 6 skipped.