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

docs/WORKFLOWS.md

@@ -71,10 +71,10 @@ contract.
 
 ## API task
 
-`taskType: api` sends one HTTP request per `spec.data` row, in parallel, and
-returns one `APIResult.items` entry per row, in order. `spec.data` is required
-and accepts `list`, `dataset`, `graph_template`, and `dataframe`; row metadata
-and `dataframe` table grouping are not applied.
+`taskType: api` sends one HTTP request per `spec.data` prompt, in parallel, and
+returns the responses in `APIResult.items`. `spec.data` is required; it supports
+the same data types as the vLLM executor (`list`, `dataset`, `graph_template`,
+`dataframe`).
 
 By default it routes to the Nebula endpoint and authenticates with the worker's `NEBULA_API_TOKEN`.
 
@@ -113,24 +113,19 @@ exactly `{{prompt}}` takes the prompt as-is, so a message-list row fills
 requests. Any failed row fails the task. Cancelling the task skips rows that
 have not started and marks it cancelled once in-flight requests return.
 
-```yaml
-spec:
-  taskType: api
-  data:
-    type: list
-    items:
-      - Explain vector databases
-      - Explain attention
-  api:
-    method: POST
-    body:
-      model: gpt-4o
-      messages:
-        - role: user
-          content: "{{prompt}}"
-    response:
-      parse_json: true
-```
+A body value that is exactly `{{prompt}}` is replaced by the row's prompt
+object as-is (a message list stays a list of `{"role", "content"}` dicts). An
+embedded `{{prompt}}` inside a longer string keeps string substitution: a
+string prompt is inserted verbatim, and any other prompt value is rendered as
+JSON.
+
+### Grouped results
+
+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`.
 
 ## Python task
 

pyproject.toml

@@ -120,6 +120,7 @@ runtime-analytics = [
     "pandas>=2.3.3",
     "psycopg[binary]>=3.2.12",
     "sqlalchemy>=2.0.44",
+    "tabulate>=0.9.0",
 ]
 runtime-worker-cpu = [
     { include-group = "runtime-worker-core" },

sdk/src/flowmesh/models/__init__.py

@@ -28,8 +28,10 @@
     AgentResult,
     AgentUsage,
     AnyExecutorResult,
+    APIGroupItem,
     APIItem,
     APIResult,
+    APIUsage,
     BaseExecutorResult,
     CostEstimates,
     DataProfilingResult,
@@ -106,8 +108,10 @@
 )
 
 __all__ = [
+    "APIGroupItem",
     "APIItem",
     "APIResult",
+    "APIUsage",
     "ActiveWaitBreakdown",
     "AgentBatchSummary",
     "AgentItem",

sdk/src/flowmesh/models/result/__init__.py

@@ -40,7 +40,9 @@
     AgentItem,
     AgentMetadata,
     AgentUsage,
+    APIGroupItem,
     APIItem,
+    APIUsage,
     CostEstimates,
     DataRetrievalItem,
     EchoItem,
@@ -91,8 +93,10 @@
     _model.model_rebuild()
 
 __all__ = [
+    "APIGroupItem",
     "APIItem",
     "APIResult",
+    "APIUsage",
     "AgentBatchSummary",
     "AgentItem",
     "AgentMetadata",

sdk/src/flowmesh/models/result/catalog.py

@@ -18,7 +18,9 @@
     AgentItem,
     AgentMetadata,
     AgentUsage,
+    APIGroupItem,
     APIItem,
+    APIUsage,
     CostEstimates,
     DataRetrievalItem,
     EchoItem,
@@ -211,9 +213,9 @@ class APIResult(StrictExecutorResult):
     truncated: bool = False
     headers: dict[str, str] | None = None
     response_json: Any = Field(default=None, alias="json")
-    usage: dict[str, Any] | None = None
+    usage: APIUsage | None = None
     text: str | None = None
-    items: list[APIItem] = Field(default_factory=list)
+    items: list[APIItem | APIGroupItem] = Field(default_factory=list)
 
 
 class SSHResult(StrictExecutorResult):

sdk/src/flowmesh/models/result/payloads.py

@@ -21,6 +21,17 @@ class GenerationUsage(StrictModel):
     latency_sec: float
 
 
+class APIUsage(StrictModel):
+    prompt_tokens: int
+    completion_tokens: int
+    reasoning_tokens: int
+    calls: int
+    failures: int
+    retries: int
+    truncated_calls: int
+    wall_sec: float
+
+
 class EmbeddingUsage(StrictModel):
     prompt_tokens: int
     total_tokens: int
@@ -174,3 +185,11 @@ class APIItem(StrictModel):
     usage: dict[str, Any] | None = None
     text: str | None = None
     prompt: str | None = None
+
+
+class APIGroupItem(StrictModel):
+    """One group's row responses in a batched API task over grouped data; ``rows``
+    holds the group's row responses in order."""
+
+    index: int
+    rows: list[APIItem]

src/shared/schemas/result/__init__.py

@@ -40,7 +40,9 @@
     AgentItem,
     AgentMetadata,
     AgentUsage,
+    APIGroupItem,
     APIItem,
+    APIUsage,
     CostEstimates,
     DataRetrievalItem,
     EchoItem,
@@ -91,7 +93,9 @@
 
 __all__ = [
     "APIItem",
+    "APIGroupItem",
     "APIResult",
+    "APIUsage",
     "AgentBatchSummary",
     "AgentItem",
     "AgentMetadata",

src/shared/schemas/result/catalog.py

@@ -20,7 +20,9 @@
     AgentItem,
     AgentMetadata,
     AgentUsage,
+    APIGroupItem,
     APIItem,
+    APIUsage,
     CostEstimates,
     DataRetrievalItem,
     EchoItem,
@@ -245,9 +247,9 @@ class EchoResult(StrictExecutorResult):
 
 
 class APIResult(StrictExecutorResult):
-    """HTTP request output. ``response_json``/``usage``/``headers`` are the
-    upstream API's own payloads and stay open mappings. ``items`` carries one
-    entry per row."""
+    """HTTP request output. ``response_json``/``headers`` are the upstream
+    API's own payloads and stay open mappings; ``usage`` is the task's summed
+    token/call accounting. ``items`` carries one entry per row."""
 
     task_type: Literal[TaskType.API] = TaskType.API
     executor: str
@@ -257,9 +259,9 @@ class APIResult(StrictExecutorResult):
     truncated: bool = False
     headers: dict[str, str] | None = None
     response_json: Any = Field(default=None, alias="json")
-    usage: dict[str, Any] | None = None
+    usage: APIUsage | None = None
     text: str | None = None
-    items: list[APIItem] = Field(default_factory=list)
+    items: list[APIItem | APIGroupItem] = Field(default_factory=list)
 
 
 class SSHResult(StrictExecutorResult):

src/shared/schemas/result/payloads.py

@@ -19,6 +19,19 @@ class GenerationUsage(StrictModel):
     latency_sec: float
 
 
+class APIUsage(StrictModel):
+    """Token/call accounting for an API task, summed over its requests."""
+
+    prompt_tokens: int
+    completion_tokens: int
+    reasoning_tokens: int
+    calls: int
+    failures: int
+    retries: int
+    truncated_calls: int
+    wall_sec: float
+
+
 class EmbeddingUsage(StrictModel):
     """Token/latency accounting for embedding inference."""
 
@@ -230,3 +243,11 @@ class APIItem(StrictModel):
     usage: dict[str, Any] | None = None
     text: str | None = None
     prompt: str | None = None
+
+
+class APIGroupItem(StrictModel):
+    """One group's row responses in a batched API task over grouped data; ``rows``
+    holds the group's row responses in order."""
+
+    index: int
+    rows: list[APIItem]