feature/Sync-client-facade · L-260907-91b052 — no asyncio.run wrapper per call site

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - **`prepare_inputs` takes the method three ways.** Beside inline `files`, it accepts a `method_ref` address (resolved by the runner) or a stored `method_id` (resolved by the hosted platform) — exactly one per call, all three server-resolved, nothing expanded client-side. Empty is absent (`files=[]`, a blank `method_ref` / `method_id`) but the wrong type is not: none, several, or a non-string selector raises `InputPreparationError` before any request leaves the process — reading a mistyped selector as absent would let the exclusivity check pass and prepare against the wrong method. A non-string `pipe_ref` is refused on the same boundary, rather than being absorbed by the pipe defaulting. A method addressed by URL that declares a file input now has an input-preparation path; previously it had none, even though the request beneath accepted the address.
 - **`PipelexValidationReport.default_pipe_ref`** — the qualified `pipe_ref` a caller gets by omitting the pipe selector, or `null` when the closure declares none or several. Optional and read leniently: a runner that predates the field sends nothing, and `prepare_inputs` falls back to the opaque `bundle_blueprint.main_pipe`.
+- **`SyncPipelexAPIClient`**: a synchronous facade over `PipelexAPIClient` (`from pipelex_sdk.sync_client import SyncPipelexAPIClient`), so a script, a batch job or a Django view calls every method without `await` or an `asyncio.run` wrapper of its own, with the same signatures, results and errors, and the paged iterators as plain iterators. It delegates to one private async client on its own event-loop thread, so one instance can be shared across threads, and a call made inside a running event loop raises the new `SyncClientInEventLoopError` instead of freezing that loop.
 
 ### Changed
 

CLAUDE.md

