From 79b4426e2a4ac28b4a3e86c95a286dd501185e2a Mon Sep 17 00:00:00 2001 From: Rafal Leszko Date: Wed, 21 Oct 2026 10:59:49 +0200 Subject: [PATCH 1/4] feat(serverless): address an environment on invoke and get_task (RUNSERV-1238) An app id now names one copy per environment. The client takes serverless_environment, each call can override it, and unset sends nothing so the platform keeps reading production. Waiting polls in the invocation's environment, since a task carries none of its own. Signed-off-by: Rafal Leszko --- README.md | 16 +++++++ runware/client.py | 21 +++++++- runware/serverless.py | 60 +++++++++++++++++++++-- runware/types/sdk.py | 4 ++ tests/test_serverless.py | 101 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 195 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index a26e476..6daf054 100644 --- a/README.md +++ b/README.md @@ -597,6 +597,22 @@ task = await client.get_task("image-tools", accepted["id"]) A task that runs and fails comes back with `status` set to `"failed"` and its `error` set. Only a failure to start raises. +An app id names one copy of the app per environment. Calls go to `production` unless you name another environment, for the whole client or for one call. Waiting polls the task in the environment it was invoked in; when you call `get_task` yourself, pass that same environment. + +```python +client = Runware(serverless_environment="staging") + +task = await client.invoke("image-tools", "generate", {"prompt": "a red bicycle"}) + +task = await client.invoke( + "image-tools", "generate", {"prompt": "a red bicycle"}, environment="production", +) +accepted = await client.invoke( + "image-tools", "train", {"epochs": 10}, environment="dev", options=InvokeOptions(wait=False), +) +task = await client.get_task("image-tools", accepted["id"], environment="dev") +``` + ## Content metadata `client.content.*` exposes Runware's curated model catalog as read-only metadata — names, AIRs, headlines, capabilities, pricing, examples. Public information, no extra cost. diff --git a/runware/client.py b/runware/client.py index 7dda35b..ab86148 100644 --- a/runware/client.py +++ b/runware/client.py @@ -425,6 +425,7 @@ async def invoke( payload: dict[str, Any] | None = None, # pyright: ignore[reportExplicitAny] *, task_id: str | None = None, + environment: str | None = None, options: InvokeOptions | None = None, ) -> ServerlessTask: """ @@ -444,12 +445,21 @@ async def invoke( Sending the same id again returns the task it already names instead of starting a second, so a call whose response was lost is safe to repeat. + ``environment`` selects which copy of the app to call, falling back to + ``config.serverless_environment`` and then to ``production``. Waiting + polls the task in that same environment. + Raises a ``RunwareError`` when the task fails to start. A task that runs and fails comes back with ``status`` ``failed`` and its ``error`` set, because that is an outcome rather than a broken call. """ return await self._serverless.invoke( - app_id, endpoint_path, payload, task_id=task_id, options=options, + app_id, + endpoint_path, + payload, + task_id=task_id, + environment=environment, + options=options, ) async def get_task( @@ -457,6 +467,8 @@ async def get_task( app_id: str, task_id: str, options: GetTaskOptions | None = None, + *, + environment: str | None = None, ) -> ServerlessTask: """ Read one serverless task by id. @@ -464,8 +476,13 @@ async def get_task( ``invoke`` already waits for you, so reach for this when you want to poll yourself: after ``invoke`` with ``wait=False``, or from a different process than the one that submitted the task. + + Pass the ``environment`` the task was invoked in. It falls back to + ``config.serverless_environment`` and then to ``production``. """ - return await self._serverless.get_task(app_id, task_id, options) + return await self._serverless.get_task( + app_id, task_id, options, environment=environment, + ) # ----------------------------------------------------------- internal diff --git a/runware/serverless.py b/runware/serverless.py index 13b91d5..659569b 100644 --- a/runware/serverless.py +++ b/runware/serverless.py @@ -36,6 +36,7 @@ _APP_ID_PATTERN = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") _ENDPOINT_PATH_PATTERN = re.compile(r"^[a-z]([a-z0-9-]{0,62}[a-z0-9])?$") +_ENVIRONMENT_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,62}$") _TASK_ID_PATTERN = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", ) @@ -121,6 +122,21 @@ def validate_endpoint_path(endpoint_path: object) -> str: return endpoint_path +def validate_environment(environment: object) -> str | None: + """Check an environment name against the platform's own pattern.""" + if environment is None: + return None + if not isinstance(environment, str) or not _ENVIRONMENT_PATTERN.match(environment): + raise create_runware_error( + "invalidParameter", + f'environment "{environment}" is invalid: use 1 to 63 lowercase ' + + "letters, digits, hyphens, underscores or periods, starting with " + + "a letter or digit", + parameter="environment", + ) + return environment + + def validate_task_id(task_id: object) -> str: """Check a task id is a canonical lowercase UUID.""" if not isinstance(task_id, str) or not _TASK_ID_PATTERN.match(task_id): @@ -230,6 +246,17 @@ def __init__( def _origin(self) -> str: return self._config.serverless_base_url.rstrip("/") + def _environment(self, environment: str | None) -> str | None: + """The per-call environment, else the client's, else none.""" + return validate_environment( + environment if environment is not None + else self._config.serverless_environment, + ) + + def _app_url(self, app: str, path: str, environment: str | None) -> str: + url = f"{self._origin}/v1/apps/{app}/{path}" + return url if environment is None else f"{url}?environment={environment}" + async def invoke( self, app_id: str, @@ -237,6 +264,7 @@ async def invoke( payload: dict[str, Any] | None = None, # pyright: ignore[reportExplicitAny] *, task_id: str | None = None, + environment: str | None = None, options: InvokeOptions | None = None, ) -> ServerlessTask: """ @@ -248,6 +276,10 @@ async def invoke( ) print(task["output"]) + ``environment`` selects which copy of the app to call, falling back to + ``config.serverless_environment`` and then to ``production``. Waiting + polls the task in that same environment. + ``delivery_method`` decides how the wait happens, not whether you get a result: ``async`` (the default) takes an acknowledgement and polls, ``sync`` holds one request open and is the fast path for work that @@ -266,6 +298,7 @@ async def invoke( app = validate_app_id(app_id) endpoint = validate_endpoint_path(endpoint_path) resolved_task_id = _resolve_task_id(task_id) + resolved_environment = self._environment(environment) delivery_method = options.delivery_method if options.wait else "async" route = "invoke-sync" if delivery_method == "sync" else "invoke-async" @@ -285,7 +318,7 @@ async def invoke( started_at = time.monotonic() _status, response_body = await self._request( "POST", - f"{self._origin}/v1/apps/{app}/{route}/{endpoint}", + self._app_url(app, f"{route}/{endpoint}", resolved_environment), body, timeout_ms, options.cancel_event, @@ -296,13 +329,17 @@ async def invoke( if not options.wait or is_terminal(task["status"]): return task - return await self._wait_for_task(task, options, started_at) + return await self._wait_for_task( + task, resolved_environment, options, started_at, + ) async def get_task( self, app_id: str, task_id: str, options: GetTaskOptions | None = None, + *, + environment: str | None = None, ) -> ServerlessTask: """ Read one serverless task by id. @@ -310,13 +347,17 @@ async def get_task( ``invoke`` already waits for you, so reach for this when you want to poll yourself: after ``invoke`` with ``wait=False``, or from a different process than the one that submitted the task. + + Pass the ``environment`` the task was invoked in. It falls back to + ``config.serverless_environment`` and then to ``production``. """ options = options or GetTaskOptions() app = validate_app_id(app_id) resolved = validate_task_id(task_id) + resolved_environment = self._environment(environment) _status, body = await self._request( "GET", - f"{self._origin}/v1/apps/{app}/tasks/{resolved}", + self._app_url(app, f"tasks/{resolved}", resolved_environment), None, options.timeout or self._config.timeout, options.cancel_event, @@ -326,12 +367,16 @@ async def get_task( async def _wait_for_task( self, task: ServerlessTask, + environment: str | None, options: InvokeOptions, started_at: float, ) -> ServerlessTask: """ Poll until the task reaches a terminal state. Never resubmits: the task is already running, and a second invocation would be a second charge. + + Polls in the environment the task was invoked in. A task carries no + environment of its own, and a poll without one would read production. """ interval = ( options.poll_interval @@ -344,13 +389,17 @@ async def _wait_for_task( while not is_terminal(current["status"]): if time.monotonic() - started_at > budget_s: + environment_arg = ( + "" if environment is None + else f', environment="{environment}"' + ) raise create_runware_error( "timeout", f'Task {current["id"]} is still running after ' + f"{options.poll_timeout or self._config.poll_timeout}ms. " + "It was not cancelled: poll " - + f'get_task("{current["appId"]}", "{current["id"]}") ' - + "to pick it up.", + + f'get_task("{current["appId"]}", "{current["id"]}"' + + f"{environment_arg}) to pick it up.", task_uuid=current["id"], ) @@ -364,6 +413,7 @@ async def _wait_for_task( timeout=options.timeout, cancel_event=options.cancel_event, ), + environment=environment, ) not_found_since = None except RunwareError as error: diff --git a/runware/types/sdk.py b/runware/types/sdk.py index 0ec7b8c..87b3da1 100644 --- a/runware/types/sdk.py +++ b/runware/types/sdk.py @@ -57,6 +57,10 @@ class SDKConfig: http_base_url: str = "https://api.runware.ai/v1" # Serverless API origin, without a version path. Used by invoke and get_task. serverless_base_url: str = "https://api.serverless.runware.ai" + # Serverless environment invoke and get_task address, such as "staging". + # An app id names one copy per environment. None sends no environment, + # which the platform reads as "production". + serverless_environment: str | None = None timeout: int = 1_200_000 poll_timeout: int = 1_200_000 diff --git a/tests/test_serverless.py b/tests/test_serverless.py index edbc62f..5a01911 100644 --- a/tests/test_serverless.py +++ b/tests/test_serverless.py @@ -527,3 +527,104 @@ async def test_get_task_honors_a_per_call_timeout_option() -> None: result = await api.get_task("my-app", TASK_ID, GetTaskOptions(timeout=1_000)) assert result["id"] == TASK_ID + + +# ---------------------------------------------------------------- environment + + +STAGING_ASYNC_URL = f"{ASYNC_URL}?environment=staging" +STAGING_TASK_URL = f"{TASK_URL}?environment=staging" + + +@pytest.mark.asyncio +async def test_sends_no_environment_when_none_is_set() -> None: + session = MockSession() + session.add("POST", ASYNC_URL, MockResponse(status=200, payload=task())) + session.add("GET", TASK_URL, MockResponse(status=200, payload=task())) + api = make_api(session) + + await api.invoke("my-app", "generate") + await api.get_task("my-app", TASK_ID) + + assert [entry.url for entry in session.request_log] == [ASYNC_URL, TASK_URL] + + +@pytest.mark.asyncio +async def test_addresses_the_client_environment() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=200, payload=task())) + session.add("GET", STAGING_TASK_URL, MockResponse(status=200, payload=task())) + api = make_api(session, serverless_environment="staging") + + await api.invoke("my-app", "generate") + await api.get_task("my-app", TASK_ID) + + assert [entry.url for entry in session.request_log] == [ + STAGING_ASYNC_URL, STAGING_TASK_URL, + ] + + +@pytest.mark.asyncio +async def test_a_per_call_environment_overrides_the_client_one() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=200, payload=task())) + session.add("GET", STAGING_TASK_URL, MockResponse(status=200, payload=task())) + api = make_api(session, serverless_environment="production") + + await api.invoke("my-app", "generate", environment="staging") + await api.get_task("my-app", TASK_ID, environment="staging") + + assert [entry.url for entry in session.request_log] == [ + STAGING_ASYNC_URL, STAGING_TASK_URL, + ] + + +@pytest.mark.asyncio +async def test_polls_in_the_environment_the_task_was_invoked_in() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=202, payload=pending())) + session.add("GET", STAGING_TASK_URL, MockResponse(status=200, payload=pending())) + session.add("GET", STAGING_TASK_URL, MockResponse(status=200, payload=task())) + api = make_api(session) + + result = await api.invoke( + "my-app", "generate", environment="staging", options=no_wait(), + ) + + assert result["status"] == "completed" + assert session.count("GET", STAGING_TASK_URL) == 2 + assert session.count("GET", TASK_URL) == 0 + + +@pytest.mark.asyncio +async def test_names_the_environment_when_the_poll_budget_runs_out() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=202, payload=pending())) + api = make_api(session, poll_timeout=0) + + with pytest.raises(RunwareError, match=r'environment="staging"\) to pick it up'): + _ = await api.invoke( + "my-app", "generate", environment="staging", options=no_wait(), + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("environment", ["", "Staging", "-dev", "a/b", "a" * 64]) +async def test_rejects_an_environment_that_cannot_exist(environment: str) -> None: + session = MockSession() + api = make_api(session) + + with pytest.raises(RunwareError) as excinfo: + _ = await api.invoke("my-app", "generate", environment=environment) + assert excinfo.value.parameter == "environment" + assert session.request_log == [] + + +@pytest.mark.asyncio +async def test_rejects_an_invalid_client_environment_before_any_request() -> None: + session = MockSession() + api = make_api(session, serverless_environment="Staging") + + with pytest.raises(RunwareError, match='environment "Staging" is invalid'): + _ = await api.get_task("my-app", TASK_ID) + assert session.request_log == [] From 4c6d7be0b030b8fa8131a181e446bae31e470548 Mon Sep 17 00:00:00 2001 From: Rafal Leszko Date: Wed, 21 Oct 2026 11:13:43 +0200 Subject: [PATCH 2/4] fix(serverless): check the client environment when built and test the wrappers (RUNSERV-1238) Review follow-ups: an invalid serverless_environment now fails when the client is built and names the config field, the name must match in full (no trailing newline), and the query goes through urlencode. Tests cover the public Runware wrappers, the name boundaries and the smoke test's optional RUNWARE_SERVERLESS_TEST_ENVIRONMENT. Signed-off-by: Rafal Leszko --- runware/serverless.py | 20 ++++++++--- tests/integration/test_smoke.py | 8 ++++- tests/test_serverless.py | 64 +++++++++++++++++++++++++++++---- 3 files changed, 79 insertions(+), 13 deletions(-) diff --git a/runware/serverless.py b/runware/serverless.py index 659569b..885ffd2 100644 --- a/runware/serverless.py +++ b/runware/serverless.py @@ -19,6 +19,7 @@ import uuid from collections.abc import Awaitable, Callable from typing import Any, cast +from urllib.parse import urlencode import aiohttp @@ -122,17 +123,21 @@ def validate_endpoint_path(endpoint_path: object) -> str: return endpoint_path -def validate_environment(environment: object) -> str | None: +def validate_environment( + environment: object, parameter: str = "environment", +) -> str | None: """Check an environment name against the platform's own pattern.""" if environment is None: return None - if not isinstance(environment, str) or not _ENVIRONMENT_PATTERN.match(environment): + if not isinstance(environment, str) or not _ENVIRONMENT_PATTERN.fullmatch( + environment, + ): raise create_runware_error( "invalidParameter", - f'environment "{environment}" is invalid: use 1 to 63 lowercase ' + f'{parameter} "{environment}" is invalid: use 1 to 63 lowercase ' + "letters, digits, hyphens, underscores or periods, starting with " + "a letter or digit", - parameter="environment", + parameter=parameter, ) return environment @@ -239,6 +244,9 @@ class ServerlessApi: def __init__( self, session_provider: SessionProvider, config: SDKConfig, ) -> None: + _ = validate_environment( + config.serverless_environment, "serverless_environment", + ) self._session_provider: SessionProvider = session_provider self._config: SDKConfig = config @@ -255,7 +263,9 @@ def _environment(self, environment: str | None) -> str | None: def _app_url(self, app: str, path: str, environment: str | None) -> str: url = f"{self._origin}/v1/apps/{app}/{path}" - return url if environment is None else f"{url}?environment={environment}" + if environment is None: + return url + return f"{url}?{urlencode({'environment': environment})}" async def invoke( self, diff --git a/tests/integration/test_smoke.py b/tests/integration/test_smoke.py index d0e1dbc..a623365 100644 --- a/tests/integration/test_smoke.py +++ b/tests/integration/test_smoke.py @@ -32,6 +32,8 @@ SERVERLESS_BASE = os.environ.get("RUNWARE_SERVERLESS_BASE_URL") SERVERLESS_APP = os.environ.get("RUNWARE_SERVERLESS_TEST_APP") SERVERLESS_ENDPOINT = os.environ.get("RUNWARE_SERVERLESS_TEST_ENDPOINT") +# Optional: the environment of the app copy to call. Unset means production. +SERVERLESS_ENVIRONMENT = os.environ.get("RUNWARE_SERVERLESS_TEST_ENVIRONMENT") or None SERVERLESS_PAYLOAD: dict[str, object] = json.loads( os.environ.get("RUNWARE_SERVERLESS_TEST_PAYLOAD") or "{}", ) @@ -42,7 +44,11 @@ def serverless_client() -> Runware: - return Runware(serverless_base_url=SERVERLESS_BASE or "", transport="rest") + return Runware( + serverless_base_url=SERVERLESS_BASE or "", + serverless_environment=SERVERLESS_ENVIRONMENT, + transport="rest", + ) IMAGE_MODEL = "runware:400@2" # Flux 2 Klein 9b — cheap and fast TEXT_MODEL = "google:gemma@4-31b" # cheap and fast diff --git a/tests/test_serverless.py b/tests/test_serverless.py index 5a01911..beb4f92 100644 --- a/tests/test_serverless.py +++ b/tests/test_serverless.py @@ -8,9 +8,10 @@ import aiohttp import pytest +from runware import Runware from runware.errors import RunwareError from runware.serverless import ServerlessApi -from runware.types.sdk import SDKConfig +from runware.types.sdk import RuntimeDependencies, SDKConfig from runware.types.serverless import GetTaskOptions, InvokeOptions from ._mocks import MockResponse, MockSession @@ -609,7 +610,9 @@ async def test_names_the_environment_when_the_poll_budget_runs_out() -> None: @pytest.mark.asyncio -@pytest.mark.parametrize("environment", ["", "Staging", "-dev", "a/b", "a" * 64]) +@pytest.mark.parametrize( + "environment", ["", "Staging", "-dev", ".dev", "_dev", "a/b", "staging\n", "a" * 64], +) async def test_rejects_an_environment_that_cannot_exist(environment: str) -> None: session = MockSession() api = make_api(session) @@ -621,10 +624,57 @@ async def test_rejects_an_environment_that_cannot_exist(environment: str) -> Non @pytest.mark.asyncio -async def test_rejects_an_invalid_client_environment_before_any_request() -> None: +@pytest.mark.parametrize( + "environment", ["pr-12", "feature_x", "v1.2", "1st", "a" * 63], +) +async def test_accepts_every_name_the_platform_allows(environment: str) -> None: + url = f"{ASYNC_URL}?environment={environment}" session = MockSession() - api = make_api(session, serverless_environment="Staging") + session.add("POST", url, MockResponse(status=200, payload=task())) + api = make_api(session) - with pytest.raises(RunwareError, match='environment "Staging" is invalid'): - _ = await api.get_task("my-app", TASK_ID) - assert session.request_log == [] + await api.invoke("my-app", "generate", environment=environment) + + assert session.request_log[0].url == url + + +def test_rejects_an_invalid_client_environment_when_built() -> None: + with pytest.raises(RunwareError) as excinfo: + _ = make_api(MockSession(), serverless_environment="Staging") + assert excinfo.value.parameter == "serverless_environment" + assert 'serverless_environment "Staging" is invalid' in str(excinfo.value) + + +@pytest.mark.asyncio +async def test_names_no_environment_in_the_hint_when_none_is_set() -> None: + session = MockSession() + session.add("POST", ASYNC_URL, MockResponse(status=202, payload=pending())) + api = make_api(session, poll_timeout=0) + + with pytest.raises(RunwareError, match=rf'get_task\("my-app", "{TASK_ID}"\) to pick it up'): + _ = await api.invoke("my-app", "generate", options=no_wait()) + + +@pytest.mark.asyncio +async def test_the_client_passes_a_per_call_environment_through() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=200, payload=task())) + session.add("GET", STAGING_TASK_URL, MockResponse(status=200, payload=task())) + client = Runware( + api_key="test-key", + serverless_base_url=ORIGIN, + transport="rest", + dependencies=RuntimeDependencies( + session=cast("aiohttp.ClientSession", session), + ), + ) + + try: + await client.invoke("my-app", "generate", environment="staging") + await client.get_task("my-app", TASK_ID, environment="staging") + finally: + await client.close() + + assert [entry.url for entry in session.request_log] == [ + STAGING_ASYNC_URL, STAGING_TASK_URL, + ] From c3d6b57ef006a503b212f567e0996d9f2352fc36 Mon Sep 17 00:00:00 2001 From: Rafal Leszko Date: Wed, 21 Oct 2026 11:16:06 +0200 Subject: [PATCH 3/4] test(serverless): cover the client-level environment through Runware (RUNSERV-1238) Signed-off-by: Rafal Leszko --- tests/test_serverless.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/test_serverless.py b/tests/test_serverless.py index beb4f92..22dc28c 100644 --- a/tests/test_serverless.py +++ b/tests/test_serverless.py @@ -678,3 +678,25 @@ async def test_the_client_passes_a_per_call_environment_through() -> None: assert [entry.url for entry in session.request_log] == [ STAGING_ASYNC_URL, STAGING_TASK_URL, ] + + +@pytest.mark.asyncio +async def test_the_client_addresses_its_own_environment() -> None: + session = MockSession() + session.add("POST", STAGING_ASYNC_URL, MockResponse(status=200, payload=task())) + client = Runware( + api_key="test-key", + serverless_base_url=ORIGIN, + serverless_environment="staging", + transport="rest", + dependencies=RuntimeDependencies( + session=cast("aiohttp.ClientSession", session), + ), + ) + + try: + await client.invoke("my-app", "generate") + finally: + await client.close() + + assert session.request_log[0].url == STAGING_ASYNC_URL From d8d3b636e442ce687ed3fc76f287badc6addca1d Mon Sep 17 00:00:00 2001 From: Rafal Leszko Date: Wed, 21 Oct 2026 13:29:20 +0200 Subject: [PATCH 4/4] feat(serverless): stop retrying an environment paused at its budget (RUNSERV-1238) A paused environment answers invoke with a 429 that holds until the budget is raised or the UTC month turns, so retrying it, or sleeping on its Retry-After, only delays the error. It now raises a quota error at once; any other 429 is still retried. Signed-off-by: Rafal Leszko --- README.md | 2 +- runware/errors.py | 1 + runware/serverless.py | 14 ++++++++---- tests/test_serverless.py | 47 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 59 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 6daf054..cbc114c 100644 --- a/README.md +++ b/README.md @@ -597,7 +597,7 @@ task = await client.get_task("image-tools", accepted["id"]) A task that runs and fails comes back with `status` set to `"failed"` and its `error` set. Only a failure to start raises. -An app id names one copy of the app per environment. Calls go to `production` unless you name another environment, for the whole client or for one call. Waiting polls the task in the environment it was invoked in; when you call `get_task` yourself, pass that same environment. +An app id names one copy of the app per environment. Calls go to `production` unless you name another environment, for the whole client or for one call. Waiting polls the task in the environment it was invoked in; when you call `get_task` yourself, pass that same environment. An environment paused at its monthly budget refuses an invocation with a `quota` error, which is not retried: it holds until the budget is raised or the UTC month turns. ```python client = Runware(serverless_environment="staging") diff --git a/runware/errors.py b/runware/errors.py index 8e530db..cc6a6ca 100644 --- a/runware/errors.py +++ b/runware/errors.py @@ -158,6 +158,7 @@ def derive_code(raw: str) -> ErrorCode: or "Balance" in raw or raw == "quotaExceeded" or raw == "paymentRequired" + or raw == "environmentPaused" ): return "quota" if "RateLimit" in raw or raw == "rateLimitExceeded": diff --git a/runware/serverless.py b/runware/serverless.py index 885ffd2..14c93bd 100644 --- a/runware/serverless.py +++ b/runware/serverless.py @@ -57,6 +57,10 @@ _TERMINAL = frozenset({"completed", "failed"}) +# The 429 an environment paused at its monthly budget answers invoke with. It +# holds until the budget is raised or the UTC month turns, so it is not retried. +_ENVIRONMENT_PAUSED_TYPE = "https://docs.runware.ai/serverless/errors#environment-paused" + def is_terminal(status: str) -> bool: """Whether a task status is final.""" @@ -208,10 +212,12 @@ def _problem_to_error( if lines: message = message + "\n" + "\n".join(lines) - error = create_runware_error( - _raw_code_for_status(status), message, status_code=status, - ) problem_type = problem.get("type") + raw_code = ( + "environmentPaused" if problem_type == _ENVIRONMENT_PAUSED_TYPE + else _raw_code_for_status(status) + ) + error = create_runware_error(raw_code, message, status_code=status) if isinstance(problem_type, str) and problem_type != "about:blank": error.problem_type = problem_type request_id = problem.get("requestId") @@ -488,7 +494,7 @@ async def _request( @staticmethod def _should_retry(error: BaseException) -> bool: if is_runware_error(error): - if error.code == "aborted": + if error.code in ("aborted", "quota"): return False if error.code == "timeout": return True diff --git a/tests/test_serverless.py b/tests/test_serverless.py index 22dc28c..8eff877 100644 --- a/tests/test_serverless.py +++ b/tests/test_serverless.py @@ -700,3 +700,50 @@ async def test_the_client_addresses_its_own_environment() -> None: await client.close() assert session.request_log[0].url == STAGING_ASYNC_URL + + +# ------------------------------------------------------------------- budget + + +PAUSED_PROBLEM = { + "type": "https://docs.runware.ai/serverless/errors#environment-paused", + "title": "Too Many Requests", + "status": 429, + "detail": 'environment "staging" is paused at its monthly budget', +} + + +@pytest.mark.asyncio +async def test_does_not_retry_an_environment_paused_at_its_budget() -> None: + session = MockSession() + session.add( + "POST", + STAGING_ASYNC_URL, + MockResponse(status=429, payload=PAUSED_PROBLEM, headers={"Retry-After": "0"}), + ) + api = make_api(session, max_retries=3) + + with pytest.raises(RunwareError, match="paused at its monthly budget") as excinfo: + _ = await api.invoke("my-app", "generate", environment="staging") + + assert session.count("POST", STAGING_ASYNC_URL) == 1 + assert excinfo.value.code == "quota" + assert excinfo.value.retryable is False + assert excinfo.value.status_code == 429 + + +@pytest.mark.asyncio +async def test_still_retries_any_other_429() -> None: + session = MockSession() + session.add( + "POST", + ASYNC_URL, + MockResponse(status=429, payload={"status": 429}, headers={"Retry-After": "0"}), + ) + session.add("POST", ASYNC_URL, MockResponse(status=200, payload=task())) + api = make_api(session, max_retries=2) + + result = await api.invoke("my-app", "generate") + + assert result["status"] == "completed" + assert session.count("POST", ASYNC_URL) == 2