From fb210c06e9d5bc219d2fd5da5302b0f52bf6a637 Mon Sep 17 00:00:00 2001 From: Louis Choquel Date: Sat, 12 Sep 2026 22:45:28 +0200 Subject: [PATCH 1/3] feat: add SyncPipelexAPIClient, a synchronous facade over PipelexAPIClient A script, a batch job or a Django view can now call every client method without await and without writing its own asyncio.run wrapper. The facade wraps and delegates to one private PipelexAPIClient running on a daemon event-loop thread, so there is still exactly one HTTP implementation; one instance is shareable across threads, and a call from inside a running event loop raises SyncClientInEventLoopError before anything is sent. A parity test pins every public coroutine's sync twin to an identical signature. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 1 + CLAUDE.md | 2 +- README.md | 19 +- docs/architecture.md | 15 +- pipelex_sdk/errors.py | 15 + pipelex_sdk/sync_client.py | 537 ++++++++++++++++++++++++++ tests/unit/conftest.py | 11 + tests/unit/test_sync_client.py | 285 ++++++++++++++ tests/unit/test_sync_client_parity.py | 74 ++++ 9 files changed, 953 insertions(+), 6 deletions(-) create mode 100644 pipelex_sdk/sync_client.py create mode 100644 tests/unit/test_sync_client.py create mode 100644 tests/unit/test_sync_client_parity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e0c392b..67b6eb2 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CLAUDE.md b/CLAUDE.md index 668a501..f532aa1 100644 --- a/CLAUDE.md +++ b/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 diff --git a/README.md b/README.md index 6bdfcc1..445a051 100644 --- a/README.md +++ b/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. diff --git a/docs/architecture.md b/docs/architecture.md index 02be059..d7f23dd 100644 --- a/docs/architecture.md +++ b/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. 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. A closed facade is reusable and starts a fresh loop at its next call. Breaking out of a sync iterator finalizes the async generator beneath it at once. +- **Interrupts.** Ctrl+C while a call is blocked 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 and stays resumable by id. +- **Processes.** The loop thread does not survive a `fork`, 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. diff --git a/pipelex_sdk/errors.py b/pipelex_sdk/errors.py index 375f31a..8cde5ea 100644 --- a/pipelex_sdk/errors.py +++ b/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. + """ diff --git a/pipelex_sdk/sync_client.py b/pipelex_sdk/sync_client.py new file mode 100644 index 0000000..f59dad0 --- /dev/null +++ b/pipelex_sdk/sync_client.py @@ -0,0 +1,537 @@ +"""`SyncPipelexAPIClient` — a synchronous facade over `PipelexAPIClient`. + +For the callers that have no event loop of their own — a script, a batch job, a Django view, +a notebook cell run without top-level `await` — so none of them writes an `asyncio.run` +wrapper around each call. + +The facade wraps and delegates; it is not a second transport. Every public method calls the +same-named coroutine on one private `PipelexAPIClient`, so there is exactly one HTTP +implementation, and the retry, error-mapping, paging and polling logic is never restated. +Each method has the same signature as its async twin, returns what the coroutine returns, and +raises what it raises; the two paged iterators come back as plain `Iterator`s. A unit test +pins the one-to-one signature match, so a coroutine added to the async client without a twin +here fails the suite. + +How the calls run. The facade owns one event loop, started lazily on a daemon thread at the +first call and kept until `close()`. `asyncio.run` per call is the obvious version and the +wrong one: the wrapped client's httpx connection pool belongs to the loop that opened it, so a +fresh loop per call either breaks on the second call or reopens the pool every time. A single +long-lived loop keeps the pool warm. Running it on its own thread, rather than driving it from +the caller's thread with `asyncio.Runner`, is what makes one facade shareable across threads: +calls from several threads run concurrently on the loop instead of queueing behind each +other, which matters when one of them is a twenty-minute `start_and_wait`. + +A caller that already runs an event loop 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; such a caller wants `PipelexAPIClient` and `await`. + +Interrupting a blocked call (Ctrl+C) cancels the coroutine on the facade's loop and re-raises +in the caller. As on the async client, cancelling a `wait_for_result` or `start_and_wait` stops +the waiting, never the run itself, which keeps executing server-side and stays resumable by id. +""" + +from __future__ import annotations + +import asyncio +import threading +from collections.abc import AsyncGenerator +from typing import TYPE_CHECKING, Any, Self, TypeVar, cast + +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.errors import SyncClientInEventLoopError + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Coroutine, Iterator + + from mthds.protocol.models import ModelCategory, ModelDeck, VersionInfo + from mthds.protocol.pipe_output import VariableMultiplicity + from mthds.protocol.pipeline_inputs import PipelineInputs + from mthds.protocol.stuff import StuffType + from mthds.protocol.working_memory import WorkingMemoryAbstract + + from pipelex_sdk.client import MthdsFile + from pipelex_sdk.crate_models import CodegenRequest, CodegenResponse, MthdsFileItem, ResolveRequest, ResolveResponse + from pipelex_sdk.execute_result import PipelexExecuteResult + from pipelex_sdk.prepare_inputs import PreparedInputs + from pipelex_sdk.product_models import ( + BillingPortalResponse, + ChangePlanResponse, + CheckoutResponse, + GatewayApiKey, + GatewayApiKeyStatus, + InvoiceView, + Membership, + MembershipsResponse, + MethodData, + MethodDeletionAccepted, + MethodPage, + MethodSummary, + MethodWriteInput, + OnboardingSubmission, + PipelexApiKeyCreated, + PipelexApiKeyList, + PipelineRun, + PlanView, + ResolvedStorageUrl, + RunDetail, + RunPage, + SubscriptionResponse, + UpdateRunInput, + UploadedFile, + UploadInput, + UserProfile, + ) + from pipelex_sdk.runs import PipelexRunResultStart, RunRead, RunResults, RunResultState, WaitForResultOptions + from pipelex_sdk.upload import UploadRecord, UploadSource + from pipelex_sdk.validation_models import PipelexValidationResult + +_ResultT = TypeVar("_ResultT") +_ItemT = TypeVar("_ItemT") + +_LOOP_THREAD_NAME = "pipelex-sdk-sync-client" + + +class SyncPipelexAPIClient: + """Synchronous facade over `PipelexAPIClient` — the same surface, without `await`. + + Construction takes the same arguments and resolves credentials exactly as + `PipelexAPIClient` does (it builds one). Use it as a context manager, or call `close()` + when done; a facade is reusable after `close()`, restarting its loop at the next call. + One instance may be shared across threads. It must not be called from a thread that runs + an event loop (`SyncClientInEventLoopError`), which includes a `WaitForResultOptions.on_poll` + callback: that callback runs on the facade's own loop. + + The semantics of every method, its errors included, are those of the same-named method on + `PipelexAPIClient`, which documents them. + """ + + def __init__(self, api_key: str | None = None, base_url: str | None = None, request_timeout_seconds: float | None = None) -> None: + self._async_client = PipelexAPIClient(api_key=api_key, base_url=base_url, request_timeout_seconds=request_timeout_seconds) + # Guards the loop's lifecycle: starting it, detaching it at close, and submitting to it. + # Submission happens under the lock so a call can never land on a loop `close()` has + # already detached; only the blocking wait for the result happens outside it. + self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._loop_thread: threading.Thread | None = None + + @property + def base_url(self) -> str: + return self._async_client.base_url + + @property + def origin_url(self) -> str: + return self._async_client.origin_url + + @property + def request_timeout_seconds(self) -> float: + return self._async_client.request_timeout_seconds + + # ── Lifecycle ────────────────────────────────────────────────────── + + def __enter__(self) -> Self: + return self + + def __exit__(self, *exc_info: object) -> None: + self.close() + + def close(self) -> None: + """Close the HTTP client and stop the facade's event loop. + + A call still in flight on another thread is cancelled and raises + `concurrent.futures.CancelledError` there. Closing a facade that never made a call, or + closing it twice, does nothing. + + Raises: + SyncClientInEventLoopError: If called from a thread that runs an event loop. + """ + if _inside_running_event_loop(): + raise SyncClientInEventLoopError(_in_event_loop_message("SyncPipelexAPIClient.close")) + with self._lock: + loop = self._loop + loop_thread = self._loop_thread + self._loop = None + self._loop_thread = None + if loop is None or loop_thread is None: + return + try: + asyncio.run_coroutine_threadsafe(self._shutdown(), loop).result() + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join() + loop.close() + + async def _shutdown(self) -> None: + """Runs on the facade's loop: cancel what is in flight, then release every resource.""" + current_task = asyncio.current_task() + in_flight = [task for task in asyncio.all_tasks() if task is not current_task] + for task in in_flight: + task.cancel() + await asyncio.gather(*in_flight, return_exceptions=True) + await self._async_client.close() + await asyncio.get_running_loop().shutdown_asyncgens() + + def _ensure_loop(self) -> asyncio.AbstractEventLoop: + """Return the facade's loop, starting it on its daemon thread if needed. Caller holds the lock.""" + if self._loop is not None: + return self._loop + loop = asyncio.new_event_loop() + loop_thread = threading.Thread(target=_run_loop_forever, args=(loop,), name=_LOOP_THREAD_NAME, daemon=True) + loop_thread.start() + self._loop = loop + self._loop_thread = loop_thread + return loop + + def _run(self, coroutine: Coroutine[Any, Any, _ResultT]) -> _ResultT: + """Run one coroutine of the wrapped client on the facade's loop and block for its result.""" + if _inside_running_event_loop(): + call_name = str(getattr(coroutine, "__qualname__", "PipelexAPIClient call")) + # Close it rather than drop it: an unawaited coroutine would otherwise surface as a + # "never awaited" RuntimeWarning pointing away from the real mistake. + coroutine.close() + raise SyncClientInEventLoopError(_in_event_loop_message(call_name)) + with self._lock: + future = asyncio.run_coroutine_threadsafe(coroutine, self._ensure_loop()) + try: + return future.result() + except BaseException: + # Reached on an interrupt (KeyboardInterrupt) while blocked, as well as on a normal + # error; cancelling an already finished future is a no-op, so this only stops work + # that is still running. + future.cancel() + raise + + def _iterate(self, iterator: AsyncIterator[_ItemT]) -> Iterator[_ItemT]: + """Drive an async iterator of the wrapped client one item per round trip to the loop.""" + try: + while True: + next_item = self._run(_next_item(iterator)) + if not next_item: + return + yield next_item[0] + finally: + self._close_async_generator(iterator) + + def _close_async_generator(self, iterator: AsyncIterator[_ItemT]) -> None: + """Finalize an async generator left part-way, e.g. by a `break` out of the sync loop. + + Skipped when there is no loop any more (`close()` already finalized every async + generator on it) or when this runs inside an event loop (a blocking call is not allowed + there; the loop finalizes the generator at shutdown instead). + """ + if not isinstance(iterator, AsyncGenerator) or _inside_running_event_loop(): + return + # The wrapped iterators are async generators that are only ever advanced, never sent to. + generator = cast("AsyncGenerator[_ItemT, None]", iterator) + with self._lock: + loop = self._loop + if loop is None: + return + future = asyncio.run_coroutine_threadsafe(_close_generator(generator), loop) + future.result() + + # ── Protocol routes ──────────────────────────────────────────────── + + def execute( + self, + pipe_code: str | None = None, + mthds_contents: list[str] | None = None, + inputs: PipelineInputs | WorkingMemoryAbstract[StuffType] | None = None, + output_name: str | None = None, + output_multiplicity: VariableMultiplicity | None = None, + dynamic_output_concept_ref: str | None = None, + extra: dict[str, Any] | None = None, + *, + method_ref: str | None = None, + method_id: str | None = None, + ) -> PipelexExecuteResult: + return self._run( + self._async_client.execute( + pipe_code=pipe_code, + mthds_contents=mthds_contents, + inputs=inputs, + output_name=output_name, + output_multiplicity=output_multiplicity, + dynamic_output_concept_ref=dynamic_output_concept_ref, + extra=extra, + method_ref=method_ref, + method_id=method_id, + ) + ) + + def start( + self, + pipe_code: str | None = None, + mthds_contents: list[str] | None = None, + inputs: PipelineInputs | WorkingMemoryAbstract[StuffType] | None = None, + output_name: str | None = None, + output_multiplicity: VariableMultiplicity | None = None, + dynamic_output_concept_ref: str | None = None, + extra: dict[str, Any] | None = None, + *, + method_ref: str | None = None, + method_id: str | None = None, + ) -> PipelexRunResultStart: + return self._run( + self._async_client.start( + pipe_code=pipe_code, + mthds_contents=mthds_contents, + inputs=inputs, + output_name=output_name, + output_multiplicity=output_multiplicity, + dynamic_output_concept_ref=dynamic_output_concept_ref, + extra=extra, + method_ref=method_ref, + method_id=method_id, + ) + ) + + def validate( + self, + mthds_contents: list[str] | None = None, + allow_signatures: bool = False, + mthds_sources: list[str] | None = None, + render: list[str] | None = None, + views: list[str] | None = None, + *, + method_ref: str | None = None, + method_id: str | None = None, + ) -> PipelexValidationResult: + return self._run( + self._async_client.validate( + mthds_contents=mthds_contents, + allow_signatures=allow_signatures, + mthds_sources=mthds_sources, + render=render, + views=views, + method_ref=method_ref, + method_id=method_id, + ) + ) + + def validate_files( + self, + files: list[MthdsFile], + allow_signatures: bool = False, + render: list[str] | None = None, + views: list[str] | None = None, + ) -> PipelexValidationResult: + return self._run(self._async_client.validate_files(files=files, allow_signatures=allow_signatures, render=render, views=views)) + + def models(self, category: ModelCategory | None = None) -> ModelDeck: + return self._run(self._async_client.models(category=category)) + + def version(self) -> VersionInfo: + return self._run(self._async_client.version()) + + # ── Durable run lifecycle ────────────────────────────────────────── + + def get_run_status(self, run_id: str) -> RunRead: + return self._run(self._async_client.get_run_status(run_id)) + + def get_run_result(self, run_id: str) -> RunResultState: + return self._run(self._async_client.get_run_result(run_id)) + + def wait_for_result(self, run_id: str, options: WaitForResultOptions | None = None) -> RunResults: + return self._run(self._async_client.wait_for_result(run_id, options=options)) + + def start_and_wait( + self, + pipe_code: str | None = None, + mthds_contents: list[str] | None = None, + inputs: PipelineInputs | WorkingMemoryAbstract[StuffType] | None = None, + output_name: str | None = None, + output_multiplicity: VariableMultiplicity | None = None, + dynamic_output_concept_ref: str | None = None, + extra: dict[str, Any] | None = None, + wait_options: WaitForResultOptions | None = None, + *, + method_ref: str | None = None, + method_id: str | None = None, + ) -> RunResults: + return self._run( + self._async_client.start_and_wait( + pipe_code=pipe_code, + mthds_contents=mthds_contents, + inputs=inputs, + output_name=output_name, + output_multiplicity=output_multiplicity, + dynamic_output_concept_ref=dynamic_output_concept_ref, + extra=extra, + wait_options=wait_options, + method_ref=method_ref, + method_id=method_id, + ) + ) + + # ── Product surface ──────────────────────────────────────────────── + + def get_me(self) -> UserProfile: + return self._run(self._async_client.get_me()) + + def list_methods(self, *, q: str | None = None, limit: int | None = None, cursor: str | None = None) -> MethodPage: + return self._run(self._async_client.list_methods(q=q, limit=limit, cursor=cursor)) + + def iterate_methods(self, *, q: str | None = None, limit: int | None = None) -> Iterator[MethodSummary]: + return self._iterate(self._async_client.iterate_methods(q=q, limit=limit)) + + def get_method(self, method_id: str) -> MethodData: + return self._run(self._async_client.get_method(method_id)) + + def create_method(self, write_input: MethodWriteInput) -> MethodData: + return self._run(self._async_client.create_method(write_input)) + + def update_method(self, method_id: str, write_input: MethodWriteInput) -> MethodData: + return self._run(self._async_client.update_method(method_id, write_input)) + + def delete_method(self, method_id: str) -> MethodDeletionAccepted: + return self._run(self._async_client.delete_method(method_id)) + + def list_memberships(self) -> MembershipsResponse: + return self._run(self._async_client.list_memberships()) + + def create_organization(self, name: str) -> Membership: + return self._run(self._async_client.create_organization(name)) + + def rename_organization(self, org_id: str, name: str) -> Membership: + return self._run(self._async_client.rename_organization(org_id, name)) + + def get_subscription(self) -> SubscriptionResponse: + return self._run(self._async_client.get_subscription()) + + def list_plans(self) -> list[PlanView]: + return self._run(self._async_client.list_plans()) + + def list_invoices(self) -> list[InvoiceView]: + return self._run(self._async_client.list_invoices()) + + def create_checkout(self, plan: str) -> CheckoutResponse: + return self._run(self._async_client.create_checkout(plan)) + + def change_plan(self, plan: str) -> ChangePlanResponse: + return self._run(self._async_client.change_plan(plan)) + + def get_billing_portal(self) -> BillingPortalResponse: + return self._run(self._async_client.get_billing_portal()) + + def list_pipelex_api_keys(self) -> PipelexApiKeyList: + return self._run(self._async_client.list_pipelex_api_keys()) + + def create_pipelex_api_key(self, label: str) -> PipelexApiKeyCreated: + return self._run(self._async_client.create_pipelex_api_key(label)) + + def revoke_pipelex_api_key(self, key_id: str) -> None: + self._run(self._async_client.revoke_pipelex_api_key(key_id)) + + def rotate_pipelex_api_key(self, key_id: str) -> PipelexApiKeyCreated: + return self._run(self._async_client.rotate_pipelex_api_key(key_id)) + + def create_gateway_api_key(self, promo_code: str | None) -> GatewayApiKey: + return self._run(self._async_client.create_gateway_api_key(promo_code)) + + def get_gateway_api_key(self) -> GatewayApiKeyStatus: + return self._run(self._async_client.get_gateway_api_key()) + + def submit_onboarding(self, submission: OnboardingSubmission) -> None: + self._run(self._async_client.submit_onboarding(submission)) + + def resolve_storage_url(self, uri: str) -> ResolvedStorageUrl: + return self._run(self._async_client.resolve_storage_url(uri)) + + def upload(self, upload_input: UploadInput) -> UploadedFile: + return self._run(self._async_client.upload(upload_input)) + + # ── Crate routes ─────────────────────────────────────────────────── + + def resolve(self, request: ResolveRequest) -> ResolveResponse: + return self._run(self._async_client.resolve(request)) + + def codegen(self, request: CodegenRequest) -> CodegenResponse: + return self._run(self._async_client.codegen(request)) + + # ── Input preparation ────────────────────────────────────────────── + + def upload_file(self, source: UploadSource, *, filename: str | None = None, content_type: str | None = None) -> UploadRecord: + return self._run(self._async_client.upload_file(source, filename=filename, content_type=content_type)) + + def prepare_inputs( + self, + *, + files: list[MthdsFileItem] | None = None, + method_ref: str | None = None, + method_id: str | None = None, + pipe_ref: str | None = None, + inputs: dict[str, Any], + ) -> PreparedInputs: + return self._run(self._async_client.prepare_inputs(files=files, method_ref=method_ref, method_id=method_id, pipe_ref=pipe_ref, inputs=inputs)) + + # ── Run records ──────────────────────────────────────────────────── + + def list_runs( + self, + method_id: str, + *, + created_from: str | None = None, + created_to: str | None = None, + limit: int | None = None, + cursor: str | None = None, + ) -> RunPage: + return self._run(self._async_client.list_runs(method_id, created_from=created_from, created_to=created_to, limit=limit, cursor=cursor)) + + def iterate_runs( + self, + method_id: str, + *, + created_from: str | None = None, + created_to: str | None = None, + limit: int | None = None, + ) -> Iterator[PipelineRun]: + return self._iterate(self._async_client.iterate_runs(method_id, created_from=created_from, created_to=created_to, limit=limit)) + + def get_run_detail(self, run_id: str) -> RunDetail: + return self._run(self._async_client.get_run_detail(run_id)) + + def update_run(self, run_id: str, update_input: UpdateRunInput) -> None: + self._run(self._async_client.update_run(run_id, update_input)) + + # ── Health ───────────────────────────────────────────────────────── + + def health(self) -> dict[str, Any]: + return self._run(self._async_client.health()) + + +def _run_loop_forever(loop: asyncio.AbstractEventLoop) -> None: + """The facade's loop thread: own the loop until `close()` stops it.""" + asyncio.set_event_loop(loop) + loop.run_forever() + + +def _inside_running_event_loop() -> bool: + try: + asyncio.get_running_loop() + except RuntimeError: + return False + return True + + +def _in_event_loop_message(call_name: str) -> str: + return ( + f"{call_name}() was called from a thread that is running an event loop, where a blocking call would freeze " + "that loop for the whole request. Use PipelexAPIClient and await the call there instead." + ) + + +async def _next_item(iterator: AsyncIterator[_ItemT]) -> list[_ItemT]: + """The iterator's next item in a one-element list, or an empty list once it is exhausted. + + Exhaustion is returned rather than raised because `StopAsyncIteration` is not an exception + to carry across the thread boundary: the sync generator turns the empty list into its own + ordinary return. + """ + try: + return [await anext(iterator)] + except StopAsyncIteration: + return [] + + +async def _close_generator(generator: AsyncGenerator[Any, Any]) -> None: + await generator.aclose() diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index 84dde2e..20be09a 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -12,8 +12,11 @@ import pytest from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.sync_client import SyncPipelexAPIClient if TYPE_CHECKING: + from collections.abc import Iterator + from pytest_mock import MockerFixture, MockType BASE_URL = "http://localhost:8081" @@ -36,6 +39,14 @@ def api_client() -> PipelexAPIClient: return PipelexAPIClient(api_key="test-token", base_url=BASE_URL) +@pytest.fixture +def sync_api_client() -> Iterator[SyncPipelexAPIClient]: + """A sync facade over a test client, closed at teardown so no loop thread outlives the test.""" + client = SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) + yield client + client.close() + + @pytest.fixture def wire_response() -> ResponseBuilder: def _build(status_code: int, *, json_body: object | None = None) -> httpx.Response: diff --git a/tests/unit/test_sync_client.py b/tests/unit/test_sync_client.py new file mode 100644 index 0000000..a713df3 --- /dev/null +++ b/tests/unit/test_sync_client.py @@ -0,0 +1,285 @@ +"""Tests for `SyncPipelexAPIClient` — the synchronous facade over `PipelexAPIClient`. + +What is pinned here is the facade's own machinery, not the routes: calls run on one private +loop on one other thread, arguments and errors cross that boundary unchanged, several threads +share a facade without queueing, a caller inside an event loop is refused before anything is +sent, the paged iterators finalize when abandoned, and `close()` releases everything and leaves +the facade reusable. The one-to-one signature match with the async client is +`test_sync_client_parity.py`. +""" + +from __future__ import annotations + +import asyncio +import concurrent.futures +import gc +import threading +import warnings +from typing import TYPE_CHECKING, Any, cast + +import pytest +from mthds.protocol.exceptions import PipelineRequestError + +from pipelex_sdk.errors import RunTimeoutError, SyncClientInEventLoopError +from pipelex_sdk.runs import WaitForResultOptions +from pipelex_sdk.sync_client import SyncPipelexAPIClient +from tests.unit.conftest import BASE_URL + +if TYPE_CHECKING: + from collections.abc import AsyncIterator, Generator + + from pytest_mock import MockerFixture + + from tests.unit.conftest import ResponseBuilder, SendPatcher + +# How long a test waits on a cross-thread handshake before failing instead of hanging. +_HANDSHAKE_TIMEOUT_SECONDS = 5.0 + + +def _method(method_id: str) -> dict[str, Any]: + return {"method_id": method_id, "name": f"Method {method_id}", "created_at": "t"} + + +class TestSyncClient: + def test_a_call_runs_through_the_real_transport( + self, + sync_api_client: SyncPipelexAPIClient, + wire_response: ResponseBuilder, + patch_send: SendPatcher, + ) -> None: + send = patch_send(sync_api_client._async_client, wire_response(200, json_body={"status": "ok"})) + + assert sync_api_client.health() == {"status": "ok"} + assert send.await_count == 1 + assert send.call_args.args == ("GET", f"{BASE_URL}/health") + + def test_arguments_reach_the_async_twin_unchanged(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + sentinel = object() + start_and_wait = mocker.patch.object(sync_api_client._async_client, "start_and_wait", mocker.AsyncMock(return_value=sentinel)) + wait_options = WaitForResultOptions(interval_seconds=0.5) + + result = sync_api_client.start_and_wait( + pipe_code="my_pipe", + inputs={"topic": "quantum computing"}, + extra={"custom": True}, + wait_options=wait_options, + method_id="mt_123", + ) + + assert result is sentinel + start_and_wait.assert_awaited_once_with( + pipe_code="my_pipe", + mthds_contents=None, + inputs={"topic": "quantum computing"}, + output_name=None, + output_multiplicity=None, + dynamic_output_concept_ref=None, + extra={"custom": True}, + wait_options=wait_options, + method_ref=None, + method_id="mt_123", + ) + + def test_the_async_error_is_raised_as_is(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + run_timeout = RunTimeoutError("run r1 did not finish", run_id="r1", timeout_seconds=1.0) + mocker.patch.object(sync_api_client._async_client, "wait_for_result", mocker.AsyncMock(side_effect=run_timeout)) + + with pytest.raises(RunTimeoutError) as exc_info: + sync_api_client.wait_for_result("r1") + + assert exc_info.value is run_timeout + + def test_every_call_runs_on_one_loop_on_one_other_thread(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + seen: list[tuple[int, asyncio.AbstractEventLoop]] = [] + + async def _record() -> dict[str, Any]: + await asyncio.sleep(0) + seen.append((threading.get_ident(), asyncio.get_running_loop())) + return {} + + mocker.patch.object(sync_api_client._async_client, "health", mocker.AsyncMock(side_effect=_record)) + + sync_api_client.health() + sync_api_client.health() + + assert len(seen) == 2 + assert seen[0] == seen[1] + assert seen[0][0] != threading.get_ident() + + def test_calls_from_several_threads_run_concurrently(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + """The first call waits for the second to have run: if calls queued, it would time out.""" + first_started = threading.Event() + second_ran: list[asyncio.Event] = [] + outcomes: list[str] = [] + + async def _wait_for_the_second(run_id: str) -> str: + released = asyncio.Event() + second_ran.append(released) + first_started.set() + await asyncio.wait_for(released.wait(), timeout=_HANDSHAKE_TIMEOUT_SECONDS) + return run_id + + async def _release_the_first(run_id: str) -> str: + await asyncio.sleep(0) + second_ran[0].set() + return run_id + + mocker.patch.object(sync_api_client._async_client, "get_run_status", mocker.AsyncMock(side_effect=_wait_for_the_second)) + mocker.patch.object(sync_api_client._async_client, "get_run_result", mocker.AsyncMock(side_effect=_release_the_first)) + + def _first() -> None: + outcomes.append(str(sync_api_client.get_run_status("first"))) + + first_thread = threading.Thread(target=_first) + first_thread.start() + assert first_started.wait(_HANDSHAKE_TIMEOUT_SECONDS) + outcomes.append(str(sync_api_client.get_run_result("second"))) + first_thread.join(_HANDSHAKE_TIMEOUT_SECONDS) + + assert sorted(outcomes) == ["first", "second"] + + def test_a_call_inside_a_running_event_loop_is_refused_before_anything_is_sent( + self, + sync_api_client: SyncPipelexAPIClient, + wire_response: ResponseBuilder, + patch_send: SendPatcher, + ) -> None: + send = patch_send(sync_api_client._async_client, wire_response(200, json_body={"status": "ok"})) + + async def _async_caller() -> None: + await asyncio.sleep(0) + sync_api_client.health() + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + with pytest.raises(SyncClientInEventLoopError, match=r"PipelexAPIClient\.health\(\).*await"): + asyncio.run(_async_caller()) + gc.collect() + + assert send.await_count == 0 + assert sync_api_client._loop is None + assert not [warning for warning in caught if "never awaited" in str(warning.message)] + + def test_close_inside_a_running_event_loop_is_refused(self, sync_api_client: SyncPipelexAPIClient) -> None: + async def _async_caller() -> None: + await asyncio.sleep(0) + sync_api_client.close() + + with pytest.raises(SyncClientInEventLoopError, match=r"SyncPipelexAPIClient\.close\(\)"): + asyncio.run(_async_caller()) + + def test_iterate_methods_follows_every_page( + self, + sync_api_client: SyncPipelexAPIClient, + wire_response: ResponseBuilder, + patch_send: SendPatcher, + ) -> None: + patch_send( + sync_api_client._async_client, + wire_response(200, json_body={"items": [_method("m1"), _method("m2")], "next_cursor": "c1"}), + wire_response(200, json_body={"items": [_method("m3")], "next_cursor": None}), + ) + + method_ids = [summary.method_id for summary in sync_api_client.iterate_methods(limit=2)] + + assert method_ids == ["m1", "m2", "m3"] + + def test_abandoning_an_iterator_finalizes_the_async_generator(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + finalized: list[bool] = [] + + async def _endless_methods(**_kwargs: Any) -> AsyncIterator[str]: + try: + while True: + await asyncio.sleep(0) + yield "m" + finally: + finalized.append(True) + + mocker.patch.object(sync_api_client._async_client, "iterate_methods", _endless_methods) + + iterator = sync_api_client.iterate_methods() + assert next(iterator) == "m" + # The public type is `Iterator`; closing it early is the generator protocol a `break` triggers. + cast("Generator[Any, None, None]", iterator).close() + + assert finalized == [True] + + def test_close_releases_everything_and_the_facade_is_reusable( + self, + sync_api_client: SyncPipelexAPIClient, + wire_response: ResponseBuilder, + patch_send: SendPatcher, + ) -> None: + patch_send( + sync_api_client._async_client, + wire_response(200, json_body={"status": "ok"}), + wire_response(200, json_body={"status": "again"}), + ) + sync_api_client.health() + sync_api_client._async_client.start_client() # the real transport is patched out, so open the HTTP client by hand + first_thread = sync_api_client._loop_thread + assert first_thread is not None + + sync_api_client.close() + + assert not first_thread.is_alive() + assert sync_api_client._loop is None + assert sync_api_client._async_client.client is None + assert sync_api_client.health() == {"status": "again"} + assert sync_api_client._loop_thread is not None + assert sync_api_client._loop_thread is not first_thread + + def test_close_cancels_a_call_in_flight_on_another_thread(self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture) -> None: + started = threading.Event() + raised: list[BaseException] = [] + + async def _never_finishes(_run_id: str) -> None: + started.set() + await asyncio.Event().wait() + + mocker.patch.object(sync_api_client._async_client, "get_run_status", mocker.AsyncMock(side_effect=_never_finishes)) + + def _blocked_caller() -> None: + try: + sync_api_client.get_run_status("r1") + except BaseException as exc: # the test inspects whatever crossed the thread boundary + raised.append(exc) + + caller_thread = threading.Thread(target=_blocked_caller) + caller_thread.start() + assert started.wait(_HANDSHAKE_TIMEOUT_SECONDS) + + sync_api_client.close() + caller_thread.join(_HANDSHAKE_TIMEOUT_SECONDS) + + assert not caller_thread.is_alive() + assert len(raised) == 1 + assert isinstance(raised[0], concurrent.futures.CancelledError) + + def test_the_context_manager_closes_on_exit(self, wire_response: ResponseBuilder, patch_send: SendPatcher) -> None: + with SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) as client: + patch_send(client._async_client, wire_response(200, json_body={"status": "ok"})) + client.health() + loop_thread = client._loop_thread + assert loop_thread is not None + + assert client._loop is None + assert not loop_thread.is_alive() + + def test_closing_an_unused_facade_twice_does_nothing(self) -> None: + client = SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) + + client.close() + client.close() + + assert client._loop is None + assert client._loop_thread is None + + def test_construction_is_the_async_clients(self) -> None: + client = SyncPipelexAPIClient(api_key="test-token", base_url=f"{BASE_URL}/", request_timeout_seconds=30.0) + + assert client.base_url == BASE_URL + assert client.origin_url == BASE_URL + assert client.request_timeout_seconds == 30.0 + with pytest.raises(PipelineRequestError): + SyncPipelexAPIClient(base_url=f"{BASE_URL}/v1") diff --git a/tests/unit/test_sync_client_parity.py b/tests/unit/test_sync_client_parity.py new file mode 100644 index 0000000..c95115c --- /dev/null +++ b/tests/unit/test_sync_client_parity.py @@ -0,0 +1,74 @@ +"""The sync facade mirrors the async client one-to-one, and this module is what keeps it so. + +Every public coroutine and async generator on `PipelexAPIClient` — inherited protocol routes +included — must have a same-named method on `SyncPipelexAPIClient` with an identical parameter +list and the same return annotation, an `AsyncIterator[X]` becoming `Iterator[X]`. A method +added to the async client without its twin fails here rather than shipping a facade that +silently lags, and the facade may not grow a method the async client does not have. +""" + +from __future__ import annotations + +import inspect +from typing import TYPE_CHECKING, Any + +import pytest + +from pipelex_sdk.client import PipelexAPIClient +from pipelex_sdk.sync_client import SyncPipelexAPIClient + +if TYPE_CHECKING: + from collections.abc import Callable + +_ASYNC_ITERATOR_PREFIX = "AsyncIterator[" +_SYNC_ITERATOR_PREFIX = "Iterator[" + + +def _async_surface() -> dict[str, Callable[..., Any]]: + surface: dict[str, Callable[..., Any]] = {} + for name in dir(PipelexAPIClient): + if name.startswith("_"): + continue + member = getattr(PipelexAPIClient, name) + if inspect.iscoroutinefunction(member) or inspect.isasyncgenfunction(member): + surface[name] = member + return surface + + +def _sync_methods() -> dict[str, Callable[..., Any]]: + methods: dict[str, Callable[..., Any]] = {} + for name in dir(SyncPipelexAPIClient): + if name.startswith("_"): + continue + member = inspect.getattr_static(SyncPipelexAPIClient, name) + if inspect.isfunction(member): + methods[name] = member + return methods + + +_ASYNC_METHOD_NAMES = sorted(_async_surface()) + + +class TestSyncClientParity: + def test_every_async_method_has_a_sync_twin(self) -> None: + missing = sorted(set(_async_surface()) - set(_sync_methods())) + assert missing == [] + + def test_the_facade_adds_no_method_the_async_client_lacks(self) -> None: + extra = sorted(set(_sync_methods()) - set(_async_surface())) + assert extra == [] + + @pytest.mark.parametrize("method_name", _ASYNC_METHOD_NAMES) + def test_the_signature_matches(self, method_name: str) -> None: + async_method = _async_surface()[method_name] + sync_method = _sync_methods()[method_name] + async_signature = inspect.signature(async_method) + sync_signature = inspect.signature(sync_method) + + assert list(sync_signature.parameters.values()) == list(async_signature.parameters.values()) + + expected_return = async_signature.return_annotation + if inspect.isasyncgenfunction(async_method): + assert str(expected_return).startswith(_ASYNC_ITERATOR_PREFIX) + expected_return = _SYNC_ITERATOR_PREFIX + str(expected_return).removeprefix(_ASYNC_ITERATOR_PREFIX) + assert sync_signature.return_annotation == expected_return From c74205204bd6c0428b2cfcfa984b174de2945cde Mon Sep 17 00:00:00 2001 From: Louis Choquel Date: Sat, 12 Sep 2026 23:12:35 +0200 Subject: [PATCH 2/3] fix: hold SyncPipelexAPIClient's close, interrupts and iterators to their contract under concurrency The first review round on the sync facade (L-260907-91b052) reproduced these defects, and each now has a regression test that fails on the previous code. - A call made on another thread while close() ran started a second loop on the HTTP client the first loop was still closing, and failed with "Cannot send a request, as the client has been closed". close() now holds the facade's lock until the old loop is gone. The lock is reentrant, so an abandoned iterator that a garbage collection finalizes under it no longer deadlocks its own thread. - A KeyboardInterrupt or SystemExit raised on the loop, such as an on_poll callback calling sys.exit(), stopped the loop and left both the caller and close() blocked for good. Both are now carried back to the caller, and the loop survives. - Ctrl+C during a page fetch surfaced as RuntimeError("aclose(): asynchronous generator is already running") instead of the interrupt. The async generator is now finalized only when its iterator is abandoned between two items. - A paged iterator left open across close() ended silently, as if the listing were complete. It now raises concurrent.futures.CancelledError at its next item. - A partly read iterator left in a module global kept the interpreter from exiting, because it waited on a daemon loop thread that had already stopped. - The in-loop refusal named the wrapped coroutine (MthdsAPIClient.models, _next_item) instead of the facade method that was called. Co-Authored-By: Claude Opus 5 (1M context) --- docs/architecture.md | 8 +- pipelex_sdk/sync_client.py | 155 +++++++++++++------ pyproject.toml | 2 + tests/unit/test_sync_client.py | 267 ++++++++++++++++++++++++++++++++- 4 files changed, 378 insertions(+), 54 deletions(-) diff --git a/docs/architecture.md b/docs/architecture.md index d7f23dd..47f652a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -219,10 +219,10 @@ The wire models are snake_case Pydantic v2. Response models are extension-open ( `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. 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. A closed facade is reusable and starts a fresh loop at its next call. Breaking out of a sync iterator finalizes the async generator beneath it at once. -- **Interrupts.** Ctrl+C while a call is blocked 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 and stays resumable by id. -- **Processes.** The loop thread does not survive a `fork`, 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. +- **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 diff --git a/pipelex_sdk/sync_client.py b/pipelex_sdk/sync_client.py index f59dad0..0d280ea 100644 --- a/pipelex_sdk/sync_client.py +++ b/pipelex_sdk/sync_client.py @@ -1,8 +1,9 @@ """`SyncPipelexAPIClient` — a synchronous facade over `PipelexAPIClient`. -For the callers that have no event loop of their own — a script, a batch job, a Django view, -a notebook cell run without top-level `await` — so none of them writes an `asyncio.run` -wrapper around each call. +For the callers that have no event loop of their own — a script, a batch job, a Django view — +so none of them writes an `asyncio.run` wrapper around each call. A Jupyter notebook is not one +of them: its kernel runs every cell inside its own event loop, so a notebook awaits +`PipelexAPIClient` directly. The facade wraps and delegates; it is not a second transport. Every public method calls the same-named coroutine on one private `PipelexAPIClient`, so there is exactly one HTTP @@ -27,13 +28,18 @@ Interrupting a blocked call (Ctrl+C) cancels the coroutine on the facade's loop and re-raises in the caller. As on the async client, cancelling a `wait_for_result` or `start_and_wait` stops -the waiting, never the run itself, which keeps executing server-side and stays resumable by id. +the waiting, never the run itself, which keeps executing server-side. A `KeyboardInterrupt` or +`SystemExit` raised on the loop itself, by a `WaitForResultOptions.on_poll` callback for +instance, reaches the caller the same way instead of stopping the loop. """ from __future__ import annotations import asyncio +import concurrent.futures +import sys import threading +import types from collections.abc import AsyncGenerator from typing import TYPE_CHECKING, Any, Self, TypeVar, cast @@ -96,7 +102,8 @@ class SyncPipelexAPIClient: Construction takes the same arguments and resolves credentials exactly as `PipelexAPIClient` does (it builds one). Use it as a context manager, or call `close()` - when done; a facade is reusable after `close()`, restarting its loop at the next call. + when done: a facade that is never closed keeps its loop thread until the process exits. A + facade is reusable after `close()`, restarting its loop at the next call. One instance may be shared across threads. It must not be called from a thread that runs an event loop (`SyncClientInEventLoopError`), which includes a `WaitForResultOptions.on_poll` callback: that callback runs on the facade's own loop. @@ -107,10 +114,13 @@ class SyncPipelexAPIClient: def __init__(self, api_key: str | None = None, base_url: str | None = None, request_timeout_seconds: float | None = None) -> None: self._async_client = PipelexAPIClient(api_key=api_key, base_url=base_url, request_timeout_seconds=request_timeout_seconds) - # Guards the loop's lifecycle: starting it, detaching it at close, and submitting to it. - # Submission happens under the lock so a call can never land on a loop `close()` has - # already detached; only the blocking wait for the result happens outside it. - self._lock = threading.Lock() + # Guards the loop's whole lifecycle: starting it, submitting to it, and closing it. `close()` + # holds it until the old loop is gone, so a call made meanwhile waits and then starts a + # fresh loop instead of reaching the wrapped client while the old loop is still closing it. + # Only the blocking wait for a call's result happens outside it. Reentrant, because a + # garbage collection can finalize an abandoned iterator, which takes the lock, on a thread + # that already holds it. + self._lock = threading.RLock() self._loop: asyncio.AbstractEventLoop | None = None self._loop_thread: threading.Thread | None = None @@ -138,27 +148,33 @@ def close(self) -> None: """Close the HTTP client and stop the facade's event loop. A call still in flight on another thread is cancelled and raises - `concurrent.futures.CancelledError` there. Closing a facade that never made a call, or - closing it twice, does nothing. + `concurrent.futures.CancelledError` there, and so does the next item of a paged iterator + left open across the close. A call made on another thread while `close()` runs waits for + it, then starts a fresh loop. Closing a facade that never made a call, or closing it twice, + does nothing. Raises: SyncClientInEventLoopError: If called from a thread that runs an event loop. """ if _inside_running_event_loop(): - raise SyncClientInEventLoopError(_in_event_loop_message("SyncPipelexAPIClient.close")) + raise SyncClientInEventLoopError(_in_event_loop_message("close")) with self._lock: loop = self._loop loop_thread = self._loop_thread + if loop is None or loop_thread is None: + return self._loop = None self._loop_thread = None - if loop is None or loop_thread is None: - return - try: - asyncio.run_coroutine_threadsafe(self._shutdown(), loop).result() - finally: - loop.call_soon_threadsafe(loop.stop) - loop_thread.join() - loop.close() + try: + asyncio.run_coroutine_threadsafe(self._shutdown(), loop).result() + finally: + loop.call_soon_threadsafe(loop.stop) + loop_thread.join() + loop.close() + # Normally `_shutdown` has already released it. When an interrupt cut the shutdown + # short, an HTTP client bound to the loop just closed would otherwise be reused by + # the next call's fresh loop, since the transport only opens one when there is none. + self._async_client.client = None async def _shutdown(self) -> None: """Runs on the facade's loop: cancel what is in flight, then release every resource.""" @@ -181,18 +197,24 @@ def _ensure_loop(self) -> asyncio.AbstractEventLoop: self._loop_thread = loop_thread return loop - def _run(self, coroutine: Coroutine[Any, Any, _ResultT]) -> _ResultT: - """Run one coroutine of the wrapped client on the facade's loop and block for its result.""" + def _run(self, coroutine: Coroutine[Any, Any, _ResultT], *, call_name: str | None = None) -> _ResultT: + """Run one coroutine of the wrapped client on the facade's loop and block for its result. + + `call_name` is the facade method to name in a refusal; it defaults to the coroutine's own + name, which is the same-named twin's. + """ if _inside_running_event_loop(): - call_name = str(getattr(coroutine, "__qualname__", "PipelexAPIClient call")) + refused_call = call_name or str(getattr(coroutine, "__name__", "call")) # Close it rather than drop it: an unawaited coroutine would otherwise surface as a # "never awaited" RuntimeWarning pointing away from the real mistake. coroutine.close() - raise SyncClientInEventLoopError(_in_event_loop_message(call_name)) + raise SyncClientInEventLoopError(_in_event_loop_message(refused_call)) with self._lock: - future = asyncio.run_coroutine_threadsafe(coroutine, self._ensure_loop()) + future = asyncio.run_coroutine_threadsafe(_contained(coroutine), self._ensure_loop()) try: return future.result() + except _LoopExitCarrier as carrier: + raise carrier.escaped from None except BaseException: # Reached on an interrupt (KeyboardInterrupt) while blocked, as well as on a normal # error; cancelling an already finished future is a no-op, so this only stops work @@ -202,23 +224,35 @@ def _run(self, coroutine: Coroutine[Any, Any, _ResultT]) -> _ResultT: def _iterate(self, iterator: AsyncIterator[_ItemT]) -> Iterator[_ItemT]: """Drive an async iterator of the wrapped client one item per round trip to the loop.""" - try: - while True: - next_item = self._run(_next_item(iterator)) - if not next_item: - return + call_name = str(getattr(iterator, "__name__", "iterate")) + while True: + # Nothing is left to finalize when this raises: the async iterator has already ended, or + # is being ended by the cancellation `_run` requested on its way out. Closing it here + # would race that cancellation and replace the caller's exception with a RuntimeError. + next_item = self._run(_next_item(iterator), call_name=call_name) + if next_item is None: + msg = f"SyncPipelexAPIClient was closed while this {call_name}() iterator was open" + raise concurrent.futures.CancelledError(msg) + if not next_item: + return + try: yield next_item[0] - finally: - self._close_async_generator(iterator) + except BaseException: + # Abandoned between two items: a `break` followed by the iterator being dropped, an + # explicit `close()`, garbage collection, or an exception thrown in. + self._close_async_generator(iterator) + raise def _close_async_generator(self, iterator: AsyncIterator[_ItemT]) -> None: - """Finalize an async generator left part-way, e.g. by a `break` out of the sync loop. + """Finalize an async generator abandoned between two items. - Skipped when there is no loop any more (`close()` already finalized every async - generator on it) or when this runs inside an event loop (a blocking call is not allowed - there; the loop finalizes the generator at shutdown instead). + Skipped when there is no loop any more (`close()` already finalized every async generator + on it), when this runs inside an event loop (a blocking call is not allowed there; the loop + finalizes the generator at shutdown instead), and while the interpreter is finalizing: an + iterator left in a module global is collected at exit, after the daemon loop thread has + stopped running, and waiting on that thread would keep the process from ever exiting. """ - if not isinstance(iterator, AsyncGenerator) or _inside_running_event_loop(): + if not isinstance(iterator, AsyncGenerator) or _inside_running_event_loop() or sys.is_finalizing(): return # The wrapped iterators are async generators that are only ever advanced, never sent to. generator = cast("AsyncGenerator[_ItemT, None]", iterator) @@ -513,25 +547,60 @@ def _inside_running_event_loop() -> bool: return True -def _in_event_loop_message(call_name: str) -> str: +def _in_event_loop_message(method_name: str) -> str: return ( - f"{call_name}() was called from a thread that is running an event loop, where a blocking call would freeze " - "that loop for the whole request. Use PipelexAPIClient and await the call there instead." + f"SyncPipelexAPIClient.{method_name}() was called from a thread that is running an event loop, where a blocking " + "call would freeze that loop for the whole request. Use PipelexAPIClient and await the call there instead." ) -async def _next_item(iterator: AsyncIterator[_ItemT]) -> list[_ItemT]: - """The iterator's next item in a one-element list, or an empty list once it is exhausted. +class _LoopExitCarrier(Exception): # ruff: ignore[error-suffix-on-exception-name] — a carrier, never raised to a caller + """Carries a `KeyboardInterrupt` or `SystemExit` raised on the facade's loop to the calling thread. + + asyncio lets those two escape the task that raised them and stop the loop outright, before the + task's outcome reaches the caller's future: the caller would block forever, and so would + `close()`, submitting its shutdown to a loop nothing runs. Wrapped in an ordinary exception, the + task ends like any failing task, and `_run` re-raises the original in the calling thread. + """ + + def __init__(self, escaped: KeyboardInterrupt | SystemExit) -> None: + super().__init__(escaped) + self.escaped = escaped + + +async def _contained(coroutine: Coroutine[Any, Any, _ResultT]) -> _ResultT: + try: + return await coroutine + except (KeyboardInterrupt, SystemExit) as exc: + raise _LoopExitCarrier(exc) from exc + + +async def _next_item(iterator: AsyncIterator[_ItemT]) -> list[_ItemT] | None: + """The iterator's next item in a one-element list, an empty list once it is exhausted, or `None` + when it had been closed before this call. Exhaustion is returned rather than raised because `StopAsyncIteration` is not an exception to carry across the thread boundary: the sync generator turns the empty list into its own - ordinary return. + ordinary return. A closed async generator is told apart from an exhausted one because + `close()` finalizes every async generator on its loop, and resuming one of those raises + `StopAsyncIteration` too: read as an ordinary end, it would hand the caller a silently + truncated listing. """ + if _async_generator_finished(iterator): + return None try: return [await anext(iterator)] except StopAsyncIteration: return [] +def _async_generator_finished(iterator: AsyncIterator[Any]) -> bool: + """Whether an async generator has run to its end or been closed; the facade never resumes one it saw end.""" + if not isinstance(iterator, types.AsyncGeneratorType): + return False + # CPython drops the generator's frame once it has returned, raised or been closed. + return iterator.ag_frame is None + + async def _close_generator(generator: AsyncGenerator[Any, Any]) -> None: await generator.aclose() diff --git a/pyproject.toml b/pyproject.toml index 220359a..c94d868 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -339,6 +339,8 @@ convention = "google" "import-private-name", # Unit tests legitimately import private module helpers under test (e.g. _parse_error_body) "unused-method-argument", # Test-double methods match a Protocol signature; an unused param (e.g. a fake validate ignoring `render`) is intentional "float-equality-comparison", # Tests assert exact float literals that round-trip exactly; `pytest.approx` would only add noise + "suspicious-subprocess-import", # Exit-time behaviour is only observable in a child interpreter the test spawns itself + "subprocess-without-shell-equals-true", # That child's argv is `sys.executable` and a script literal the test wrote, never outside input ] "examples/**/*.py" = [ "implicit-namespace-package", # Runnable demo scripts, not an importable package diff --git a/tests/unit/test_sync_client.py b/tests/unit/test_sync_client.py index a713df3..8fbf31c 100644 --- a/tests/unit/test_sync_client.py +++ b/tests/unit/test_sync_client.py @@ -3,9 +3,13 @@ What is pinned here is the facade's own machinery, not the routes: calls run on one private loop on one other thread, arguments and errors cross that boundary unchanged, several threads share a facade without queueing, a caller inside an event loop is refused before anything is -sent, the paged iterators finalize when abandoned, and `close()` releases everything and leaves -the facade reusable. The one-to-one signature match with the async client is -`test_sync_client_parity.py`. +sent, the paged iterators finalize when abandoned, interrupts and exits reach the caller as they +were raised, and `close()` releases everything, holds off the calls made meanwhile, and leaves +the facade reusable. The one-to-one match with the async client is `test_sync_client_parity.py`. + +A test whose regression would leave a thread blocked for good runs that call through +`_outcome_in_thread`, and builds its own facade instead of taking the fixture, so a regression +fails the test instead of hanging the suite at teardown. """ from __future__ import annotations @@ -13,20 +17,23 @@ import asyncio import concurrent.futures import gc +import subprocess +import sys +import textwrap import threading import warnings -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Protocol, cast import pytest from mthds.protocol.exceptions import PipelineRequestError from pipelex_sdk.errors import RunTimeoutError, SyncClientInEventLoopError -from pipelex_sdk.runs import WaitForResultOptions +from pipelex_sdk.runs import PollInfo, WaitForResultOptions from pipelex_sdk.sync_client import SyncPipelexAPIClient from tests.unit.conftest import BASE_URL if TYPE_CHECKING: - from collections.abc import AsyncIterator, Generator + from collections.abc import AsyncIterator, Callable, Coroutine, Generator from pytest_mock import MockerFixture @@ -34,12 +41,58 @@ # How long a test waits on a cross-thread handshake before failing instead of hanging. _HANDSHAKE_TIMEOUT_SECONDS = 5.0 +# How long a child interpreter gets to run a short script and exit before the test fails. +_SUBPROCESS_TIMEOUT_SECONDS = 60.0 def _method(method_id: str) -> dict[str, Any]: return {"method_id": method_id, "name": f"Method {method_id}", "created_at": "t"} +def _outcome_in_thread(call: Callable[[], object]) -> object: + """Run a call on a daemon thread and return what it returned or raised. + + A call still blocked after the handshake timeout fails the test rather than hanging it, and + the stuck thread, being a daemon, does not keep the test run from exiting. + """ + outcome: list[object] = [] + + def _target() -> None: + try: + outcome.append(call()) + except BaseException as exc: # the test inspects whatever crossed the thread boundary + outcome.append(exc) + + thread = threading.Thread(target=_target, daemon=True) + thread.start() + thread.join(_HANDSHAKE_TIMEOUT_SECONDS) + assert outcome, "the call was still blocked" + return outcome[0] + + +class _Lock(Protocol): + def acquire(self, blocking: bool = ..., timeout: float = ...) -> bool: ... + + def release(self) -> None: ... + + +class _LockReportingContention: + """Stands in for the facade's lock, and reports that a caller found it held before waiting for it.""" + + def __init__(self, lock: _Lock, found_held: threading.Event) -> None: + self._lock = lock + self._found_held = found_held + + def __enter__(self) -> bool: + if not self._lock.acquire(blocking=False): + self._found_held.set() + self._lock.acquire() + return True + + def __exit__(self, *exc_info: object) -> None: + self._lock.release() + + class TestSyncClient: def test_a_call_runs_through_the_real_transport( self, @@ -152,7 +205,7 @@ async def _async_caller() -> None: with warnings.catch_warnings(record=True) as caught: warnings.simplefilter("always") - with pytest.raises(SyncClientInEventLoopError, match=r"PipelexAPIClient\.health\(\).*await"): + with pytest.raises(SyncClientInEventLoopError, match=r"^SyncPipelexAPIClient\.health\(\).*await"): asyncio.run(_async_caller()) gc.collect() @@ -160,6 +213,18 @@ async def _async_caller() -> None: assert sync_api_client._loop is None assert not [warning for warning in caught if "never awaited" in str(warning.message)] + def test_a_refusal_names_the_facade_method_that_was_called(self, sync_api_client: SyncPipelexAPIClient) -> None: + """Inherited routes and the paged iterators included, where the coroutine's own name would mislead.""" + + async def _async_caller() -> None: + await asyncio.sleep(0) + with pytest.raises(SyncClientInEventLoopError, match=r"^SyncPipelexAPIClient\.models\(\)"): + sync_api_client.models() + with pytest.raises(SyncClientInEventLoopError, match=r"^SyncPipelexAPIClient\.iterate_methods\(\)"): + next(sync_api_client.iterate_methods()) + + asyncio.run(_async_caller()) + def test_close_inside_a_running_event_loop_is_refused(self, sync_api_client: SyncPipelexAPIClient) -> None: async def _async_caller() -> None: await asyncio.sleep(0) @@ -204,6 +269,194 @@ async def _endless_methods(**_kwargs: Any) -> AsyncIterator[str]: assert finalized == [True] + def test_an_interrupt_during_a_page_fetch_reaches_the_caller_as_raised( + self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture + ) -> None: + """The interrupt comes out of the blocking wait once the page fetch is in flight, where a real Ctrl+C lands. + + The page's cleanup spans several loop steps, as releasing an httpx response does: closing the + async generator while that cleanup still runs raised a RuntimeError in place of the interrupt. + """ + fetching = threading.Event() + unwound = threading.Event() + + async def _pages(**_kwargs: Any) -> AsyncIterator[str]: + yield "m1" + try: + fetching.set() + await asyncio.Event().wait() + yield "never" + finally: + for _ in range(3): + await asyncio.sleep(0) + unwound.set() + + mocker.patch.object(sync_api_client._async_client, "iterate_methods", _pages) + real_result = concurrent.futures.Future[Any].result + blocking_waits: list[None] = [] + + def _interrupt_the_second_wait(future: concurrent.futures.Future[Any], timeout: float | None = None) -> Any: + blocking_waits.append(None) + if len(blocking_waits) == 2: + assert fetching.wait(_HANDSHAKE_TIMEOUT_SECONDS) + raise KeyboardInterrupt + return real_result(future, timeout) + + mocker.patch.object(concurrent.futures.Future, "result", autospec=True, side_effect=_interrupt_the_second_wait) + + iterator = sync_api_client.iterate_methods() + assert next(iterator) == "m1" + with pytest.raises(KeyboardInterrupt): + next(iterator) + + assert unwound.wait(_HANDSHAKE_TIMEOUT_SECONDS) + + def test_a_system_exit_raised_on_the_loop_reaches_the_caller_and_the_loop_survives(self, mocker: MockerFixture) -> None: + client = SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) + + async def _poll_once(_run_id: str, options: WaitForResultOptions | None = None) -> None: + await asyncio.sleep(0) + if options is not None and options.on_poll is not None: + options.on_poll(PollInfo(attempt=1, elapsed_seconds=0.0)) + + def _exit_from_the_callback(_info: PollInfo) -> None: + raise SystemExit(3) + + mocker.patch.object(client._async_client, "wait_for_result", mocker.AsyncMock(side_effect=_poll_once)) + mocker.patch.object(client._async_client, "health", mocker.AsyncMock(return_value={"status": "ok"})) + + outcome = _outcome_in_thread(lambda: client.wait_for_result("r1", options=WaitForResultOptions(on_poll=_exit_from_the_callback))) + + assert isinstance(outcome, SystemExit) + assert outcome.code == 3 + assert _outcome_in_thread(client.health) == {"status": "ok"} + client.close() + + def test_an_iterator_left_open_across_close_raises_rather_than_ending_early( + self, sync_api_client: SyncPipelexAPIClient, mocker: MockerFixture + ) -> None: + async def _five_methods(**_kwargs: Any) -> AsyncIterator[str]: + for index in range(5): + await asyncio.sleep(0) + yield f"m{index}" + + mocker.patch.object(sync_api_client._async_client, "iterate_methods", _five_methods) + + iterator = sync_api_client.iterate_methods() + assert next(iterator) == "m0" + sync_api_client.close() + + with pytest.raises(concurrent.futures.CancelledError, match="was closed while this"): + next(iterator) + + def test_an_iterator_collected_while_its_facade_holds_the_lock_does_not_deadlock(self, mocker: MockerFixture) -> None: + client = SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) + + async def _endless_methods(**_kwargs: Any) -> AsyncIterator[str]: + while True: + await asyncio.sleep(0) + yield "m" + + mocker.patch.object(client._async_client, "iterate_methods", _endless_methods) + mocker.patch.object(client._async_client, "health", mocker.AsyncMock(return_value={"status": "ok"})) + real_submit = asyncio.run_coroutine_threadsafe + + def _collect_then_submit(coroutine: Coroutine[Any, Any, Any], loop: asyncio.AbstractEventLoop) -> concurrent.futures.Future[Any]: + gc.collect() # stands in for an automatic collection, which any allocation made under the lock can trigger + return real_submit(coroutine, loop) + + # Automatic collection is off so the abandoned iterator is collected under the lock, not earlier. + gc.disable() + try: + iterator = client.iterate_methods() + assert next(iterator) == "m" + cycle: list[object] = [iterator] + cycle.append(cycle) + del iterator, cycle + mocker.patch.object(asyncio, "run_coroutine_threadsafe", side_effect=_collect_then_submit) + outcome = _outcome_in_thread(client.health) + finally: + gc.enable() + + assert outcome == {"status": "ok"} + client.close() + + def test_an_iterator_left_in_a_module_global_does_not_keep_the_interpreter_from_exiting(self) -> None: + """At exit the iterator is collected after the daemon loop thread has stopped; waiting on it hung the process.""" + script = textwrap.dedent( + f""" + import asyncio + from pipelex_sdk.sync_client import SyncPipelexAPIClient + + async def _endless_methods(**_kwargs): + while True: + await asyncio.sleep(0) + yield "m" + + client = SyncPipelexAPIClient(api_key="test-token", base_url="{BASE_URL}") + client._async_client.iterate_methods = _endless_methods + methods = client.iterate_methods() + next(methods) + """ + ) + + completed = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=_SUBPROCESS_TIMEOUT_SECONDS, check=False) + + assert completed.returncode == 0, completed.stderr + + def test_a_call_made_while_close_runs_waits_for_it_then_starts_a_fresh_loop( + self, + sync_api_client: SyncPipelexAPIClient, + mocker: MockerFixture, + ) -> None: + """Calling in the middle of a close started a second loop on the client the first was still closing.""" + closing = threading.Event() + close_released: list[asyncio.Event] = [] + caller_settled = threading.Event() + events: list[str] = [] + + async def _slow_close() -> None: + if close_released: # a later close, the fixture's teardown among them, has nothing to wait for + return + released = asyncio.Event() + close_released.append(released) + closing.set() + await released.wait() + events.append("closed") + + async def _health() -> dict[str, Any]: + await asyncio.sleep(0) + events.append("health") + return {} + + mocker.patch.object(sync_api_client._async_client, "health", mocker.AsyncMock(side_effect=_health)) + mocker.patch.object(sync_api_client._async_client, "close", mocker.AsyncMock(side_effect=_slow_close)) + sync_api_client.health() + events.clear() + first_loop = sync_api_client._loop + assert first_loop is not None + + closer = threading.Thread(target=sync_api_client.close) + closer.start() + assert closing.wait(_HANDSHAKE_TIMEOUT_SECONDS) + mocker.patch.object(sync_api_client, "_lock", _LockReportingContention(sync_api_client._lock, caller_settled)) + + def _caller() -> None: + try: + sync_api_client.health() + finally: + caller_settled.set() + + caller = threading.Thread(target=_caller) + caller.start() + # Settled: the caller found the lock held, or ran its whole call without having to wait. + assert caller_settled.wait(_HANDSHAKE_TIMEOUT_SECONDS) + first_loop.call_soon_threadsafe(close_released[0].set) + closer.join(_HANDSHAKE_TIMEOUT_SECONDS) + caller.join(_HANDSHAKE_TIMEOUT_SECONDS) + + assert events == ["closed", "health"] + def test_close_releases_everything_and_the_facade_is_reusable( self, sync_api_client: SyncPipelexAPIClient, From d0c0fbcbfe33fc538ddc83d27048a21d9585ceea Mon Sep 17 00:00:00 2001 From: Louis Choquel Date: Sat, 12 Sep 2026 23:12:35 +0200 Subject: [PATCH 3/3] test: pin that every sync twin forwards each argument and hands back the result The parity test compared signatures only, so a twin that dropped or swapped an argument passed the whole suite; mutating iterate_methods to forward limit=None proved it. Each twin is now called with a distinct sentinel per parameter and must hand every one to its coroutine under its own name, then return what the coroutine returned. Co-Authored-By: Claude Opus 5 (1M context) --- tests/unit/test_sync_client_parity.py | 53 ++++++++++++++++++++++++++- 1 file changed, 51 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_sync_client_parity.py b/tests/unit/test_sync_client_parity.py index c95115c..85a6bc7 100644 --- a/tests/unit/test_sync_client_parity.py +++ b/tests/unit/test_sync_client_parity.py @@ -4,11 +4,15 @@ included — must have a same-named method on `SyncPipelexAPIClient` with an identical parameter list and the same return annotation, an `AsyncIterator[X]` becoming `Iterator[X]`. A method added to the async client without its twin fails here rather than shipping a facade that -silently lags, and the facade may not grow a method the async client does not have. +silently lags, and the facade may not grow a method the async client does not have. Matching +signatures do not prove the body forwards them, so every twin is also called with a distinct +sentinel per parameter, and must hand each one to its coroutine under its own name and hand the +result back. """ from __future__ import annotations +import asyncio import inspect from typing import TYPE_CHECKING, Any @@ -16,9 +20,12 @@ from pipelex_sdk.client import PipelexAPIClient from pipelex_sdk.sync_client import SyncPipelexAPIClient +from tests.unit.conftest import BASE_URL if TYPE_CHECKING: - from collections.abc import Callable + from collections.abc import AsyncIterator, Callable + + from pytest_mock import MockerFixture _ASYNC_ITERATOR_PREFIX = "AsyncIterator[" _SYNC_ITERATOR_PREFIX = "Iterator[" @@ -47,6 +54,8 @@ def _sync_methods() -> dict[str, Callable[..., Any]]: _ASYNC_METHOD_NAMES = sorted(_async_surface()) +# `close()` is the facade's own lifecycle, pinned in `test_sync_client.py`, rather than a forwarded call. +_FORWARDED_METHOD_NAMES = [name for name in _ASYNC_METHOD_NAMES if name != "close"] class TestSyncClientParity: @@ -72,3 +81,43 @@ def test_the_signature_matches(self, method_name: str) -> None: assert str(expected_return).startswith(_ASYNC_ITERATOR_PREFIX) expected_return = _SYNC_ITERATOR_PREFIX + str(expected_return).removeprefix(_ASYNC_ITERATOR_PREFIX) assert sync_signature.return_annotation == expected_return + + @pytest.mark.parametrize("method_name", _FORWARDED_METHOD_NAMES) + def test_every_argument_and_the_result_cross_unchanged(self, method_name: str, mocker: MockerFixture) -> None: + async_method = _async_surface()[method_name] + signature = inspect.signature(async_method) + arguments: dict[str, object] = {name: object() for name in list(signature.parameters)[1:]} + returned = object() + received: list[dict[str, Any]] = [] + + def _record(*args: Any, **kwargs: Any) -> None: + bound = signature.bind(None, *args, **kwargs) + received.append({name: value for name, value in bound.arguments.items() if name != "self"}) + + async def _coroutine_twin(*args: Any, **kwargs: Any) -> object: + await asyncio.sleep(0) + _record(*args, **kwargs) + return returned + + async def _generator_twin(*args: Any, **kwargs: Any) -> AsyncIterator[object]: + await asyncio.sleep(0) + _record(*args, **kwargs) + yield returned + + is_generator = inspect.isasyncgenfunction(async_method) + client = SyncPipelexAPIClient(api_key="test-token", base_url=BASE_URL) + mocker.patch.object(client._async_client, method_name, _generator_twin if is_generator else _coroutine_twin) + try: + result = getattr(client, method_name)(**arguments) + if is_generator: + result = list(result) + finally: + client.close() + + assert received == [arguments] + if is_generator: + assert result == [returned] + elif signature.return_annotation == "None": + assert result is None + else: + assert result is returned