feat: API executor runs DataMixin prompts, reads python stages and reports token usage

[timzsu]

Purpose

Lumilake's ingestion workflow runs as API-backed LLM ops mixed with deterministic list steps. The deterministic steps (explode a paper into passages, filter, split, pair into ragged groups, collapse groups) now run as FlowMesh python tasks. The batch API executor still served only list data, though. A row-wise op, an aggregate op and grouped rows couldn't run through it the way they run on the vLLM path, and nothing let an API task consume a python stage's groups. Nor did an API task report what it cost in model calls and tokens. This PR closes all three gaps.

Changes

• API executor runs DataMixin prompts (api_executor.py, result schemas).
  • A dataframe spec issues one request per row. A graph_template spec issues one request over all rows, or one per group when the upstream is grouped.
  • A graph_template that reads a grouped column directly issues one prompt per group, carrying the whole group. The grouped column is repeated whole across its group's rows and never counts toward the row count, while an ungrouped column still expands to one value per row.
  • Grouped dataframe data returns one APIGroupItem {index, rows: [APIItem...]} per table, the way the vLLM executor regroups with _populate_table. List data stays one item per row.
• Expressions read results through lists (graph_templates.py). _evaluate_expr maps attribute access over lists of result models, DataFrames and nested lists of groups, and it resolves a declared field by name or alias, never a method. So items.rows.json.choices[0].message.content (row-wise upstream) and items.json.choices[0].message.content (aggregate upstream) read API content downstream. A python stage is read the same way at value.items.output.
• One grouping rule for the vLLM and API paths (graph_templates.py, mixins/data.py).
  • _evaluate_expr returns whether the value is grouped. It is grouped when the first attribute access over the items list yields one list per item, so each item whose output is a list of records is one group, and groups may be ragged. A list of DataFrames groups by item.
  • The dataframe and graph-template paths group on that flag instead of guessing from the value's shape, and the dataframe branch reuses _build_grouped_dataframes.
  • A cell whose per-row value is a list stays one cell, and index access into a per-item list (items.json.choices[0]) does not group.
• Zero rows run zero requests. A grouped upstream with an empty group builds an empty table instead of a one-row table of blanks. A grouped spec with no rows at all returns an empty result instead of failing with "produced no rows". The task's status code comes from the first row found in any group, or 0 when there is none.
• Grouped API results keep the json wire key. A plain dump of an APIResult holding APIGroupItems emits json for every row, which is where Lumilake reads items[].rows[].json.
• The SDK mirrors APIGroupItem and APIUsage, and APIResult.items admits APIItem | APIGroupItem.
• Token usage per API task (api_executor.py, result schemas). APIResult.usage is a typed APIUsage summed over the task's requests: prompt, completion and reasoning tokens, calls, failures, retries, truncated calls and wall time. A call whose response fails parsing, or that is cancelled, is still counted. Token counts are read only when they are integers, so an odd usage payload cannot fail a finished call.
• Per-call statistics in the API executor. One debug line per finished call (row, attempts, status or exception, wall time, token counts, finish reason, backend), one debug summary per task (calls, failures, retries, latency p50/p95/max, summed tokens, per-backend counts), and a 60 s INFO heartbeat while calls are outstanding. No request or response content is logged. One lock guards the statistics, the heartbeat and the summary, and a call that exhausts its connection retries logs its real attempt count.
• A cancelled grouped task with no rows reports cancelled. The task checks for a cancel once all its requests have returned. So a zero-row grouped task, or a cancel that lands during the last request, ends cancelled.
• Declare tabulate (pyproject.toml, uv.lock, worker requirements). Graph-template rendering calls DataFrame.to_markdown, which needs tabulate, but nothing declared it. It was only in the lock through GPU packages, so a CPU worker failed any aggregate over a dataframe column with Missing optional dependency 'tabulate'. It goes in runtime-analytics, beside pandas, which the CPU worker image installs.
• Docs. docs/WORKFLOWS.md documents grouped API results and how a dataframe API task reads a python stage (value.items.output.<field>, one group per item whose output is a list of records).

Design

No new task type. The API executor reuses DataMixin's prompt collection, and whole-list functions stay in main's python task. The result shape matches the vLLM executor's, so Lumilake reads a row-wise op the same way on the local path and the API path. Usage is reported per task, on the task's result. A client that wants a workflow's total sums its tasks' results, which also survives a Redis restart; the server keeps no per-workflow usage.

Test Plan

The repo's pre-commit gate (gitleaks, isort, black, ruff, codespell, mypy on the CI environment from uv sync --all-packages --group ci --frozen, requirements sync), then tests/server tests/worker tests/shared tests/sdk. The tests cover:

• dataframe rows, a graph_template aggregate (a plain APIItem, read at items.json), ragged groups, and a graph_template reading a grouped column directly;
• attribute mapping over lists, DataFrames and nested groups, and the json alias;
• a python stage read at value.items.output: rows, ragged groups of three and two, an aggregate over those groups, and an empty group;
• grouping decided from upstream structure: a list-valued cell and index access into a per-item list;
• empty groups and a grouped spec with no rows, with their lineage, and the status code over grouped results;
• the json wire key on a plain dump in the worker and the SDK, and SDK parity with the shared schema for APIGroupItem and APIUsage;
• per-task usage: tokens and calls summed across rows, a retried 503 then 200 counted as one call with one retry, a single call's usage shape, and no usage when there are no rows;
• a cancelled zero-row grouped task raising cancellation.

Then end to end on an isolated stack built from this branch, against https://lum.id/llm/v1/chat/completions with qwen3.8-27b: a python stage T0 emits two ragged groups of questions, a row-wise dataframe API task T1 reads them at value.items.output.q, a graph_template aggregate T2 reads T1 at items.rows.json.choices[0].message.content, a python task T3 collapses T1's groups into one item per answer, and an API stage T4 whose condition is never met.

Test Result

• tests/server tests/worker tests/shared tests/sdk at 349dfae: 2431 passed, 16 skipped, 1 deselected (test_mp_executor_cleans_up_vllm, a GPU test whose vLLM engine does not start in this environment, with or without these changes).
• Pre-commit: all hooks pass across the repo.
• End to end at 349dfae, workflow wfl-e824a49e: DONE, 5 completed, 0 failed.
  • T0 emitted two groups of three and two question records.
  • T1 returned two APIGroupItems of three and two rows, every row HTTP 200 with its answer.
  • T2 returned two plain APIItems, both HTTP 200.
  • T3 returned five items, one per answer.
  • T4 was skipped: it never reached a worker and carries no usage.
  • T1 and T2 carry usage with calls 5 and 2, failures 0, retries 0, and non-zero tokens. Summed client-side: calls 7, prompt 186, completion 408, reasoning 370.
  • GET /workflows/{id} has no usage field.
  • T1 and T2 ran with timeout_sec: 300 (default 60) because the upstream model endpoint was slow during an earlier run.

[kaiitunnz]

Leave comments. PTAL.

[kaiitunnz]

Leave some comments. PTAL.

[kaiitunnz]

This decides grouping from shape again after the first access, which the docstring rules out. items.json.choices.message.content comes back grouped (one group per item), while items.json.choices[0].message.content doesn't. With this change, the worker tests for graph templates, dataframe grouping and the API executor still pass.

Suggested change

	if not grouped:
	if not grouped and not mapped_items:

[kaiitunnz]

v0.1.9 stored the upstream usage dict here (prompt_tokens, completion_tokens, total_tokens), so those API results no longer validate. I checked with the SDK's AnyExecutorResult. Could the summed usage go in a new field, or should this PR be marked [BREAKING]?

[kaiitunnz]

With max_len starting at 0, an empty group fails when any column has a single repeated value (e.g. a constant data: list of one item). The length-1 column no longer matches 0 and raises "same number of rows per group". When max_len == 0, a one-value column should become [].

[kaiitunnz]

Message arguments to a function step still expand grouped columns per row, while the step runs once per group. With groups [a0, a1] and [b0], the step gets a0 and a1, and b0 is dropped. Pass the grouped labels through here.

[kaiitunnz]

failures is always 0. Every failed call raises and fails the task, and a failed task uploads no result, so no stored APIUsage ever counts a failure, and clients summing per-task usage never see failed calls. Drop the field, or report usage on failure too.

[kaiitunnz]

nit: run() now keeps nine counters as closure variables, copies each into a snapshot, and passes ten arguments to _log_summary, which makes it hard to follow. Could a small tracker class with a lock, record() and snapshot() -> APIUsage hold them? failed: bool here also shadows the failed event above.

[kaiitunnz]

nit: CODE_STYLE.md rules out a bare # type: ignore; add the error code. The same applies to lines 320 and 322.

[kaiitunnz]

The first sentence repeats the paragraph above. Keep only the rule for an embedded placeholder. Consider merging the two paragraphs.

[kaiitunnz]

A python task's return value is PythonResult.value, so the path is value.items.output.<field>. The per-group behavior of graph_template is also missing.

Suggested change

	A `dataframe` spec returns one `APIGroupItem` per table, with that table's
	responses in `rows`. A column is grouped when it reads one list of records per
	upstream item, such as an upstream API task's `items.rows` or a python stage's
	`items.output.<field>`; each list becomes one table. Downstream stages read the
	responses through `items.rows`.
	A `dataframe` spec returns one `APIGroupItem` per table, with that table's
	responses in `rows`. A column is grouped when it reads one list of records per
	upstream item, such as an upstream API task's `items.rows` or a python stage's
	`value.items.output.<field>`; each list becomes one table. Downstream stages
	read the responses through `items.rows`. A `graph_template` over a grouped
	column sends one request per group and returns one `APIItem` for each.

[kaiitunnz]

This test is ad hoc. Consider extending assert_fields_match or add assert_field_types_match.