diff --git a/CHANGELOG.md b/CHANGELOG.md index 212a0cb..fc5e1a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## [v0.3.0] - 2026-07-05 + +### Changed + +- **One output accessor across both execution modes: `result.main_stuff` (Breaking).** Reading a run's output no longer depends on which path ran. Both the durable result (`RunResults`, from `wait_for_result` / `start_and_wait`) and the blocking result (now a `PipelexExecuteResult`, from `execute`) expose a resolved, non-null `.main_stuff` — so a caller writes `result = await client.(...); output = result.main_stuff` uniformly, with no `main_stuff or pipe_output` fallback and no working-memory spelunking. + - `RunResults.main_stuff` is now **required and non-null** for a completed run (was `Any = None`). On the hosted path it is the `main_stuff.json` S3 artifact; on the blocking path the SDK resolves it out of the returned working memory via the response's `main_stuff_name` extension field. The full working memory still rides `pipe_output` (blocking path only) for consumers that want it. + - `execute()` now returns a `PipelexExecuteResult` — the protocol's raw execute response enriched with the same resolved `.main_stuff` accessor. It remains a `DictRunResultExecute` subtype, so existing field access (`pipeline_run_id`, `pipe_output`) is unchanged. + - *(Migration: read `result.main_stuff` instead of digging through `pipe_output` / falling back from `main_stuff` to `pipe_output`.)* + +### Added + +- **`MissingMainStuffError`.** A completed run that cannot deliver a main stuff now raises this typed error (derives from `PipelineRequestError`, carries `run_id`) instead of silently yielding a null output: the hosted results endpoint answered a `200` that omits `main_stuff` or sends it null, or a blocking `execute` response named a `main_stuff_name` absent from its working-memory root. The results path checks the decoded payload before validating into the now-required `RunResults` model, so an omitted key surfaces as `MissingMainStuffError` rather than a raw Pydantic validation error. A falsy-but-present main stuff (empty list, `0`) is a valid output and does not raise. + ## [v0.2.0] - 2026-07-02 ### Added diff --git a/README.md b/README.md index 6916a6f..5759113 100644 --- a/README.md +++ b/README.md @@ -45,9 +45,9 @@ async def main() -> None: inputs={"topic": "quantum computing"}, ) - # 3. Read the output. Hosted runs carry `main_stuff`; the bare-runner - # fallback carries the native `pipe_output`. - print(result.main_stuff or result.pipe_output) + # 3. Read the output. Every completed run delivers a resolved `main_stuff` + # (the full working memory also rides `pipe_output` on the blocking path). + print(result.main_stuff) ``` ### Long runs: start + poll explicitly diff --git a/docs/architecture.md b/docs/architecture.md index 7372af2..2c76757 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -75,6 +75,8 @@ The protocol `execute` is **overridden** to add one Pipelex-API behavior the bar Everything else stays the inherited regime: the protocol's optional 202 async-degrade still raises `RunStillRunningError` (from the base `execute`), and every other non-2xx keeps `httpx.HTTPStatusError` — consistent with the other inherited protocol routes (decision #5). The `_execute_blocking` bare-runner fallback (below) calls this same overridden `execute`, so it inherits the translation; off-platform there is no gateway cap, so the `503/504`-after-28s condition effectively never fires there. +The override also **enriches the return type**: it re-validates the base result into a `PipelexExecuteResult` (`pipelex_sdk/execute_result.py`), a `DictRunResultExecute` subtype that adds a resolved `.main_stuff` accessor (dug out of `pipe_output`'s working memory via the response's `main_stuff_name`, raising `MissingMainStuffError` if unlocatable). This gives blocking and durable results the **same output accessor** — `result.main_stuff` — so callers never branch on which path ran. `_map_run_result_to_run_results` reads that accessor too, keeping the resolution single-sourced. + ## Run lifecycle (hosted extension) The durable run lifecycle (`pipelex_sdk/runs.py` + the client's lifecycle methods) is a **hosted-API extension, not part of the MTHDS Protocol**. Long method runs outlive the hosted gateway's ~30s synchronous cap, so a caller submits a run (`POST /v1/start`), then polls a self-healing endpoint by bare `pipeline_run_id` until it reaches a terminal state. All state lives behind the id (DynamoDB + Temporal on the platform), so a caller can drop the poll loop and resume later with just the id. A bare runner has no run store and `404`s these routes, which the client translates into a clear `RunLifecycleUnavailableError`. @@ -85,7 +87,7 @@ The durable run lifecycle (`pipelex_sdk/runs.py` + the client's lifecycle method - `RunStatus` — the hosted status enum, with `is_terminal` / `is_success` predicates (exhaustive `match`). - `RunRead` — a run record read through the self-healing status path (adds `degraded` + `retry_after_seconds`). -- `RunResults` — result artifacts. Hosted runs carry `main_stuff` (+ `graph_spec`); the bare-runner blocking fallback carries `pipe_output` (the runner's native execute response). Consumers read `main_stuff or pipe_output` (the documented hosted/bare output-shape difference). Extension-open, so any other server artifact is preserved. +- `RunResults` — result artifacts. `main_stuff` (the resolved main output content) is always present for a completed run: on the hosted path it is the `main_stuff.json` S3 artifact; on the bare-runner blocking path the SDK resolves it from the returned working memory via the response's `main_stuff_name`, so both paths deliver the same shape. Consumers read `main_stuff` directly. The full working memory still rides `pipe_output` (blocking path only) for consumers that want it, and `graph_spec` rides the hosted path. A completed run that cannot deliver a main stuff raises `MissingMainStuffError`. Extension-open, so any other server artifact is preserved. - `RunResultState` — the single-shot result outcome, a union discriminated on `state` (`running` / `completed` / `failed`). - `WaitForResultOptions` / `PollInfo` — poll-loop tuning and progress info. Async-native cancellation is via `asyncio.CancelledError` (cancel the awaiting task), so there is no `signal` field. @@ -103,7 +105,7 @@ These poll GETs go through `_send_or_unreachable`, so a transport failure surfac - A cached `GET /v1/version` handshake (`_supports_run_lifecycle`) classifies the runner. `VersionInfo.implementation == "pipelex-api"` ⇒ a bare runner (no run store); anything else ⇒ assumed hosted. The outcome is cached for the client's lifetime; a failed handshake assumes hosted and lets `start` surface the real error. - **Hosted:** durable `start` (202 ack) → `wait_for_result` (poll to terminal). -- **Bare runner:** the blocking `POST /v1/execute` (`_execute_blocking`), which has no gateway cap off-platform and returns the native `pipe_output`, mapped onto `RunResults` (`main_stuff = None`). +- **Bare runner:** the blocking `POST /v1/execute` (`_execute_blocking`), which has no gateway cap off-platform and returns the native `pipe_output`, mapped onto `RunResults` — the SDK resolves `main_stuff` out of the working memory via the response's `main_stuff_name` and keeps the full working memory on `pipe_output`. - **Self-heal:** a runner can look hosted yet lack the durable routes (`implementation` is an optional extension a compliant bare runner may omit). The client's `start` override translates a bare-runner missing-route `404` into `RunLifecycleUnavailableError` — raised **before any run is created** — so `start_and_wait` falls back to the blocking path without risking a double-run, and caches the negative so later calls skip the durable attempt. ### Run/lifecycle errors diff --git a/pipelex_sdk/client.py b/pipelex_sdk/client.py index bde5dcc..a160bc7 100644 --- a/pipelex_sdk/client.py +++ b/pipelex_sdk/client.py @@ -35,11 +35,13 @@ from pipelex_sdk.errors import ( ApiResponseError, ApiUnreachableError, + MissingMainStuffError, PipelineExecuteTimeoutError, RunFailedError, RunLifecycleUnavailableError, RunTimeoutError, ) +from pipelex_sdk.execute_result import PipelexExecuteResult from pipelex_sdk.product_models import ( BillingPortalResponse, ChangePlanResponse, @@ -77,7 +79,6 @@ from mthds.protocol.pipeline_inputs import PipelineInputs from mthds.protocol.stuff import StuffType from mthds.protocol.working_memory import WorkingMemoryAbstract - from mthds.runners.api.models import DictRunResultExecute from pipelex_sdk.product_models import ( MethodWriteInput, @@ -317,9 +318,13 @@ async def execute( output_multiplicity: VariableMultiplicity | None = None, dynamic_output_concept_ref: str | None = None, extra: dict[str, Any] | None = None, - ) -> DictRunResultExecute: + ) -> PipelexExecuteResult: """Execute a method synchronously and wait for its completion — `POST /v1/execute`. + Returns a `PipelexExecuteResult` — the protocol's raw execute response enriched with a + resolved `.main_stuff` accessor, so a blocking result reads its output the same way as a + durable one (`result.main_stuff`) instead of digging through `pipe_output`. + Identical to the inherited protocol `execute`, except a failure consistent with the hosted gateway's ~30s synchronous ceiling — a gateway `503`/`504`, or a client-side request timeout, after at least ~28s have elapsed — is translated into a clear @@ -337,7 +342,7 @@ async def execute( """ started_at = monotonic() try: - return await super().execute( + result = await super().execute( pipe_code=pipe_code, mthds_contents=mthds_contents, inputs=inputs, @@ -351,6 +356,9 @@ async def execute( if _is_gateway_timeout(exc, elapsed_seconds): raise PipelineExecuteTimeoutError(_execute_timeout_message(elapsed_seconds), elapsed_seconds=elapsed_seconds) from exc raise + # Re-validate the base result into the enriched subclass (adds the `.main_stuff` accessor; + # the `main_stuff_name` extension + working memory ride `model_extra`/`pipe_output`). + return PipelexExecuteResult.model_validate(result.model_dump()) # ── Protocol surface: `start` override (bare-runner 404 → typed error) ── @@ -524,7 +532,15 @@ async def get_run_result(self, run_id: str) -> RunResultState: self._raise_if_lifecycle_unavailable(response, url) response.raise_for_status() - result = RunResults.model_validate(response.json()) + # Inspect the decoded payload before validating: `main_stuff` is a required field on `RunResults`, + # so a `200` that omits the key would raise a raw Pydantic error instead of the typed + # `MissingMainStuffError`. `.get(...) is None` covers both the missing-key and explicit-null cases + # for the same un-deliverable-output condition (a present-but-falsy main stuff — `[]`, `0` — stays). + payload = response.json() + if isinstance(payload, dict) and cast("dict[str, Any]", payload).get("main_stuff") is None: + msg = f"Completed run '{run_id}' returned no main stuff — a completed run always delivers a main stuff." + raise MissingMainStuffError(msg, run_id=run_id) + result = RunResults.model_validate(payload) return RunResultCompleted(pipeline_run_id=run_id, result=result) async def wait_for_result(self, run_id: str, options: WaitForResultOptions | None = None) -> RunResults: @@ -927,16 +943,17 @@ def _extract_run_status_from_message(message: str) -> RunStatus: return RunStatus.FAILED -def _map_run_result_to_run_results(response: DictRunResultExecute) -> RunResults: +def _map_run_result_to_run_results(response: PipelexExecuteResult) -> RunResults: """Map the protocol's blocking `POST /v1/execute` response onto the lifecycle's `RunResults`. - The bare-runner path returns `pipe_output` (native runner shape); `main_stuff` and `graph_spec` - are hosted-durable artifacts and stay `None` here. Consumers read `main_stuff or pipe_output` - (the documented hosted/bare output-shape difference). + `response.main_stuff` resolves the main output out of the returned working memory (and raises + `MissingMainStuffError` if the run named no locatable main stuff), so the durable and blocking + paths hand back the same `main_stuff` content shape. The full working memory rides `pipe_output` + (blocking only). """ return RunResults( pipeline_run_id=response.pipeline_run_id, - main_stuff=None, + main_stuff=response.main_stuff, graph_spec=None, pipe_output=response.pipe_output.model_dump(), ) diff --git a/pipelex_sdk/errors.py b/pipelex_sdk/errors.py index 05a5cc2..997e93a 100644 --- a/pipelex_sdk/errors.py +++ b/pipelex_sdk/errors.py @@ -131,6 +131,22 @@ def __init__(self, message: str, run_id: str, timeout_seconds: float) -> None: self.timeout_seconds = timeout_seconds +class MissingMainStuffError(PipelineRequestError): + """Raised when a completed run cannot deliver its main stuff. + + Every completed run delivers a main stuff (the pipelex >= 0.37 wire invariant), so the SDK + hands consumers a non-null `RunResults.main_stuff`. This surfaces the contract violation when it + cannot: the hosted results endpoint answered a `200` with a null `main_stuff`, or a blocking + `execute` response named a `main_stuff_name` whose stuff is absent from the returned working + memory. `run_id` locates the run. (A falsy-but-present main stuff — an empty list, `0` — is a + valid output and does NOT raise; only a genuinely absent one does.) + """ + + def __init__(self, message: str, run_id: str) -> None: + super().__init__(message) + self.run_id = run_id + + class RunLifecycleUnavailableError(PipelineRequestError): """Raised when the durable run lifecycle (`/v1/runs/*`) is not served by the configured `PIPELEX_BASE_URL`. diff --git a/pipelex_sdk/execute_result.py b/pipelex_sdk/execute_result.py new file mode 100644 index 0000000..a053c4e --- /dev/null +++ b/pipelex_sdk/execute_result.py @@ -0,0 +1,48 @@ +"""The blocking `execute()` result — a `DictRunResultExecute` that resolves its `.main_stuff`. + +Kept in its own module (not `runs.py`) so it can import `MissingMainStuffError` from +`errors` without forming an import cycle (`errors` type-imports `runs`). +""" + +from __future__ import annotations + +from typing import Any + +from mthds.runners.api.models import DictRunResultExecute + +from pipelex_sdk.errors import MissingMainStuffError + + +class PipelexExecuteResult(DictRunResultExecute): + """The SDK's blocking `execute()` result — a `DictRunResultExecute` that also exposes the + resolved main output as `.main_stuff`. + + The protocol's raw execute response carries the working memory (`pipe_output`) and names the + main output via `main_stuff_name`, but not the output itself. The neutral `mthds` model leaves + `main_stuff_name` in its extension bag; this Pipelex-branded subclass declares it as a typed + field (Pipelex owns that concept) and digs the output out on access, so callers read + `result.main_stuff` exactly the same way as on the durable path (`RunResults.main_stuff`) — one + output accessor across both execution modes, no working-memory spelunking. + """ + + #: The working-memory `root` key the completed execute response names as its main stuff + #: (pipelex >= 0.37 always sends it). `None` only if a runner omits it, in which case + #: `.main_stuff` raises `MissingMainStuffError`. + main_stuff_name: str | None = None + + @property + def main_stuff(self) -> Any: + """The resolved main output content, dug out of the working memory via `main_stuff_name`. + Raises `MissingMainStuffError` if the completed run named no locatable main stuff. A + falsy-but-present value (empty list, `0`) is a valid output and is returned as-is. + """ + main_stuff_name = self.main_stuff_name + stuff = self.pipe_output.working_memory.root.get(main_stuff_name) if main_stuff_name is not None else None + if stuff is None: + msg = ( + f"Blocking run '{self.pipeline_run_id}' delivered no locatable main stuff " + f"(main_stuff_name={main_stuff_name!r} is absent from the working-memory root) — " + "a completed run always delivers a main stuff." + ) + raise MissingMainStuffError(msg, run_id=self.pipeline_run_id) + return stuff.content diff --git a/pipelex_sdk/runs.py b/pipelex_sdk/runs.py index c897f9b..99af881 100644 --- a/pipelex_sdk/runs.py +++ b/pipelex_sdk/runs.py @@ -118,24 +118,29 @@ class RunRead(RunPublic): class RunResults(BaseModel): """Result artifacts for a completed run — `GET /v1/runs/{pipeline_run_id}/results`. - Hosted: `main_stuff` + `graph_spec` (S3 artifacts relayed VERBATIM; - `main_stuff` is polymorphic — a list output renders to a top-level array, a - structured output to an object — so both are typed as opaque JSON (`Any`), - never `dict`; either may be `None` mid-write). Bare-runner blocking fallback: - the runner's native execute response rides `pipe_output`. Consumers read - `main_stuff or pipe_output` (the documented hosted/bare output-shape - difference). Extension-open (`extra="allow"`): any other server artifact is - preserved without being named by the SDK. + `main_stuff` is the resolved main output content and is ALWAYS present for a + completed run (the pipelex >= 0.37 main-stuff invariant): on the hosted path + it is the `main_stuff.json` S3 artifact relayed verbatim; on the bare-runner + blocking path the SDK resolves it from the returned working memory via the + run's `main_stuff_name`, so both paths deliver the same content shape. + Consumers read `main_stuff` directly — no shape-guessing. A completed run that + cannot deliver a main stuff raises `MissingMainStuffError`. Extension-open + (`extra="allow"`): any other server artifact (e.g. the hosted `working_memory`) + is preserved without being named by the SDK. """ model_config = ConfigDict(extra="allow") pipeline_run_id: str + #: The resolved main output content — always present for a completed run. Typed `Any` because the + #: content is polymorphic (a list output renders to a top-level array, a structured output to an + #: object) and may be a valid falsy value (empty list, `0`); it is never absent for a completed run. + main_stuff: Any #: Method graph spec (`graphspec.json`); `None` if missing mid-write or on the bare-runner path. graph_spec: Any = None - #: Main output stuff (`main_stuff.json`); `None` if missing mid-write or on the bare-runner path. - main_stuff: Any = None - #: Bare runner's native pipe output (blocking-execute fallback only); `None` on the hosted path. + #: Bare runner's native pipe output — the full working memory (`{"root": ..., "aliases": ...}`), + #: blocking-execute path only; `None` on the hosted path. Supplementary to `main_stuff`, which is + #: already resolved out of it; kept for consumers that need the whole working memory. pipe_output: dict[str, Any] | None = None diff --git a/pyproject.toml b/pyproject.toml index 8bc0520..e2ded0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "pipelex-sdk" -version = "0.2.0" +version = "0.3.0" description = "The Python client for the Pipelex hosted API — the MTHDS Protocol surface plus the durable run lifecycle and the Pipelex product surface, built on the `mthds` protocol base." authors = [{ name = "Evotis S.A.S.", email = "oss@pipelex.com" }] maintainers = [{ name = "Pipelex staff", email = "oss@pipelex.com" }] diff --git a/tests/unit/test_client_execute.py b/tests/unit/test_client_execute.py index 24e18e3..18f34d5 100644 --- a/tests/unit/test_client_execute.py +++ b/tests/unit/test_client_execute.py @@ -13,13 +13,22 @@ from pytest_mock import MockerFixture from pipelex_sdk.client import PipelexAPIClient -from pipelex_sdk.errors import PipelineExecuteTimeoutError, RunStillRunningError +from pipelex_sdk.errors import MissingMainStuffError, PipelineExecuteTimeoutError, RunStillRunningError _BASE_URL = "http://localhost:8081" +# A completed execute response: `main_stuff_name` ("result") names the working-memory root key the +# `.main_stuff` accessor resolves to. _EXECUTE_BODY: dict[str, object] = { "pipeline_run_id": "run-x", - "pipe_output": {"working_memory": {"root": {}, "aliases": {}}, "pipeline_run_id": "run-x"}, + "main_stuff_name": "result", + "pipe_output": { + "working_memory": { + "root": {"result": {"concept": "native.Text", "content": {"text": "hi"}}}, + "aliases": {"main_stuff": "result"}, + }, + "pipeline_run_id": "run-x", + }, } @@ -74,12 +83,31 @@ def test_fast_503_stays_inherited_http_status_error(self, mocker: MockerFixture) asyncio.run(client.execute(pipe_code="p")) assert not isinstance(exc_info.value, PipelineExecuteTimeoutError) - def test_success_passes_through_untranslated(self, mocker: MockerFixture) -> None: + def test_success_resolves_main_stuff(self, mocker: MockerFixture) -> None: client = self._client() mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=_response(200, json=_EXECUTE_BODY))) result = asyncio.run(client.execute(pipe_code="p")) assert result.pipeline_run_id == "run-x" + # `.main_stuff` resolves the output out of the working memory — same accessor as the durable path. + assert result.main_stuff == {"text": "hi"} + + def test_main_stuff_raises_when_unlocatable(self, mocker: MockerFixture) -> None: + client = self._client() + # `main_stuff_name` names "missing", absent from the working-memory root. + body: dict[str, object] = { + "pipeline_run_id": "run-x", + "main_stuff_name": "missing", + "pipe_output": { + "working_memory": {"root": {"other": {"concept": "native.Text", "content": {}}}, "aliases": {}}, + "pipeline_run_id": "run-x", + }, + } + mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=_response(200, json=body))) + + result = asyncio.run(client.execute(pipe_code="p")) + with pytest.raises(MissingMainStuffError): + _ = result.main_stuff def test_202_degrade_stays_run_still_running_error(self, mocker: MockerFixture) -> None: client = self._client() diff --git a/tests/unit/test_client_lifecycle.py b/tests/unit/test_client_lifecycle.py index 1a7f3cd..417d134 100644 --- a/tests/unit/test_client_lifecycle.py +++ b/tests/unit/test_client_lifecycle.py @@ -9,6 +9,7 @@ from pipelex_sdk.client import PipelexAPIClient from pipelex_sdk.errors import ( + MissingMainStuffError, RunFailedError, RunLifecycleUnavailableError, RunStillRunningError, @@ -129,6 +130,32 @@ def test_get_run_result_completed_keeps_polymorphic_main_stuff(self, mocker: Moc assert state.result.main_stuff == [{"color": "red"}, {"color": "blue"}] assert state.result.graph_spec == {"nodes": []} + @pytest.mark.parametrize( + "body", + [ + pytest.param({"pipeline_run_id": "run_1"}, id="main_stuff_key_omitted"), + pytest.param({"pipeline_run_id": "run_1", "main_stuff": None}, id="main_stuff_null"), + ], + ) + def test_get_run_result_completed_without_main_stuff_raises_typed_error(self, mocker: MockerFixture, body: dict[str, object]) -> None: + """A 200 that omits main_stuff (or sends it null) raises MissingMainStuffError, not a raw Pydantic error.""" + client = self._client() + mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=_response(200, json=body))) + + with pytest.raises(MissingMainStuffError) as exc_info: + asyncio.run(client.get_run_result("run_1")) + assert exc_info.value.run_id == "run_1" + + def test_get_run_result_completed_keeps_falsy_main_stuff(self, mocker: MockerFixture) -> None: + """A present-but-falsy main_stuff (an empty list) is a valid output and does NOT raise.""" + client = self._client() + body: dict[str, object] = {"pipeline_run_id": "run_1", "main_stuff": []} + mocker.patch.object(client, "_send", mocker.AsyncMock(return_value=_response(200, json=body))) + + state = asyncio.run(client.get_run_result("run_1")) + assert isinstance(state, RunResultCompleted) + assert state.result.main_stuff == [] + def test_get_run_result_running_honors_retry_after(self, mocker: MockerFixture) -> None: """A 202 maps to RunResultRunning with the server's Retry-After hint.""" client = self._client() diff --git a/tests/unit/test_client_run_fallback.py b/tests/unit/test_client_run_fallback.py index 300d333..612ecb8 100644 --- a/tests/unit/test_client_run_fallback.py +++ b/tests/unit/test_client_run_fallback.py @@ -12,7 +12,7 @@ from pytest_mock import MockerFixture from pipelex_sdk.client import PipelexAPIClient -from pipelex_sdk.errors import ApiUnreachableError, RunLifecycleUnavailableError +from pipelex_sdk.errors import ApiUnreachableError, MissingMainStuffError, RunLifecycleUnavailableError _BASE_URL = "http://localhost:8081" @@ -23,9 +23,18 @@ # discover the missing lifecycle at runtime (start 404s) and self-heal to the blocking path. _BASE_ONLY_VERSION = {"protocol_version": "0.6.0", "runner_version": "9.9.9"} +# A completed blocking-execute response: `main_stuff_name` (an extension field) names the +# working-memory root key of the main stuff, which the SDK resolves into `RunResults.main_stuff`. _EXECUTE_BODY: dict[str, object] = { "pipeline_run_id": "run-x", - "pipe_output": {"working_memory": {"root": {}, "aliases": {}}, "pipeline_run_id": "run-x"}, + "main_stuff_name": "result", + "pipe_output": { + "working_memory": { + "root": {"result": {"concept": "native.Text", "content": {"text": "hello"}}}, + "aliases": {"main_stuff": "result"}, + }, + "pipeline_run_id": "run-x", + }, } @@ -89,6 +98,24 @@ def test_caches_the_version_handshake_across_calls(self, mocker: MockerFixture) asyncio.run(client.start_and_wait(pipe_code="p")) assert _urls(send).count(f"{_BASE_URL}/v1/version") == 1 + def test_hosted_completed_with_null_main_stuff_raises(self, mocker: MockerFixture) -> None: + """A hosted 200 whose `main_stuff` is null is a completed run that delivered nothing — hard fail.""" + client = self._client() + mocker.patch.object( + client, + "_send", + mocker.AsyncMock( + side_effect=[ + _response(200, json=_HOSTED_VERSION), + _response(202, json={"pipeline_run_id": "run-1", "state": "STARTED", "created_at": "t0"}), + _response(200, json={"pipeline_run_id": "run-1", "main_stuff": None, "graph_spec": {"n": 1}}), + ] + ), + ) + + with pytest.raises(MissingMainStuffError): + asyncio.run(client.start_and_wait(pipe_code="p")) + # ── Bare runner (blocking execute fallback) ────────────────── def test_bare_runner_falls_back_to_blocking_execute(self, mocker: MockerFixture) -> None: @@ -102,10 +129,34 @@ def test_bare_runner_falls_back_to_blocking_execute(self, mocker: MockerFixture) result = asyncio.run(client.start_and_wait(pipe_code="p", mthds_contents=["x"])) assert result.pipeline_run_id == "run-x" - assert result.main_stuff is None - assert result.pipe_output == {"working_memory": {"root": {}, "aliases": {}}, "pipeline_run_id": "run-x"} + # The SDK resolves `main_stuff` out of the working memory via `main_stuff_name` ("result") — + # its content, the same shape the hosted path relays; the full working memory rides pipe_output. + assert result.main_stuff == {"text": "hello"} + assert result.pipe_output is not None + assert result.pipe_output["working_memory"]["root"]["result"]["content"] == {"text": "hello"} assert _urls(send) == [f"{_BASE_URL}/v1/version", f"{_BASE_URL}/v1/execute"] + def test_blocking_fallback_raises_when_main_stuff_unlocatable(self, mocker: MockerFixture) -> None: + """A completed blocking response whose `main_stuff_name` names no root stuff is a hard fail.""" + client = self._client() + # `main_stuff_name` points at "answer", but the working-memory root has no such stuff. + bad_body: dict[str, object] = { + "pipeline_run_id": "run-y", + "main_stuff_name": "answer", + "pipe_output": { + "working_memory": {"root": {"other": {"concept": "native.Text", "content": {}}}, "aliases": {}}, + "pipeline_run_id": "run-y", + }, + } + mocker.patch.object( + client, + "_send", + mocker.AsyncMock(side_effect=[_response(200, json=_BARE_VERSION), _response(200, json=bad_body)]), + ) + + with pytest.raises(MissingMainStuffError): + asyncio.run(client.start_and_wait(pipe_code="p", mthds_contents=["x"])) + def test_fallback_forwards_extra_extension_args(self, mocker: MockerFixture) -> None: """An `extra` extension arg rides the blocking execute body as a top-level field — not dropped.""" client = self._client() diff --git a/uv.lock b/uv.lock index 960876d..cb3f991 100644 --- a/uv.lock +++ b/uv.lock @@ -350,7 +350,7 @@ wheels = [ [[package]] name = "pipelex-sdk" -version = "0.2.0" +version = "0.3.0" source = { editable = "." } dependencies = [ { name = "backports-strenum", marker = "python_full_version < '3.11'" },