diff --git a/CHANGELOG.md b/CHANGELOG.md index ab76c57..ad05570 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,17 @@ # Changelog +## [v0.15.0] - 2026-07-22 + +### Added + +- **Cost reports on every run.** Each result-producing command now prints a per-call cost report (pipe, model, tokens in→out, USD cost, and a total) to stderr, after the JSON result on stdout. A new shared `piper/usage.py` reads the run's per-call usage off the SDK result into a `RunUsage` (`usage_from_results` for the durable modes' typed `RunResults.tokens_usages`; `usage_from_execute` for blocking, which lifts the same records off its execute result) and renders it with `print_cost_report`. It respects the SDK's usage semantics: an unpriced call (`cost is None` — mock / own-GPU / dry-run) is excluded from the total rather than counted as zero, and token categories are never summed. Detached prints the report from `wait` / `result` (collection time), not from the start-only demo commands. + +### Changed + +- **Cleaner file upload for `summarize-pdf` (breaking):** the document is now uploaded to hosted storage up front with `client.upload_file` (new `inputs.upload_document_input`), and the run request carries only the returned `pipelex-storage://` URI — replacing the previous inline base64 `data:` URL. `inputs.build_document_input` becomes the pure envelope builder `build_document_input(path, uri)`; the upload is a separate step, so a file or capability error surfaces before any run is created. `piper/errors.py` gains hints for the file-upload error family (unsupported capability, rejected asset, authentication, invalid local source). +- **Lifecycle helpers return `(main_stuff, RunUsage)` (breaking):** `execute_pipe`, `start_and_wait`, and detached's `attend_run` now return the resolved main output *and* the run's usage, instead of `main_stuff` alone, so the commands can print the cost report. The demo commands unpack the pair; the SDK output accessor itself is unchanged. +- Bumped the `pipelex-sdk` floor from `>=0.4.0` to `>=0.5.0` (for `upload_file` and typed run usage). + ## [v0.14.1] - 2026-07-15 - **Multi-file bundle support in the mode lifecycle helpers:** `execute_pipe`, `start_and_wait`, and `start_pipe` now take `mthds_contents: list[str]` (the bundle's `.mthds` files as strings — one entry for a single-file bundle, several for a multi-file one) instead of a single `bundle: str`, passed straight through to the SDK. Multi-file bundles cannot be concatenated into one string (duplicate top-level TOML keys), so the list is the interface. The three demos stay single-file (`mthds_contents=[bundle]`); a method dir with several `.mthds` files is read with `[p.read_text() for p in sorted((METHODS_DIR / "").glob("*.mthds"))]`. The package-data glob broadens to `methods/*/*.mthds` so multi-file bundles ship. diff --git a/CLAUDE.md b/CLAUDE.md index 75d2f51..e0013fe 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -38,15 +38,15 @@ This starter calls the **hosted Pipelex API** via the `pipelex-sdk` package (`Pi - Credentials/endpoint come from `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` (see `.env.example`). `python-dotenv` loads `.env` when running the CLI or tests. - **The execution mode is the command group, not an option.** There are exactly three, each a self-contained Typer sub-package: `piper blocking …` (`client.execute` — one call, dies at the hosted ~30s cap), `piper attended …` (`client.start` + `client.wait_for_result` — durable, you wait), `piper detached …` (`client.start` only — durable, you collect it later with `piper detached status|result|wait `). Attended and detached start the *same* durable run; the axis they name is who waits. There is no default mode and no `--mode` option: `piper/cli.py` is a thin assembler (`load_dotenv` callback + three `add_typer` calls in reading order) and nothing else. -- **Each mode file is a copy-paste unit; lifecycle code is never shared.** `piper//cli.py` holds that mode's whole story: its `typer.Typer`, its consoles (results → stdout, progress → stderr), its one public lifecycle helper (`execute_pipe` / `start_and_wait` / `start_pipe`, plus `attend_run` + the fetchers in detached), its demo commands, and a private `_run()` that wraps `asyncio.run` and catches SDK errors once. The **only** shared modules are the two that are orthogonal to execution: `piper/inputs.py` (text-or-file input with a built-in **sample fallback** so every demo runs with zero arguments — `read_text_input` returns `TextInput(text, is_sample)` and the demo prints a stderr notice when the sample was used; plus file → `{"concept": "Document", "content": …}` envelope with the file base64-encoded into a `data:` URL, and the `SAMPLE_*` constants) and `piper/errors.py` (SDK error → message + hint, hints naming the mode groups; it reads the RFC 7807 **problem+json** body off raw protocol-route `httpx.HTTPStatusError`s and branches on the structured `error_type`, e.g. `StartRequiresAsyncOrchestration` → "use `piper blocking`"). Do not introduce a shared runner — the dispatch indirection is exactly what this layout removed. See `docs/cli-architecture.md`. +- **Each mode file is a copy-paste unit; lifecycle code is never shared.** `piper//cli.py` holds that mode's whole story: its `typer.Typer`, its consoles (results → stdout, progress → stderr), its one public lifecycle helper (`execute_pipe` / `start_and_wait` / `start_pipe`, plus `attend_run` + the fetchers in detached), its demo commands, and a private `_run()` that wraps `asyncio.run` and catches SDK errors once. The **only** shared modules are those orthogonal to execution: `piper/inputs.py` (text-or-file input with a built-in **sample fallback** so every demo runs with zero arguments — `read_text_input` returns `TextInput(text, is_sample)` and the demo prints a stderr notice when the sample was used; plus file → `{"concept": "Document", "content": …}` envelope — `upload_document_input` uploads the file to hosted storage with `client.upload_file` and wraps the returned `pipelex-storage://` URI, `build_document_input(path, uri)` being the pure envelope builder — and the `SAMPLE_*` constants), `piper/errors.py` (SDK error → message + hint, hints naming the mode groups; it reads the RFC 7807 **problem+json** body off raw protocol-route `httpx.HTTPStatusError`s and branches on the structured `error_type`, e.g. `StartRequiresAsyncOrchestration` → "use `piper blocking`"; it also hints the file-upload error family), and `piper/usage.py` (cost report: `usage_from_results` / `usage_from_execute` read the per-call usage off the SDK result into a `RunUsage`, `print_cost_report` renders it to **stderr**). Do not introduce a shared runner — the dispatch indirection is exactly what this layout removed. See `docs/cli-architecture.md`. - **Full demo matrix, guarded.** All three demos exist in all three modes: `extract-entities` (text in), `summarize-pdf` (a *file* in), `generate-image` (prompt in). `generate-image` is the deliberate slow case that overruns the ~30s blocking cap — `piper blocking generate-image` is *expected to fail*, and that is the teaching moment for the durable modes. The near-duplication across mode files is the pedagogy (diff two mode files and only the lifecycle helper differs); `tests/unit/test_mode_symmetry.py` keeps it from drifting. `samples/sample-invoice.pdf` is shipped for `summarize-pdf`. -- The SDK resolves the main output on every result-producing path (`client.execute` returns a `PipelexExecuteResult`, the durable path a `RunResults`, both exposing a resolved `.main_stuff`, typed `Any`; a completed run with no main stuff raises `MissingMainStuffError`). So the result-producing lifecycle helpers (`execute_pipe`, `start_and_wait`, detached's `attend_run`) return `main_stuff`, and the blocking/attended demo commands narrow it inline — e.g. `ExtractedEntities.model_validate(main_stuff)` — into the generated model. Detached is the exception by design: `start_pipe` returns only the run id (the demos print it bare), and the run-id commands (`wait`/`result`) print the output generically — no model narrowing, since at collection time the command doesn't know which method the run executed. There is no per-example wrapper layer. +- The SDK resolves the main output on every result-producing path (`client.execute` returns a `PipelexExecuteResult`, the durable path a `RunResults`, both exposing a resolved `.main_stuff`, typed `Any`; a completed run with no main stuff raises `MissingMainStuffError`). So the result-producing lifecycle helpers (`execute_pipe`, `start_and_wait`, detached's `attend_run`) return a `(main_stuff, RunUsage)` pair — `main_stuff` plus the run's per-call cost/token usage — and the blocking/attended demo commands narrow `main_stuff` inline — e.g. `ExtractedEntities.model_validate(main_stuff)` — into the generated model, then print the usage as a cost report to stderr. Cost usage is `RunResults`-typed on the durable modes but only raw on blocking's execute result (lifted by `usage_from_execute`; the typed-on-execute follow-up is tracked in `wip/sdk-qol/typed-tokens-usages-on-execute.md`). Detached is the exception by design: `start_pipe` returns only the run id (the demos print it bare, no cost — the run isn't done), and the run-id commands (`wait`/`result`) print the output generically **and** its cost report — no model narrowing, since at collection time the command doesn't know which method the run executed. There is no per-example wrapper layer. - The modes spell out lifecycles the SDK could hide: `client.start_and_wait()` is a self-healing one-liner that picks the path for you (the production shortcut). The starter writes them out because teaching the difference is the point. - **The typed models are generated, never hand-written.** `pipelex codegen` projects each bundle's concepts into `piper/generated//models.py` (stamped, locked by a sibling `codegen.lock`); the mode CLIs and the e2e tests import from there. Do NOT edit generated files — edit the bundle, then `make codegen` (regenerates models + `inputs.template.json` scaffolds) and `make codegen-check` (offline drift check). The `pipelex` CLI is not a dependency of this starter; point the `PIPELEX` make variable at a pipelex install that ships `codegen`. `piper/generated` is excluded from ruff (reformatting would trip the drift check) but fully type-checked. See `docs/codegen.md`. ## Project Structure -- Package: `piper/` (Python 3.11+, target 3.11) — root `cli.py` + one sub-package per execution mode (`blocking/`, `attended/`, `detached/`) + the two shared modules (`inputs.py`, `errors.py`). New mode sub-packages must be added to `[tool.setuptools] packages` in `pyproject.toml`. +- Package: `piper/` (Python 3.11+, target 3.11) — root `cli.py` + one sub-package per execution mode (`blocking/`, `attended/`, `detached/`) + the shared modules (`inputs.py`, `errors.py`, `usage.py`). New mode sub-packages must be added to `[tool.setuptools] packages` in `pyproject.toml`. - Tests: `tests/` (unit = offline per-mode CLI tests patching each mode's public lifecycle helper, plus mode-symmetry / error-mapping / generated-client tests; integration = offline boot/bundle checks + API `validate`; e2e = full run via the API, one execution mode per demo so all three get end-to-end coverage) - Dependency manager: uv (>=0.7.2) - Pipelex dependency: `pipelex-sdk` package from PyPI (the API client — see pyproject.toml). The `pipelex` runtime is **not** a dependency. diff --git a/README.md b/README.md index 1e8a135..c98d6ca 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ It ships a handful of demo methods, each exposed as a `piper` CLI command: - **`summarize-pdf`** — given a document (PDF), produce a title, document type, and key points. Shows how to feed a *file* to a pipe. - **`generate-image`** — given a text prompt, generate an image. Being slow, it's the example that best shows the split between the execution modes (image generation routinely outlives the hosted ~30s blocking cap). -Each prints its result as JSON. +Each prints its result as JSON on stdout, and a short **cost report** (per-call model + USD cost) to stderr — so the result stays pipeable while you still see what the run consumed. ### Use this template @@ -157,11 +157,11 @@ flowchart TD 1. **Read the bundle.** `piper` reads `methods/extract-entities/main.mthds` from disk and constructs a `PipelexAPIClient`, which picks up `PIPELEX_BASE_URL` / `PIPELEX_API_KEY` from the environment. A method dir may hold a single `main.mthds` or several `.mthds` files (a multi-file bundle split across pipes) — read them all with `[p.read_text() for p in sorted((METHODS_DIR / "").glob("*.mthds"))]`. 2. **Run it on the API.** The bundle's files are sent together as *content* (`mthds_contents`, one string per file), so nothing method-specific needs to live in the runtime — edit the `.mthds` file(s) and re-run, no redeploy. -3. **Narrow the result.** The SDK resolves the run's `main_stuff`; the command validates it into the generated `ExtractedEntities` model (`ExtractedEntities.model_validate(main_stuff)`), printed as JSON. +3. **Narrow the result.** The SDK resolves the run's `main_stuff`; the command validates it into the generated `ExtractedEntities` model (`ExtractedEntities.model_validate(main_stuff)`), printed as JSON. It also returns the run's per-call usage, which the command prints as a cost report to stderr (via `piper/usage.py`). The typed models are **not hand-written**: they are generated from the `.mthds` bundles by `pipelex codegen` into `piper/generated/` (stamped, with a `codegen.lock` per method). Edit a bundle → `make codegen` regenerates the models and input templates → `make codegen-check` verifies offline that nothing is stale or hand-edited. See [docs/codegen.md](docs/codegen.md). -The other demos run through the exact same path — they differ only in their inputs and output shapes. `summarize-pdf` sends a `Document` envelope (`inputs.build_document_input()` base64-encodes the file into a `data:` URL); `generate-image` returns the built-in `Image` content. +The other demos run through the exact same path — they differ only in their inputs and output shapes. `summarize-pdf` feeds a *file* to a pipe: because a hosted run can't see your filesystem, `inputs.upload_document_input()` uploads the PDF first (`client.upload_file`) and the run request carries only the returned `pipelex-storage://` URI, wrapped in a `Document` envelope — the bytes never ride the request. `generate-image` returns the built-in `Image` content. ## Execution modes: blocking, attended, detached @@ -260,7 +260,7 @@ Each mode is a **self-contained copy-paste unit**: `piper//cli.py` holds t | `attended` | `start_and_wait()` | `client.start` + `client.wait_for_result` | | `detached` | `start_pipe()` (+ `attend_run()` for `wait`) | `client.start` (then `client.wait_for_result` later) | -The only things shared across modes are the two concerns that have nothing to do with execution: `piper/inputs.py` (encode the inputs) and `piper/errors.py` (present an SDK error). **Lifecycle code is never shared** — that is what keeps each mode file copy-pasteable, and it is why the demo commands are near-duplicated across the three: diff `blocking/cli.py` against `attended/cli.py` and the only difference is the lifecycle helper. (`tests/unit/test_mode_symmetry.py` guards that duplication against drift.) +The only things shared across modes are the concerns that have nothing to do with execution: `piper/inputs.py` (prepare the inputs — incl. uploading a file), `piper/errors.py` (present an SDK error), and `piper/usage.py` (read + print the cost report). **Lifecycle code is never shared** — that is what keeps each mode file copy-pasteable, and it is why the demo commands are near-duplicated across the three: diff `blocking/cli.py` against `attended/cli.py` and the only difference is the lifecycle helper. (`tests/unit/test_mode_symmetry.py` guards that duplication against drift.) The modes also spell out lifecycles the SDK could hide: `client.start_and_wait()` is a self-healing one-liner that picks the right path by itself — the production shortcut when you don't care which one runs. This starter branches explicitly because teaching the difference is the point. @@ -272,8 +272,9 @@ piper/ blocking/cli.py # the whole blocking mode in one file (execute_pipe) attended/cli.py # the whole attended mode in one file (start_and_wait) detached/cli.py # the whole detached mode in one file (start_pipe + wait/status/result) - inputs.py # SHARED: text-or-file input, and file → Document envelope + inputs.py # SHARED: text-or-file input, and file → upload → Document envelope errors.py # SHARED: maps SDK errors to CLI messages + hints + usage.py # SHARED: reads per-call usage and prints the cost report (stderr) generated/ # typed clients generated from the bundles (`make codegen`) — do not edit extract_entities/ # models.py (stamped) + codegen.lock summarize_pdf/ @@ -291,7 +292,7 @@ tests/ .env.example # PIPELEX_BASE_URL + PIPELEX_API_KEY ``` -Only two modules are shared across the modes, and neither knows anything about execution: see [docs/cli-architecture.md](docs/cli-architecture.md) for the sharing rule and the anatomy of a mode file. +Only these modules are shared across the modes, and none knows anything about execution mode: see [docs/cli-architecture.md](docs/cli-architecture.md) for the sharing rule and the anatomy of a mode file. ## Useful commands diff --git a/docs/cli-architecture.md b/docs/cli-architecture.md index ccf5295..e853240 100644 --- a/docs/cli-architecture.md +++ b/docs/cli-architecture.md @@ -16,20 +16,21 @@ piper detached wait # or status, result ``` piper/ cli.py # the assembler: load .env, mount the three groups in reading order. Nothing else. - inputs.py # SHARED — encode the inputs + inputs.py # SHARED — prepare the inputs (text/file; incl. the hosted file upload) errors.py # SHARED — present an SDK error + usage.py # SHARED — read + print the run's cost report blocking/cli.py # the whole blocking mode attended/cli.py # the whole attended mode detached/cli.py # the whole detached mode + the run-id lifecycle commands ``` -Each mode file is a **copy-paste unit**: that file plus `inputs.py` plus `errors.py` is all the *lifecycle* code you need to lift the mode into your own project. A runnable copy also carries the method-specific artifacts the demos reference — the bundles (`piper/methods/`), the generated models (`piper/generated/`), and the sample document — but those are exactly what you replace with your own method anyway. No other code in `piper/` is load-bearing for a mode. +Each mode file is a **copy-paste unit**: that file plus `inputs.py` plus `errors.py` plus `usage.py` is all the *lifecycle* code you need to lift the mode into your own project. A runnable copy also carries the method-specific artifacts the demos reference — the bundles (`piper/methods/`), the generated models (`piper/generated/`), and the sample document — but those are exactly what you replace with your own method anyway. No other code in `piper/` is load-bearing for a mode. ## The sharing rule > Share what is orthogonal to execution. **Never share lifecycle code.** -Input encoding and error presentation don't care how a run is executed, so they are shared. The lifecycle — what you call, in what order, and what you do while waiting — *is* the mode, so it lives in the mode's own file, in full, even when that means near-duplicating a demo command across three files. +Input preparation, error presentation, and cost-report formatting don't care how a run is executed, so they are shared. The lifecycle — what you call, in what order, and what you do while waiting — *is* the mode, so it lives in the mode's own file, in full, even when that means near-duplicating a demo command across three files. (Cost *extraction* is the one seam that differs per mode — the durable modes read a typed `RunResults.tokens_usages`, blocking lifts the same records raw off its execute result — so each helper picks the right `usage.py` reader; the *formatting* is shared.) That duplication is deliberate, and it is the pedagogy: diff `blocking/cli.py` against `attended/cli.py` and the only thing that differs is the lifecycle helper. The moment two modes reach for a shared `runner` helper, a dispatch layer is back between the command and the SDK, and the reading path grows a hop. (The earlier version of this starter had exactly that: a `--mode` option → `_dispatch()` → `match ExecutionMode` → `runner.py` → the SDK. Four layers to answer "how do I call the API?".) @@ -42,19 +43,26 @@ All three have the same four-part shape, so they diff cleanly: 1. **Module docstring** — the mode's contract in a paragraph, plus its copy-paste contract. 2. **App + consoles** — its own `typer.Typer`, its own `Console()` (stdout, for results — pipeable) and `Console(stderr=True)` (stderr, for progress chatter). 3. **The lifecycle helper** — one public async function that *is* the mode (`detached` adds the run-id lifecycle helpers described below). It gets a public name because it is the featured code, and it is what the unit tests patch and the e2e tests call directly. -4. **The demo commands + a private `_run()`** — each command reads its input, reads its bundle, awaits the lifecycle helper through `_run()` (`asyncio.run` + the single `except (PipelineRequestError, httpx.HTTPStatusError)` that presents via `piper/errors.py`), narrows the result into its *generated* model, prints JSON. +4. **The demo commands + a private `_run()`** — each command reads its input, reads its bundle, awaits the lifecycle helper through `_run()` (`asyncio.run` + the single `except (PipelineRequestError, httpx.HTTPStatusError)` that presents via `piper/errors.py`), narrows the result into its *generated* model, prints JSON to stdout, and prints the run's cost report to stderr. The lifecycle helpers take the bundle as `mthds_contents: list[str]` — one string per `.mthds` file — and pass it straight to the SDK. The three demos are single-file methods, so each reads its `main.mthds` and wraps it as `mthds_contents=[bundle]`. A method dir may instead hold several `.mthds` files (a multi-file bundle split across pipes with `signature_for` cross-file declarations); read them all with `[p.read_text() for p in sorted((METHODS_DIR / "").glob("*.mthds"))]` and hand that list to the helper unchanged. Concatenating the files into one string would be invalid TOML — the list is the interface for exactly this reason. | Mode | Lifecycle helper | SDK calls | What the demo prints | | --- | --- | --- | --- | -| `blocking` | `execute_pipe()` | `client.execute` | the result, as JSON | -| `attended` | `start_and_wait()` | `client.start` + `client.wait_for_result` | the run id (stderr), then the result as JSON | -| `detached` | `start_pipe()` | `client.start` | the run id, bare, on **stdout** | +| `blocking` | `execute_pipe()` | `client.execute` | the result as JSON, then a cost report (stderr) | +| `attended` | `start_and_wait()` | `client.start` + `client.wait_for_result` | the run id (stderr), then the result as JSON + a cost report (stderr) | +| `detached` | `start_pipe()` | `client.start` | the run id, bare, on **stdout** (no cost — the run isn't done yet) | -`detached` additionally owns the run-id lifecycle: `attend_run()` backs `wait`, and thin fetchers back `status` and `result`. It is the only mode with more than the demos, because "start now, collect later" is only a complete story if you can come back for the result. +`detached` additionally owns the run-id lifecycle: `attend_run()` backs `wait`, and thin fetchers back `status` and `result`. It is the only mode with more than the demos, because "start now, collect later" is only a complete story if you can come back for the result. The cost report lands where the *result* does, so in detached mode it prints from `wait` and `result` (collection time), not from the start-only demo commands. -The result-producing helpers — `execute_pipe()`, `start_and_wait()`, and detached's `attend_run()` — return the run's resolved `main_stuff` (the SDK types it `Any` — the content is polymorphic). In blocking and attended mode each demo command narrows it with `Model.model_validate(main_stuff)`, which is what restores type safety; the models are generated from the bundles (see [codegen.md](codegen.md)). Detached is different by design: `start_pipe()` returns only the run id, and the run-id commands (`wait`, `result`) print the output generically — at collection time the command doesn't know which method the run executed, so there is no model to narrow into. +The result-producing helpers — `execute_pipe()`, `start_and_wait()`, and detached's `attend_run()` — return a `(main_stuff, usage)` pair: the run's resolved `main_stuff` (the SDK types it `Any` — the content is polymorphic) and its `RunUsage` (the per-call cost/token records, read via `piper/usage.py`). In blocking and attended mode each demo command narrows `main_stuff` with `Model.model_validate(main_stuff)`, which is what restores type safety (the models are generated from the bundles — see [codegen.md](codegen.md)), then prints `usage` as a cost report to stderr. Detached is different by design: `start_pipe()` returns only the run id, and the run-id commands (`wait`, `result`) print the output generically — at collection time the command doesn't know which method the run executed, so there is no model to narrow into. + +### Cost reports and file upload + +Two SDK capabilities show up in every result-producing path: + +- **Cost report.** A completed run reports what its inference calls consumed. `piper/usage.py` reads that off the SDK result (`usage_from_results` for the durable modes' typed `RunResults.tokens_usages`; `usage_from_execute` for blocking, which still lifts the records raw off `pipe_output`) and `print_cost_report` renders a per-call table plus a USD total — always to **stderr**, so stdout stays the clean, pipeable result. It respects the SDK's semantics: an unpriced call (`cost is None`: mock / own-GPU / dry-run) is not a zero-cost call, and token categories are never summed (`input_cached` is a subset of `input`). +- **File upload.** `summarize-pdf` feeds a *file* to a pipe. A hosted run cannot see your filesystem, so `inputs.py`'s `upload_document_input` uploads the file first (`client.upload_file`) and the run request carries only the returned `pipelex-storage://` URI — never the bytes. Preparation is a step of its own, before the run: an unreadable file or an upload-incapable deployment fails before any run is created, presented through the same `piper/errors.py` path as every other SDK error. ## Two conventions worth copying diff --git a/piper/attended/cli.py b/piper/attended/cli.py index 41802d8..b21d813 100644 --- a/piper/attended/cli.py +++ b/piper/attended/cli.py @@ -11,8 +11,8 @@ right path by itself — that is the production shortcut. This starter spells the lifecycle out because teaching the difference is the point. -Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py`. The three mode -packages never share lifecycle code, so the demo commands below are deliberate +Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py` + `piper/usage.py`. The +three mode packages never share lifecycle code, so the demo commands below are deliberate mirrors of their `blocking` / `detached` twins: only `start_and_wait()` differs. """ @@ -31,7 +31,8 @@ from piper.generated.extract_entities.models import ExtractedEntities from piper.generated.generate_image.models import Image from piper.generated.summarize_pdf.models import DocumentSummary -from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, build_document_input, read_text_input +from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, read_text_input, upload_document_input +from piper.usage import RunUsage, print_cost_report, usage_from_results ResultT = TypeVar("ResultT") @@ -44,14 +45,15 @@ progress_console = Console(stderr=True) -async def start_and_wait(*, pipe_code: str, mthds_contents: list[str], inputs: dict[str, Any]) -> Any: +async def start_and_wait(*, pipe_code: str, mthds_contents: list[str], inputs: dict[str, Any]) -> tuple[Any, RunUsage]: """The whole attended lifecycle: start a durable run, print its id, wait here for the result. Credentials come from `PIPELEX_API_KEY` / `PIPELEX_BASE_URL`. `mthds_contents` is the bundle's `.mthds` files as strings — one entry for a single-file bundle, several for a multi-file one. The SDK resolves the method's main output for you: `.main_stuff` is the content the pipe named as its result (a completed run that names none raises - `MissingMainStuffError`). + `MissingMainStuffError`). Alongside it we return the run's `RunUsage` — the per-call + cost/token records the command prints as a cost report. """ async with PipelexAPIClient() as client: start_result = await client.start(pipe_code=pipe_code, mthds_contents=mthds_contents, inputs=inputs) @@ -70,7 +72,7 @@ def on_poll(info: PollInfo) -> None: # Ctrl-C: the run keeps executing server-side — it has just become a detached run. progress_console.print(f"\nInterrupted — the run is still executing. Resume with: [bold]piper detached wait {run_id}[/bold]") raise - return results.main_stuff + return results.main_stuff, usage_from_results(results) @app.command(name="extract-entities") @@ -83,10 +85,11 @@ def extract_entities( if resolved.is_sample: progress_console.print(f"[dim]No text given — using the sample: {resolved.text!r}. Pass your own as an argument or via --file.[/dim]") bundle = (METHODS_DIR / "extract-entities" / "main.mthds").read_text() - main_stuff = _run(start_and_wait(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": resolved.text})) + main_stuff, usage = _run(start_and_wait(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": resolved.text})) # Narrow into the generated typed model (validates the concept's shape), then print it as JSON. entities = ExtractedEntities.model_validate(main_stuff) output_console.print_json(data=entities.model_dump()) + print_cost_report(progress_console, usage) @app.command(name="summarize-pdf") @@ -101,10 +104,12 @@ def summarize_pdf( if file is None: progress_console.print(f"[dim]No file given — using the sample: {document.name}. Pass a path to summarize your own document.[/dim]") bundle = (METHODS_DIR / "summarize-pdf" / "main.mthds").read_text() - inputs = {"document": build_document_input(document)} - main_stuff = _run(start_and_wait(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs)) + # Upload the file first (a separate step from the run) — the run request carries only its URI. + inputs = {"document": _run(upload_document_input(document))} + main_stuff, usage = _run(start_and_wait(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs)) summary = DocumentSummary.model_validate(main_stuff) output_console.print_json(data=summary.model_dump()) + print_cost_report(progress_console, usage) @app.command(name="generate-image") @@ -122,11 +127,12 @@ def generate_image( if resolved.is_sample: progress_console.print(f"[dim]No prompt given — using the sample: {resolved.text!r}. Pass your own as an argument or via --file.[/dim]") bundle = (METHODS_DIR / "generate-image" / "main.mthds").read_text() - main_stuff = _run(start_and_wait(pipe_code="generate_image", mthds_contents=[bundle], inputs={"image_prompt": resolved.text})) + main_stuff, usage = _run(start_and_wait(pipe_code="generate_image", mthds_contents=[bundle], inputs={"image_prompt": resolved.text})) # On the hosted path the runtime returns a storage `url` (`pipelex-storage://…`) # *and* a web-renderable `public_url` (a signed URL); the model keeps both. image = Image.model_validate(main_stuff) output_console.print_json(data=image.model_dump()) + print_cost_report(progress_console, usage) def _run(coro: Coroutine[Any, Any, ResultT]) -> ResultT: diff --git a/piper/blocking/cli.py b/piper/blocking/cli.py index a7107e8..e0f9b36 100644 --- a/piper/blocking/cli.py +++ b/piper/blocking/cli.py @@ -4,8 +4,8 @@ Behind the hosted gateway a run longer than ~30s is cut off (run `generate-image` to see it happen) — that is what `piper attended` and `piper detached` are for. -Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py`. The three mode -packages never share lifecycle code, so the demo commands below are deliberate +Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py` + `piper/usage.py`. The +three mode packages never share lifecycle code, so the demo commands below are deliberate mirrors of their `attended` / `detached` twins: only `execute_pipe()` differs. """ @@ -23,7 +23,8 @@ from piper.generated.extract_entities.models import ExtractedEntities from piper.generated.generate_image.models import Image from piper.generated.summarize_pdf.models import DocumentSummary -from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, build_document_input, read_text_input +from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, read_text_input, upload_document_input +from piper.usage import RunUsage, print_cost_report, usage_from_execute ResultT = TypeVar("ResultT") @@ -36,19 +37,20 @@ progress_console = Console(stderr=True) -async def execute_pipe(*, pipe_code: str, mthds_contents: list[str], inputs: dict[str, Any]) -> Any: +async def execute_pipe(*, pipe_code: str, mthds_contents: list[str], inputs: dict[str, Any]) -> tuple[Any, RunUsage]: """The whole blocking lifecycle: one call, and the result comes back in the response. Credentials come from `PIPELEX_API_KEY` / `PIPELEX_BASE_URL`. `mthds_contents` is the bundle's `.mthds` files as strings — one entry for a single-file bundle, several for a multi-file one. The SDK resolves the method's main output for you: `.main_stuff` is the content the pipe named as its result (a completed run that names none raises - `MissingMainStuffError`). + `MissingMainStuffError`). Alongside it we return the run's `RunUsage` — the per-call + cost/token records the command prints as a cost report. """ async with PipelexAPIClient() as client: with progress_console.status("Running…"): result = await client.execute(pipe_code=pipe_code, mthds_contents=mthds_contents, inputs=inputs) - return result.main_stuff + return result.main_stuff, usage_from_execute(result) @app.command(name="extract-entities") @@ -61,10 +63,11 @@ def extract_entities( if resolved.is_sample: progress_console.print(f"[dim]No text given — using the sample: {resolved.text!r}. Pass your own as an argument or via --file.[/dim]") bundle = (METHODS_DIR / "extract-entities" / "main.mthds").read_text() - main_stuff = _run(execute_pipe(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": resolved.text})) + main_stuff, usage = _run(execute_pipe(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": resolved.text})) # Narrow into the generated typed model (validates the concept's shape), then print it as JSON. entities = ExtractedEntities.model_validate(main_stuff) output_console.print_json(data=entities.model_dump()) + print_cost_report(progress_console, usage) @app.command(name="summarize-pdf") @@ -79,10 +82,12 @@ def summarize_pdf( if file is None: progress_console.print(f"[dim]No file given — using the sample: {document.name}. Pass a path to summarize your own document.[/dim]") bundle = (METHODS_DIR / "summarize-pdf" / "main.mthds").read_text() - inputs = {"document": build_document_input(document)} - main_stuff = _run(execute_pipe(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs)) + # Upload the file first (a separate step from the run) — the run request carries only its URI. + inputs = {"document": _run(upload_document_input(document))} + main_stuff, usage = _run(execute_pipe(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs)) summary = DocumentSummary.model_validate(main_stuff) output_console.print_json(data=summary.model_dump()) + print_cost_report(progress_console, usage) @app.command(name="generate-image") @@ -100,11 +105,12 @@ def generate_image( if resolved.is_sample: progress_console.print(f"[dim]No prompt given — using the sample: {resolved.text!r}. Pass your own as an argument or via --file.[/dim]") bundle = (METHODS_DIR / "generate-image" / "main.mthds").read_text() - main_stuff = _run(execute_pipe(pipe_code="generate_image", mthds_contents=[bundle], inputs={"image_prompt": resolved.text})) + main_stuff, usage = _run(execute_pipe(pipe_code="generate_image", mthds_contents=[bundle], inputs={"image_prompt": resolved.text})) # On the hosted path the runtime returns a storage `url` (`pipelex-storage://…`) # *and* a web-renderable `public_url` (a signed URL); the model keeps both. image = Image.model_validate(main_stuff) output_console.print_json(data=image.model_dump()) + print_cost_report(progress_console, usage) def _run(coro: Coroutine[Any, Any, ResultT]) -> ResultT: diff --git a/piper/detached/cli.py b/piper/detached/cli.py index cca9b11..6a89985 100644 --- a/piper/detached/cli.py +++ b/piper/detached/cli.py @@ -11,8 +11,8 @@ Same durable run as `piper attended`; the only difference is who waits. -Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py`. The three mode -packages never share lifecycle code, so the demo commands below are deliberate +Copy-paste unit: this file + `piper/inputs.py` + `piper/errors.py` + `piper/usage.py`. The +three mode packages never share lifecycle code, so the demo commands below are deliberate mirrors of their `blocking` / `attended` twins: only `start_pipe()` differs — plus the run-lifecycle commands, which exist only here. """ @@ -29,7 +29,8 @@ from rich.console import Console from piper.errors import present_error -from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, build_document_input, read_text_input +from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT, SAMPLE_INVOICE, read_text_input, upload_document_input +from piper.usage import RunUsage, print_cost_report, usage_from_results ResultT = TypeVar("ResultT") @@ -54,8 +55,8 @@ async def start_pipe(*, pipe_code: str, mthds_contents: list[str], inputs: dict[ return start_result.pipeline_run_id -async def attend_run(run_id: str) -> Any: - """Poll an already-started run to completion and resolve its main output.""" +async def attend_run(run_id: str) -> tuple[Any, RunUsage]: + """Poll an already-started run to completion; resolve its main output and its `RunUsage`.""" async with PipelexAPIClient() as client: short_id = run_id[:8] with progress_console.status(f"Run {short_id}… in progress") as status: @@ -69,7 +70,7 @@ def on_poll(info: PollInfo) -> None: # Ctrl-C: the run keeps executing server-side — nothing is lost, just wait on it again. progress_console.print(f"\nInterrupted — the run is still executing. Resume with: [bold]piper detached wait {run_id}[/bold]") raise - return results.main_stuff + return results.main_stuff, usage_from_results(results) async def fetch_run_status(run_id: str) -> RunRead: @@ -110,7 +111,8 @@ def summarize_pdf( if file is None: progress_console.print(f"[dim]No file given — using the sample: {document.name}. Pass a path to summarize your own document.[/dim]") bundle = (METHODS_DIR / "summarize-pdf" / "main.mthds").read_text() - inputs = {"document": build_document_input(document)} + # Upload the file first (a separate step from the run) — the run request carries only its URI. + inputs = {"document": _run(upload_document_input(document))} run_id = _run(start_pipe(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs)) _print_run_id(run_id) @@ -136,8 +138,9 @@ def generate_image( @app.command(name="wait") def wait(run_id: Annotated[str, typer.Argument(help="The pipeline run id printed when the run started.")]) -> None: """Poll a run to completion, then print its result.""" - main_stuff = _run(attend_run(run_id)) + main_stuff, usage = _run(attend_run(run_id)) _print_main_stuff(main_stuff) + print_cost_report(progress_console, usage) @app.command(name="status") @@ -161,6 +164,7 @@ def result(run_id: Annotated[str, typer.Argument(help="The pipeline run id print ) case RunResultCompleted(): _print_main_stuff(state.result.main_stuff) + print_cost_report(progress_console, usage_from_results(state.result)) case RunResultFailed(): progress_console.print(f"[red]Run {state.pipeline_run_id} ended with status {state.status}: {state.message}[/red]") raise typer.Exit(1) diff --git a/piper/errors.py b/piper/errors.py index 6128925..afc6405 100644 --- a/piper/errors.py +++ b/piper/errors.py @@ -20,10 +20,15 @@ from pipelex_sdk.errors import ( ApiResponseError, ApiUnreachableError, + InputPreparationError, + InvalidLocalSourceError, PipelineExecuteTimeoutError, + RejectedAssetError, RunFailedError, RunLifecycleUnavailableError, RunTimeoutError, + UnsupportedUploadCapabilityError, + UploadAuthenticationError, ) @@ -79,9 +84,28 @@ def present_error(exc: PipelineRequestError | httpx.HTTPStatusError) -> ErrorPre message=f"Gave up waiting for run {exc.run_id} after {exc.timeout_seconds:.0f}s — the run is still executing server-side.", hint=f"Resume waiting with `piper detached wait {exc.run_id}`.", ) + # File upload (summarize-pdf) preparation errors — raised before any run is created. + if isinstance(exc, InputPreparationError): + return _present_upload_error(exc) return ErrorPresentation(message=str(exc), hint=None) +def _present_upload_error(exc: InputPreparationError) -> ErrorPresentation: + """Present a file-upload (input-preparation) failure. The SDK already gives each a clear + message; here we add the actionable hint per semantic category.""" + if isinstance(exc, UnsupportedUploadCapabilityError): + hint = "File upload is a hosted capability — point PIPELEX_BASE_URL at https://api.pipelex.com (a bare runner may not serve /v1/upload)." + elif isinstance(exc, UploadAuthenticationError): + hint = "Set PIPELEX_API_KEY in your environment or .env file — get a key at https://app.pipelex.com" + elif isinstance(exc, RejectedAssetError): + hint = "The server rejected the file (usually past the service size cap) — try a smaller file." + elif isinstance(exc, InvalidLocalSourceError): + hint = "Check the path points at a readable file." + else: + hint = "Check PIPELEX_BASE_URL and that the API is reachable." + return ErrorPresentation(message=str(exc), hint=hint) + + def _present_http_status_error(exc: httpx.HTTPStatusError) -> ErrorPresentation: """Present a raw protocol-route HTTP error, reading its RFC 7807 problem+json body. diff --git a/piper/inputs.py b/piper/inputs.py index d753139..af099d7 100644 --- a/piper/inputs.py +++ b/piper/inputs.py @@ -8,22 +8,25 @@ - `read_text_input()` — a text argument, a `--file` pointing at one, or (when you give neither) a built-in sample, so every demo runs with zero arguments. -- `build_document_input()` — a local file, base64-encoded into a `data:` URL and - wrapped in the `Document` envelope the API expects: `{"concept": "Document", - "content": {"url": ..., "filename": ..., "mime_type": ...}}`. The API decodes - it server-side and uploads it to storage, so the CLI never has to host the file - itself. This mirrors `buildDocumentInput` in the JS starter's `fileEncoding.ts`. +- `upload_document_input()` — a local file, uploaded to hosted Pipelex storage with + `client.upload_file`, then wrapped in the `Document` envelope the API expects: + `{"concept": "Document", "content": {"url": , "filename": + ..., "mime_type": ...}}`. A hosted run cannot see your filesystem, so the file is + uploaded first and the run request carries only the URI, never the bytes. Preparation + is deliberately a separate step from running: a file error surfaces before any run + exists, and the uploaded asset is reusable across retries and modes. The pure + `build_document_input(path, uri)` assembles the envelope once the URI is known. The built-in samples let a fresh clone show a working result before you have any input of your own; they are also the values the README documents. """ -import base64 import mimetypes from pathlib import Path from typing import Any, NamedTuple import typer +from pipelex_sdk.client import PipelexAPIClient DEFAULT_MIME_TYPE = "application/octet-stream" @@ -67,20 +70,33 @@ def read_text_input(*, text: str | None, file: Path | None, sample: str) -> Text return TextInput(text=sample, is_sample=True) -def build_document_input(path: Path) -> dict[str, Any]: - """Read a file from disk and build its `Document` input envelope. +def build_document_input(path: Path, uri: str) -> dict[str, Any]: + """Build the `Document` input envelope for an already-uploaded asset. - The bytes are base64-encoded into a `data:` URL; the MIME type is guessed - from the extension (falling back to `application/octet-stream`). - - Raises: - FileNotFoundError: `path` does not point at a readable file. + `uri` is the `pipelex-storage://` reference `client.upload_file` returned for the file; + the envelope carries that URI (not the bytes). The MIME type is guessed from the + extension (falling back to `application/octet-stream`). Pure and offline — the upload + itself happens in `upload_document_input`. """ - data = path.read_bytes() mime_type = mimetypes.guess_type(path.name)[0] or DEFAULT_MIME_TYPE - encoded = base64.b64encode(data).decode("ascii") - data_url = f"data:{mime_type};base64,{encoded}" return { "concept": "Document", - "content": {"url": data_url, "filename": path.name, "mime_type": mime_type}, + "content": {"url": uri, "filename": path.name, "mime_type": mime_type}, } + + +async def upload_document_input(path: Path) -> dict[str, Any]: + """Upload a local document to hosted Pipelex storage and build its `Document` input envelope. + + A hosted run cannot read your filesystem, so a local file is uploaded first: + `client.upload_file` stores it and returns a `pipelex-storage://` URI, which the envelope + references — the run request never carries the file's bytes. Credentials come from + `PIPELEX_API_KEY` / `PIPELEX_BASE_URL`. + + Upload is a hosted Pipelex capability; a deployment without it raises + `UnsupportedUploadCapabilityError`, and an unreadable path raises `InvalidLocalSourceError` + (both descend from `PipelineRequestError`, so the mode's `_run()` presents them cleanly). + """ + async with PipelexAPIClient() as client: + record = await client.upload_file(path) + return build_document_input(path, record.uri) diff --git a/piper/usage.py b/piper/usage.py new file mode 100644 index 0000000..7baf72a --- /dev/null +++ b/piper/usage.py @@ -0,0 +1,110 @@ +"""Read and print what a run consumed — shared by every execution mode. + +Cost reporting is orthogonal to *how* a run is executed, so — like input encoding +(`piper/inputs.py`) and error presentation (`piper/errors.py`) — it lives here rather +than in one of the mode packages. + +Two responsibilities: + +- **Reading usage off an SDK result.** The durable modes (`attended`, `detached`) get a + typed `RunResults.tokens_usages`; the blocking mode's `client.execute` result carries + the same records only raw in `pipe_output.model_extra`, so `usage_from_execute` lifts + and validates them by hand. Both `usage_from_results` and `usage_from_execute` return a + `RunUsage`, so a mode reads cost the same way regardless of which result it holds. (When + `wip/sdk-qol/typed-tokens-usages-on-execute.md` lands upstream, `usage_from_execute` + collapses to `result.tokens_usages`.) +- **Printing a cost report.** `print_cost_report` renders a compact per-call table plus a + USD total — always to *stderr*, so stdout stays the clean, pipeable result. + +The SDK's documented usage semantics are respected here: `cost is None` (no rate table — +mock / own-GPU / dry-run) is distinct from `cost == 0` (priced at zero); a `None` record +list (usage off, broken, or pre-artifact) is distinct from `[]` (ran, no inference); +`usage_assembly_error` is the only signal that separates "broke" from "off"; and token +categories are never summed (`input_cached` is a subset of `input`). +""" + +from typing import NamedTuple + +from pipelex_sdk.execute_result import PipelexExecuteResult +from pipelex_sdk.runs import RunResults, TokensUsageRecord +from rich.console import Console +from rich.table import Table + + +class RunUsage(NamedTuple): + """A run's per-call usage records, normalized across execution modes. + + Mirrors the `RunResults` usage pair so a caller reads cost the same way whichever mode + ran: `tokens_usages` is `None` when usage assembly produced no list (off / broke / + pre-artifact) and `[]` when it ran with no inference; `usage_assembly_error` is the only + field that tells the "broke" case apart from the other two. + """ + + tokens_usages: list[TokensUsageRecord] | None + usage_assembly_error: str | None + + +def usage_from_results(results: RunResults) -> RunUsage: + """Read the usage pair off a durable `RunResults` (attended / detached) — already typed.""" + return RunUsage(tokens_usages=results.tokens_usages, usage_assembly_error=results.usage_assembly_error) + + +def usage_from_execute(result: PipelexExecuteResult) -> RunUsage: + """Lift the usage pair off a blocking `execute` result. + + Unlike `RunResults`, `PipelexExecuteResult` does not surface usage typed yet: the records + ride the execute response's extension-open `pipe_output` as raw dicts, so we read them from + `model_extra` and validate them into `TokensUsageRecord`s ourselves (mirroring the SDK's own + `_map_run_result_to_run_results`). When the typed-usage-on-execute follow-up lands + (`wip/sdk-qol/typed-tokens-usages-on-execute.md`), this collapses to `result.tokens_usages`. + """ + extras = result.pipe_output.model_extra or {} + raw = extras.get("tokens_usages") + records = [TokensUsageRecord.model_validate(item) for item in raw] if raw is not None else None + return RunUsage(tokens_usages=records, usage_assembly_error=extras.get("usage_assembly_error")) + + +def print_cost_report(console: Console, usage: RunUsage) -> None: + """Print a run's cost report to `console` (always stderr, so stdout stays the pipeable result). + + Renders a per-call table (pipe, model, tokens in→out, USD cost) and a total. The three + "no records" cases each get their own one-line note rather than an empty table: a failed + assembly, no usage reported (off / pre-artifact), and no inference calls. + """ + if usage.usage_assembly_error is not None: + console.print(f"[dim]Cost report unavailable — usage assembly failed: {usage.usage_assembly_error}[/dim]") + return + records = usage.tokens_usages + if records is None: + console.print("[dim]No usage was reported for this run.[/dim]") + return + if not records: + console.print("[dim]No inference calls — nothing to cost.[/dim]") + return + + table = Table(title="Cost report", title_justify="left", title_style="bold", show_edge=False, pad_edge=False) + table.add_column("pipe", style="cyan") + table.add_column("model") + table.add_column("tokens (in→out)", justify="right") + table.add_column("cost (USD)", justify="right") + + priced_total = 0.0 + unpriced = 0 + for record in records: + tokens = record.nb_tokens_by_category or {} + # `input` is the joined total and `output` the generated tokens — never sum the categories. + tokens_str = f"{tokens.get('input', 0)}→{tokens.get('output', 0)}" + if record.cost is None: + unpriced += 1 + cost_str = "—" + else: + priced_total += record.cost + cost_str = f"${record.cost:.4f}" + model = record.inference_model_name or record.model_type or "—" + table.add_row(record.pipe_code or "—", model, tokens_str, cost_str) + + console.print(table) + total = f"[bold]Total: ${priced_total:.4f}[/bold]" + if unpriced: + total += f" [dim](+{unpriced} unpriced: mock / own-GPU / dry-run)[/dim]" + console.print(total) diff --git a/pyproject.toml b/pyproject.toml index f14e3f1..6549ffb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "piper" -version = "0.14.1" +version = "0.15.0" description = "Replace this with your project description" # authors = [{ name = "Your Name", email = "your.email@example.com" }] license = "MIT" @@ -17,7 +17,7 @@ classifiers = [ dependencies = [ "httpx>=0.27.0", - "pipelex-sdk>=0.4.0", + "pipelex-sdk>=0.5.0", "python-dotenv>=1.0.0", "typer>=0.15.0", ] diff --git a/tests/conftest.py b/tests/conftest.py index 1b5a542..d0cb15a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,5 +1,30 @@ +import pytest from dotenv import load_dotenv +from pipelex_sdk.upload import UploadRecord +from pytest_mock import MockerFixture # Load .env so the Pipelex API client picks up PIPELEX_BASE_URL / PIPELEX_API_KEY # (the tests marked `pipelex_api` / `inference` reach the hosted API). load_dotenv() + +# The storage URI the offline `stub_upload` fixture pretends the hosted upload returned. +STUB_UPLOAD_URI = "pipelex-storage://stub-upload" + + +@pytest.fixture +def stub_upload(mocker: MockerFixture) -> str: + """Patch the hosted file upload so `summarize-pdf` CLI tests stay offline. + + `inputs.upload_document_input` opens a `PipelexAPIClient` and calls `upload_file`; here we + replace the client with a fake whose `upload_file` returns a fixed `pipelex-storage://` URI. + The real `build_document_input` still runs, so the envelope keeps the real filename / MIME + from the path — only the network upload is stubbed. Returns the stub URI for assertions. + """ + record = UploadRecord(uri=STUB_UPLOAD_URI, filename="stub", content_type="application/octet-stream", size=0) + fake_client = mocker.AsyncMock() + fake_client.upload_file.return_value = record + async_cm = mocker.MagicMock() + async_cm.__aenter__ = mocker.AsyncMock(return_value=fake_client) + async_cm.__aexit__ = mocker.AsyncMock(return_value=None) + mocker.patch("piper.inputs.PipelexAPIClient", return_value=async_cm) + return STUB_UPLOAD_URI diff --git a/tests/e2e/test_extract_entities.py b/tests/e2e/test_extract_entities.py index 789744d..f0dfa57 100644 --- a/tests/e2e/test_extract_entities.py +++ b/tests/e2e/test_extract_entities.py @@ -18,6 +18,6 @@ async def test_blocking(self): # The blocking lifecycle end to end: one `execute` call, then narrow. # Extraction finishes well under the hosted ~30s cap, so the blocking mode owns this demo. bundle = BUNDLE_PATH.read_text() - main_stuff = await execute_pipe(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": SAMPLE_TEXT}) + main_stuff, _usage = await execute_pipe(pipe_code="extract_entities", mthds_contents=[bundle], inputs={"text": SAMPLE_TEXT}) entities = ExtractedEntities.model_validate(main_stuff) assert any("Curie" in person for person in entities.people) diff --git a/tests/e2e/test_generate_image.py b/tests/e2e/test_generate_image.py index c04b3b4..54e0953 100644 --- a/tests/e2e/test_generate_image.py +++ b/tests/e2e/test_generate_image.py @@ -19,6 +19,6 @@ async def test_detached(self): bundle = BUNDLE_PATH.read_text() run_id = await start_pipe(pipe_code="generate_image", mthds_contents=[bundle], inputs={"image_prompt": SAMPLE_PROMPT}) assert run_id - main_stuff = await attend_run(run_id) + main_stuff, _usage = await attend_run(run_id) image = Image.model_validate(main_stuff) assert image.url diff --git a/tests/e2e/test_summarize_pdf.py b/tests/e2e/test_summarize_pdf.py index 599e265..5f60ab9 100644 --- a/tests/e2e/test_summarize_pdf.py +++ b/tests/e2e/test_summarize_pdf.py @@ -4,7 +4,7 @@ from piper.attended.cli import METHODS_DIR, start_and_wait from piper.generated.summarize_pdf.models import DocumentSummary -from piper.inputs import build_document_input +from piper.inputs import upload_document_input BUNDLE_PATH = METHODS_DIR / "summarize-pdf" / "main.mthds" @@ -16,10 +16,10 @@ @pytest.mark.pipelex_api class TestSummarizePdf: async def test_attended(self): - # The attended lifecycle end to end: encode the PDF, start a durable run, poll it, narrow. + # The attended lifecycle end to end: upload the PDF, start a durable run, poll it, narrow. bundle = BUNDLE_PATH.read_text() - inputs = {"document": build_document_input(SAMPLE_PDF)} - main_stuff = await start_and_wait(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs) + inputs = {"document": await upload_document_input(SAMPLE_PDF)} + main_stuff, _usage = await start_and_wait(pipe_code="summarize_pdf", mthds_contents=[bundle], inputs=inputs) summary = DocumentSummary.model_validate(main_stuff) assert summary.title assert summary.doc_type diff --git a/tests/unit/test_attended_cli.py b/tests/unit/test_attended_cli.py index 591385d..5889e8b 100644 --- a/tests/unit/test_attended_cli.py +++ b/tests/unit/test_attended_cli.py @@ -5,11 +5,15 @@ from piper.cli import app from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT +from piper.usage import RunUsage ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} SUMMARY_CONTENT = {"title": "Q3 Report", "doc_type": "report", "key_points": ["Revenue up 12%"]} IMAGE_CONTENT = {"url": "https://example.com/cat.png", "public_url": "https://cdn.example.com/cat.png"} +# The lifecycle helpers return (main_stuff, RunUsage); these offline tests don't exercise cost. +NO_USAGE = RunUsage(tokens_usages=None, usage_assembly_error=None) + runner = CliRunner() @@ -21,7 +25,7 @@ def test_help_lists_the_demos(self): assert command in result.output def test_extract_entities_starts_waits_and_prints_result(self, mocker: MockerFixture): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=ENTITIES_CONTENT) + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(ENTITIES_CONTENT, NO_USAGE)) result = runner.invoke(app, ["attended", "extract-entities", "some text"]) assert result.exit_code == 0 attended_mock.assert_awaited_once() @@ -31,7 +35,7 @@ def test_extract_entities_starts_waits_and_prints_result(self, mocker: MockerFix assert "Marie Curie" in result.output def test_extract_entities_falls_back_to_the_sample(self, mocker: MockerFixture): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=ENTITIES_CONTENT) + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(ENTITIES_CONTENT, NO_USAGE)) result = runner.invoke(app, ["attended", "extract-entities"]) assert result.exit_code == 0 assert attended_mock.await_args is not None @@ -44,7 +48,7 @@ def test_extract_entities_rejects_both_text_and_file(self, tmp_path: Path): assert result.exit_code != 0 def test_extract_entities_reads_the_file_input(self, mocker: MockerFixture, tmp_path: Path): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=ENTITIES_CONTENT) + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(ENTITIES_CONTENT, NO_USAGE)) input_file = tmp_path / "input.txt" input_file.write_text("text from a file") result = runner.invoke(app, ["attended", "extract-entities", "--file", str(input_file)]) @@ -52,21 +56,22 @@ def test_extract_entities_reads_the_file_input(self, mocker: MockerFixture, tmp_ assert attended_mock.await_args is not None assert attended_mock.await_args.kwargs["inputs"] == {"text": "text from a file"} - def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=SUMMARY_CONTENT) + def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture, stub_upload: str): + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(SUMMARY_CONTENT, NO_USAGE)) result = runner.invoke(app, ["attended", "summarize-pdf"]) assert result.exit_code == 0 assert attended_mock.await_args is not None document_input = attended_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["filename"] == "sample-invoice.pdf" + assert document_input["content"]["url"] == stub_upload def test_summarize_pdf_rejects_a_missing_file(self, tmp_path: Path): result = runner.invoke(app, ["attended", "summarize-pdf", str(tmp_path / "nope.pdf")]) assert result.exit_code != 0 - def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=SUMMARY_CONTENT) + def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path, stub_upload: str): + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(SUMMARY_CONTENT, NO_USAGE)) pdf = tmp_path / "doc.pdf" pdf.write_bytes(b"%PDF-1.4 fake") result = runner.invoke(app, ["attended", "summarize-pdf", str(pdf)]) @@ -76,17 +81,17 @@ def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, document_input = attended_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["mime_type"] == "application/pdf" - assert document_input["content"]["url"].startswith("data:application/pdf;base64,") + assert document_input["content"]["url"] == stub_upload def test_generate_image_falls_back_to_the_sample(self, mocker: MockerFixture): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=IMAGE_CONTENT) + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(IMAGE_CONTENT, NO_USAGE)) result = runner.invoke(app, ["attended", "generate-image"]) assert result.exit_code == 0 assert attended_mock.await_args is not None assert attended_mock.await_args.kwargs["inputs"] == {"image_prompt": SAMPLE_IMAGE_PROMPT} def test_generate_image_sends_the_prompt(self, mocker: MockerFixture): - attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=IMAGE_CONTENT) + attended_mock = mocker.patch("piper.attended.cli.start_and_wait", return_value=(IMAGE_CONTENT, NO_USAGE)) result = runner.invoke(app, ["attended", "generate-image", "a cat wearing a hat"]) assert result.exit_code == 0 assert "example.com/cat.png" in result.output diff --git a/tests/unit/test_blocking_cli.py b/tests/unit/test_blocking_cli.py index 366758b..a0d8c1f 100644 --- a/tests/unit/test_blocking_cli.py +++ b/tests/unit/test_blocking_cli.py @@ -5,11 +5,15 @@ from piper.cli import app from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT +from piper.usage import RunUsage ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} SUMMARY_CONTENT = {"title": "Q3 Report", "doc_type": "report", "key_points": ["Revenue up 12%"]} IMAGE_CONTENT = {"url": "https://example.com/cat.png", "public_url": "https://cdn.example.com/cat.png"} +# The lifecycle helpers return (main_stuff, RunUsage); these offline tests don't exercise cost. +NO_USAGE = RunUsage(tokens_usages=None, usage_assembly_error=None) + runner = CliRunner() @@ -21,7 +25,7 @@ def test_help_lists_the_demos(self): assert command in result.output def test_extract_entities_executes_and_prints_result(self, mocker: MockerFixture): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=ENTITIES_CONTENT) + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(ENTITIES_CONTENT, NO_USAGE)) result = runner.invoke(app, ["blocking", "extract-entities", "some text"]) assert result.exit_code == 0 execute_mock.assert_awaited_once() @@ -31,7 +35,7 @@ def test_extract_entities_executes_and_prints_result(self, mocker: MockerFixture assert "Marie Curie" in result.output def test_extract_entities_falls_back_to_the_sample(self, mocker: MockerFixture): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=ENTITIES_CONTENT) + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(ENTITIES_CONTENT, NO_USAGE)) result = runner.invoke(app, ["blocking", "extract-entities"]) assert result.exit_code == 0 assert execute_mock.await_args is not None @@ -44,7 +48,7 @@ def test_extract_entities_rejects_both_text_and_file(self, tmp_path: Path): assert result.exit_code != 0 def test_extract_entities_reads_the_file_input(self, mocker: MockerFixture, tmp_path: Path): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=ENTITIES_CONTENT) + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(ENTITIES_CONTENT, NO_USAGE)) input_file = tmp_path / "input.txt" input_file.write_text("text from a file") result = runner.invoke(app, ["blocking", "extract-entities", "--file", str(input_file)]) @@ -52,21 +56,22 @@ def test_extract_entities_reads_the_file_input(self, mocker: MockerFixture, tmp_ assert execute_mock.await_args is not None assert execute_mock.await_args.kwargs["inputs"] == {"text": "text from a file"} - def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=SUMMARY_CONTENT) + def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture, stub_upload: str): + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(SUMMARY_CONTENT, NO_USAGE)) result = runner.invoke(app, ["blocking", "summarize-pdf"]) assert result.exit_code == 0 assert execute_mock.await_args is not None document_input = execute_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["filename"] == "sample-invoice.pdf" + assert document_input["content"]["url"] == stub_upload def test_summarize_pdf_rejects_a_missing_file(self, tmp_path: Path): result = runner.invoke(app, ["blocking", "summarize-pdf", str(tmp_path / "nope.pdf")]) assert result.exit_code != 0 - def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=SUMMARY_CONTENT) + def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path, stub_upload: str): + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(SUMMARY_CONTENT, NO_USAGE)) pdf = tmp_path / "doc.pdf" pdf.write_bytes(b"%PDF-1.4 fake") result = runner.invoke(app, ["blocking", "summarize-pdf", str(pdf)]) @@ -76,17 +81,17 @@ def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, document_input = execute_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["mime_type"] == "application/pdf" - assert document_input["content"]["url"].startswith("data:application/pdf;base64,") + assert document_input["content"]["url"] == stub_upload def test_generate_image_falls_back_to_the_sample(self, mocker: MockerFixture): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=IMAGE_CONTENT) + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(IMAGE_CONTENT, NO_USAGE)) result = runner.invoke(app, ["blocking", "generate-image"]) assert result.exit_code == 0 assert execute_mock.await_args is not None assert execute_mock.await_args.kwargs["inputs"] == {"image_prompt": SAMPLE_IMAGE_PROMPT} def test_generate_image_sends_the_prompt(self, mocker: MockerFixture): - execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=IMAGE_CONTENT) + execute_mock = mocker.patch("piper.blocking.cli.execute_pipe", return_value=(IMAGE_CONTENT, NO_USAGE)) result = runner.invoke(app, ["blocking", "generate-image", "a cat wearing a hat"]) assert result.exit_code == 0 assert "example.com/cat.png" in result.output diff --git a/tests/unit/test_detached_cli.py b/tests/unit/test_detached_cli.py index d6c540d..65703db 100644 --- a/tests/unit/test_detached_cli.py +++ b/tests/unit/test_detached_cli.py @@ -6,10 +6,14 @@ from piper.cli import app from piper.inputs import SAMPLE_ENTITIES_TEXT, SAMPLE_IMAGE_PROMPT +from piper.usage import RunUsage ENTITIES_CONTENT = {"people": ["Marie Curie"], "orgs": ["University of Paris"], "dates": ["1906"]} RUN_ID = "run-abc123" +# `attend_run` returns (main_stuff, RunUsage); this offline test doesn't exercise cost. +NO_USAGE = RunUsage(tokens_usages=None, usage_assembly_error=None) + runner = CliRunner() @@ -55,7 +59,7 @@ def test_extract_entities_reads_the_file_input(self, mocker: MockerFixture, tmp_ assert start_mock.await_args is not None assert start_mock.await_args.kwargs["inputs"] == {"text": "text from a file"} - def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture): + def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixture, stub_upload: str): start_mock = mocker.patch("piper.detached.cli.start_pipe", return_value=RUN_ID) result = runner.invoke(app, ["detached", "summarize-pdf"]) assert result.exit_code == 0 @@ -63,12 +67,13 @@ def test_summarize_pdf_falls_back_to_the_sample_invoice(self, mocker: MockerFixt document_input = start_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["filename"] == "sample-invoice.pdf" + assert document_input["content"]["url"] == stub_upload def test_summarize_pdf_rejects_a_missing_file(self, tmp_path: Path): result = runner.invoke(app, ["detached", "summarize-pdf", str(tmp_path / "nope.pdf")]) assert result.exit_code != 0 - def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path): + def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, tmp_path: Path, stub_upload: str): start_mock = mocker.patch("piper.detached.cli.start_pipe", return_value=RUN_ID) pdf = tmp_path / "doc.pdf" pdf.write_bytes(b"%PDF-1.4 fake") @@ -78,7 +83,7 @@ def test_summarize_pdf_sends_the_document_envelope(self, mocker: MockerFixture, document_input = start_mock.await_args.kwargs["inputs"]["document"] assert document_input["concept"] == "Document" assert document_input["content"]["mime_type"] == "application/pdf" - assert document_input["content"]["url"].startswith("data:application/pdf;base64,") + assert document_input["content"]["url"] == stub_upload def test_generate_image_falls_back_to_the_sample(self, mocker: MockerFixture): start_mock = mocker.patch("piper.detached.cli.start_pipe", return_value=RUN_ID) @@ -97,7 +102,7 @@ def test_generate_image_sends_the_prompt(self, mocker: MockerFixture): assert result.stdout.strip() == RUN_ID def test_wait_prints_the_raw_main_stuff(self, mocker: MockerFixture): - attend_mock = mocker.patch("piper.detached.cli.attend_run", return_value=ENTITIES_CONTENT) + attend_mock = mocker.patch("piper.detached.cli.attend_run", return_value=(ENTITIES_CONTENT, NO_USAGE)) result = runner.invoke(app, ["detached", "wait", RUN_ID]) assert result.exit_code == 0 attend_mock.assert_awaited_once_with(RUN_ID) diff --git a/tests/unit/test_errors.py b/tests/unit/test_errors.py index 2bfc167..69a06a4 100644 --- a/tests/unit/test_errors.py +++ b/tests/unit/test_errors.py @@ -3,10 +3,14 @@ import httpx from pipelex_sdk.errors import ( ApiUnreachableError, + InvalidLocalSourceError, PipelineExecuteTimeoutError, + RejectedAssetError, RunFailedError, RunLifecycleUnavailableError, RunTimeoutError, + UnsupportedUploadCapabilityError, + UploadAuthenticationError, ) from pipelex_sdk.runs import RunStatus @@ -90,3 +94,25 @@ def test_run_timeout_hints_wait(self): presentation = present_error(RunTimeoutError("too slow", run_id="run-9", timeout_seconds=1200.0)) assert presentation.hint is not None assert "piper detached wait run-9" in presentation.hint + + def test_unsupported_upload_capability_hints_the_hosted_api(self): + # summarize-pdf uploads the file; a runner without /v1/upload must point at the hosted API. + presentation = present_error(UnsupportedUploadCapabilityError("no /v1/upload route here")) + assert "no /v1/upload route here" in presentation.message + assert presentation.hint is not None + assert "PIPELEX_BASE_URL" in presentation.hint + + def test_upload_authentication_hints_api_key(self): + presentation = present_error(UploadAuthenticationError("not authorized (401)", status=401)) + assert presentation.hint is not None + assert "PIPELEX_API_KEY" in presentation.hint + + def test_rejected_asset_hints_a_smaller_file(self): + presentation = present_error(RejectedAssetError("too large", filename="huge.pdf", status=413)) + assert presentation.hint is not None + assert "smaller" in presentation.hint + + def test_invalid_local_source_hints_the_path(self): + presentation = present_error(InvalidLocalSourceError("cannot read", source="/nope.pdf")) + assert presentation.hint is not None + assert "path" in presentation.hint diff --git a/tests/unit/test_inputs.py b/tests/unit/test_inputs.py index 0e11583..997f11a 100644 --- a/tests/unit/test_inputs.py +++ b/tests/unit/test_inputs.py @@ -1,10 +1,11 @@ -import base64 from pathlib import Path import pytest import typer +from pipelex_sdk.upload import UploadRecord +from pytest_mock import MockerFixture -from piper.inputs import build_document_input, read_text_input +from piper.inputs import build_document_input, read_text_input, upload_document_input class TestInputs: @@ -38,23 +39,36 @@ def test_read_text_input_falls_back_to_the_sample(self): assert resolved.is_sample is True def test_pdf_envelope(self, tmp_path: Path): + # build_document_input is pure now: it wraps an already-uploaded storage URI. pdf = tmp_path / "invoice.pdf" - payload = b"%PDF-1.4 hello" - pdf.write_bytes(payload) - envelope = build_document_input(pdf) + envelope = build_document_input(pdf, "pipelex-storage://abc123") assert envelope["concept"] == "Document" content = envelope["content"] assert content["filename"] == "invoice.pdf" assert content["mime_type"] == "application/pdf" - expected = base64.b64encode(payload).decode("ascii") - assert content["url"] == f"data:application/pdf;base64,{expected}" + assert content["url"] == "pipelex-storage://abc123" def test_unknown_extension_falls_back_to_octet_stream(self, tmp_path: Path): blob = tmp_path / "data.unknownext" - blob.write_bytes(b"\x00\x01\x02") - envelope = build_document_input(blob) + envelope = build_document_input(blob, "pipelex-storage://blob") assert envelope["content"]["mime_type"] == "application/octet-stream" - def test_missing_file_raises(self, tmp_path: Path): - with pytest.raises(FileNotFoundError): - build_document_input(tmp_path / "nope.pdf") + async def test_upload_document_input_uploads_then_wraps_the_uri(self, mocker: MockerFixture, tmp_path: Path): + # upload_document_input uploads the file, then builds the envelope around the returned URI. + pdf = tmp_path / "doc.pdf" + pdf.write_bytes(b"%PDF-1.4 fake") + record = UploadRecord(uri="pipelex-storage://uploaded", filename="doc.pdf", content_type="application/pdf", size=12) + fake_client = mocker.AsyncMock() + fake_client.upload_file.return_value = record + async_cm = mocker.MagicMock() + async_cm.__aenter__ = mocker.AsyncMock(return_value=fake_client) + async_cm.__aexit__ = mocker.AsyncMock(return_value=None) + mocker.patch("piper.inputs.PipelexAPIClient", return_value=async_cm) + + envelope = await upload_document_input(pdf) + + fake_client.upload_file.assert_awaited_once_with(pdf) + assert envelope["concept"] == "Document" + assert envelope["content"]["url"] == "pipelex-storage://uploaded" + assert envelope["content"]["filename"] == "doc.pdf" + assert envelope["content"]["mime_type"] == "application/pdf" diff --git a/tests/unit/test_usage.py b/tests/unit/test_usage.py new file mode 100644 index 0000000..e4c900e --- /dev/null +++ b/tests/unit/test_usage.py @@ -0,0 +1,99 @@ +import io +from typing import Any + +from pipelex_sdk.execute_result import PipelexExecuteResult +from pipelex_sdk.runs import RunResults, TokensUsageRecord +from rich.console import Console + +from piper.usage import RunUsage, print_cost_report, usage_from_execute, usage_from_results + + +def _render(usage: RunUsage) -> str: + buffer = io.StringIO() + # A wide, non-terminal console → plain text, no wrapping, no ANSI styling to assert around. + print_cost_report(Console(file=buffer, width=200), usage) + return buffer.getvalue() + + +def _record(**fields: Any) -> TokensUsageRecord: + return TokensUsageRecord.model_validate(fields) + + +class TestUsage: + def test_reports_records_with_a_total(self): + usage = RunUsage( + tokens_usages=[ + _record(pipe_code="extract", inference_model_name="gpt-4o", cost=0.01, nb_tokens_by_category={"input": 100, "output": 50}), + _record(pipe_code="summarize", inference_model_name="claude", cost=0.02, nb_tokens_by_category={"input": 200, "output": 80}), + ], + usage_assembly_error=None, + ) + output = _render(usage) + assert "gpt-4o" in output + assert "claude" in output + assert "100→50" in output # the joined `input` total → `output`, never a sum of categories + assert "$0.0300" in output # 0.01 + 0.02 + + def test_marks_unpriced_calls_and_excludes_them_from_the_total(self): + usage = RunUsage( + tokens_usages=[ + _record(pipe_code="gen", model_type="img_gen", cost=None, nb_tokens_by_category=None), + _record(pipe_code="extract", inference_model_name="gpt-4o", cost=0.05, nb_tokens_by_category={"input": 10, "output": 5}), + ], + usage_assembly_error=None, + ) + output = _render(usage) + assert "$0.0500" in output # the total excludes the unpriced (cost is None) call + assert "unpriced" in output + + def test_assembly_error_is_reported(self): + output = _render(RunUsage(tokens_usages=None, usage_assembly_error="event read failed")) + assert "event read failed" in output + assert "failed" in output.lower() + + def test_none_usage_says_nothing_reported(self): + # None (off / pre-artifact) is distinct from the assembly-error case above. + output = _render(RunUsage(tokens_usages=None, usage_assembly_error=None)) + assert "No usage was reported" in output + + def test_empty_usage_says_no_inference(self): + # [] (ran, but no inference happened) is distinct from None. + output = _render(RunUsage(tokens_usages=[], usage_assembly_error=None)) + assert "No inference calls" in output + + def test_usage_from_results_passes_the_pair_through(self): + records = [_record(pipe_code="p", cost=0.01)] + results = RunResults(pipeline_run_id="run-1", main_stuff={"x": 1}, tokens_usages=records, usage_assembly_error=None) + usage = usage_from_results(results) + assert usage.tokens_usages == records + assert usage.usage_assembly_error is None + + def test_usage_from_execute_lifts_raw_records_off_pipe_output(self): + result = PipelexExecuteResult.model_validate( + { + "pipeline_run_id": "run-1", + "main_stuff_name": "out", + "pipe_output": { + "pipeline_run_id": "run-1", + "working_memory": {"root": {}, "aliases": {}}, + "tokens_usages": [{"pipe_code": "p", "inference_model_name": "gpt-4o", "cost": 0.01}], + "usage_assembly_error": None, + }, + } + ) + usage = usage_from_execute(result) + assert usage.tokens_usages is not None + assert usage.tokens_usages[0].pipe_code == "p" + assert usage.tokens_usages[0].cost == 0.01 + + def test_usage_from_execute_handles_a_response_without_usage(self): + result = PipelexExecuteResult.model_validate( + { + "pipeline_run_id": "run-1", + "main_stuff_name": "out", + "pipe_output": {"pipeline_run_id": "run-1", "working_memory": {"root": {}, "aliases": {}}}, + } + ) + usage = usage_from_execute(result) + assert usage.tokens_usages is None + assert usage.usage_assembly_error is None diff --git a/uv.lock b/uv.lock index c3497a7..82a43cf 100644 --- a/uv.lock +++ b/uv.lock @@ -283,7 +283,7 @@ wheels = [ [[package]] name = "pipelex-sdk" -version = "0.4.0" +version = "0.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "httpx" }, @@ -291,9 +291,9 @@ dependencies = [ { name = "pydantic" }, { name = "typing-extensions" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/46/ee/463270b47a166001a8875af968cdce4619e5815510066eaf6be8fc6d17d6/pipelex_sdk-0.4.0.tar.gz", hash = "sha256:c11200cab8141e1d315e530dca9ef8a1fe32371c86d4f8cb56fc1fd39952188f", size = 111049, upload-time = "2026-07-06T21:37:34.046Z" } +sdist = { url = "https://files.pythonhosted.org/packages/a3/19/ae26f50b47c53c328942e0d19b2caa457b0f82e4250685f4399d26f7d476/pipelex_sdk-0.5.0.tar.gz", hash = "sha256:b83a81c2f27b729d0f995f1ef99a009dcdd5dd0c472c901c8392cc0790290d63", size = 139941, upload-time = "2026-07-22T14:54:33.223Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/0f/b5/ce14f9bbb9717a193c6e180b8c945c132d4bc53bdcd439849de456549a01/pipelex_sdk-0.4.0-py3-none-any.whl", hash = "sha256:af4f0502cf8ef592df7d5eb26c9512491d5d35d4e31721877aaea9364856f99a", size = 33166, upload-time = "2026-07-06T21:37:32.575Z" }, + { url = "https://files.pythonhosted.org/packages/eb/f5/ce84b165fc410aec2275a9e5e4c1800a9799d69c26b7f5f707bb2d754ae4/pipelex_sdk-0.5.0-py3-none-any.whl", hash = "sha256:13bfd9b280737e32b7d7227fd9a27517b9a498483e810bce0201dd27696d10cd", size = 44078, upload-time = "2026-07-22T14:54:32.006Z" }, ] [[package]] @@ -313,7 +313,7 @@ wheels = [ [[package]] name = "piper" -version = "0.14.1" +version = "0.15.0" source = { editable = "." } dependencies = [ { name = "httpx" }, @@ -338,7 +338,7 @@ dev = [ requires-dist = [ { name = "httpx", specifier = ">=0.27.0" }, { name = "mypy", marker = "extra == 'dev'", specifier = "==1.19.1" }, - { name = "pipelex-sdk", specifier = ">=0.4.0" }, + { name = "pipelex-sdk", specifier = ">=0.5.0" }, { name = "pipelex-tools", marker = "extra == 'dev'", specifier = ">=0.7.2" }, { name = "pyright", marker = "extra == 'dev'", specifier = ">=1.1.411" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.3" },