From e0934adb9f5068651d6effa411afb869e95ec0cb Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Sat, 18 Jul 2026 17:28:59 +0200 Subject: [PATCH 1/9] feat(runs): type tokens_usages/usage_assembly_error on RunResults MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both paths populate them: the hosted durable path reads the pair GET /v1/runs/{id}/results now returns (unpacked from the runner's tokens_usages.json artifact — pipelex-platform PR #82), and the blocking fallback unpacks the same pair from the execute response's extension-open pipe_output, so result.tokens_usages reads the same regardless of which path ran. None against older platforms or pre-artifact runs. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 6 +++ pipelex_sdk/client.py | 7 +++- pipelex_sdk/runs.py | 9 +++++ tests/unit/test_client_lifecycle.py | 30 +++++++++++++++ tests/unit/test_client_run_fallback.py | 52 ++++++++++++++++++++++++++ 5 files changed, 103 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35b420c..2ce0bdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## [Unreleased] + +### Added + +- **Typed run usage: `RunResults.tokens_usages` + `RunResults.usage_assembly_error`.** The per-call usage records (token counts by category, `unit_costs` in $/1M, model id — LLM and img-gen/extract/search alike) that previously rode `model_extra` are now first-class typed fields, populated on both paths: the hosted durable path reads them from `GET /v1/runs/{id}/results` (which unpacks the runner's `tokens_usages.json` artifact), and the blocking fallback unpacks the same pair from the execute response's extension-open `pipe_output` — so `result.tokens_usages` reads the same regardless of which path ran. Both fields are `None` when usage assembly was off for the run, or against an older platform / a run delivered before the artifact existed; `usage_assembly_error` is set when the runner's usage assembly failed. + ## [v0.4.0] - 2026-07-06 ### Changed diff --git a/pipelex_sdk/client.py b/pipelex_sdk/client.py index 7561696..2fd0ec1 100644 --- a/pipelex_sdk/client.py +++ b/pipelex_sdk/client.py @@ -949,13 +949,18 @@ def _map_run_result_to_run_results(response: PipelexExecuteResult) -> RunResults `response.main_stuff` resolves the main output out of the returned working memory (and raises `MissingMainStuffError` if the run named no locatable main stuff), so the durable and blocking paths hand back the same `main_stuff` content shape. The full working memory rides `pipe_output` - (blocking only). + (blocking only). The usage pair (`tokens_usages` / `usage_assembly_error`) is unpacked from the + execute response's extension-open `pipe_output`, so `.tokens_usages` reads the same on both the + blocking and durable paths. """ + pipe_output_extras: dict[str, Any] = response.pipe_output.model_extra or {} return RunResults( pipeline_run_id=response.pipeline_run_id, main_stuff=response.main_stuff, graph_spec=None, pipe_output=response.pipe_output.model_dump(), + tokens_usages=pipe_output_extras.get("tokens_usages"), + usage_assembly_error=pipe_output_extras.get("usage_assembly_error"), ) diff --git a/pipelex_sdk/runs.py b/pipelex_sdk/runs.py index 9db5c14..706d73f 100644 --- a/pipelex_sdk/runs.py +++ b/pipelex_sdk/runs.py @@ -141,6 +141,15 @@ class RunResults(BaseModel): #: blocking-execute path only; `None` on the hosted path. Supplementary to `main_stuff`, which is #: already resolved out of it; kept for consumers that need the whole working memory. pipe_output: dict[str, Any] | None = None + #: Per-call usage records — token counts by category, `unit_costs` in $/1M, model id — for LLM and + #: img-gen/extract/search calls alike. On the hosted path this is the `tokens_usages.json` artifact's + #: record list relayed verbatim; on the blocking path it is the execute response's + #: `pipe_output.tokens_usages`. `None` when usage assembly was off for the run, or (hosted) when the + #: run was delivered before the artifact existed. + tokens_usages: list[dict[str, Any]] | None = None + #: Non-`None` when the runner's usage assembly failed for the run — distinguishes "usage broke" + #: from "usage was off" (both leave `tokens_usages` as `None`). + usage_assembly_error: str | None = None # ── Single-shot result lookup outcome (discriminated on `state`) ───── diff --git a/tests/unit/test_client_lifecycle.py b/tests/unit/test_client_lifecycle.py index 417d134..baeb39d 100644 --- a/tests/unit/test_client_lifecycle.py +++ b/tests/unit/test_client_lifecycle.py @@ -156,6 +156,36 @@ def test_get_run_result_completed_keeps_falsy_main_stuff(self, mocker: MockerFix assert isinstance(state, RunResultCompleted) assert state.result.main_stuff == [] + def test_get_run_result_completed_parses_usage_pair(self, mocker: MockerFixture) -> None: + """A 200 carrying the hosted usage pair lands on the typed fields, records verbatim; a body + without them (older platform / pre-artifact run) defaults both to None. + """ + client = self._client() + tokens_usages = [ + { + "model_type": "llm", + "inference_model_name": "test-model", + "nb_tokens_by_category": {"input": 15, "output": 4}, + "unit_costs": {"input": 3.0, "output": 15.0}, + } + ] + body: dict[str, object] = { + "pipeline_run_id": "run_1", + "main_stuff": {"answer": "42"}, + "tokens_usages": tokens_usages, + "usage_assembly_error": None, + } + mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=_response(200, json=body))) + + state = asyncio.run(client.get_run_result("run_1")) + assert isinstance(state, RunResultCompleted) + assert state.result.tokens_usages == tokens_usages + assert state.result.usage_assembly_error is None + + bare = RunResults(pipeline_run_id="run_1", main_stuff={"answer": "42"}) + assert bare.tokens_usages is None + assert bare.usage_assembly_error is None + def test_get_run_result_running_honors_retry_after(self, mocker: MockerFixture) -> None: """A 202 maps to RunResultRunning with the server's Retry-After hint.""" client = self._client() diff --git a/tests/unit/test_client_run_fallback.py b/tests/unit/test_client_run_fallback.py index 612ecb8..bbd881d 100644 --- a/tests/unit/test_client_run_fallback.py +++ b/tests/unit/test_client_run_fallback.py @@ -136,6 +136,58 @@ def test_bare_runner_falls_back_to_blocking_execute(self, mocker: MockerFixture) assert result.pipe_output["working_memory"]["root"]["result"]["content"] == {"text": "hello"} assert _urls(send) == [f"{_BASE_URL}/v1/version", f"{_BASE_URL}/v1/execute"] + def test_blocking_fallback_unpacks_usage_pair_from_pipe_output(self, mocker: MockerFixture) -> None: + """The blocking execute response carries usage inside `pipe_output` (extension-open); the SDK + unpacks it onto `RunResults.tokens_usages` / `.usage_assembly_error` so the accessor reads the + same on both paths. A body without the pair (usage off) leaves both None. + """ + client = self._client() + tokens_usages = [ + { + "model_type": "llm", + "inference_model_name": "test-model", + "nb_tokens_by_category": {"input": 15, "output": 4}, + "unit_costs": {"input": 3.0, "output": 15.0}, + } + ] + usage_body: dict[str, object] = { + "pipeline_run_id": "run-x", + "main_stuff_name": "result", + "pipe_output": { + "working_memory": { + "root": {"result": {"concept": "native.Text", "content": {"text": "hello"}}}, + "aliases": {"main_stuff": "result"}, + }, + "pipeline_run_id": "run-x", + "tokens_usages": tokens_usages, + "usage_assembly_error": None, + }, + } + mocker.patch.object( + client, + "_send", + mocker.AsyncMock(side_effect=[_response(200, json=_BARE_VERSION), _response(200, json=usage_body)]), + ) + + result = asyncio.run(client.start_and_wait(pipe_code="p", mthds_contents=["x"])) + assert result.tokens_usages == tokens_usages + assert result.usage_assembly_error is None + + def test_blocking_fallback_without_usage_pair_defaults_to_none(self, mocker: MockerFixture) -> None: + """A blocking response whose pipe_output carries no usage fields (usage off, or an older + runner) maps to None on both fields — never a validation error. + """ + client = self._client() + mocker.patch.object( + client, + "_send", + mocker.AsyncMock(side_effect=[_response(200, json=_BARE_VERSION), _response(200, json=_EXECUTE_BODY)]), + ) + + result = asyncio.run(client.start_and_wait(pipe_code="p", mthds_contents=["x"])) + assert result.tokens_usages is None + assert result.usage_assembly_error is None + def test_blocking_fallback_raises_when_main_stuff_unlocatable(self, mocker: MockerFixture) -> None: """A completed blocking response whose `main_stuff_name` names no root stuff is a hard fail.""" client = self._client() From a0418efa240de171e05406e97ca65855fe1905df Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Tue, 21 Jul 2026 23:11:51 +0200 Subject: [PATCH 2/9] feat(runs)!: mirror the TokensUsageRecord wire contract MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `RunResults.tokens_usages` was a `list[dict[str, Any]]` bag. It now validates into `TokensUsageRecord`, a client-side mirror of the wire contract specified in docs/specs/pipelex-mthds-protocol.md — every field optional and `extra="allow"`, so pre-contract durable artifacts (relayed verbatim, never migrated) still parse with `cost`/`pipe_code` None and their legacy `job_metadata`/`unit_costs` kept in `model_extra`. Enum-ish fields stay plain `str` so runtime enum churn is non-breaking. Both paths feed the same typed records: the durable path off the results body, the blocking path lifted out of the execute response's extension-open `pipe_output`. Breaking: `RunResults.pipe_output` is now `DictPipeOutputAbstract | None` rather than `dict[str, Any] | None`. The blocking path already parsed the protocol model and then discarded the types via `.model_dump()`; it now carries the parsed model through, so consumers read the working memory as attributes. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 10 +- docs/architecture.md | 1 + docs/run-usage.md | 97 ++++++++++++++++++ pipelex_sdk/client.py | 14 ++- pipelex_sdk/runs.py | 95 +++++++++++++---- tests/unit/test_client_lifecycle.py | 22 +++- tests/unit/test_client_run_fallback.py | 21 +++- tests/unit/test_runs.py | 136 ++++++++++++++++++++++++- 8 files changed, 365 insertions(+), 31 deletions(-) create mode 100644 docs/run-usage.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 2ce0bdc..c0acb6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,15 @@ ### Added -- **Typed run usage: `RunResults.tokens_usages` + `RunResults.usage_assembly_error`.** The per-call usage records (token counts by category, `unit_costs` in $/1M, model id — LLM and img-gen/extract/search alike) that previously rode `model_extra` are now first-class typed fields, populated on both paths: the hosted durable path reads them from `GET /v1/runs/{id}/results` (which unpacks the runner's `tokens_usages.json` artifact), and the blocking fallback unpacks the same pair from the execute response's extension-open `pipe_output` — so `result.tokens_usages` reads the same regardless of which path ran. Both fields are `None` when usage assembly was off for the run, or against an older platform / a run delivered before the artifact existed; `usage_assembly_error` is set when the runner's usage assembly failed. +- **Typed run usage: `RunResults.tokens_usages` + `RunResults.usage_assembly_error`.** The per-call usage records a run produces — token counts by category, the server-computed `cost` in USD, model name and id, the pipe that made the call, job-kind fields and timing, for LLM and img-gen/extract/search calls alike — are now first-class typed fields instead of riding `model_extra`. Records validate into a new `TokensUsageRecord` model (`pipelex_sdk/runs.py`) mirroring the wire contract specified in the MTHDS protocol spec. Both paths populate the pair: the hosted durable path reads it off `GET /v1/runs/{id}/results` (which unpacks the runner's `tokens_usages.json` artifact), and the blocking fallback lifts the same pair out of the execute response's extension-open `pipe_output` — so `result.tokens_usages` reads the same regardless of which path ran. + + Note that the rate table (`unit_costs`) no longer crosses the wire: a record now carries the computed `cost` for the call instead, which is `None` when the model has no rate table at all (own-GPU, mock, dry run) and `0` when a rate table priced it at zero. There is no run-level aggregate — sum the records. + + `tokens_usages` is `None` whenever usage assembly produced no list (it was off, it broke, or the run was delivered before the artifact existed) and `[]` when assembly ran and no inference happened; `usage_assembly_error` is the only field separating a broken assembly from an off one. `TokensUsageRecord` keeps every field optional and is extension-open, so durable artifacts written before the contract shipped — relayed verbatim, never migrated — still parse: `cost` and `pipe_code` come back `None`, and the legacy `job_metadata` / `unit_costs` survive in `model_extra`. Enum-ish fields (`model_type`, `job_category`, `unit_job_id`) are open sets typed as plain `str`, so runtime enum churn stays non-breaking. + +### Changed + +- **`RunResults.pipe_output` is now `DictPipeOutputAbstract | None`, was `dict[str, Any] | None` (breaking).** The blocking path already parsed the protocol model and then threw the types away with a `.model_dump()` round-trip; it now carries the parsed model straight through. Read the working memory as attributes — `result.pipe_output.working_memory.root["out"].content` — rather than nested dict keys. The durable path still leaves it `None`. ## [v0.4.0] - 2026-07-06 diff --git a/docs/architecture.md b/docs/architecture.md index 2c76757..22b8525 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -88,6 +88,7 @@ The durable run lifecycle (`pipelex_sdk/runs.py` + the client's lifecycle method - `RunStatus` — the hosted status enum, with `is_terminal` / `is_success` predicates (exhaustive `match`). - `RunRead` — a run record read through the self-healing status path (adds `degraded` + `retry_after_seconds`). - `RunResults` — result artifacts. `main_stuff` (the resolved main output content) is always present for a completed run: on the hosted path it is the `main_stuff.json` S3 artifact; on the bare-runner blocking path the SDK resolves it from the returned working memory via the response's `main_stuff_name`, so both paths deliver the same shape. Consumers read `main_stuff` directly. The full working memory still rides `pipe_output` (blocking path only) for consumers that want it, and `graph_spec` rides the hosted path. A completed run that cannot deliver a main stuff raises `MissingMainStuffError`. Extension-open, so any other server artifact is preserved. +- `TokensUsageRecord` — one client-facing usage record per inference call, carried by `RunResults.tokens_usages` on both paths. A mirror of the wire contract specified in the MTHDS protocol spec, not a shape this SDK owns: every field is optional and the model is extension-open so pre-contract artifacts (relayed verbatim, never migrated) still parse. See [`run-usage.md`](./run-usage.md) for the field reference, the cost/null semantics, and the old-artifact rules. - `RunResultState` — the single-shot result outcome, a union discriminated on `state` (`running` / `completed` / `failed`). - `WaitForResultOptions` / `PollInfo` — poll-loop tuning and progress info. Async-native cancellation is via `asyncio.CancelledError` (cancel the awaiting task), so there is no `signal` field. diff --git a/docs/run-usage.md b/docs/run-usage.md new file mode 100644 index 0000000..80ce251 --- /dev/null +++ b/docs/run-usage.md @@ -0,0 +1,97 @@ +# Run usage — reading what a run consumed + +A completed run reports what its inference calls consumed as a list of `TokensUsageRecord` objects on `RunResults`, one per inference call, in the order the calls completed. This page covers how to read them, what each field means, and the edge cases the model is deliberately shaped around. + +The wire shape is not this SDK's invention: it is specified in the MTHDS protocol spec under [TokensUsage records on run artifacts](https://github.com/Pipelex/Pipelex/blob/main/docs/specs/pipelex-mthds-protocol.md#tokensusage-records-on-run-artifacts), and `pipelex_sdk.runs.TokensUsageRecord` is a client-side mirror of it. `@pipelex/sdk` carries the same mirror in TypeScript. + +## Reading the records + +```python +result = await client.start_and_wait(pipe_code="my_domain.summarize", mthds_contents=[...]) + +if result.tokens_usages is not None: + total_cost = sum(record.cost or 0.0 for record in result.tokens_usages) + for record in result.tokens_usages: + print(record.pipe_code, record.inference_model_name, record.nb_tokens_by_category, record.cost) +``` + +The accessor is the same whichever path ran. `start_and_wait` picks a path from the `GET /v1/version` handshake: + +- **Hosted (durable) path** — the records come from the runner's `tokens_usages.json` artifact, which `GET /v1/runs/{id}/results` unpacks onto the results body as top-level keys and relays verbatim. +- **Bare runner (blocking) path** — the records ride the execute response's extension-open `pipe_output` as Pipelex extension fields; the SDK lifts them onto the same two top-level fields. + +Because the runtime emits both surfaces through one helper, the two cannot structurally diverge. + +## Field reference + +| field | type | meaning | +|---|---|---| +| `model_type` | `str \| None` | Kind of inference. Known values: `llm`, `img_gen`, `extract`, `search`. | +| `inference_model_name` | `str \| None` | Human model name (e.g. `gpt-4o`). | +| `inference_model_id` | `str \| None` | Provider/platform model id (e.g. `gpt-4o-2024-11-20`). | +| `pipe_code` | `str \| None` | The pipe that made the call — what makes per-pipe cost attribution possible. | +| `job_category` | `str \| None` | Known values: `llm_job`, `img_gen_job`, `extract_job`, `search_job`, `jinja2_job`, `mock_job`. | +| `unit_job_id` | `str \| None` | Known values: `llm_gen_text`, `llm_gen_object`, `img_gen_text_to_image`, `extract_pages`, `search_sourced_answer`, `search_structured`. | +| `nb_tokens_by_category` | `dict[str, int] \| None` | Raw provider-reported token counts, keyed by token category (`input`, `input_cached`, `output`, `output_reasoning`, …). | +| `cost` | `float \| None` | Computed USD cost of this call. | +| `started_at` | `str \| None` | ISO 8601. | +| `completed_at` | `str \| None` | ISO 8601. | + +Two traps worth naming explicitly: + +- **Token categories are not additive.** `input` is the joined total and `input_cached` is a *subset* of it. Summing every category double-counts the cached tokens. +- **Duration is not shipped.** Derive it from the `started_at` / `completed_at` pair. + +### Enum-ish fields are open sets + +`model_type`, `job_category`, `unit_job_id`, and the `nb_tokens_by_category` keys are plain strings, never frozen enums, and the values listed above are *known* values rather than an exhaustive set. This is deliberate: the runtime can add an inference kind without breaking any SDK consumer. Match on them defensively — do not assume the list is closed. + +## Cost semantics + +`cost` is a server-computed USD total for that one call. The underlying rate table never crosses the wire, so there is nothing to recompute client-side and no risk of a client's arithmetic disagreeing with the runtime's own reporting — the figure comes from the same cost engine that produces the local CLI cost table. + +- `cost is None` means the model has **no rate table at all** — an own-GPU model, a mock run, a dry run. +- `cost == 0` means a rate table existed and priced the call at zero. + +Those are different facts; `record.cost or 0.0` conflates them, which is fine for a sum but wrong for "was this call priced?". + +There is no per-category cost breakdown and no run-level aggregate on the wire. Sum the records for a run total. + +## Null and empty semantics + +`tokens_usages` is `None` whenever usage assembly produced no list at all, which happens for three different reasons: + +- usage assembly was **off** for the run; +- usage assembly **broke** (an event-read failure); +- on the hosted path, the run was **delivered before the artifact existed**. + +It is `[]` when assembly ran, succeeded, and no inference happened, and non-empty otherwise. + +`usage_assembly_error` is the **only** field that distinguishes the broken case from the other two — they are otherwise indistinguishable on the wire. A caller that needs to tell "we have no usage data because something failed" from "there was nothing to report" must branch on `usage_assembly_error`, not on `tokens_usages` alone: + +```python +if result.usage_assembly_error is not None: + log.warning("usage assembly failed for this run: %s", result.usage_assembly_error) +elif result.tokens_usages is None: + ... # usage was off, or this run predates the artifact +elif not result.tokens_usages: + ... # ran, but no inference happened +``` + +## Old artifacts parse too + +Durable artifacts written before this contract shipped are relayed verbatim and never migrated. `TokensUsageRecord` therefore keeps **every field optional** and is extension-open (`extra="allow"`) — a pre-contract record parses without raising: + +- `cost` comes back `None` (it did not exist yet — the record carried a raw `unit_costs` rate table instead); +- `pipe_code` comes back `None` (it was still nested inside a `job_metadata` object rather than flattened onto the record); +- the legacy `job_metadata` and `unit_costs` survive in `model_extra`. + +Those legacy fields are **not** contract fields. They exist on old records only, and reading them is reading a relic — a record the current runtime emits never carries them. Treat their presence as a signal that you are looking at an old artifact, not as an API. + +Conversely, a record the current runtime emits always carries the **full key set**: a field with no value is an explicit `null`, never an omitted key. You can read any field without an existence check. + +## What is deliberately absent + +The runtime's internal reporting models carry execution plumbing — `job_metadata`, `otel_context`, `trace_context`, `session_id`, `request_id`, `user_id`, `pipe_run_id`, `content_generation_job_id` — that is dropped at the boundary and must never appear on a record. This is enforced upstream by leak-regression tests in `pipelex` and a conformance leak guard that walks relayed records at any nesting depth. + +One consequence worth knowing: the record shape is **invariant** with respect to server-side telemetry and tracing settings, because the only fields that varied with them are precisely the ones the boundary drops. You never get a structurally different record because an operator changed an observability setting. diff --git a/pipelex_sdk/client.py b/pipelex_sdk/client.py index 2fd0ec1..e62a59e 100644 --- a/pipelex_sdk/client.py +++ b/pipelex_sdk/client.py @@ -948,17 +948,21 @@ def _map_run_result_to_run_results(response: PipelexExecuteResult) -> RunResults `response.main_stuff` resolves the main output out of the returned working memory (and raises `MissingMainStuffError` if the run named no locatable main stuff), so the durable and blocking - paths hand back the same `main_stuff` content shape. The full working memory rides `pipe_output` - (blocking only). The usage pair (`tokens_usages` / `usage_assembly_error`) is unpacked from the - execute response's extension-open `pipe_output`, so `.tokens_usages` reads the same on both the - blocking and durable paths. + paths hand back the same `main_stuff` content shape. The already-parsed `pipe_output` model is + carried over as-is — no `.model_dump()` round-trip — so the full working memory stays typed + (blocking only; the hosted path has none). + + The usage pair (`tokens_usages` / `usage_assembly_error`) rides the execute response's + extension-open `pipe_output` as Pipelex extension fields. Lifting it onto the two top-level + fields here is what makes `.tokens_usages` read the same on the blocking and durable paths; + `RunResults` validates the raw records into `TokensUsageRecord`s on the way in. """ pipe_output_extras: dict[str, Any] = response.pipe_output.model_extra or {} return RunResults( pipeline_run_id=response.pipeline_run_id, main_stuff=response.main_stuff, graph_spec=None, - pipe_output=response.pipe_output.model_dump(), + pipe_output=response.pipe_output, tokens_usages=pipe_output_extras.get("tokens_usages"), usage_assembly_error=pipe_output_extras.get("usage_assembly_error"), ) diff --git a/pipelex_sdk/runs.py b/pipelex_sdk/runs.py index 706d73f..9aa6cf6 100644 --- a/pipelex_sdk/runs.py +++ b/pipelex_sdk/runs.py @@ -10,11 +10,18 @@ bare runner 404s these routes, which the client translates into `RunLifecycleUnavailableError`. -These types are **owned by this SDK** (not imported from `mthds`): the run -lifecycle is a Pipelex-branded hosted surface, mirroring `pipelex-sdk-js/src/runs.ts`. -During the transition (HANDOFF Phase 2) the same shapes still exist in -`mthds-python`; that duplication is deliberate and is removed from `mthds-python` -in Phase 6, leaving these as the single home. +The lifecycle types **defined here are owned by this SDK** (not imported from +`mthds`): the run lifecycle is a Pipelex-branded hosted surface, mirroring +`pipelex-sdk-js/src/runs.ts`. During the transition (HANDOFF Phase 2) the same +shapes still exist in `mthds-python`; that duplication is deliberate and is +removed from `mthds-python` in Phase 6, leaving these as the single home. + +Two things in this module are deliberately NOT owned here, and both reuse rather +than redefine. `RunResults.pipe_output` is typed with the protocol's own +`DictPipeOutputAbstract` wire model from `mthds` — a shared wire contract the +`pipelex` runtime also builds on, not a lifecycle concept. `TokensUsageRecord` +mirrors a runtime wire contract specified in the MTHDS protocol spec; this SDK +follows that shape, it does not define it. Wire contract mirrors `pipelex-platform`: POST /v1/start -> RunResultStart (start, 202) @@ -28,6 +35,7 @@ from enum import StrEnum from typing import TYPE_CHECKING, Annotated, Any, Literal, TypeAlias +from mthds.runners.api.models import DictPipeOutputAbstract from pydantic import BaseModel, ConfigDict, Field if TYPE_CHECKING: @@ -114,6 +122,54 @@ class RunRead(RunPublic): retry_after_seconds: int | None = None +class TokensUsageRecord(BaseModel): + """One inference call's token usage — the client-facing wire record. + + Mirrors the runtime's `TokensUsageRecord`, specified in + `docs/specs/pipelex-mthds-protocol.md#tokensusage-records-on-run-artifacts`. The same + shape rides both surfaces: the durable `tokens_usages.json` artifact that the hosted + results route relays, and the blocking execute response's `pipe_output.tokens_usages`. + + Every field is optional and the model is extension-open **on purpose**. A record the + current runtime emits always carries the full key set (a field with no value is an + explicit `null`, never an omitted key), so callers may read any field without an + existence check. But durable artifacts written before the contract shipped are relayed + verbatim and never migrated: such a record parses here with `cost` and `pipe_code` unset + and keeps its legacy `job_metadata` / `unit_costs` in `model_extra`. + + The enum-ish fields are open sets on the wire and stay plain `str` here — never frozen + enums — so runtime enum churn is non-breaking for consumers. + """ + + model_config = ConfigDict(extra="allow") + + #: Kind of inference. Known values: `llm`, `img_gen`, `extract`, `search`. + model_type: str | None = None + #: Human model name (e.g. `gpt-4o`). + inference_model_name: str | None = None + #: Provider/platform model id (e.g. `gpt-4o-2024-11-20`). + inference_model_id: str | None = None + #: The pipe that made the call — what makes per-pipe cost attribution possible. + pipe_code: str | None = None + #: Known values: `llm_job`, `img_gen_job`, `extract_job`, `search_job`, `jinja2_job`, `mock_job`. + job_category: str | None = None + #: Known values: `llm_gen_text`, `llm_gen_object`, `img_gen_text_to_image`, `extract_pages`, + #: `search_sourced_answer`, `search_structured`. + unit_job_id: str | None = None + #: Raw provider-reported token counts, keyed by token category (`input`, `input_cached`, + #: `output`, `output_reasoning`, …). `input` is the joined total and `input_cached` a subset + #: of it — the categories are NOT additive, so summing them double-counts. + nb_tokens_by_category: dict[str, int] | None = None + #: Computed USD cost of this call. `None` when the model has no rate table at all (own-GPU, + #: mock, dry run); `0` means a rate table existed and priced the call at zero. The underlying + #: rate table never crosses the wire and there is no run-level aggregate — sum the records. + cost: float | None = None + #: ISO 8601 start of the call. + started_at: str | None = None + #: ISO 8601 end of the call. Duration is derivable from the pair and deliberately not shipped. + completed_at: str | None = None + + class RunResults(BaseModel): """Result artifacts for a completed run — `GET /v1/runs/{pipeline_run_id}/results`. @@ -137,18 +193,23 @@ class RunResults(BaseModel): main_stuff: Any #: Method graph spec (`graphspec.json`); `None` if missing mid-write or on the bare-runner path. graph_spec: Any = None - #: Bare runner's native pipe output — the full working memory (`{"root": ..., "aliases": ...}`), - #: blocking-execute path only; `None` on the hosted path. Supplementary to `main_stuff`, which is - #: already resolved out of it; kept for consumers that need the whole working memory. - pipe_output: dict[str, Any] | None = None - #: Per-call usage records — token counts by category, `unit_costs` in $/1M, model id — for LLM and - #: img-gen/extract/search calls alike. On the hosted path this is the `tokens_usages.json` artifact's - #: record list relayed verbatim; on the blocking path it is the execute response's - #: `pipe_output.tokens_usages`. `None` when usage assembly was off for the run, or (hosted) when the - #: run was delivered before the artifact existed. - tokens_usages: list[dict[str, Any]] | None = None - #: Non-`None` when the runner's usage assembly failed for the run — distinguishes "usage broke" - #: from "usage was off" (both leave `tokens_usages` as `None`). + #: Bare runner's native pipe output — the full working memory, blocking-execute path only; + #: `None` on the hosted path. Supplementary to `main_stuff`, which is already resolved out of + #: it; kept for consumers that need the whole working memory. Extension-open, so the Pipelex + #: extension fields the runner rides on it stay reachable via `model_extra` — including the + #: usage pair, in its **raw** form. Read `tokens_usages` below instead: same data, validated + #: into records, and present on the hosted path too (where `pipe_output` is `None`). + pipe_output: DictPipeOutputAbstract | None = None + #: Per-call usage records — token counts by category, computed `cost` in USD, model id — for + #: LLM and img-gen/extract/search calls alike. On the hosted path this is the + #: `tokens_usages.json` artifact's record list relayed verbatim; on the blocking path it is the + #: execute response's `pipe_output.tokens_usages`. `None` whenever assembly produced no list — + #: it was off, it broke (see `usage_assembly_error`), or (hosted) the run was delivered before + #: the artifact existed; `[]` when assembly ran and no inference happened. + tokens_usages: list[TokensUsageRecord] | None = None + #: Non-`None` when the runner's usage assembly failed for the run. The ONLY field that + #: separates "usage broke" from "usage was off" / "pre-artifact run" — all three leave + #: `tokens_usages` as `None`, so a caller that cares must branch on this, not on the list. usage_assembly_error: str | None = None diff --git a/tests/unit/test_client_lifecycle.py b/tests/unit/test_client_lifecycle.py index baeb39d..a42bf6a 100644 --- a/tests/unit/test_client_lifecycle.py +++ b/tests/unit/test_client_lifecycle.py @@ -22,6 +22,7 @@ RunResultRunning, RunResults, RunStatus, + TokensUsageRecord, WaitForResultOptions, ) @@ -157,16 +158,23 @@ def test_get_run_result_completed_keeps_falsy_main_stuff(self, mocker: MockerFix assert state.result.main_stuff == [] def test_get_run_result_completed_parses_usage_pair(self, mocker: MockerFixture) -> None: - """A 200 carrying the hosted usage pair lands on the typed fields, records verbatim; a body - without them (older platform / pre-artifact run) defaults both to None. + """A 200 carrying the hosted usage pair validates the relayed records into + `TokensUsageRecord`s; a body without them (older platform / pre-artifact run) defaults both + to None. """ client = self._client() tokens_usages = [ { "model_type": "llm", "inference_model_name": "test-model", + "inference_model_id": "test-model-2026-01-01", + "pipe_code": "test_domain.summarize", + "job_category": "llm_job", + "unit_job_id": "llm_gen_text", "nb_tokens_by_category": {"input": 15, "output": 4}, - "unit_costs": {"input": 3.0, "output": 15.0}, + "cost": 0.000105, + "started_at": "2026-06-20T10:00:01+00:00", + "completed_at": "2026-06-20T10:00:03+00:00", } ] body: dict[str, object] = { @@ -179,7 +187,13 @@ def test_get_run_result_completed_parses_usage_pair(self, mocker: MockerFixture) state = asyncio.run(client.get_run_result("run_1")) assert isinstance(state, RunResultCompleted) - assert state.result.tokens_usages == tokens_usages + assert state.result.tokens_usages is not None + record = state.result.tokens_usages[0] + assert isinstance(record, TokensUsageRecord) + assert record.inference_model_name == "test-model" + assert record.pipe_code == "test_domain.summarize" + assert record.nb_tokens_by_category == {"input": 15, "output": 4} + assert record.cost == 0.000105 assert state.result.usage_assembly_error is None bare = RunResults(pipeline_run_id="run_1", main_stuff={"answer": "42"}) diff --git a/tests/unit/test_client_run_fallback.py b/tests/unit/test_client_run_fallback.py index bbd881d..acbe6d0 100644 --- a/tests/unit/test_client_run_fallback.py +++ b/tests/unit/test_client_run_fallback.py @@ -13,6 +13,7 @@ from pipelex_sdk.client import PipelexAPIClient from pipelex_sdk.errors import ApiUnreachableError, MissingMainStuffError, RunLifecycleUnavailableError +from pipelex_sdk.runs import TokensUsageRecord _BASE_URL = "http://localhost:8081" @@ -132,8 +133,9 @@ def test_bare_runner_falls_back_to_blocking_execute(self, mocker: MockerFixture) # The SDK resolves `main_stuff` out of the working memory via `main_stuff_name` ("result") — # its content, the same shape the hosted path relays; the full working memory rides pipe_output. assert result.main_stuff == {"text": "hello"} + # `pipe_output` is the already-parsed protocol model, carried over without a dump round-trip. assert result.pipe_output is not None - assert result.pipe_output["working_memory"]["root"]["result"]["content"] == {"text": "hello"} + assert result.pipe_output.working_memory.root["result"].content == {"text": "hello"} assert _urls(send) == [f"{_BASE_URL}/v1/version", f"{_BASE_URL}/v1/execute"] def test_blocking_fallback_unpacks_usage_pair_from_pipe_output(self, mocker: MockerFixture) -> None: @@ -146,8 +148,14 @@ def test_blocking_fallback_unpacks_usage_pair_from_pipe_output(self, mocker: Moc { "model_type": "llm", "inference_model_name": "test-model", + "inference_model_id": "test-model-2026-01-01", + "pipe_code": "test_domain.summarize", + "job_category": "llm_job", + "unit_job_id": "llm_gen_text", "nb_tokens_by_category": {"input": 15, "output": 4}, - "unit_costs": {"input": 3.0, "output": 15.0}, + "cost": 0.000105, + "started_at": "2026-06-20T10:00:01+00:00", + "completed_at": "2026-06-20T10:00:03+00:00", } ] usage_body: dict[str, object] = { @@ -170,7 +178,14 @@ def test_blocking_fallback_unpacks_usage_pair_from_pipe_output(self, mocker: Moc ) result = asyncio.run(client.start_and_wait(pipe_code="p", mthds_contents=["x"])) - assert result.tokens_usages == tokens_usages + assert result.tokens_usages is not None + record = result.tokens_usages[0] + # Same typed record the durable path yields — the pair is validated, not passed through raw. + assert isinstance(record, TokensUsageRecord) + assert record.inference_model_name == "test-model" + assert record.pipe_code == "test_domain.summarize" + assert record.nb_tokens_by_category == {"input": 15, "output": 4} + assert record.cost == 0.000105 assert result.usage_assembly_error is None def test_blocking_fallback_without_usage_pair_defaults_to_none(self, mocker: MockerFixture) -> None: diff --git a/tests/unit/test_runs.py b/tests/unit/test_runs.py index 11c8bd2..eb98a5a 100644 --- a/tests/unit/test_runs.py +++ b/tests/unit/test_runs.py @@ -1,9 +1,45 @@ """Tests for pipelex_sdk.runs — run-lifecycle models for the hosted polling surface.""" +from typing import Any + import pytest from pydantic import TypeAdapter -from pipelex_sdk.runs import RunStatus +from pipelex_sdk.runs import RunResults, RunStatus, TokensUsageRecord + +# A record in the shape the current runtime emits: every contract field present, absent values +# sent as explicit nulls. Mirrors the conformance seed corpus +# (conformance/conformance/usage_records.py), which is what the platform arm asserts on the wire. +_RATED_RECORD: dict[str, Any] = { + "model_type": "llm", + "inference_model_name": "test-model", + "inference_model_id": "test-model-2026-01-01", + "pipe_code": "test_domain.summarize", + "job_category": "llm_job", + "unit_job_id": "llm_gen_text", + "nb_tokens_by_category": {"input": 15, "input_cached": 5, "output": 4}, + "cost": 0.000105, + "started_at": "2026-06-20T10:00:01+00:00", + "completed_at": "2026-06-20T10:00:03+00:00", +} + +# A durable artifact written BEFORE the wire contract shipped, relayed verbatim ever since: a dump +# of the runtime's internal reporting model, carrying the nested `job_metadata` and the `unit_costs` +# rate table, and lacking the computed `cost`. Old artifacts are never migrated, so the mirror must +# parse this without complaint. +_PRE_CONTRACT_RECORD: dict[str, Any] = { + "model_type": "llm", + "inference_model_name": "legacy-model", + "inference_model_id": "legacy-model-v0", + "nb_tokens_by_category": {"input": 20, "output": 6}, + "unit_costs": {"input": 3.0, "output": 15.0}, + "job_metadata": { + "pipe_code": "legacy_domain.summarize", + "job_category": "llm_job", + "session_id": "legacy-session", + "user_id": "legacy-user", + }, +} class TestRuns: @@ -29,3 +65,101 @@ def test_run_status_parses_from_string(self) -> None: """A wire string parses into the enum.""" adapter = TypeAdapter(RunStatus) assert adapter.validate_python("TIMED_OUT") == RunStatus.TIMED_OUT + + def test_tokens_usage_record_parses_every_contract_field(self) -> None: + """A current-shape record round-trips each contract field with its wire value and type.""" + record = TokensUsageRecord.model_validate(_RATED_RECORD) + + assert record.model_type == "llm" + assert record.inference_model_name == "test-model" + assert record.inference_model_id == "test-model-2026-01-01" + assert record.pipe_code == "test_domain.summarize" + assert record.job_category == "llm_job" + assert record.unit_job_id == "llm_gen_text" + assert record.nb_tokens_by_category == {"input": 15, "input_cached": 5, "output": 4} + assert record.cost == 0.000105 + assert record.started_at == "2026-06-20T10:00:01+00:00" + assert record.completed_at == "2026-06-20T10:00:03+00:00" + # Nothing rode `model_extra`: the contract field set covers the whole record. + assert record.model_extra == {} + + def test_tokens_usage_record_parses_pre_contract_record(self) -> None: + """A pre-contract artifact record parses instead of raising: the contract fields it predates + come back None, and its legacy fields survive as extras rather than being dropped. + """ + record = TokensUsageRecord.model_validate(_PRE_CONTRACT_RECORD) + + assert record.inference_model_name == "legacy-model" + assert record.nb_tokens_by_category == {"input": 20, "output": 6} + # `cost` is server-computed and did not exist when this artifact was written; `pipe_code` was + # still nested inside `job_metadata` rather than flattened onto the record. + assert record.cost is None + assert record.pipe_code is None + # The legacy fields ride `model_extra` — relayed, never reshaped. A client must not read them + # as contract fields, but the mirror must not choke on them either. + assert record.model_extra == { + "unit_costs": {"input": 3.0, "output": 15.0}, + "job_metadata": { + "pipe_code": "legacy_domain.summarize", + "job_category": "llm_job", + "session_id": "legacy-session", + "user_id": "legacy-user", + }, + } + + def test_tokens_usage_record_keeps_unrated_cost_null(self) -> None: + """An unrated call sends `cost: null` — distinct from a rate table that priced it at zero.""" + unrated = TokensUsageRecord.model_validate({**_RATED_RECORD, "cost": None}) + priced_at_zero = TokensUsageRecord.model_validate({**_RATED_RECORD, "cost": 0}) + + assert unrated.cost is None + assert priced_at_zero.cost == 0.0 + assert priced_at_zero.cost is not None + + def test_run_results_validates_usage_records(self) -> None: + """A results body's raw records become typed records; the null branch stays None.""" + results = RunResults.model_validate( + { + "pipeline_run_id": "run_1", + "main_stuff": {"answer": "42"}, + "tokens_usages": [_RATED_RECORD, _PRE_CONTRACT_RECORD], + "usage_assembly_error": None, + } + ) + + assert results.tokens_usages is not None + assert [record.inference_model_name for record in results.tokens_usages] == ["test-model", "legacy-model"] + assert [record.cost for record in results.tokens_usages] == [0.000105, None] + assert results.usage_assembly_error is None + + @pytest.mark.parametrize( + ("tokens_usages", "usage_assembly_error"), + [ + pytest.param(None, None, id="assembly-off-or-pre-artifact"), + pytest.param(None, "failed to read usage events for the run", id="assembly-broke"), + pytest.param([], None, id="assembly-ran-no-inference"), + ], + ) + def test_run_results_preserves_usage_null_semantics(self, tokens_usages: list[dict[str, Any]] | None, usage_assembly_error: str | None) -> None: + """`None` (off / broke / pre-artifact) and `[]` (ran, no inference) stay distinct, and + `usage_assembly_error` is the only field separating a broken assembly from the other nulls. + """ + results = RunResults.model_validate( + { + "pipeline_run_id": "run_1", + "main_stuff": {"answer": "42"}, + "tokens_usages": tokens_usages, + "usage_assembly_error": usage_assembly_error, + } + ) + + assert results.tokens_usages == tokens_usages + assert results.usage_assembly_error == usage_assembly_error + + def test_run_results_defaults_usage_pair_to_none(self) -> None: + """A body with no usage keys at all (older platform) leaves both fields None, never raises.""" + results = RunResults.model_validate({"pipeline_run_id": "run_1", "main_stuff": {"answer": "42"}}) + + assert results.tokens_usages is None + assert results.usage_assembly_error is None + assert results.pipe_output is None From 0b04c9304c944f6386fa71f1706aff4c11b90804 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 00:01:27 +0200 Subject: [PATCH 3/9] docs(usage): scope the contract-only claims and fix the run-usage example MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review follow-ups raised on the TypeScript mirror of this page; the same three defects were present here, so both SDKs stay in sync. - Drop a dead link to the protocol spec. It pointed at a path that does not exist in the public runtime repo, and the spec's real home is an internal repo, so there is no public target to redirect to — reference the spec and its section by name instead. While there, correct the brand: the record is a Pipelex runtime concept, not part of the MTHDS standard, which says nothing about usage reporting. - Make the example meaningful. `mthds_contents=[...]` parses in Python but passes an `Ellipsis` where a bundle source is expected; a registered `pipe_code` with `inputs` is both valid and the more idiomatic hosted call. - Scope two guarantees that the page stated absolutely while documenting their exception further down. The spec qualifies both with "on a record emitted under this contract": pre-contract artifacts are relayed verbatim, so they do carry a raw `unit_costs` rate table and `job_metadata`. Name the exemption at both claims rather than only at the compatibility section. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/run-usage.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/docs/run-usage.md b/docs/run-usage.md index 80ce251..2c70606 100644 --- a/docs/run-usage.md +++ b/docs/run-usage.md @@ -2,12 +2,12 @@ A completed run reports what its inference calls consumed as a list of `TokensUsageRecord` objects on `RunResults`, one per inference call, in the order the calls completed. This page covers how to read them, what each field means, and the edge cases the model is deliberately shaped around. -The wire shape is not this SDK's invention: it is specified in the MTHDS protocol spec under [TokensUsage records on run artifacts](https://github.com/Pipelex/Pipelex/blob/main/docs/specs/pipelex-mthds-protocol.md#tokensusage-records-on-run-artifacts), and `pipelex_sdk.runs.TokensUsageRecord` is a client-side mirror of it. `@pipelex/sdk` carries the same mirror in TypeScript. +The wire shape is not this SDK's invention: it is specified in Pipelex's protocol spec, under "TokensUsage records on run artifacts", and `pipelex_sdk.runs.TokensUsageRecord` is a client-side mirror of it. `@pipelex/sdk` carries the same mirror in TypeScript. The record is a Pipelex runtime concept rather than part of the MTHDS standard — the MTHDS protocol itself says nothing about usage reporting. ## Reading the records ```python -result = await client.start_and_wait(pipe_code="my_domain.summarize", mthds_contents=[...]) +result = await client.start_and_wait(pipe_code="my_domain.summarize", inputs={"text": "..."}) if result.tokens_usages is not None: total_cost = sum(record.cost or 0.0 for record in result.tokens_usages) @@ -48,7 +48,7 @@ Two traps worth naming explicitly: ## Cost semantics -`cost` is a server-computed USD total for that one call. The underlying rate table never crosses the wire, so there is nothing to recompute client-side and no risk of a client's arithmetic disagreeing with the runtime's own reporting — the figure comes from the same cost engine that produces the local CLI cost table. +`cost` is a server-computed USD total for that one call. The rate table behind it is not a contract field and does not cross the wire, so there is nothing to recompute client-side and no risk of a client's arithmetic disagreeing with the runtime's own reporting — the figure comes from the same cost engine that produces the local CLI cost table. (Pre-contract artifacts are the one exception: they carry a raw `unit_costs` table, which is a relic rather than an API — see [Old artifacts parse too](#old-artifacts-parse-too).) - `cost is None` means the model has **no rate table at all** — an own-GPU model, a mock run, a dry run. - `cost == 0` means a rate table existed and priced the call at zero. @@ -92,6 +92,6 @@ Conversely, a record the current runtime emits always carries the **full key set ## What is deliberately absent -The runtime's internal reporting models carry execution plumbing — `job_metadata`, `otel_context`, `trace_context`, `session_id`, `request_id`, `user_id`, `pipe_run_id`, `content_generation_job_id` — that is dropped at the boundary and must never appear on a record. This is enforced upstream by leak-regression tests in `pipelex` and a conformance leak guard that walks relayed records at any nesting depth. +The runtime's internal reporting models carry execution plumbing — `job_metadata`, `otel_context`, `trace_context`, `session_id`, `request_id`, `user_id`, `pipe_run_id`, `content_generation_job_id` — that is dropped at the boundary: on a record emitted under this contract, finding one of these is reading a leak, not a contract field. This is enforced upstream by leak-regression tests in `pipelex` and a conformance leak guard that walks relayed records at any nesting depth. Pre-contract artifacts are the documented exemption — relayed verbatim, they legitimately still carry `job_metadata` and `unit_costs`, and the leak guard does not run on them. One consequence worth knowing: the record shape is **invariant** with respect to server-side telemetry and tracing settings, because the only fields that varied with them are precisely the ones the boundary drops. You never get a structurally different record because an operator changed an observability setting. From 9ca1147879d1a34f29f98a230a01653667614f1b Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 02:27:51 +0200 Subject: [PATCH 4/9] docs: document planned upload_file/prepare_inputs preparation surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 0 of the hosted input-upload & SDK-preparation project: promote the approved contract into this repo's own docs before the code lands. New docs/input-preparation.md is the Python-flavored mirror of the JS SDK contract — upload_file/prepare_inputs, str/Path/bytes sources, signature-driven asset identification, upload-record guarantees, error/capability outcomes, inherited storage policy, and stability across the future endpoint move. Notes the build_inputs parity gap prepare_inputs closes. Pointer added from architecture.md's storage bullet. Docs-only; no code touched. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Y9RPphjd62DywHtok2fX1q --- docs/architecture.md | 2 +- docs/input-preparation.md | 100 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 docs/input-preparation.md diff --git a/docs/architecture.md b/docs/architecture.md index 22b8525..c357c95 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -138,7 +138,7 @@ The wire models are snake_case Pydantic v2. Response models are extension-open ( - **Pipelex API keys** — `list_pipelex_api_keys()`; `create_pipelex_api_key(label)` and `rotate_pipelex_api_key(id)` return the plaintext `api_key` **once**; `revoke_pipelex_api_key(id)`. Creation surfaces a **409 `pipelex_api_key_limit_reached`** when the per-account limit is hit. Rotation sends no body. - **Gateway (LLM inference) key** — `create_gateway_api_key(promo_code)` **always sends a JSON body** (even with `promo_code=None` → `{"promo_code": null}`); the server 422s an empty body. `get_gateway_api_key()` → status (`gateway_api_key` is `None` until provisioned). - **Onboarding** — `submit_onboarding(OnboardingSubmission)` (`POST /v1/onboarding/submit`, empty 2xx body); absent optional fields are dropped. -- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle. +- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle. The higher-level `upload_file` / `prepare_inputs` preparation surface built on top of `upload` is a planned addition — see [input-preparation.md](./input-preparation.md). - **Run records** — `list_runs(method_id)` → `list[PipelineRun]` (the catalog-style list, distinct from the lifecycle status/result routes); `update_run(run_id, UpdateRunInput)` (admin/manual status patch, empty 2xx body). ## Health probe diff --git a/docs/input-preparation.md b/docs/input-preparation.md new file mode 100644 index 0000000..7664a5b --- /dev/null +++ b/docs/input-preparation.md @@ -0,0 +1,100 @@ +# Input preparation (`upload_file` / `prepare_inputs`) + +> **Status: planned surface — not yet implemented.** This document records the approved cross-repository contract (design source: `wip/upload/README.md` in the workspace, tracked in `TODOS.md`) so the SDK's own docs carry the part it owns before the code lands. The raw `upload()` primitive described in [architecture.md](./architecture.md) already exists; `upload_file` and `prepare_inputs` are the higher-level surface built on top of it, and are the Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`. + +## Why this exists + +A hosted run cannot see the caller's filesystem. Turning caller-local assets into run-ready inputs is therefore the SDK's job, not the runner's — the SDK process is the only component that can read the local file or hold the bytes. Today that work is re-implemented by every consumer (read file → base64 → `POST /v1/upload` → rewrite the input to the returned URI); `fenix-pipelex` is an early real-world example. `prepare_inputs` makes it one reusable, explicit operation. + +Preparation is **explicit and separate from running.** `execute` / `start` never silently upload local files. The payoff: file-access errors happen *before* a run exists, prepared inputs are inspectable and reusable across a model sweep or retries without re-uploading, and `start` keeps a deterministic JSON-input contract. + +## Parity note + +This is the Python side of one cross-language contract. The behavior matrix, pass-through rules, `Dynamic` handling, dedup, and failure categories are identical to the JS SDK; only the accepted source types differ per language. The two SDKs must agree semantically. See the JS counterpart's `docs/input-preparation.md` for the mirror. + +The Python SDK currently has **no `/v1/build/*` coverage** — `prepare_inputs` adds the `build_inputs` counterpart it needs to resolve the declared signature (the JS SDK already exposes `buildInputs`). + +## The two operations + +### `upload_file` — single-asset convenience + +Uploads one asset and returns its upload record. It is the language-native convenience over the raw `upload()` wire call (base64 JSON body), assembling the record client-side. + +- **Accepted sources:** `str` and `pathlib.Path` filesystem paths, and raw `bytes`. +- A string that is an **HTTP(S) URL** or an existing **`pipelex-storage://` URI** is not a local asset and passes through (see pass-through rules below). +- Open file objects and streams are **deferred** — they can be added later without removing anything. + +The returned **upload record** guarantees, beyond the source identity: + +| Field | Guarantee | +| --- | --- | +| `uri` | The `pipelex-storage://` reference for the uploaded asset. | +| content type (MIME) | Known client-side at upload time. | +| size (bytes) | Known client-side at upload time. | +| filename | Already in the wire model. | +| checksum | **Best-effort, not guaranteed.** Within-preparation dedup relies on source identity, not hashing; cross-preparation dedup is a hosted storage-policy concern (Phase 5). | + +The MIME type and size are known client-side, so the record is assembled without extending the `/v1/upload` response. + +### `prepare_inputs` — signature-driven input preparation + +``` +prepare_inputs(method_ref, pipe, inputs) → PreparedInputs +``` + +Takes the **method reference** (bundle files or catalog `method_id`) plus the target **pipe**, resolves the pipe's declared input signature, interprets the caller's compact `inputs` top-down against that signature, uploads the file-bearing values, and returns `PreparedInputs`: + +- `inputs` — a **copy** of the caller's inputs with each asset reference replaced by the canonical content shape carrying `pipelex-storage://` in its `url` field (see "Rewritten-input shape" below). Copy-on-write: the caller's original object is never mutated. +- `uploads` — one upload record per prepared asset (the `upload_file` record shape), exposing `uri` so callers can log which source became which reference without reverse-engineering the rewritten object. + +The prepared `inputs` are passed to the existing run lifecycle unchanged. + +## Signature-driven asset identification + +The SDK **must not** guess that every string resembling a path is an asset — that would make ordinary text inputs environment-dependent and could upload unintended files. Interpretation comes from the method's **declared signature**, never from a value's shape alone. This mirrors the runtime's own top-down interpretation (`pipelex/pipelex/core/memory/input_shaper.py`, `InputShaper`) combined with the file-reference resolution of `pipelex/pipelex/pipeline/input_normalizer.py`, so local and hosted execution read the same compact inputs the same way. + +The declared signature is resolved via the explicit inputs template (`build_inputs` with `explicit=True`), which carries concept identity, canonical content shape, and multiplicity per input. + +Interpretation per declared input: + +- A bare string, `Path`, or `bytes` value at an **Image/Document-declared** input is a **file reference**: local paths, data URLs, and bytes are uploaded and rewritten to `pipelex-storage://` URIs; HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged. +- The **identical** bare string at a **Text-declared** input is text and is never touched. +- **Canonical image/document content structures** are recognized by their URL-bearing fields wherever they appear, including nested in structured objects and lists — exactly as the runtime normalizer walks them. The refining case matters: a concept refining `Image`/`PDF` is classified by the **canonical content shape**, not by the concept ref alone. +- Inputs declared **`Dynamic`** are not path-interpreted (the signature genuinely cannot guide them); they accept canonical content structures or already-prepared references only. +- A repeated reference to the **same source** within one preparation is uploaded once and rewritten consistently (within-preparation dedup by source identity). + +### Pass-through rules + +| Source at a file-bearing input | Action | +| --- | --- | +| Local path (`str`/`Path`) / data URL / `bytes` | Upload → rewrite to `pipelex-storage://` | +| Existing `pipelex-storage://` URI | Already prepared — pass through unchanged | +| HTTP(S) URL | Pass through unchanged, **unless** the caller explicitly asks to ingest it into Pipelex storage | + +## Rewritten-input shape: `url` carries the URI + +The runtime's canonical image/document content stores its reference in a **`url`** field. Preparation emits inputs the runtime interprets natively, so a rewritten input keeps the canonical content shape with `url` holding the `pipelex-storage://` value — exactly what the runtime's `input_normalizer` writes. + +The "uploaded reference is named `uri`" decision applies to the **upload surface**: the raw upload result and each upload record expose the storage reference as `uri`. Preparation must **not** invent a `uri` field inside rewritten inputs — that would produce inputs the runtime does not recognize. + +## Error and capability behavior + +Upload is a **hosted Pipelex-product capability**, even though the SDK can be pointed at other base URLs. A deployment that does not support upload must raise a specific, actionable exception — preparation must never silently leave a local path in place and let a later run fail obscurely. + +The contract distinguishes at least these semantic outcomes (exact typed exception classes are settled during implementation): + +- **invalid local source** — missing or unreadable path; +- **rejected asset** — the server refused it (e.g. a `413` past the service-defined size cap — see "Storage policy" — surfaced as a clear rejection, not a raw transport error); +- **unsupported server capability** — the configured deployment has no upload route; +- **authentication / authorization failure** — `401` / `403`; +- **transport failure** — network / server fault. + +All preparation failures are raised **before any run is created**. + +## Storage policy (inherited, Phase 1) + +The SDK ships against **today's route behavior**: a service-defined size cap (hosted default 50 MiB via `MAX_UPLOAD_MIB`, rejected with `413`), auth required, per-user keys, and nothing else — no MIME validation, retention, quotas, dedup, or cleanup. The SDK documents limits as **service-defined** and surfaces server rejections as clear "rejected asset" errors; it does **not** hardcode a client-side cap. Real storage policy (retention, quotas, org scoping, cleanup) is a later hosted-owner deliverable. + +## Stability across the future endpoint move + +The public abstraction sits deliberately **above** the HTTP route. Callers depend on `upload_file` / `prepare_inputs`, the `uri` result field, and the `pipelex-storage://` scheme — never on which backend service owns the route. The current transport is `POST /v1/upload` on `pipelex-api`; when hosted storage upload later moves to `pipelex-platform` (together with its paired resolution route, as one storage domain), the public path and wire shape are kept compatible so released SDK versions keep working, and any wire-protocol change is absorbed inside the SDK's upload transport. `upload_file`, `prepare_inputs`, and the prepared run-input shape stay stable across that move. From fe766d63a286b5a29a2ab5e7f723dd43e82dabb4 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:22:52 +0200 Subject: [PATCH 5/9] feat: upload_file + prepare_inputs + build_inputs route (parity with @pipelex/sdk) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the hosted upload-capability surface, the Python counterpart of @pipelex/sdk's uploadFile/prepareInputs, plus the build_inputs route the signature step needs: - client.upload_file(source, *, filename=None, content_type=None) -> UploadRecord — accepts str/Path paths and raw bytes; record (uri, content_type, size, filename) assembled client-side (mimetypes for MIME). - client.prepare_inputs(*, files, pipe_ref=None, inputs) -> PreparedInputs — resolves the pipe's declared signature via build_inputs(explicit=True), template-guided walk (file signal = a canonical {url} content dict, mirroring the runtime's input_normalizer), uploads file-bearing values, returns a copy-on-write rewrite carrying pipelex-storage:// in `url` plus one UploadRecord per prepared asset. http(s)/pipelex-storage:// pass through; data URLs and local/byte sources upload; dedup by source identity; all failures before any run. An unrecognized value at a file position fails as a typed InputPreparationError. - client.build_inputs(BuildInputsRequest) -> BuildInputsResponse — closes the /v1/build/* parity gap the Python SDK had. The response is an is_valid discriminated union parsed through BuildInputsResponseAdapter (a TypeAdapter), mirroring the repo's PipelexValidationResultAdapter precedent, so a malformed 200 raises a clean ValidationError rather than being read as a valid verdict. - Typed exception family mirrors the JS SDK: InputPreparationError base + InvalidLocalSource / RejectedAsset / UnsupportedUploadCapability / UploadAuthentication / UploadTransport. Scope: prepare_inputs takes the closure as inline files; catalog method_id and opt-in http(s) ingest deferred (additive). Matrix-derived parity tests + real-client wiring tests. make agent-check + make agent-test green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Y9RPphjd62DywHtok2fX1q --- CHANGELOG.md | 2 + docs/input-preparation.md | 6 +- pipelex_sdk/build_models.py | 76 +++++++++++ pipelex_sdk/client.py | 48 +++++++ pipelex_sdk/errors.py | 52 +++++++ pipelex_sdk/prepare_inputs.py | 194 ++++++++++++++++++++++++++ pipelex_sdk/upload.py | 127 +++++++++++++++++ pyproject.toml | 1 + tests/unit/test_build_inputs.py | 83 +++++++++++ tests/unit/test_prepare_inputs.py | 220 ++++++++++++++++++++++++++++++ tests/unit/test_upload.py | 141 +++++++++++++++++++ 11 files changed, 948 insertions(+), 2 deletions(-) create mode 100644 pipelex_sdk/build_models.py create mode 100644 pipelex_sdk/prepare_inputs.py create mode 100644 pipelex_sdk/upload.py create mode 100644 tests/unit/test_build_inputs.py create mode 100644 tests/unit/test_prepare_inputs.py create mode 100644 tests/unit/test_upload.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c0acb6b..741b3fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,8 @@ ### Added +- **Input preparation: `upload_file` and `prepare_inputs` (hosted upload capability).** The Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`, in parity. `client.upload_file(source, *, filename=None, content_type=None)` uploads one local asset — a filesystem path (`str`/`Path`) or raw `bytes` — and returns an `UploadRecord` (`uri`, `content_type`, `size`, `filename`) assembled client-side. `client.prepare_inputs(*, files, pipe_ref=None, inputs)` resolves the target pipe's declared signature from the explicit inputs template, interprets the caller's compact `inputs` top-down against it (the file signal is the canonical Image/Document content shape — a `{"url": …}` dict — mirroring the runtime's `input_normalizer`), uploads the file-bearing values, and returns `PreparedInputs`: a copy-on-write rewrite of `inputs` with each asset reference replaced by canonical content carrying `pipelex-storage://` in `url`, plus one `UploadRecord` per prepared asset. HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged; data URLs and local/byte sources are uploaded; the same source referenced twice uploads once (dedup by source identity); all failures are raised before any run is created. Failures are typed per category: `InvalidLocalSourceError`, `RejectedAssetError`, `UnsupportedUploadCapabilityError`, `UploadAuthenticationError`, `UploadTransportError` (all extend `InputPreparationError`). `prepare_inputs` takes the method closure as inline `files`; catalog `method_id` resolution and opt-in `http(s)` ingest are deferred and additive. See [`docs/input-preparation.md`](./docs/input-preparation.md). +- **`build_inputs` route (`POST /v1/build/inputs`).** Closes the `/v1/build/*` parity gap the Python SDK had — `client.build_inputs(BuildInputsRequest)` projects a pipe's declared inputs as a fill-in template, returning a 200 verdict discriminated on `is_valid` (`BuildInputsValidReport` | `CrateInvalidReport`); a no-verdict condition throws `ApiResponseError`. It is the signature source `prepare_inputs` reads (with `explicit=True`). Models live in `pipelex_sdk/build_models.py`. - **Typed run usage: `RunResults.tokens_usages` + `RunResults.usage_assembly_error`.** The per-call usage records a run produces — token counts by category, the server-computed `cost` in USD, model name and id, the pipe that made the call, job-kind fields and timing, for LLM and img-gen/extract/search calls alike — are now first-class typed fields instead of riding `model_extra`. Records validate into a new `TokensUsageRecord` model (`pipelex_sdk/runs.py`) mirroring the wire contract specified in the MTHDS protocol spec. Both paths populate the pair: the hosted durable path reads it off `GET /v1/runs/{id}/results` (which unpacks the runner's `tokens_usages.json` artifact), and the blocking fallback lifts the same pair out of the execute response's extension-open `pipe_output` — so `result.tokens_usages` reads the same regardless of which path ran. Note that the rate table (`unit_costs`) no longer crosses the wire: a record now carries the computed `cost` for the call instead, which is `None` when the model has no rate table at all (own-GPU, mock, dry run) and `0` when a rate table priced it at zero. There is no run-level aggregate — sum the records. diff --git a/docs/input-preparation.md b/docs/input-preparation.md index 7664a5b..c478724 100644 --- a/docs/input-preparation.md +++ b/docs/input-preparation.md @@ -1,6 +1,8 @@ # Input preparation (`upload_file` / `prepare_inputs`) -> **Status: planned surface — not yet implemented.** This document records the approved cross-repository contract (design source: `wip/upload/README.md` in the workspace, tracked in `TODOS.md`) so the SDK's own docs carry the part it owns before the code lands. The raw `upload()` primitive described in [architecture.md](./architecture.md) already exists; `upload_file` and `prepare_inputs` are the higher-level surface built on top of it, and are the Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`. +> **Status: implemented** (`pipelex_sdk/upload.py`, `pipelex_sdk/prepare_inputs.py`, `pipelex_sdk/build_models.py`). This document records the contract (design source: `wip/upload/README.md` in the workspace, tracked in `TODOS.md`). `upload_file` and `prepare_inputs` are the Python counterpart of `@pipelex/sdk`'s `uploadFile` / `prepareInputs`, built on the raw `upload()` wire call. This work also added the `build_inputs` route (the signature source), which the Python SDK previously lacked. +> +> **Current scope.** `prepare_inputs` takes the method closure as inline `files` (the signature source). Two pieces are deliberately deferred and additive (they do not change this contract): resolving a closure from a catalog `method_id`, and the opt-in ingest of `http(s)` URLs into storage — for now an `http(s)` URL at a file position always passes through unchanged. Kept in parity with `@pipelex/sdk`. ## Why this exists @@ -12,7 +14,7 @@ Preparation is **explicit and separate from running.** `execute` / `start` never This is the Python side of one cross-language contract. The behavior matrix, pass-through rules, `Dynamic` handling, dedup, and failure categories are identical to the JS SDK; only the accepted source types differ per language. The two SDKs must agree semantically. See the JS counterpart's `docs/input-preparation.md` for the mirror. -The Python SDK currently has **no `/v1/build/*` coverage** — `prepare_inputs` adds the `build_inputs` counterpart it needs to resolve the declared signature (the JS SDK already exposes `buildInputs`). +The Python SDK previously had **no `/v1/build/*` coverage** — this change added the `build_inputs` counterpart `prepare_inputs` needs to resolve the declared signature (the JS SDK already exposes `buildInputs`). ## The two operations diff --git a/pipelex_sdk/build_models.py b/pipelex_sdk/build_models.py new file mode 100644 index 0000000..047160b --- /dev/null +++ b/pipelex_sdk/build_models.py @@ -0,0 +1,76 @@ +"""Wire models for the `/v1/build/inputs` route — the signature source `prepare_inputs` +reads to resolve a pipe's declared inputs. + +The Python SDK had no `/v1/build/*` coverage; `prepare_inputs` needs the explicit +inputs template, so this adds the `build_inputs` counterpart of `pipelex-sdk-js`'s +`buildInputs` (only this route — the other build projections are not needed here). +A produced verdict is a `200` discriminated on `is_valid`; a no-verdict condition +(unknown `pipe_ref`, auth, server fault) throws `ApiResponseError`. +""" + +from __future__ import annotations + +from typing import Annotated, Any, Literal, TypeAlias + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter + +from pipelex_sdk.validation_models import ValidationErrorItem + +InputsTemplateFormat = Literal["json", "toml"] + + +class MthdsFileItem(BaseModel): + """One MTHDS file in a build closure. `source` is an optional provenance label the + server threads onto diagnostics raised from this file. + """ + + content: str + source: str | None = None + + +class BuildInputsRequest(BaseModel): + """Request for `POST /v1/build/inputs`. The closure is supplied as inline `files`.""" + + files: list[MthdsFileItem] + pipe_ref: str | None = None + format: InputsTemplateFormat = "json" + explicit: bool = False + + +class BuildInputsValidReport(BaseModel): + """The `is_valid: true` arm. The template rides `inputs` (json) or `inputs_toml` (toml).""" + + model_config = ConfigDict(extra="allow") + + is_valid: Literal[True] + pipe_ref: str + requested_pipe_ref: str | None = None + message: str + format: InputsTemplateFormat + explicit: bool + inputs: dict[str, Any] | None = None + inputs_toml: str | None = None + + +class CrateInvalidReport(BaseModel): + """The `is_valid: false` arm shared by the build routes — an unresolvable closure is a + produced verdict on a `200`, never a thrown error. Branch on `is_valid`, not transport. + """ + + model_config = ConfigDict(extra="allow") + + is_valid: Literal[False] + validation_errors: list[ValidationErrorItem] + message: str + + +BuildInputsResponse: TypeAlias = Annotated[ + BuildInputsValidReport | CrateInvalidReport, + Field(discriminator="is_valid"), +] + +# The single parse path for a 200 `/build/inputs` body — discriminated on `is_valid`, built once at +# import (TypeAdapter construction is expensive), mirroring `PipelexValidationResultAdapter`. A +# malformed 200 (or an empty body) raises a clean `pydantic.ValidationError` rather than being +# mistaken for a valid verdict. +BuildInputsResponseAdapter: TypeAdapter[BuildInputsResponse] = TypeAdapter(BuildInputsResponse) # pylint: disable=invalid-name diff --git a/pipelex_sdk/client.py b/pipelex_sdk/client.py index e62a59e..114b48b 100644 --- a/pipelex_sdk/client.py +++ b/pipelex_sdk/client.py @@ -32,6 +32,7 @@ from pydantic_core import to_json from typing_extensions import override +from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, BuildInputsResponseAdapter, MthdsFileItem from pipelex_sdk.errors import ( ApiResponseError, ApiUnreachableError, @@ -42,6 +43,8 @@ RunTimeoutError, ) from pipelex_sdk.execute_result import PipelexExecuteResult +from pipelex_sdk.prepare_inputs import PreparedInputs +from pipelex_sdk.prepare_inputs import prepare_inputs as _prepare_inputs_impl from pipelex_sdk.product_models import ( BillingPortalResponse, ChangePlanResponse, @@ -71,6 +74,8 @@ RunStatus, WaitForResultOptions, ) +from pipelex_sdk.upload import UploadRecord, UploadSource +from pipelex_sdk.upload import upload_file as _upload_file_impl from pipelex_sdk.validation_models import PipelexValidationResultAdapter, ValidationErrorItem if TYPE_CHECKING: @@ -817,6 +822,49 @@ async def upload(self, upload_input: UploadInput) -> UploadedFile: body = upload_input.model_dump(mode="json", exclude_none=True) return UploadedFile.model_validate(await self._request_product("POST", "upload", body=body)) + async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse: + """Project a pipe's declared inputs as a fill-in template — `POST /v1/build/inputs`. + + Returns a 200 verdict: branch on `is_valid` before reading the arm — an unresolvable + closure comes back as `is_valid: false` with `validation_errors`, not a thrown error. + A no-verdict condition (unknown `pipe_ref`, auth, server fault) raises `ApiResponseError`. + This is the signature source `prepare_inputs` reads (with `explicit=True`). + """ + body = request.model_dump(mode="json", exclude_none=True) + raw = await self._request_product("POST", "build/inputs", body=body) + return BuildInputsResponseAdapter.validate_python(raw) + + async def upload_file( + self, + source: UploadSource, + *, + filename: str | None = None, + content_type: str | None = None, + ) -> UploadRecord: + """Upload one local asset and return its `UploadRecord` — the single-asset convenience + over `upload`. `source` is a filesystem path (`str`/`Path`) or raw `bytes`. The record + guarantees `uri`, `content_type`, `size`, and `filename`. Transport failures surface as + the semantic input-preparation errors (rejected asset, auth, unsupported capability, + transport). See `docs/input-preparation.md`. + """ + return await _upload_file_impl(self, source, filename=filename, content_type=content_type) + + async def prepare_inputs( + self, + *, + files: list[MthdsFileItem], + pipe_ref: str | None = None, + inputs: dict[str, Any], + ) -> PreparedInputs: + """Prepare a pipe's inputs — resolve the declared signature, upload the file-bearing + assets, and return copy-on-write rewritten inputs (canonical content carrying + `pipelex-storage://` in `url`) plus one upload record per prepared asset. HTTP(S) URLs + and existing `pipelex-storage://` URIs pass through unchanged; all failures are raised + before any run is created. The caller supplies the method closure as inline `files`. + See `docs/input-preparation.md`. + """ + return await _prepare_inputs_impl(self, files=files, pipe_ref=pipe_ref, inputs=inputs) + async def list_runs(self, method_id: str) -> list[PipelineRun]: """List a method's runs — `GET /v1/runs?method_id={methodId}`.""" result = await self._request_product("GET", f"{_RUNS}?method_id={quote(method_id, safe='')}") diff --git a/pipelex_sdk/errors.py b/pipelex_sdk/errors.py index 997e93a..f7afb63 100644 --- a/pipelex_sdk/errors.py +++ b/pipelex_sdk/errors.py @@ -161,3 +161,55 @@ class RunLifecycleUnavailableError(PipelineRequestError): def __init__(self, message: str, api_url: str) -> None: super().__init__(message) self.api_url = api_url + + +class InputPreparationError(PipelineRequestError): + """Base class for every failure raised by input preparation (`upload_file` / + `prepare_inputs`). + + Catch this to handle any preparation failure; catch a subclass to branch on the + semantic category. All preparation failures are raised BEFORE any run is created — + a run never triggers a hidden upload. Mirrors `pipelex-sdk-js`'s + `InputPreparationError` family. + """ + + +class InvalidLocalSourceError(InputPreparationError): + """A local asset could not be turned into bytes — a missing or unreadable path. + `source` is the offending path. + """ + + def __init__(self, message: str, source: str) -> None: + super().__init__(message) + self.source = source + + +class RejectedAssetError(InputPreparationError): + """The server refused the asset — most commonly a `413` past the service-defined + size cap. The SDK imposes no client-side cap; it surfaces the server's rejection. + `filename` and `status` locate it. + """ + + def __init__(self, message: str, filename: str, status: int) -> None: + super().__init__(message) + self.filename = filename + self.status = status + + +class UnsupportedUploadCapabilityError(InputPreparationError): + """The configured deployment does not support upload (no `/v1/upload` route, seen + as a `404`). Upload is a hosted Pipelex-product capability even though the SDK can + be pointed at other base URLs. + """ + + +class UploadAuthenticationError(InputPreparationError): + """Upload was not authorized — a `401`/`403` from the upload route.""" + + def __init__(self, message: str, status: int) -> None: + super().__init__(message) + self.status = status + + +class UploadTransportError(InputPreparationError): + """A network or server fault reaching the upload route (unreachable host, `5xx`).""" diff --git a/pipelex_sdk/prepare_inputs.py b/pipelex_sdk/prepare_inputs.py new file mode 100644 index 0000000..da07862 --- /dev/null +++ b/pipelex_sdk/prepare_inputs.py @@ -0,0 +1,194 @@ +"""`prepare_inputs` — signature-driven input preparation. Resolves the target pipe's +declared inputs via the explicit inputs template, interprets the caller's compact inputs +top-down against it, uploads the file-bearing values, and returns rewritten inputs +(canonical content carrying `pipelex-storage://` in `url`) plus one upload record per +prepared asset. Python counterpart of `pipelex-sdk-js`'s `prepareInputs`. + +The classification mirrors the runtime: `pipelex`'s `input_normalizer` walks +Image/Document contents (recognized by their `url`-bearing shape, incl. nested in +structured content) and `resolve_uri` decides upload vs pass-through. The declared +signature comes from the explicit template (`build_inputs`, `explicit=True`), whose +canonical content shape is the classifier — the file signal is a value that is a dict +containing a `url` key. See the shared behavior matrix (`wip/upload/behavior-matrix.md`) +and `docs/input-preparation.md`. +""" + +from __future__ import annotations + +import base64 +import re +from pathlib import Path +from typing import TYPE_CHECKING, Any, Protocol, cast +from urllib.parse import unquote + +from pydantic import BaseModel + +from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, CrateInvalidReport, MthdsFileItem +from pipelex_sdk.errors import InputPreparationError +from pipelex_sdk.upload import UploadRecord, UploadSource, upload_file + +if TYPE_CHECKING: + from pipelex_sdk.product_models import UploadedFile, UploadInput + +PIPELEX_STORAGE_SCHEME = "pipelex-storage://" +_HTTP_URL_RE = re.compile(r"^https?://", re.IGNORECASE) + + +class PreparedInputs(BaseModel): + """The result of `prepare_inputs`: rewritten inputs (copy-on-write) plus upload records. + + `inputs` is a copy of the caller's inputs with each file-bearing value rewritten to + canonical content carrying `pipelex-storage://` in `url`. `uploads` carries one record + per uploaded asset — pass-through references (http(s), existing storage URIs) produce none. + """ + + inputs: dict[str, Any] + uploads: list[UploadRecord] + + +class _PrepareClient(Protocol): + """The client surface `prepare_inputs` needs: raw `upload` plus the `build_inputs` signature source.""" + + async def upload(self, upload_input: UploadInput) -> UploadedFile: ... + + async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse: ... + + +class _PrepareContext: + """Mutable state threaded through one preparation walk.""" + + def __init__(self, client: _PrepareClient) -> None: + self.client = client + self.uploads: list[UploadRecord] = [] + # Dedup by source identity: same source (str/bytes/Path value) uploads once. + self.dedup: dict[UploadSource, str] = {} + + +def _is_file_content(node: Any) -> bool: + """A canonical Image/Document content is a dict carrying a `url` key.""" + return isinstance(node, dict) and "url" in node + + +def _decode_data_url(data_url: str) -> tuple[bytes, str]: + """Decode a `data:` URL into bytes plus its MIME type.""" + comma = data_url.find(",") + if comma < 0: + msg = f"Malformed data URL (no comma separator): {data_url[:32]}…" + raise InputPreparationError(msg) + header = data_url[5:comma] # strip "data:" + payload = data_url[comma + 1 :] + content_type = header.split(";")[0] or "application/octet-stream" + if ";base64" in header.lower(): + return base64.b64decode(payload), content_type + return unquote(payload).encode("utf-8"), content_type + + +async def _do_resolve_source(ctx: _PrepareContext, source: Any) -> str: + """Resolve one source to the URL/URI to write.""" + if isinstance(source, str): + if source.startswith(PIPELEX_STORAGE_SCHEME): + return source # already prepared + if _HTTP_URL_RE.match(source): + return source # reachable URL — pass through + if source.startswith("data:"): + data, content_type = _decode_data_url(source) + record = await upload_file(ctx.client, data, content_type=content_type) + ctx.uploads.append(record) + return record.uri + # Anything else is a local filesystem path. + record = await upload_file(ctx.client, source) + ctx.uploads.append(record) + return record.uri + if isinstance(source, (bytes, Path)): + record = await upload_file(ctx.client, source) + ctx.uploads.append(record) + return record.uri + # An unrecognized value sits at a file-bearing position (neither a source string, + # bytes/Path, nor a canonical {url} content dict). Fail with a typed error rather than + # passing an unusable value through to a later run. + msg = ( + "Unsupported value at a file input: expected a path (str/Path), bytes, a data URL, " + f"an http(s)/pipelex-storage:// URL, or canonical {{url}} content; got {type(source).__name__}." + ) + raise InputPreparationError(msg) + + +async def _resolve_source(ctx: _PrepareContext, source: Any) -> str: + """Resolve a source, deduped by identity (same source uploads once).""" + hashable = isinstance(source, (str, bytes, Path)) + if hashable and source in ctx.dedup: + return ctx.dedup[source] + resolved = await _do_resolve_source(ctx, source) + if hashable: + ctx.dedup[source] = resolved + return resolved + + +async def _resolve_file_position(ctx: _PrepareContext, caller_value: Any) -> Any: + """Resolve a value known to sit at a file position into canonical content with a rewritten `url`.""" + if isinstance(caller_value, dict) and "url" in caller_value: + content = cast("dict[str, Any]", caller_value) + resolved = await _resolve_source(ctx, content["url"]) + return {**content, "url": resolved} + resolved = await _resolve_source(ctx, caller_value) + return {"url": resolved} + + +async def _resolve_node(ctx: _PrepareContext, template_node: Any, caller_value: Any) -> Any: + """Template-guided walk: a template node that is canonical file content marks a file position.""" + if _is_file_content(template_node): + return await _resolve_file_position(ctx, caller_value) + if isinstance(template_node, list) and template_node: + element_template = cast("list[Any]", template_node)[0] + if isinstance(caller_value, list): + items = cast("list[Any]", caller_value) + return [await _resolve_node(ctx, element_template, item) for item in items] + return caller_value # shape mismatch — leave it for the run to reject + if isinstance(template_node, dict) and isinstance(caller_value, dict): + template_dict = cast("dict[str, Any]", template_node) + caller_dict = cast("dict[str, Any]", caller_value) + result: dict[str, Any] = dict(caller_dict) + for key in template_dict: + if key in caller_dict: + result[key] = await _resolve_node(ctx, template_dict[key], caller_dict[key]) + return result + return caller_value # scalar (text/number/…) or shape mismatch — pass through + + +async def prepare_inputs( + client: _PrepareClient, + *, + files: list[MthdsFileItem], + pipe_ref: str | None = None, + inputs: dict[str, Any], +) -> PreparedInputs: + """Prepare a pipe's inputs: upload local/byte/data-URL assets at the signature's + file-bearing positions and return copy-on-write rewritten inputs plus upload records. + + HTTP(S) URLs and existing `pipelex-storage://` URIs pass through unchanged. All failures + are raised before any run is created. The declared signature is resolved from the inline + `files` closure; a closure that does not resolve raises `InputPreparationError`. No-verdict + conditions from the signature route (unknown `pipe_ref`, auth, server fault) surface as the + build route's `ApiResponseError`. + """ + report = await client.build_inputs(BuildInputsRequest(files=files, pipe_ref=pipe_ref, format="json", explicit=True)) + if isinstance(report, CrateInvalidReport): + first = report.validation_errors[0].message if report.validation_errors else report.message + msg = f"Cannot prepare inputs: the method signature did not resolve — {first}" + raise InputPreparationError(msg) + if report.format != "json" or report.inputs is None: + msg = f'Cannot prepare inputs: expected a JSON inputs template, got "{report.format}".' + raise InputPreparationError(msg) + template = report.inputs + + ctx = _PrepareContext(client) + rewritten = dict(inputs) + for name, caller_value in inputs.items(): + entry = template.get(name) + if not isinstance(entry, dict) or "content" not in entry: + # Not a declared input (or an unexpected envelope) — pass through untouched. + continue + content = cast("dict[str, Any]", entry)["content"] + rewritten[name] = await _resolve_node(ctx, content, caller_value) + + return PreparedInputs(inputs=rewritten, uploads=ctx.uploads) diff --git a/pipelex_sdk/upload.py b/pipelex_sdk/upload.py new file mode 100644 index 0000000..fcc4776 --- /dev/null +++ b/pipelex_sdk/upload.py @@ -0,0 +1,127 @@ +"""`upload_file` — the single-asset upload convenience over the raw `upload()` wire +call. Accepts `str`/`pathlib.Path` filesystem paths and raw `bytes`, and returns an +`UploadRecord` assembled client-side. Python counterpart of `pipelex-sdk-js`'s +`uploadFile`. See `docs/input-preparation.md`. + +The MIME type and size are known client-side at upload time, so the record is built +without extending the `/v1/upload` response. +""" + +from __future__ import annotations + +import base64 +import mimetypes +from pathlib import Path +from typing import Protocol + +from pydantic import BaseModel + +from pipelex_sdk.errors import ( + ApiResponseError, + ApiUnreachableError, + InputPreparationError, + InvalidLocalSourceError, + RejectedAssetError, + UnsupportedUploadCapabilityError, + UploadAuthenticationError, + UploadTransportError, +) +from pipelex_sdk.product_models import UploadedFile, UploadInput + +DEFAULT_CONTENT_TYPE = "application/octet-stream" +DEFAULT_FILENAME = "upload.bin" + +# A local asset `upload_file` accepts. A bare string in `upload_file` is a filesystem +# path; `prepare_inputs` classifies strings (url vs path) before they reach here. +UploadSource = str | Path | bytes + + +class UploadRecord(BaseModel): + """The record `upload_file` returns for a prepared asset. Beyond the source identity it + guarantees the resulting `uri`, the MIME `content_type`, the `size` in bytes, and the + `filename`. A content checksum is deliberately not included — best-effort at most, and + within-preparation dedup keys on source identity. + """ + + uri: str + filename: str + content_type: str + size: int + + +class _UploadClient(Protocol): + """The client surface `upload_file` needs — the raw base64 `upload` wire call.""" + + async def upload(self, upload_input: UploadInput) -> UploadedFile: ... + + +def _guess_content_type(filename: str) -> str: + """MIME guess from a filename extension; `application/octet-stream` when unknown.""" + guessed, _ = mimetypes.guess_type(filename) + return guessed or DEFAULT_CONTENT_TYPE + + +def _read_path(path: Path) -> bytes: + """Read a filesystem path into bytes, mapping read failures to `InvalidLocalSourceError`.""" + try: + return path.read_bytes() + except OSError as exc: + msg = f'Local file cannot be read: "{path}" ({type(exc).__name__}).' + raise InvalidLocalSourceError(msg, source=str(path)) from exc + + +def _to_asset_bytes(source: UploadSource, filename: str | None, content_type: str | None) -> tuple[bytes, str, str]: + """Normalize any accepted asset form into bytes plus a filename and MIME.""" + if isinstance(source, bytes): + resolved_name = filename or DEFAULT_FILENAME + return source, resolved_name, content_type or _guess_content_type(resolved_name) + path = Path(source) + data = _read_path(path) + resolved_name = filename or path.name or DEFAULT_FILENAME + return data, resolved_name, content_type or _guess_content_type(resolved_name) + + +def _map_upload_error(error: ApiResponseError | ApiUnreachableError, filename: str) -> InputPreparationError: + """Translate a raw `upload()` transport error into the matching preparation error.""" + if isinstance(error, ApiUnreachableError): + msg = f'Upload of "{filename}" could not reach the Pipelex API ({error.code or "unreachable"}).' + return UploadTransportError(msg) + match error.status: + case 413: + detail = error.server_message or "asset exceeds the service size limit" + return RejectedAssetError(f'The server rejected "{filename}": {detail}.', filename=filename, status=error.status) + case 401 | 403: + return UploadAuthenticationError( + f'Upload of "{filename}" was not authorized ({error.status}). Check the configured Pipelex API key.', + status=error.status, + ) + case 404: + return UnsupportedUploadCapabilityError( + "The configured Pipelex deployment does not support file upload (no /v1/upload route). Upload is a hosted Pipelex capability." + ) + case _: + detail = error.server_message or error.status_text + return UploadTransportError(f'Upload of "{filename}" failed ({error.status}): {detail}.') + + +async def upload_file( + client: _UploadClient, + source: UploadSource, + *, + filename: str | None = None, + content_type: str | None = None, +) -> UploadRecord: + """Upload one local asset and return its `UploadRecord`. + + `source` is a filesystem path (`str`/`Path`) or raw `bytes`. Maps the raw `upload()` + transport errors onto the semantic input-preparation errors: a `413` is a rejected + asset, `401`/`403` an auth failure, `404` an unsupported upload capability, an + unreachable host a transport failure. + """ + data, resolved_name, resolved_type = _to_asset_bytes(source, filename, content_type) + encoded = base64.b64encode(data).decode("ascii") + try: + uploaded = await client.upload(UploadInput(filename=resolved_name, data=encoded, content_type=resolved_type)) + except (ApiResponseError, ApiUnreachableError) as exc: + raise _map_upload_error(exc, resolved_name) from exc + return UploadRecord(uri=uploaded.uri, filename=uploaded.filename, content_type=resolved_type, size=len(data)) diff --git a/pyproject.toml b/pyproject.toml index 3157342..92e83a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -318,6 +318,7 @@ convention = "google" "INP001", # Allow test files to not have __init__.py in their directories (avoids namespace collisions) "SLF001", # Unit tests legitimately probe private transport/error helpers (e.g. _request_product, _request_json) "PLC2701", # Unit tests legitimately import private module helpers under test (e.g. _parse_error_body) + "ARG002", # Test-double methods match a Protocol signature; an unused param (e.g. a fake build_inputs ignoring `request`) is intentional ] "examples/**/*.py" = [ "INP001", # Runnable demo scripts, not an importable package diff --git a/tests/unit/test_build_inputs.py b/tests/unit/test_build_inputs.py new file mode 100644 index 0000000..3f9adb7 --- /dev/null +++ b/tests/unit/test_build_inputs.py @@ -0,0 +1,83 @@ +"""The `build_inputs` route — the signature source `prepare_inputs` reads. Pins the verb + +path + body, the 200-verdict discipline (branch on `is_valid`), and the no-verdict throw. + +Ports the relevant slice of `pipelex-sdk-js/tests/build-routes.test.ts` for `/v1/build/inputs`. +`_send` is mocked; a produced verdict is a 200 discriminated on `is_valid`, a no-verdict +condition throws `ApiResponseError`. +""" + +import asyncio +import json + +import httpx +import pytest +from pytest_mock import MockerFixture, MockType + +from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.errors import ApiResponseError + +_BASE_URL = "http://localhost:8081" + + +def _response(status_code: int, *, json_body: object | None = None) -> httpx.Response: + request = httpx.Request("POST", f"{_BASE_URL}/x") + if json_body is not None: + return httpx.Response(status_code, json=json_body, request=request) + return httpx.Response(status_code, request=request) + + +class TestBuildInputs: + def _client(self) -> PipelexAPIClient: + return PipelexAPIClient(api_key="test-token", base_url=_BASE_URL) + + def _mock_send(self, mocker: MockerFixture, client: PipelexAPIClient, response: httpx.Response) -> MockType: + return mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=response)) + + def test_posts_files_and_flags_to_build_inputs(self, mocker: MockerFixture) -> None: + client = self._client() + valid = { + "is_valid": True, + "pipe_ref": "demo.main", + "message": "ok", + "format": "json", + "explicit": True, + "inputs": {"photo": {"concept": "demo.Photo", "content": {"url": "https://mock/p.png"}}}, + } + send = self._mock_send(mocker, client, _response(200, json_body=valid)) + + report = asyncio.run( + client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content='domain = "demo"', source="b.mthds")], format="json", explicit=True)) + ) + + call = send.call_args + assert call.args[0] == "POST" + assert call.args[1] == f"{_BASE_URL}/v1/build/inputs" + body = json.loads(call.kwargs["content"]) + assert body == {"files": [{"content": 'domain = "demo"', "source": "b.mthds"}], "format": "json", "explicit": True} + assert isinstance(report, BuildInputsValidReport) + assert report.pipe_ref == "demo.main" + assert report.inputs is not None + + def test_invalid_closure_is_a_200_verdict(self, mocker: MockerFixture) -> None: + client = self._client() + invalid = { + "is_valid": False, + "message": "closure did not validate", + "validation_errors": [{"category": "blueprint_validation", "message": "unknown pipe type"}], + } + self._mock_send(mocker, client, _response(200, json_body=invalid)) + + report = asyncio.run(client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content="x")]))) + + assert isinstance(report, CrateInvalidReport) + assert report.validation_errors[0].message == "unknown pipe type" + + def test_no_verdict_422_raises_api_response_error(self, mocker: MockerFixture) -> None: + client = self._client() + problem = {"detail": "Unknown pipe_ref", "error_type": "PipeNotFound"} + self._mock_send(mocker, client, _response(422, json_body=problem)) + + with pytest.raises(ApiResponseError) as exc_info: + asyncio.run(client.build_inputs(BuildInputsRequest(files=[MthdsFileItem(content="x")], pipe_ref="demo.nope"))) + assert exc_info.value.status == 422 diff --git a/tests/unit/test_prepare_inputs.py b/tests/unit/test_prepare_inputs.py new file mode 100644 index 0000000..713fd94 --- /dev/null +++ b/tests/unit/test_prepare_inputs.py @@ -0,0 +1,220 @@ +"""`prepare_inputs` — signature-driven input preparation. Cases derive from the shared +behavior matrix (`wip/upload/behavior-matrix.md`): file-bearing positions are found from the +explicit template's canonical content shape (a `{"url": …}` dict), assets are uploaded and +rewritten to `pipelex-storage://` in `url`, http(s)/storage references pass through, dedup +keys on source identity, and the call is copy-on-write. + +Ports `pipelex-sdk-js/tests/prepare-inputs.test.ts`. The fake client returns a canned explicit +template from `build_inputs` and a counting `upload`; one wiring test drives the real client. +""" + +import asyncio +from pathlib import Path +from typing import Any + +import httpx +import pytest +from pytest_mock import MockerFixture + +from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponse, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.errors import ApiResponseError, InputPreparationError, RejectedAssetError +from pipelex_sdk.prepare_inputs import prepare_inputs +from pipelex_sdk.product_models import UploadedFile, UploadInput + +_BASE_URL = "http://localhost:8081" +_FILES = [MthdsFileItem(content='domain = "demo"')] + + +def _entry(concept: str, content: Any) -> dict[str, Any]: + return {"concept": concept, "content": content} + + +class _FakePrepareClient: + """Fake client: `build_inputs` returns the given envelope template; `upload` counts calls.""" + + def __init__(self, template: dict[str, Any], *, report: BuildInputsResponse | None = None, upload_error: Exception | None = None) -> None: + self._template = template + self._report = report + self._upload_error = upload_error + self.upload_calls: list[UploadInput] = [] + self._counter = 0 + + async def build_inputs(self, request: BuildInputsRequest) -> BuildInputsResponse: + if self._report is not None: + return self._report + return BuildInputsValidReport(is_valid=True, pipe_ref="demo.main", message="ok", format="json", explicit=True, inputs=self._template) + + async def upload(self, upload_input: UploadInput) -> UploadedFile: + if self._upload_error is not None: + raise self._upload_error + self._counter += 1 + self.upload_calls.append(upload_input) + return UploadedFile(uri=f"pipelex-storage://user/assets/{self._counter}.bin", filename=upload_input.filename) + + +class TestPrepareInputs: + def test_uploads_top_level_image_bytes(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1, 2, 3])})) + + assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}} + assert len(prepared.uploads) == 1 + assert prepared.uploads[0].uri == "pipelex-storage://user/assets/1.bin" + + def test_passes_http_url_through(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "https://example.com/real.png"})) + + assert prepared.inputs == {"photo": {"url": "https://example.com/real.png"}} + assert prepared.uploads == [] + assert client.upload_calls == [] + + def test_passes_existing_storage_uri_through(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "pipelex-storage://user/assets/already.png"})) + + assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/already.png"}} + assert prepared.uploads == [] + + def test_decodes_and_uploads_data_url(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "data:image/png;base64,AQIDBA=="})) + + assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}} + assert client.upload_calls[0].content_type == "image/png" + assert client.upload_calls[0].data == "AQIDBA==" + + def test_uploads_each_element_of_declared_multiple(self) -> None: + client = _FakePrepareClient({"exhibits": _entry("demo.Exhibit", [{"url": "https://mock/d.pdf"}])}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"exhibits": [bytes([1]), bytes([2])]})) + + assert prepared.inputs == {"exhibits": [{"url": "pipelex-storage://user/assets/1.bin"}, {"url": "pipelex-storage://user/assets/2.bin"}]} + assert len(prepared.uploads) == 2 + + def test_leaves_text_input_untouched(self) -> None: + client = _FakePrepareClient({"question": _entry("demo.Question", {"text": "text_value"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"question": "notes/summary.txt"})) + + assert prepared.inputs == {"question": "notes/summary.txt"} + assert client.upload_calls == [] + + def test_uploads_only_nested_image_of_structured_input(self) -> None: + client = _FakePrepareClient( + {"dossier": _entry("demo.Dossier", {"title": "title_value", "cover": {"url": "https://mock/c.png", "mime_type": "image/png"}})} + ) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"dossier": {"title": "Q3 report", "cover": bytes([7, 7])}})) + + assert prepared.inputs == {"dossier": {"title": "Q3 report", "cover": {"url": "pipelex-storage://user/assets/1.bin"}}} + assert len(prepared.uploads) == 1 + + def test_does_not_path_interpret_bare_string_at_dynamic_input(self) -> None: + client = _FakePrepareClient({"freeform": _entry("native.Anything", {"whatever": "value"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"freeform": "resembles/a/path"})) + + assert prepared.inputs == {"freeform": "resembles/a/path"} + assert client.upload_calls == [] + + def test_uploads_canonical_image_nested_in_dynamic(self) -> None: + client = _FakePrepareClient({"data": _entry("native.Composite", {"text": "t", "images": [{"url": "https://mock/i.png"}]})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"data": {"text": "hi", "images": [bytes([5])]}})) + + assert prepared.inputs == {"data": {"text": "hi", "images": [{"url": "pipelex-storage://user/assets/1.bin"}]}} + assert len(prepared.uploads) == 1 + + def test_dedups_by_source_identity(self) -> None: + client = _FakePrepareClient({"exhibits": _entry("demo.Exhibit", [{"url": "https://mock/d.pdf"}])}) + shared = bytes([9, 9, 9]) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"exhibits": [shared, shared]})) + + assert len(client.upload_calls) == 1 + exhibits = prepared.inputs["exhibits"] + assert exhibits[0]["url"] == exhibits[1]["url"] + + def test_is_copy_on_write(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + original = {"photo": bytes([1, 2, 3])} + + asyncio.run(prepare_inputs(client, files=_FILES, inputs=original)) + + assert original["photo"] == bytes([1, 2, 3]) + + def test_passes_through_undeclared_input(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "https://example.com/p.png", "stray": "left alone"})) + + assert prepared.inputs["stray"] == "left alone" + + def test_uploads_real_local_path(self, tmp_path: Path) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + path = tmp_path / "shot.png" + path.write_bytes(bytes([1, 2, 3, 4])) + + prepared = asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": str(path)})) + + assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}} + assert client.upload_calls[0].content_type == "image/png" + + def test_raises_for_unrecognized_value_at_file_position(self) -> None: + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + # A plain object that is neither a canonical {url} content nor bytes — a realistic + # caller typo — must surface as a typed error, not pass through unresolved. + with pytest.raises(InputPreparationError): + asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": {"mimeType": "image/png", "bytes": [1, 2, 3]}})) + assert client.upload_calls == [] + + def test_raises_when_signature_does_not_resolve(self) -> None: + report = CrateInvalidReport(is_valid=False, message="closure did not validate", validation_errors=[]) + client = _FakePrepareClient({}, report=report) + + with pytest.raises(InputPreparationError): + asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1])})) + + def test_surfaces_rejected_asset_before_returning(self) -> None: + error = ApiResponseError( + "HTTP 413", api_url=f"{_BASE_URL}/v1/upload", status=413, status_text="Payload Too Large", response_body="", server_message="too big" + ) + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}, upload_error=error) + + with pytest.raises(RejectedAssetError): + asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": bytes([1])})) + + def test_wires_through_the_real_client(self, mocker: MockerFixture) -> None: + client = PipelexAPIClient(api_key="test-token", base_url=_BASE_URL) + build_body = { + "is_valid": True, + "pipe_ref": "demo.main", + "message": "ok", + "format": "json", + "explicit": True, + "inputs": {"photo": {"concept": "demo.Photo", "content": {"url": "https://mock/p.png"}}}, + } + upload_body = {"uri": "pipelex-storage://user/assets/1.bin", "filename": "upload.bin"} + request = httpx.Request("POST", f"{_BASE_URL}/x") + mocker.patch.object( + client, + "_send", + mocker.AsyncMock( + side_effect=[ + httpx.Response(200, json=build_body, request=request), + httpx.Response(200, json=upload_body, request=request), + ] + ), + ) + + prepared = asyncio.run(client.prepare_inputs(files=_FILES, inputs={"photo": bytes([1, 2, 3])})) + + assert prepared.inputs == {"photo": {"url": "pipelex-storage://user/assets/1.bin"}} + assert len(prepared.uploads) == 1 diff --git a/tests/unit/test_upload.py b/tests/unit/test_upload.py new file mode 100644 index 0000000..4025216 --- /dev/null +++ b/tests/unit/test_upload.py @@ -0,0 +1,141 @@ +"""`upload_file` — the single-asset upload convenience over the raw `upload()` wire call. +Pins accepted asset forms (path str/Path, bytes), the client-side record assembly +(uri/filename/content_type/size), base64 correctness, and the mapping of raw transport +errors onto the semantic preparation errors. + +Ports `pipelex-sdk-js/tests/upload.test.ts`. Uses a fake `upload` client double for the +logic, plus one wiring test through the real `PipelexAPIClient` with a mocked `_send`. +""" + +import asyncio +import base64 +import json +from pathlib import Path + +import httpx +import pytest +from pytest_mock import MockerFixture + +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.errors import ( + ApiResponseError, + ApiUnreachableError, + InvalidLocalSourceError, + RejectedAssetError, + UnsupportedUploadCapabilityError, + UploadAuthenticationError, + UploadTransportError, +) +from pipelex_sdk.product_models import UploadedFile, UploadInput +from pipelex_sdk.upload import upload_file + +_BASE_URL = "http://localhost:8081" + + +class _FakeUploadClient: + """A fake `upload` client: records the wire body, returns a canned URI (or raises).""" + + def __init__(self, *, uri: str = "pipelex-storage://user/assets/abc.bin", error: Exception | None = None) -> None: + self.calls: list[UploadInput] = [] + self._uri = uri + self._error = error + + async def upload(self, upload_input: UploadInput) -> UploadedFile: + if self._error is not None: + raise self._error + self.calls.append(upload_input) + return UploadedFile(uri=self._uri, filename=upload_input.filename) + + +def _api_error(status: int, server_message: str = "boom") -> ApiResponseError: + return ApiResponseError( + f"HTTP {status}", api_url=f"{_BASE_URL}/v1/upload", status=status, status_text="Error", response_body="", server_message=server_message + ) + + +class TestUploadFile: + def test_uploads_bytes_with_base64_and_full_record(self) -> None: + client = _FakeUploadClient() + data = bytes([1, 2, 3, 4, 5]) + + record = asyncio.run(upload_file(client, data, filename="blob.png", content_type="image/png")) + + assert len(client.calls) == 1 + assert client.calls[0].data == base64.b64encode(data).decode("ascii") + assert client.calls[0].content_type == "image/png" + assert record.uri == "pipelex-storage://user/assets/abc.bin" + assert record.filename == "blob.png" + assert record.content_type == "image/png" + assert record.size == 5 + + def test_reads_a_local_path_str_deriving_filename_and_mime(self, tmp_path: Path) -> None: + client = _FakeUploadClient() + path = tmp_path / "diagram.png" + path.write_bytes(bytes([10, 20, 30])) + + record = asyncio.run(upload_file(client, str(path))) + + assert client.calls[0].filename == "diagram.png" + assert client.calls[0].content_type == "image/png" + assert record.size == 3 + + def test_accepts_a_pathlib_path(self, tmp_path: Path) -> None: + client = _FakeUploadClient() + path = tmp_path / "report.pdf" + path.write_bytes(bytes([1, 2])) + + record = asyncio.run(upload_file(client, path)) + + assert client.calls[0].filename == "report.pdf" + assert client.calls[0].content_type == "application/pdf" + assert record.size == 2 + + def test_missing_path_raises_invalid_local_source(self, tmp_path: Path) -> None: + client = _FakeUploadClient() + missing = tmp_path / "nope.png" + with pytest.raises(InvalidLocalSourceError): + asyncio.run(upload_file(client, missing)) + + def test_413_maps_to_rejected_asset(self) -> None: + client = _FakeUploadClient(error=_api_error(413, "too big")) + with pytest.raises(RejectedAssetError) as exc_info: + asyncio.run(upload_file(client, bytes([1]), filename="big.pdf")) + assert exc_info.value.filename == "big.pdf" + assert exc_info.value.status == 413 + + @pytest.mark.parametrize("status", [401, 403]) + def test_401_403_map_to_upload_authentication(self, status: int) -> None: + client = _FakeUploadClient(error=_api_error(status)) + with pytest.raises(UploadAuthenticationError): + asyncio.run(upload_file(client, bytes([1]))) + + def test_404_maps_to_unsupported_capability(self) -> None: + client = _FakeUploadClient(error=_api_error(404)) + with pytest.raises(UnsupportedUploadCapabilityError): + asyncio.run(upload_file(client, bytes([1]))) + + def test_other_status_and_unreachable_map_to_transport(self) -> None: + client_500 = _FakeUploadClient(error=_api_error(500)) + with pytest.raises(UploadTransportError): + asyncio.run(upload_file(client_500, bytes([1]))) + + client_down = _FakeUploadClient(error=ApiUnreachableError("down", api_url=_BASE_URL, code="ECONNREFUSED")) + with pytest.raises(UploadTransportError): + asyncio.run(upload_file(client_down, bytes([1]))) + + def test_wires_through_the_real_client(self, mocker: MockerFixture) -> None: + client = PipelexAPIClient(api_key="test-token", base_url=_BASE_URL) + uploaded = {"uri": "pipelex-storage://user/assets/z.bin", "filename": "x.png"} + request = httpx.Request("POST", f"{_BASE_URL}/v1/upload") + send = mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=httpx.Response(200, json=uploaded, request=request))) + + record = asyncio.run(client.upload_file(bytes([1, 2, 3]), filename="x.png", content_type="image/png")) + + call = send.call_args + assert call.args[0] == "POST" + assert call.args[1] == f"{_BASE_URL}/v1/upload" + body = json.loads(call.kwargs["content"]) + assert body["filename"] == "x.png" + assert body["data"] == base64.b64encode(bytes([1, 2, 3])).decode("ascii") + assert record.uri == "pipelex-storage://user/assets/z.bin" + assert record.size == 3 From 1fc24690d2ae109ee832243f698b6696e57496cb Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:08:54 +0200 Subject: [PATCH 6/9] fix(prepare_inputs): harden data-URL decoding + doc accuracy (review feedback) Addresses the bot-review findings on PR #10 that are genuine correctness or doc-accuracy issues; deliberately defers the rest. Fixed: - Malformed base64 data URL no longer escapes the typed-error contract. Decode with `validate=True` (junk chars are rejected, not silently discarded into corrupted bytes) and map `binascii.Error` to `InputPreparationError`, so all preparation failures stay within the documented exception family. (Codex, Greptile, cubic all flagged this.) - Percent-encoded binary data URLs keep their exact bytes: decode with `urllib.parse.unquote_to_bytes` instead of `unquote(...).encode("utf-8")`, which corrupted any byte >= 0x80. - docs/input-preparation.md: the URL / storage-URI pass-through is a `prepare_inputs`-level behavior, not a feature of `upload_file` (which treats every string as a filesystem path); removed the `checksum` row that listed a field `UploadRecord` does not have. - docs/architecture.md: upload_file/prepare_inputs are now implemented, not a "planned addition". Deferred (not bugs): offloading the file read/base64 off the event loop (perf nicety on an inherently I/O-bound op), a template-required-per-format validator on BuildInputsValidReport (prepare_inputs already raises a clean typed error), and splitting a P3 two-scenario transport-error test. Regression tests: malformed base64 (bad padding + non-alphabet) raises a typed error and uploads nothing; percent-encoded binary round-trips its exact bytes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_019cpRcAL7MEQcsx2hHvGxC2 --- docs/architecture.md | 2 +- docs/input-preparation.md | 7 +++---- pipelex_sdk/prepare_inputs.py | 22 ++++++++++++++++++---- tests/unit/test_prepare_inputs.py | 26 ++++++++++++++++++++++++++ 4 files changed, 48 insertions(+), 9 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index c357c95..5f8b652 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -138,7 +138,7 @@ The wire models are snake_case Pydantic v2. Response models are extension-open ( - **Pipelex API keys** — `list_pipelex_api_keys()`; `create_pipelex_api_key(label)` and `rotate_pipelex_api_key(id)` return the plaintext `api_key` **once**; `revoke_pipelex_api_key(id)`. Creation surfaces a **409 `pipelex_api_key_limit_reached`** when the per-account limit is hit. Rotation sends no body. - **Gateway (LLM inference) key** — `create_gateway_api_key(promo_code)` **always sends a JSON body** (even with `promo_code=None` → `{"promo_code": null}`); the server 422s an empty body. `get_gateway_api_key()` → status (`gateway_api_key` is `None` until provisioned). - **Onboarding** — `submit_onboarding(OnboardingSubmission)` (`POST /v1/onboarding/submit`, empty 2xx body); absent optional fields are dropped. -- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle. The higher-level `upload_file` / `prepare_inputs` preparation surface built on top of `upload` is a planned addition — see [input-preparation.md](./input-preparation.md). +- **Storage** — `resolve_storage_url(uri)` → presigned URL; `upload(UploadInput)` → the stored file handle. The higher-level `upload_file` / `prepare_inputs` preparation surface built on top of `upload` is now available — see [input-preparation.md](./input-preparation.md). - **Run records** — `list_runs(method_id)` → `list[PipelineRun]` (the catalog-style list, distinct from the lifecycle status/result routes); `update_run(run_id, UpdateRunInput)` (admin/manual status patch, empty 2xx body). ## Health probe diff --git a/docs/input-preparation.md b/docs/input-preparation.md index c478724..6d65eed 100644 --- a/docs/input-preparation.md +++ b/docs/input-preparation.md @@ -22,8 +22,8 @@ The Python SDK previously had **no `/v1/build/*` coverage** — this change adde Uploads one asset and returns its upload record. It is the language-native convenience over the raw `upload()` wire call (base64 JSON body), assembling the record client-side. -- **Accepted sources:** `str` and `pathlib.Path` filesystem paths, and raw `bytes`. -- A string that is an **HTTP(S) URL** or an existing **`pipelex-storage://` URI** is not a local asset and passes through (see pass-through rules below). +- **Accepted sources:** `str` and `pathlib.Path` filesystem paths, and raw `bytes`. `upload_file` treats every string as a **filesystem path** — a URL handed directly to it is read as a local file and fails. +- The URL / storage-URI **pass-through** (leaving **HTTP(S) URLs** and existing **`pipelex-storage://` URIs** untouched) is a `prepare_inputs`-level behavior (see pass-through rules below), not a feature of `upload_file` itself: `prepare_inputs` classifies each string (URL vs local path) against the declared signature *before* it reaches `upload_file`. - Open file objects and streams are **deferred** — they can be added later without removing anything. The returned **upload record** guarantees, beyond the source identity: @@ -34,9 +34,8 @@ The returned **upload record** guarantees, beyond the source identity: | content type (MIME) | Known client-side at upload time. | | size (bytes) | Known client-side at upload time. | | filename | Already in the wire model. | -| checksum | **Best-effort, not guaranteed.** Within-preparation dedup relies on source identity, not hashing; cross-preparation dedup is a hosted storage-policy concern (Phase 5). | -The MIME type and size are known client-side, so the record is assembled without extending the `/v1/upload` response. +The MIME type and size are known client-side, so the record is assembled without extending the `/v1/upload` response. There is deliberately **no checksum field**: within-preparation dedup keys on source identity (not hashing), and cross-preparation dedup is a hosted storage-policy concern (Phase 5). ### `prepare_inputs` — signature-driven input preparation diff --git a/pipelex_sdk/prepare_inputs.py b/pipelex_sdk/prepare_inputs.py index da07862..7857ad8 100644 --- a/pipelex_sdk/prepare_inputs.py +++ b/pipelex_sdk/prepare_inputs.py @@ -16,10 +16,11 @@ from __future__ import annotations import base64 +import binascii import re from pathlib import Path from typing import TYPE_CHECKING, Any, Protocol, cast -from urllib.parse import unquote +from urllib.parse import unquote_to_bytes from pydantic import BaseModel @@ -70,7 +71,15 @@ def _is_file_content(node: Any) -> bool: def _decode_data_url(data_url: str) -> tuple[bytes, str]: - """Decode a `data:` URL into bytes plus its MIME type.""" + """Decode a `data:` URL into bytes plus its MIME type. + + A base64 payload is decoded with `validate=True` so junk characters are rejected rather + than silently discarded (which would upload corrupted bytes), and a decode failure (bad + padding or non-alphabet input) surfaces as a typed `InputPreparationError` — never a raw + `binascii.Error` escaping the preparation contract. A non-base64 payload decodes straight + to bytes via `unquote_to_bytes`, so percent-encoded binary keeps its exact bytes (decoding + it as UTF-8 text first would corrupt any byte ≥ 0x80). + """ comma = data_url.find(",") if comma < 0: msg = f"Malformed data URL (no comma separator): {data_url[:32]}…" @@ -79,8 +88,13 @@ def _decode_data_url(data_url: str) -> tuple[bytes, str]: payload = data_url[comma + 1 :] content_type = header.split(";")[0] or "application/octet-stream" if ";base64" in header.lower(): - return base64.b64decode(payload), content_type - return unquote(payload).encode("utf-8"), content_type + try: + decoded = base64.b64decode(payload, validate=True) + except binascii.Error as exc: + msg = f"Malformed data URL: the base64 payload is not valid ({exc})." + raise InputPreparationError(msg) from exc + return decoded, content_type + return unquote_to_bytes(payload), content_type async def _do_resolve_source(ctx: _PrepareContext, source: Any) -> str: diff --git a/tests/unit/test_prepare_inputs.py b/tests/unit/test_prepare_inputs.py index 713fd94..18c79a8 100644 --- a/tests/unit/test_prepare_inputs.py +++ b/tests/unit/test_prepare_inputs.py @@ -9,6 +9,7 @@ """ import asyncio +import base64 from pathlib import Path from typing import Any @@ -89,6 +90,31 @@ def test_decodes_and_uploads_data_url(self) -> None: assert client.upload_calls[0].content_type == "image/png" assert client.upload_calls[0].data == "AQIDBA==" + @pytest.mark.parametrize( + "data_url", + [ + "data:image/png;base64,AQI", # bad padding — binascii.Error + "data:image/png;base64,AQID!!!!", # non-alphabet junk — rejected by validate=True + ], + ) + def test_malformed_base64_data_url_raises_typed_error(self, data_url: str) -> None: + # A malformed base64 data URL must surface as the typed `InputPreparationError` + # (never a raw binascii.Error), and must never upload silently-corrupted bytes. + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + with pytest.raises(InputPreparationError): + asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": data_url})) + assert client.upload_calls == [] + + def test_percent_encoded_binary_data_url_keeps_exact_bytes(self) -> None: + # A non-base64 data URL carrying percent-encoded binary must upload its exact bytes; + # decoding as UTF-8 text first would corrupt any byte >= 0x80 (e.g. %FF). + client = _FakePrepareClient({"photo": _entry("demo.Photo", {"url": "https://mock/p.png"})}) + + asyncio.run(prepare_inputs(client, files=_FILES, inputs={"photo": "data:application/octet-stream,%00%ff%01"})) + + assert base64.b64decode(client.upload_calls[0].data) == bytes([0x00, 0xFF, 0x01]) + def test_uploads_each_element_of_declared_multiple(self) -> None: client = _FakePrepareClient({"exhibits": _entry("demo.Exhibit", [{"url": "https://mock/d.pdf"}])}) From e292fc3535f3ef5b4b2e020703c8c62f78673840 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:39:58 +0200 Subject: [PATCH 7/9] fix: offload upload read off the loop + enforce build-inputs template shape MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second pass on PR #10 review-bot feedback — the two remaining substantive items, both promoted from "defer" to "fix now" after independent verification showed each closes a documented JS-parity gap in new code at near-zero cost. - upload_file: offload the synchronous whole-file read via asyncio.to_thread so a large read no longer blocks the event loop before the first await. Matches the JS SDK (which reads off-loop via node:fs/promises) and the repo's async-only invariant. base64 stays inline on purpose — CPython's binascii holds the GIL, so threading it would not free the loop (JS also encodes inline); `_to_asset_bytes`/`_read_path` stay sync. (Greptile P2 + cubic P2.) - BuildInputsValidReport: add a model_validator enforcing the template-present- per-format invariant (inputs for json, inputs_toml for toml, not the other). This makes the adapter's documented malformed-200 guarantee true for the template shape too — an `is_valid: true` body missing its template now raises a clean ValidationError at parse time instead of surfacing one layer down in prepare_inputs with a misleading "got json" message. Mirrors the JS sibling's per-format-required modeling. (cubic P2.) - Tests: assert the read is offloaded (to_thread awaited with _to_asset_bytes, real behavior preserved via wraps); assert the validator rejects the three malformed template shapes and accepts a well-formed toml report. Parametrized the two-scenario transport-error test for per-case attribution (cubic P3). make agent-check + make agent-test green. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_019cpRcAL7MEQcsx2hHvGxC2 --- pipelex_sdk/build_models.py | 27 +++++++++++++++++++++++++-- pipelex_sdk/upload.py | 7 ++++++- tests/unit/test_build_inputs.py | 26 +++++++++++++++++++++++++- tests/unit/test_upload.py | 33 ++++++++++++++++++++++++++------- 4 files changed, 82 insertions(+), 11 deletions(-) diff --git a/pipelex_sdk/build_models.py b/pipelex_sdk/build_models.py index 047160b..a1c4086 100644 --- a/pipelex_sdk/build_models.py +++ b/pipelex_sdk/build_models.py @@ -10,9 +10,9 @@ from __future__ import annotations -from typing import Annotated, Any, Literal, TypeAlias +from typing import Annotated, Any, Literal, Self, TypeAlias -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, model_validator from pipelex_sdk.validation_models import ValidationErrorItem @@ -51,6 +51,29 @@ class BuildInputsValidReport(BaseModel): inputs: dict[str, Any] | None = None inputs_toml: str | None = None + @model_validator(mode="after") + def _template_matches_format(self) -> Self: + # Honor the adapter's malformed-200 guarantee for the template shape too: a valid + # verdict must carry the template field its `format` selects (and not the other). + # Without this, an `is_valid: true` body missing both templates would parse as a valid + # report and only fail one layer down in `prepare_inputs`. + match self.format: + case "json": + if self.inputs is None: + msg = "inputs is required when format is 'json'" + raise ValueError(msg) + if self.inputs_toml is not None: + msg = "inputs_toml must be absent when format is 'json'" + raise ValueError(msg) + case "toml": + if self.inputs_toml is None: + msg = "inputs_toml is required when format is 'toml'" + raise ValueError(msg) + if self.inputs is not None: + msg = "inputs must be absent when format is 'toml'" + raise ValueError(msg) + return self + class CrateInvalidReport(BaseModel): """The `is_valid: false` arm shared by the build routes — an unresolvable closure is a diff --git a/pipelex_sdk/upload.py b/pipelex_sdk/upload.py index fcc4776..dc8edcc 100644 --- a/pipelex_sdk/upload.py +++ b/pipelex_sdk/upload.py @@ -9,6 +9,7 @@ from __future__ import annotations +import asyncio import base64 import mimetypes from pathlib import Path @@ -118,7 +119,11 @@ async def upload_file( asset, `401`/`403` an auth failure, `404` an unsupported upload capability, an unreachable host a transport failure. """ - data, resolved_name, resolved_type = _to_asset_bytes(source, filename, content_type) + # Offload the (possibly large) synchronous file read off the event loop — `read_bytes` + # releases the GIL during the underlying os.read, so other coroutines run during disk I/O. + # base64 stays inline on purpose: CPython's binascii holds the GIL, so threading it would + # not free the loop (and this matches the JS SDK, which also reads off-loop but encodes inline). + data, resolved_name, resolved_type = await asyncio.to_thread(_to_asset_bytes, source, filename, content_type) encoded = base64.b64encode(data).decode("ascii") try: uploaded = await client.upload(UploadInput(filename=resolved_name, data=encoded, content_type=resolved_type)) diff --git a/tests/unit/test_build_inputs.py b/tests/unit/test_build_inputs.py index 3f9adb7..3e07a78 100644 --- a/tests/unit/test_build_inputs.py +++ b/tests/unit/test_build_inputs.py @@ -11,9 +11,10 @@ import httpx import pytest +from pydantic import ValidationError from pytest_mock import MockerFixture, MockType -from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem +from pipelex_sdk.build_models import BuildInputsRequest, BuildInputsResponseAdapter, BuildInputsValidReport, CrateInvalidReport, MthdsFileItem from pipelex_sdk.client import PipelexAPIClient from pipelex_sdk.errors import ApiResponseError @@ -73,6 +74,29 @@ def test_invalid_closure_is_a_200_verdict(self, mocker: MockerFixture) -> None: assert isinstance(report, CrateInvalidReport) assert report.validation_errors[0].message == "unknown pipe type" + @pytest.mark.parametrize( + "body", + [ + # format=json but carrying no template at all — the flagged malformed-200 shape. + {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "json", "explicit": True}, + # format=json but carrying the toml template (mismatched/opposite shape). + {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "json", "explicit": True, "inputs_toml": "x = 1"}, + # format=toml but carrying no toml template. + {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "toml", "explicit": True}, + ], + ) + def test_valid_report_without_matching_template_is_rejected(self, body: dict[str, object]) -> None: + # A valid verdict must carry the template its `format` selects — the adapter's + # malformed-200 guarantee, now honored for the template shape too. + with pytest.raises(ValidationError): + BuildInputsResponseAdapter.validate_python(body) + + def test_valid_toml_report_is_accepted(self) -> None: + body = {"is_valid": True, "pipe_ref": "demo.main", "message": "ok", "format": "toml", "explicit": True, "inputs_toml": "photo = 1"} + report = BuildInputsResponseAdapter.validate_python(body) + assert isinstance(report, BuildInputsValidReport) + assert report.inputs_toml == "photo = 1" + def test_no_verdict_422_raises_api_response_error(self, mocker: MockerFixture) -> None: client = self._client() problem = {"detail": "Unknown pipe_ref", "error_type": "PipeNotFound"} diff --git a/tests/unit/test_upload.py b/tests/unit/test_upload.py index 4025216..6c8b227 100644 --- a/tests/unit/test_upload.py +++ b/tests/unit/test_upload.py @@ -27,7 +27,7 @@ UploadTransportError, ) from pipelex_sdk.product_models import UploadedFile, UploadInput -from pipelex_sdk.upload import upload_file +from pipelex_sdk.upload import _to_asset_bytes, upload_file _BASE_URL = "http://localhost:8081" @@ -114,14 +114,33 @@ def test_404_maps_to_unsupported_capability(self) -> None: with pytest.raises(UnsupportedUploadCapabilityError): asyncio.run(upload_file(client, bytes([1]))) - def test_other_status_and_unreachable_map_to_transport(self) -> None: - client_500 = _FakeUploadClient(error=_api_error(500)) + @pytest.mark.parametrize( + "error", + [ + _api_error(500), + ApiUnreachableError("down", api_url=_BASE_URL, code="ECONNREFUSED"), + ], + ) + def test_non_semantic_failures_map_to_transport(self, error: Exception) -> None: + client = _FakeUploadClient(error=error) with pytest.raises(UploadTransportError): - asyncio.run(upload_file(client_500, bytes([1]))) + asyncio.run(upload_file(client, bytes([1]))) - client_down = _FakeUploadClient(error=ApiUnreachableError("down", api_url=_BASE_URL, code="ECONNREFUSED")) - with pytest.raises(UploadTransportError): - asyncio.run(upload_file(client_down, bytes([1]))) + def test_reads_the_local_file_off_the_event_loop(self, mocker: MockerFixture, tmp_path: Path) -> None: + # The (possibly large) file read is offloaded via asyncio.to_thread so it never blocks + # the event loop. `wraps` keeps the real behavior; we only assert the offload happened. + to_thread_spy = mocker.patch("pipelex_sdk.upload.asyncio.to_thread", wraps=asyncio.to_thread) + client = _FakeUploadClient() + path = tmp_path / "shot.png" + path.write_bytes(bytes([1, 2, 3, 4])) + + record = asyncio.run(upload_file(client, path)) + + assert to_thread_spy.await_count == 1 + await_args = to_thread_spy.await_args + assert await_args is not None + assert await_args.args[0] is _to_asset_bytes + assert record.size == 4 # real behavior preserved through the wrapped call def test_wires_through_the_real_client(self, mocker: MockerFixture) -> None: client = PipelexAPIClient(api_key="test-token", base_url=_BASE_URL) From bba5074a29db944323f9d67f87460fdcbaf7149d Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:07:39 +0200 Subject: [PATCH 8/9] Release v0.5.0 --- CHANGELOG.md | 2 +- pyproject.toml | 2 +- uv.lock | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 741b3fb..596169d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog -## [Unreleased] +## [v0.5.0] - 2026-07-22 ### Added diff --git a/pyproject.toml b/pyproject.toml index 92e83a9..9769741 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pipelex-sdk" -version = "0.4.0" +version = "0.5.0" description = "The Python client for the Pipelex hosted API — the MTHDS Protocol surface plus the durable run lifecycle and the Pipelex product surface, built on the `mthds` protocol base." authors = [{ name = "Evotis S.A.S.", email = "oss@pipelex.com" }] maintainers = [{ name = "Pipelex staff", email = "oss@pipelex.com" }] diff --git a/uv.lock b/uv.lock index d9e79d6..9e47f0e 100644 --- a/uv.lock +++ b/uv.lock @@ -303,7 +303,7 @@ wheels = [ [[package]] name = "pipelex-sdk" -version = "0.4.0" +version = "0.5.0" source = { editable = "." } dependencies = [ { name = "httpx" }, From 4ef1d1e45fe33dc894acd071bc817c3b53e54284 Mon Sep 17 00:00:00 2001 From: Louis Choquel <8851983+lchoquel@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:48:04 +0200 Subject: [PATCH 9/9] docs(wip): capture PR #11 deferred review-agent findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Triaged the unresolved SWE-bot comments on the v0.5.0 release PR against the JS SDK (@pipelex/sdk) and the pipelex runtime. Records the one confirmed-but-deferred bug (nested asset under a structured url field is not uploaded — parity-bound, needs an upstream template-contract change) plus the server-side 422-vs-413 upload seam, and dismisses the two false positives. No source changes. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_019cpRcAL7MEQcsx2hHvGxC2 --- wip/pr-11-review-notes.md | 87 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) create mode 100644 wip/pr-11-review-notes.md diff --git a/wip/pr-11-review-notes.md b/wip/pr-11-review-notes.md new file mode 100644 index 0000000..a318f66 --- /dev/null +++ b/wip/pr-11-review-notes.md @@ -0,0 +1,87 @@ +# PR #11 — deferred review-agent findings + +Follow-up notes from triaging the SWE-bot review comments on [PR #11](https://github.com/Pipelex/pipelex-sdk-python/pull/11) (Release v0.5.0). Each item below was verified read-only against the code and against the two repos this SDK is contractually bound to — `../pipelex-sdk-js/` (`@pipelex/sdk`, the parity counterpart) and `../pipelex/` (the runtime whose `input_normalizer` the walk mirrors). + +The other flagged comment on the PR was a false positive and needed no follow-up (recorded under "Dismissed" below). + +--- + +## 1. Nested asset under a structured `url` field is not uploaded (confirmed, deferred) + +**Reported by:** codex — thread on `pipelex_sdk/prepare_inputs.py:153`. + +**Status:** Confirmed latent bug. Deferred because a correct fix is a cross-repo contract decision, not a local patch, and the SDK is at exact parity with `@pipelex/sdk`. + +### The bug + +`_resolve_node` classifies a template node as file content purely by shape: + +```python +def _is_file_content(node: Any) -> bool: + return isinstance(node, dict) and "url" in node +``` + +For a structured concept that has a top-level field literally named `url` **and** a sibling file-bearing field — e.g. `Article { url: str, cover: Image }` — the explicit template renders as: + +```json +{"url": "https://mock.invalid/url", "cover": {"url": "https://mock.invalid/url"}} +``` + +(The template generator special-cases any field named `url` or `*_url` at `../pipelex/pipelex/core/concepts/concept_representation_generator.py:314-320`, so `url` is not reserved to Image/Document.) + +`_is_file_content` then fires on the **top-level** `url` (`prepare_inputs.py:153`), so `_resolve_file_position` resolves only the article's text URL and returns early. The walk never recurses into `cover`, so a caller value like `{"url": "https://example.com/a", "cover": }` leaves the `cover` bytes **unuploaded** — the hosted run receives raw bytes at a nested Image position. + +### Why the runtime gets this right and the SDK does not + +The runtime `../pipelex/pipelex/pipeline/input_normalizer.py:61-93` dispatches on the Python **type** of the value: `isinstance(value, (ImageContent, DocumentContent))` vs `isinstance(value, StructuredContent)` (which recurses every field). It never keys on the presence of a `url` dict key. The SDK's shape-only heuristic is a documented approximation of that type-based classifier (`prepare_inputs.py:7-12`, `docs/input-preparation.md`); the two coincide for the common case and diverge exactly on this shape. + +### Why it is not cleanly fixable at the SDK layer + +- **No type info at nested positions.** Only the top-level template envelope carries a `"concept"` key, and it is dropped when the walk reads `entry["content"]` (`prepare_inputs.py:205`). At a nested `{"url": ...}` the SDK has nothing but shape to go on — it cannot tell an `ImageContent` from a structured concept whose single field is `url: str`. +- **Shape refinement is ambiguous.** A single-key `{"url": str}` structured concept is indistinguishable from single-key image content; and multi-key canonical image content is deliberately supported (the existing test `tests/unit/test_prepare_inputs.py:136` feeds a two-key `{"url", "mime_type"}` cover). So neither "single-key only" nor "keys ⊆ image/document vocabulary" is reliable without hardcoding the runtime's field vocabulary into the SDK — fragile and still collision-prone. +- **Parity constraint.** The Python code is a faithful port of `../pipelex-sdk-js/src/prepare-inputs.ts:67-69,172-174` (identical `isFileContent` + identical short-circuit). Any behavioral change must land in both SDKs in the same coordinated change; a Python-only fix would break the parity invariant. + +### Recommended real fix (upstream, coordinated) + +Thread concept/type information into the **nested** positions of the explicit inputs template in `../pipelex/` (so a nested node self-identifies as Image/Document vs structured), then update both SDKs to classify by that tag instead of by the `url` key. That is a template-contract change and should be decided with the runtime + JS SDK owners together. + +### Fragile interim mitigation (only if forced, must be mirrored in JS) + +Make `_is_file_content` treat a dict as file content only when `"url" in node` **and** every other key is drawn from the known Image/Document optional-field set (`public_url, mime_type, filename, title, snippet, caption, width, height, source_prompt, source_negative_prompt`). This recovers the `{url, cover}` case while preserving the multi-key image case in the current tests. It does **not** fix the single-key `{url: str}` structured concept (still shape-indistinguishable), so it is a partial mitigation, not a fix — which is why the honest call is to defer and raise the contract question upstream. + +### Repro (documentation only — not added to the suite) + +A failing test would break release CI, so this is recorded here rather than committed. In `tests/unit/test_prepare_inputs.py` style (`_FakePrepareClient`, `asyncio.run`, single `TestPrepareInputs` class): + +- template entry: `_entry("demo.Article", {"url": "https://mock.invalid/url", "cover": {"url": "https://mock/c.png"}})` +- inputs: `{"article": {"url": "https://example.com/a", "cover": bytes([7, 7])}}` +- expected once fixed: `cover` rewritten to a `pipelex-storage://` url, top-level `url` passed through, `len(uploads) == 1`. + +Under today's code this asserts 0 uploads and the cover bytes leak — cleanly documenting the gap. + +--- + +## 2. Oversized upload surfaces as `UploadTransportError`, not `RejectedAssetError` (needs-judgment, server-side) + +**Reported by:** greptile (P1) — thread on `pipelex_sdk/upload.py:101-103`. The literal comment ("400/422 should be `RejectedAssetError`") is a **false positive for this PR** (see below), but verification surfaced a real cross-repo seam worth a decision. + +### Why the literal comment is a false positive + +`_map_upload_error` (`upload.py:90-105`) maps `413 → RejectedAssetError`, `401|403 → UploadAuthenticationError`, `404 → UnsupportedUploadCapabilityError`, and everything else → `UploadTransportError`. `../pipelex-sdk-js/src/upload.ts:196-241` is byte-for-byte identical (only 413 maps to a rejected asset). Mapping 400/422 → `RejectedAssetError` in Python alone would diverge from `@pipelex/sdk`, which is the repo's controlling invariant. So the flagged line is correct-by-design. + +### The real seam + +`pipelex-api` rejects an oversized upload with **422**, not 413: the base64 `data` field has a Pydantic `max_length=MAX_UPLOAD_BASE64_CHARS` constraint (`pipelex-api/api/routes/uploader.py`), which FastAPI turns into a 422 request-validation error (asserted by `pipelex-api/tests/unit/test_uploader.py`). The explicit `len(data) > MAX_UPLOAD_BYTES` → 413 path is only reachable in the narrow band where the char count passes but decoded bytes marginally exceed the cap. A base64-decode failure returns 400. + +Consequence: the documented "asset too big → `RejectedAssetError`" category is effectively **unreachable in the common case**, in *both* SDKs — the most common oversized rejection comes back as a transport error. + +### Options (pick one, coordinated) + +- **Preferred:** make `pipelex-api`'s size rejection surface as **413** (align the Pydantic-`max_length` rejection with the explicit 413 check) so the existing "413 == rejected asset" contract holds end-to-end. No SDK change; parity preserved. +- **Alternative:** treat 422 as a rejection at the SDK layer — extend `case 413:` → `case 413 | 422:` — but only if landed in **both** `pipelex_sdk/upload.py` and `../pipelex-sdk-js/src/upload.ts` together, with matching tests. `RejectedAssetError` carries `status`, so callers could still tell 413 from 422. + +--- + +## Dismissed (no follow-up needed) + +**Empty-list template skips uploads** — greptile (P2), `prepare_inputs.py:155-160`. Can't-happen: the explicit template never emits an empty list. Top-level multiplicity wraps the content in a one-element exemplar (`../pipelex/pipelex/core/concepts/concept.py:225-226`) and nested `list[T]` fields render one example item (`../pipelex/pipelex/core/concepts/concept_representation_generator.py:210-236`). Line 156's `template_node[0]` already relies on non-empty, and JS carries the identical `> 0` guard (`prepare-inputs.ts:175`). Recorded here so it is not re-flagged.