feat: batch multiple HTTP requests in one API executor task

[timzsu]

Purpose

One API executor task issued exactly one HTTP request, so a multi-row op became N separate FlowMesh tasks. The serving endpoint answers request-by-request, so a batch is N requests — issue them in parallel from one task.

Changes

• api_executor.py — spec.data is required and there is a single execution path; a one-row batch is just N=1. The invariant request is built once and each row substitutes its prompt into a {{prompt}} slot. Rows are issued on a ThreadPoolExecutor bounded by _MAX_CONCURRENCY = 8, and the httpx pool is sized from the capped concurrency so requests cannot queue on connections.
• HTTP errors are classified before strict payload parsing, so a 4xx/5xx is reported as an HTTP error rather than as a parse failure — previously a retryable 503 became permanent.
• Task-scoped cancellation — check-and-reset is atomic under a lock, rechecked before run() and again in flight.
• APIItem / APIResult.items added to the shared schemas and mirrored in the SDK with the identical alias="json", registered in _RESULT_MODEL_NAMES so the drift guard covers it.
• n8n_parser.py, api_two_stage.yaml, docs/EXECUTORS.md, docs/WORKFLOWS.md.

Design

One path, not two. No if batching: branch, no second request builder, no second result shape.

Result shape. APIResult is extended rather than given a sibling: the discriminated union keys on task_type and admits one model per tag. Lumilake's result normaliser already reads items before falling back to text, so the consumer needs no change.

Placeholder {{prompt}}, deliberately not ${prompt}. ${...} is the server-side stage-reference syntax resolved at dispatch, before a task reaches a worker; per-row substitution happens worker-side, after that. A ${prompt} token would be walked by the stage resolver looking for a stage named "prompt".

Row alignment. Results are collected keyed by row index and rebuilt in input order, so completion order cannot disturb it. Any row failing fails the whole task; partial success is deliberately not offered.

Cancellation, stated precisely. A request already inside the synchronous HTTP call is NOT aborted. Cancelling prevents queued rows from starting and marks the task cancelled once in-flight requests return. The docstring previously overclaimed this and was corrected.

Test Plan

End-to-end on a stack built from this branch, against the office serving endpoint https://lum.id/llm:

1. Submit a multi-row API workflow and confirm it is one task.
2. Confirm each output row answers its own prompt and index == position.
3. Force a 5xx with raise_for_status: false and confirm the per-row status survives.
4. Force a 503 and confirm it fails with the concrete orchestrator error, retryable.
5. Cancel a running task and confirm it reaches cancelled.
6. Paired A/B at equal N and prompts, varying only spec.api.concurrency, repeated — never a single unpaired timing.

Test Result

All seven ran and passed against https://lum.id/llm with qwen3.8-27b.

• One task, failed_tasks: [], 8 rows, 8 distinct answers, index == position on every row — row alignment on real HTTP rather than httpx.MockTransport.
• 5xx preserved per-row status_code: 500; the 503 surfaced API request returned status 503 (row 0): service unavailable, the orchestrator's own error and not a batch wrapper.
• Cancellation reached cancelled tasks on the apibatch server.
• Parallelism, paired at N=8: serial median 6.44s (range 6.43–7.75), parallel median 2.66s (range 2.35–4.12). Ranges do not overlap. An earlier one-shot pair had given 5.89s serial against 8.23s parallel — a cold-start artefact that reversed under repetition, which is why the pairing is repeated rather than single.

---

Pre-submission Checklist

• I have read the contribution guidelines.
• I have run pre-commit run --all-files and fixed any issues.
• I have added or updated tests covering my changes (if applicable).
• I have verified that uv run pytest tests/ passes locally.
• If I changed shared schemas or proto definitions, I have checked downstream compatibility across Server and Worker.
• If I changed the SDK or CLI, I have verified the affected packages work (uv sync --all-packages --group ci --frozen).
• If this is a breaking change, I have prefixed the PR title with [BREAKING] and described migration steps above.
• I have updated documentation or config examples if user-facing behavior changed.