@@ -14,7 +14,7 @@ It is the **hosted superset**: the five normative MTHDS Protocol routes (inherit
 - **Inheritance, not re-implementation.** `class PipelexAPIClient(MthdsAPIClient)`. Reuse the base transport (`_send`, `_url`), body-builders, the reusable protocol methods, `runner_type`, and the async context-manager. Add lifecycle/product/health on top. The base's single-underscore transport methods are treated as a documented **protected extension surface** — do not rename or fork them.
 - **Brand boundary (MTHDS vs Pipelex).** MTHDS = the open standard's brand; Pipelex = the runtime/product brand. Protocol routes and their models belong to `mthds` and keep neutral names; Pipelex-specific surfaces (lifecycle, product routes, implementation envelopes) live here. Name by which brand owns the concept.
 - **Credentials.** Resolution is Pipelex-only — this SDK **never** reads the `mthds` resolver (`MTHDS_API_KEY` / `MTHDS_BASE_URL` / `~/.mthds/config`): that config is a `(base_url, api_key)` credential pair for whatever runner the vendor-neutral `mthds` tooling targets, and borrowing its key while ignoring its URL would send a foreign key to the hosted API. The **API key** resolves `api_key` arg → `PIPELEX_API_KEY` → anonymous (token is **optional**). The **base URL** resolves `base_url` arg → `PIPELEX_BASE_URL` → the hosted default `https://api.pipelex.com`. Both chains match the JS SDK exactly. `request_timeout_seconds` (constructor arg, default 20 min) sets the per-instance blocking-execute ceiling.
-- **Async-only.** httpx `AsyncClient`, `async def` throughout. No sync facade in v0.1.
+- **Async-native, with one sync facade that delegates.** `PipelexAPIClient` is httpx `AsyncClient` and `async def` throughout. `SyncPipelexAPIClient` (`pipelex_sdk/sync_client.py`) wraps it through one private event loop on a daemon thread and is never a second transport: no retry, error-mapping, paging or polling logic is restated there. A public coroutine added to `PipelexAPIClient` gets its sync twin, with the identical signature, in the same change; `tests/unit/test_sync_client_parity.py` fails otherwise.
 - **No barrel.** `__init__.py` files stay empty — no re-exports, no docstrings. Import via full paths (`from pipelex_sdk.client import PipelexAPIClient`).
 
 ## Workflow

README.md

@@ -22,7 +22,7 @@ The SDK never reads the `mthds` resolver (`MTHDS_API_KEY` / `MTHDS_BASE_URL` / `
 
 `request_timeout_seconds` (constructor argument, default 20 min) sets the per-instance blocking-execute ceiling the inherited protocol routes (`execute` / `start` / `validate` / `models` / `version`) use.
 
-The client is **async-only** (httpx `AsyncClient`) and is an async context manager.
+`PipelexAPIClient` is async (httpx `AsyncClient`) and is an async context manager. Code with no event loop of its own uses `SyncPipelexAPIClient`, the same surface without `await` (see "Synchronous callers" below).
 
 ## Quickstart
 
@@ -78,6 +78,20 @@ ack = await client.start(pipe_code="long_pipe", inputs={...})
 result = await client.wait_for_result(ack.pipeline_run_id)
 ```
 
+### Synchronous callers
+
+A script, a batch job or a Django view calls the same methods without `await`, and without an `asyncio.run` wrapper of its own, through `SyncPipelexAPIClient`:
+
+```python
+from pipelex_sdk.sync_client import SyncPipelexAPIClient
+
+with SyncPipelexAPIClient() as client:
+    result = client.start_and_wait(pipe_code="my_pipe", inputs={"topic": "quantum computing"})
+    print(result.main_stuff)
+```
+
+It delegates every call to a private `PipelexAPIClient` running on its own event-loop thread, so the signatures, results and errors are the async client's, and one instance can be shared across threads. It refuses to run inside a running event loop and raises `SyncClientInEventLoopError` there, because a blocking call would freeze that loop: async code awaits `PipelexAPIClient` instead.
+
 ### Product routes: branch on `err.code`, not the HTTP status
 
 The hosted product routes raise a typed `ApiResponseError` carrying the RFC 9457 `code` discriminant. Branch on `err.code`, which is decoupled from the transport status:
@@ -100,10 +114,11 @@ except ApiResponseError as exc:
 There is no barrel import — package `__init__.py` files stay empty. Import each symbol from its module:
 
 - **Client & construction** — `from pipelex_sdk.client import PipelexAPIClient, DEFAULT_API_BASE_URL, MthdsFile`
+- **Synchronous facade** — `from pipelex_sdk.sync_client import SyncPipelexAPIClient`
 - **Run lifecycle types** — `from pipelex_sdk.runs import RunStatus, RunPublic, RunRead, RunResults, RunResultState, WaitForResultOptions, PollInfo`
 - **Product wire models** — `from pipelex_sdk.product_models import UserProfile, MethodData, MethodWriteInput, Membership, MembershipsResponse, SubscriptionResponse, PlanView, InvoiceView, OnboardingSubmission, UploadInput, UploadedFile, PipelineRun, ...`
 - **Validation verdict types** — `from pipelex_sdk.validation_models import PipelexValidationResult, PipelexValidationReport, PipelexInvalidReport, ValidationErrorItem, SuggestedFix, VALIDATION_VIEW_INPUT_FORM, ...`
-- **Typed errors** — `from pipelex_sdk.errors import ApiResponseError, ApiUnreachableError, PipelineExecuteTimeoutError, PagingNotTerminatingError, RunFailedError, RunTimeoutError, RunLifecycleUnavailableError, RunStillRunningError, ...`
+- **Typed errors** — `from pipelex_sdk.errors import ApiResponseError, ApiUnreachableError, PipelineExecuteTimeoutError, PagingNotTerminatingError, RunFailedError, RunTimeoutError, RunLifecycleUnavailableError, RunStillRunningError, SyncClientInEventLoopError, ...`
 - **Version** — `from pipelex_sdk.version import __version__`
 - **Protocol surface** (the MTHDS standard's wire types) comes from the `mthds` dependency — e.g. `from mthds.protocol.exceptions import PipelineRequestError`, `from mthds.protocol.models import ValidationResult` (the neutral verdict union that `PipelexValidationResult` narrows).
 - **Input-form descriptors and pipe I/O contracts** come from `mthds` too, because they are the standard's artifacts and this SDK only carries them: `from mthds.protocol.input_form import InputForm, InputFormField, ListField, TextField, ...` and `from mthds.protocol.pipe_io_contracts import PipeIOContracts, PipeInputContract, PresenceMarker, IOMultiplicity, ...`. `PipelexValidationReport.input_form` and `.pipe_io_contracts` are typed with them, so a node narrows on its `kind` and a slot's presence and multiplicity read as enums — but `pipelex_sdk` does not re-export the vocabulary, and importing it from here is the one supported path.

docs/architecture.md

@@ -46,7 +46,7 @@ A token is **optional** (anonymous access is allowed; protocol routes work again
 
 ## Conventions
 
-- **Async-only** — httpx `AsyncClient`, `async def` throughout. No sync facade in v0.1.
+- **Async-native, with a synchronous facade** — httpx `AsyncClient` and `async def` throughout `PipelexAPIClient`; `SyncPipelexAPIClient` wraps it for callers with no event loop of their own (see "Synchronous facade" below).
 - **No barrel** — package `__init__.py` files stay empty; consumers import via full paths (`from pipelex_sdk.client import PipelexAPIClient`). The public import paths are documented in the README.
 - **Wire format** — snake_case JSON fields on Pydantic v2 models.
 
@@ -214,12 +214,21 @@ The wire models are snake_case Pydantic v2. Response models are extension-open (
 
 `health()` → `GET {origin}/health`. The one route served at the **origin**, NOT under the `/v1` prefix — the origin is derived from the base URL (`_origin_of`, exposed as `self.origin_url`), so a base URL of `https://api.pipelex.com/v1/...` still probes `https://api.pipelex.com/health`. It is **out-of-protocol**: the MTHDS Protocol defines no health route, and `/health` is neither a protocol nor a product surface. It rides `_request_json` (the plainer regime), so a non-2xx raises `PipelineRequestError` rather than the product `ApiResponseError` — liveness needs no `code` taxonomy. Transport failures still map to `ApiUnreachableError`. (Checkpoint-5 decision: kept the plainer regime — see "Error regimes" above.)
 
+## Synchronous facade
+
+`SyncPipelexAPIClient` (`pipelex_sdk/sync_client.py`) is the same surface without `await`, for a script, a batch job, a Django view, or any other caller with no event loop of its own. It wraps and delegates: every public method calls the same-named coroutine on one private `PipelexAPIClient`, so there is one HTTP implementation and none of the retry, error-mapping, paging or polling logic is restated. Each method keeps its twin's signature, result and errors, and `iterate_methods` / `iterate_runs` come back as plain `Iterator`s. `tests/unit/test_sync_client_parity.py` compares the two classes signature by signature in both directions, so a coroutine added to the async client without its twin fails the suite. The facade has no JavaScript counterpart, because Node has no synchronous half to serve; it is ergonomics, not parity.
+
+- **One private loop on a daemon thread.** The loop starts at the first call and lives until `close()`. Calling `asyncio.run` once per call was rejected: the wrapped client's httpx connection pool belongs to the loop that opened it, so a fresh loop per call breaks the pool or rebuilds it every time. Driving a long-lived loop from the caller's thread with `asyncio.Runner` was rejected too, because a facade shared across threads would then queue every call behind the one in flight, a twenty-minute `start_and_wait` included. With the loop on its own thread, calls from several threads run concurrently, and one instance can be shared.
+- **A caller inside a running event loop is refused.** It gets `SyncClientInEventLoopError` before anything is sent. The call would work mechanically, but it would block that loop for the whole request and freeze everything else scheduled on it; async code awaits `PipelexAPIClient` instead. A Jupyter notebook is such a caller, since its kernel runs every cell inside its own event loop. The error is a `RuntimeError`, the class `asyncio.run()` raises for the same mistake, rather than a `PipelineRequestError`, so a handler written for API failures does not swallow a programming error. A `WaitForResultOptions.on_poll` callback runs on the facade's loop, so it falls under the same rule.
+- **Close and reuse.** `close()`, or leaving the `with` block, cancels any call still in flight on another thread (it raises `concurrent.futures.CancelledError` there), closes the HTTP client, finalizes async generators left part-way and joins the thread. It holds the facade's lock until all of that is done, so a call made on another thread meanwhile waits and then starts the fresh loop, rather than reaching the HTTP client the old loop is still closing. A closed facade is reusable and starts a fresh loop at its next call. A paged iterator left open across the close raises `concurrent.futures.CancelledError` at its next item instead of ending early, because the close finalized the async generator beneath it. Closing a sync iterator, or dropping it after a `break`, finalizes that generator. A facade that is never closed keeps its loop thread until the process exits.
+- **Interrupts and exits.** Ctrl+C while a call is blocked, a paged iterator's page fetch included, cancels the coroutine on the loop and re-raises in the caller. As on the async client, that stops the waiting, never the run, which keeps executing server-side. A `KeyboardInterrupt` or `SystemExit` raised on the loop itself, by a `WaitForResultOptions.on_poll` callback for instance, is carried back to the caller the same way; left alone, asyncio would let it stop the loop and leave the caller blocked for good. An iterator still open when the interpreter exits is not finalized, since the daemon loop thread has already stopped by then.
+- **Processes.** The loop thread does not survive a `fork`. A facade that has made a call must not be used in a forked child, whose call would wait on a loop no thread runs, so a pre-forking server (a preloading gunicorn, for example) should make a facade's first call in the worker rather than in the parent.
+
 ## Out of scope
 
 - The `/v1/build/*` helpers — `build_output`, `build_runner`, `concept`, `pipe_spec`. `build_inputs` shipped in 0.5.0 and was removed again once `prepare_inputs`, its only caller, moved onto `validate` + the input-form descriptor: this SDK no longer touches `/v1/build/*`, which the workspace is retiring (`wip/build-retirement/`).
 - Organization *switch* (a WorkOS session operation, not a `/v1` route).
 - A `~/.pipelex/config` file reader (env-only for now, matching the JS SDK).
-- A synchronous client facade.
 
 ## Versioning
 
@@ -249,4 +258,4 @@ Each stays deferred rather than silently missing. Everything else — the protoc
 
 **Conscious exclusions:** the surfaces listed at the top of this section, and the organization *switch* (a WorkOS session op, not a `/v1` route).
 
-**Intentional divergences from the JS SDK** (Python house style / clean inheritance): no barrel (`__init__.py` stays empty; import via full paths); inheritance on `MthdsAPIClient` rather than the JS composition-of-types; async-only; `__version__` derived from installed metadata rather than a hand-synced constant.
+**Intentional divergences from the JS SDK** (Python house style / clean inheritance): no barrel (`__init__.py` stays empty; import via full paths); inheritance on `MthdsAPIClient` rather than the JS composition-of-types; a synchronous facade (`SyncPipelexAPIClient`) with no JS counterpart; `__version__` derived from installed metadata rather than a hand-synced constant.

pipelex_sdk/errors.py

@@ -15,6 +15,11 @@
 - `PagingNotTerminatingError` — a paged-list iterator hit its runaway backstop, meaning
   the server never stopped handing out cursors.
 
+`SyncClientInEventLoopError` is the one exception to that lineage: it is a usage error
+(`SyncPipelexAPIClient` called where a blocking call would freeze a running event loop), so
+it derives from `RuntimeError`, the class `asyncio.run()` raises for the same mistake, and a
+handler written for API failures does not swallow it.
+
 The run-lifecycle errors (`RunFailedError`, `RunTimeoutError`,
 `RunLifecycleUnavailableError`) are owned here (ported from `mthds-python` in
 HANDOFF Phase 2, and removed from `mthds-python` in Phase 6). `RunStillRunningError`
@@ -229,3 +234,13 @@ def __init__(self, message: str, status: int) -> None:
 
 class UploadTransportError(InputPreparationError):
     """A network or server fault reaching the upload route (unreachable host, `5xx`)."""
+
+
+class SyncClientInEventLoopError(RuntimeError):
+    """Raised when `SyncPipelexAPIClient` is called from a thread that runs an event loop.
+
+    The call would block that loop for the whole request, freezing everything else scheduled
+    on it, so it is refused before anything is sent. An async caller wants `PipelexAPIClient`
+    and `await`. Deliberately not a `PipelineRequestError`: it is a programming error, not a
+    failed request.
+    """