From dea0cb1a2efad9d3962426add4d9929955966781 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:15:39 -0800 Subject: [PATCH] fix: harden JSON execution boundaries --- README.md | 12 ++ requirements.txt | 1 - runpod/_serialization.py | 21 +++ runpod/apps/build.py | 1 - runpod/apps/handles.py | 13 +- runpod/apps/images.py | 8 +- runpod/apps/protocol.py | 48 +++-- runpod/apps/serialization.py | 47 ++--- runpod/apps/targets.py | 13 +- runpod/apps/tasks.py | 19 +- runpod/serverless/modules/rp_fastapi.py | 17 +- runpod/serverless/modules/rp_http.py | 17 +- runpod/serverless/modules/rp_job.py | 10 +- tests/test_apps/test_dispatch.py | 28 ++- tests/test_apps/test_tasks.py | 18 +- tests/test_serialization.py | 231 ++++++++++++++++++++++++ 16 files changed, 398 insertions(+), 106 deletions(-) create mode 100644 runpod/_serialization.py create mode 100644 tests/test_serialization.py diff --git a/README.md b/README.md index 6f5a3fc4d..b3fa882c0 100644 --- a/README.md +++ b/README.md @@ -84,6 +84,18 @@ make setup make test ``` +## Remote values + +Queue and task calls use finite JSON-compatible arguments, return values, +and stream chunks. JSON conversion turns tuples into lists and supported +dictionary keys into strings, including in same-worker remote calls. +Bytes, custom objects, circular references, and non-finite numbers produce +explicit errors. `.local()` remains a direct Python call. + +Successful function responses contain `json_result`, including for a null +return value. SDK and worker runtimes share the JSON protocol validators; +use compatible package releases when building worker images. + ## ⚡ | Serverless Worker (SDK) This python package can also be used to create a serverless worker that can be deployed to Runpod as a custom endpoint API. diff --git a/requirements.txt b/requirements.txt index 0bfa95d23..566603c1c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,7 +5,6 @@ backoff >= 2.2.1 boto3 >= 1.43.118 certifi >= 2026.10.19 click >= 8.5.0 -cloudpickle >= 3.1.2 colorama >= 0.4.6, < 0.4.7 cryptography >= 50.0.2 fastapi[all] >= 0.162.0 diff --git a/runpod/_serialization.py b/runpod/_serialization.py new file mode 100644 index 000000000..7d4beb7e9 --- /dev/null +++ b/runpod/_serialization.py @@ -0,0 +1,21 @@ +"""finite json validation and wire-value conversion for callers and workers.""" + +import json +from typing import Any + + +def _json_text(value: Any, label: str) -> str: + try: + return json.dumps(value, allow_nan=False) + except (TypeError, ValueError, RecursionError) as exc: + raise TypeError(f"{label} must be json-serializable: {exc}") from exc + + +def validate_json(value: Any, label: str = "value") -> None: + """require values encodable as finite json.""" + _json_text(value, label) + + +def normalize_json(value: Any, label: str = "value") -> Any: + """apply json's container and key conversions without a network roundtrip.""" + return json.loads(_json_text(value, label)) diff --git a/runpod/apps/build.py b/runpod/apps/build.py index efb91dac6..91575e67f 100644 --- a/runpod/apps/build.py +++ b/runpod/apps/build.py @@ -50,7 +50,6 @@ "aiohttp[speedups]", "aiohttp-retry", "backoff", - "cloudpickle", "py-cpuinfo", "requests", "tomli", diff --git a/runpod/apps/handles.py b/runpod/apps/handles.py index 0ab02e680..e4e804f61 100644 --- a/runpod/apps/handles.py +++ b/runpod/apps/handles.py @@ -20,7 +20,6 @@ def load_model(): ... # worker startup hook import asyncio import inspect -import json from typing import ( TYPE_CHECKING, Any, @@ -32,6 +31,7 @@ def load_model(): ... # worker startup hook Type, ) +from .._serialization import normalize_json from .context import Context, block, current_context from .discovery_state import DiscoveryInvocationError, in_discovery from .errors import InvalidResourceError @@ -39,7 +39,6 @@ def load_model(): ... # worker startup hook from .job import Job from .markers import INIT_ATTR, is_init, route_of from .schedule import SCHEDULE_ATTR -from .serialization import require_json_value from .spec import ResourceKind, ResourceSpec, RouteSpec if TYPE_CHECKING: @@ -126,10 +125,12 @@ def _call_local_sync(self, *args: Any, **kwargs: Any) -> Any: @staticmethod def _json_value(value: Any) -> Any: - require_json_value( - value, "remote arguments and results must be JSON-serializable" - ) - return json.loads(json.dumps(value, allow_nan=False)) + try: + return normalize_json(value) + except TypeError: + raise ValueError( + "remote arguments and results must be JSON-serializable" + ) from None @staticmethod async def _await_local_result(result: Any) -> Any: diff --git a/runpod/apps/images.py b/runpod/apps/images.py index fb00e1947..343f58fa7 100644 --- a/runpod/apps/images.py +++ b/runpod/apps/images.py @@ -38,11 +38,9 @@ def runtime_tag() -> str: def local_python_version() -> str: """the local interpreter's version, which the worker must match. - dev sessions and tasks ship cloudpickle payloads produced on the - client interpreter; those are not reliably portable across python - versions, so a silent fallback would surface as cryptic - deserialization failures on the worker. mismatches fail here, - loudly, instead. + live execution ships python source from the client. matching the + interpreter keeps syntax and language behavior consistent between + local development and the worker. """ version = f"{sys.version_info.major}.{sys.version_info.minor}" if version not in SUPPORTED_PYTHON_VERSIONS: diff --git a/runpod/apps/protocol.py b/runpod/apps/protocol.py index bc9f34bbf..dea7d5cfa 100644 --- a/runpod/apps/protocol.py +++ b/runpod/apps/protocol.py @@ -1,21 +1,20 @@ """wire protocol for remote function execution. this is the contract between the sdk and the worker runtime images. -the worker repo imports these shapes from the runpod package so client -and worker can never drift. +the worker runtime and client share these shapes and validation rules. live mode ships function source with every request; deployed mode omits the code (the worker unpacked the build and resolves the function by -name). args/kwargs are base64 cloudpickle unless serialization_format -is "json". +name). arguments, results, and stream chunks are json values. """ -from dataclasses import asdict, dataclass, field +from dataclasses import asdict, dataclass, field, fields from typing import Any, Dict, List, Optional +from .._serialization import normalize_json, validate_json + EXECUTION_FUNCTION = "function" -FORMAT_CLOUDPICKLE = "cloudpickle" FORMAT_JSON = "json" @@ -31,10 +30,30 @@ class FunctionRequest: system_dependencies: Optional[List[str]] = None execution_type: str = EXECUTION_FUNCTION accelerate_downloads: bool = True - serialization_format: str = FORMAT_CLOUDPICKLE + serialization_format: str = FORMAT_JSON + + def validate(self) -> None: + if self.serialization_format != FORMAT_JSON: + raise ValueError("serialization_format must be 'json'") + if not isinstance(self.args, list) or not isinstance(self.kwargs, dict): + raise TypeError("args must be a list and kwargs must be a dictionary") + validate_json( + {f.name: getattr(self, f.name) for f in fields(self)}, "function request" + ) + + @classmethod + def from_input(cls, data: Dict[str, Any]) -> "FunctionRequest": + if not isinstance(data, dict): + raise TypeError("function request must be a dictionary") + request = cls(**{f.name: data[f.name] for f in fields(cls) if f.name in data}) + request.validate() + request.args = normalize_json(request.args, "arguments") + request.kwargs = normalize_json(request.kwargs, "keyword arguments") + return request def to_input(self) -> Dict[str, Any]: """serialize as the job input dict.""" + self.validate() data = asdict(self) return {k: v for k, v in data.items() if v is not None} @@ -44,17 +63,24 @@ class FunctionResponse: """the worker's result for one function execution.""" success: bool - result: Optional[str] = None json_result: Any = None error: Optional[str] = None stdout: Optional[str] = None @classmethod def from_output(cls, output: Dict[str, Any]) -> "FunctionResponse": + if not isinstance(output, dict) or not isinstance(output.get("success"), bool): + raise TypeError("function response must contain a boolean success field") + if output.get("serialization_format", FORMAT_JSON) != FORMAT_JSON: + raise ValueError("serialization_format must be 'json'") + if output.get("result") is not None: + raise ValueError("function responses must use json_result") + if output["success"] and "json_result" not in output: + raise ValueError("successful function response is missing json_result") + validate_json(output, "function response") return cls( - success=output.get("success", False), - result=output.get("result"), - json_result=output.get("json_result"), + success=output["success"], + json_result=normalize_json(output.get("json_result"), "return value"), error=output.get("error"), stdout=output.get("stdout"), ) diff --git a/runpod/apps/serialization.py b/runpod/apps/serialization.py index a7c8f5bb1..0972e2859 100644 --- a/runpod/apps/serialization.py +++ b/runpod/apps/serialization.py @@ -1,52 +1,37 @@ """argument and result serialization for remote execution. remote calls use finite json-compatible arguments and results. this -module validates that contract, provides cloudpickle codecs for explicitly -encoded values, and extracts function source for live workers. +module validates that contract, decodes json response envelopes, and extracts +function source for live workers. """ import ast -import base64 import inspect import os import textwrap -from typing import Any, Callable, Dict, List +from typing import Any, Callable -import cloudpickle +from .._serialization import validate_json +from .errors import RemoteExecutionError +from .protocol import FunctionResponse -def serialize_arg(arg: Any) -> str: - return base64.b64encode(cloudpickle.dumps(arg)).decode("utf-8") - - -def serialize_args(args: tuple) -> List[str]: - return [serialize_arg(a) for a in args] - - -def serialize_kwargs(kwargs: dict) -> Dict[str, str]: - return {k: serialize_arg(v) for k, v in kwargs.items()} - - -def deserialize_result(result_b64: str) -> Any: +def unwrap_function_response(output: dict) -> Any: try: - return cloudpickle.loads(base64.b64decode(result_b64)) - except ModuleNotFoundError as exc: - from .errors import RemoteExecutionError - + response = FunctionResponse.from_output(output) + except (TypeError, ValueError) as exc: + raise RemoteExecutionError(f"invalid remote response: {exc}") from exc + if not response.success: raise RemoteExecutionError( - f"the remote result contains an object from the " - f"'{exc.name}' package, which is not installed locally. " - f"return plain python types (e.g. str(...)) or install " - f"'{exc.name}' on this machine." - ) from exc + f"remote execution failed: {response.error or 'unknown'}" + ) + return response.json_result def require_json_value(value: Any, message: str) -> None: try: - import json - - json.dumps(value, allow_nan=False) - except (TypeError, ValueError): + validate_json(value) + except TypeError: raise ValueError(message) from None diff --git a/runpod/apps/targets.py b/runpod/apps/targets.py index 780139dc9..011641d53 100644 --- a/runpod/apps/targets.py +++ b/runpod/apps/targets.py @@ -15,7 +15,7 @@ from ..user_agent import USER_AGENT from .errors import EndpointNotFound, RemoteExecutionError -from .protocol import FORMAT_JSON, FunctionRequest, FunctionResponse +from .protocol import FORMAT_JSON, FunctionRequest from .serialization import require_json_value from .spec import ResourceSpec from .utils.events import emit @@ -696,16 +696,9 @@ def unwrap(self, data: Dict[str, Any]) -> Any: def _unwrap_response(self, output: Any) -> Any: require_json_value(output, "queue results must be JSON-serializable") if isinstance(output, dict) and "success" in output: - response = FunctionResponse.from_output(output) - if not response.success: - raise RemoteExecutionError( - f"remote execution failed: {response.error or 'unknown'}" - ) - if response.result is not None: - from .serialization import deserialize_result + from .serialization import unwrap_function_response - return deserialize_result(response.result) - return response.json_result + return unwrap_function_response(output) return output def _monitor(self): diff --git a/runpod/apps/tasks.py b/runpod/apps/tasks.py index 78af5dc08..9663aee31 100644 --- a/runpod/apps/tasks.py +++ b/runpod/apps/tasks.py @@ -21,7 +21,6 @@ from .api import AppsApiClient, is_capacity_error from .context import CLEANUP_TIMEOUT -from .errors import RemoteExecutionError from .spec import ResourceSpec log = logging.getLogger(__name__) @@ -264,8 +263,11 @@ async def set_timeout(self, timeout: float) -> None: await self._post("/timeout", {"timeout": timeout}) async def _post(self, path: str, request: Dict[str, Any]) -> None: + from .serialization import require_json_value + timeout = request.get("timeout") _validate_timeout(timeout) + require_json_value(request, "task request body must be JSON-serializable") url = f"{_proxy_url(self.pod_id)}{path}" attempts = 6 async with aiohttp.ClientSession() as session: @@ -473,19 +475,10 @@ async def _terminate_pod(self): def unwrap_task_response(response: Dict[str, Any]) -> Any: """extract the function result from a runner response.""" - if not response.get("success"): - raise RemoteExecutionError( - f"task execution failed: {response.get('error', 'unknown')}" - ) - if response.get("result") is not None: - from .serialization import deserialize_result - - return deserialize_result(response["result"]) - from .serialization import require_json_value + from .serialization import require_json_value, unwrap_function_response - result = response.get("json_result") - require_json_value(result, "task results must be JSON-serializable") - return result + require_json_value(response, "task results must be JSON-serializable") + return unwrap_function_response(response) class TaskJob: diff --git a/runpod/serverless/modules/rp_fastapi.py b/runpod/serverless/modules/rp_fastapi.py index f5b7c0c6e..490702163 100644 --- a/runpod/serverless/modules/rp_fastapi.py +++ b/runpod/serverless/modules/rp_fastapi.py @@ -267,7 +267,6 @@ def __init__(self, config: dict[str, Any]): "/runsync", self._sim_runsync, methods=["POST"], - response_model_exclude_none=True, summary="Mimics the behavior of the runsync endpoint.", description=RUNSYNC_DESCRIPTION, tags=["Synchronously Submit Request & Get Job Results"], @@ -285,7 +284,6 @@ def __init__(self, config: dict[str, Any]): "/status/{job_id}", self._sim_status, methods=["POST"], - response_model_exclude_none=True, summary="Mimics the behavior of the status endpoint.", description=STATUS_DESCRIPTION, tags=["Check Job Results"], @@ -348,6 +346,9 @@ async def _sim_runsync(self, job_request: DefaultRequest) -> JobOutput: generator_output = run_job_generator(self.config["handler"], job.__dict__) job_output = {"output": []} async for stream_output in generator_output: + if stream_output.get("error"): + job_output = stream_output + break job_output["output"].append(stream_output["output"]) else: job_output = await run_job(self.config["handler"], job.__dict__) @@ -366,7 +367,7 @@ async def _sim_runsync(self, job_request: DefaultRequest) -> JobOutput: thread.start() return jsonable_encoder( - {"id": job.id, "status": "COMPLETED", "output": job_output["output"]} + {"id": job.id, "status": "COMPLETED", "output": job_output.get("output", {})} ) # ---------------------------------- stream ---------------------------------- # @@ -384,6 +385,11 @@ async def _sim_stream(self, job_id: str) -> StreamOutput: generator_output = run_job_generator(self.config["handler"], job.__dict__) stream_accumulator = [] async for stream_output in generator_output: + if stream_output.get("error"): + job_list.remove(job.id) + return jsonable_encoder( + {"id": job_id, "status": "FAILED", "error": stream_output["error"]} + ) stream_accumulator.append({"output": stream_output["output"]}) else: return jsonable_encoder( @@ -423,6 +429,9 @@ async def _sim_status(self, job_id: str) -> JobOutput: generator_output = run_job_generator(self.config["handler"], job.__dict__) job_output = {"output": []} async for stream_output in generator_output: + if stream_output.get("error"): + job_output = stream_output + break job_output["output"].append(stream_output["output"]) else: job_output = await run_job(self.config["handler"], job.__dict__) @@ -443,5 +452,5 @@ async def _sim_status(self, job_id: str) -> JobOutput: thread.start() return jsonable_encoder( - {"id": job_id, "status": "COMPLETED", "output": job_output["output"]} + {"id": job_id, "status": "COMPLETED", "output": job_output.get("output", {})} ) diff --git a/runpod/serverless/modules/rp_http.py b/runpod/serverless/modules/rp_http.py index 3d82d35b6..1f9dfeed7 100644 --- a/runpod/serverless/modules/rp_http.py +++ b/runpod/serverless/modules/rp_http.py @@ -8,6 +8,7 @@ from aiohttp import ClientError from aiohttp_retry import FibonacciRetry, RetryClient +from runpod._serialization import validate_json from runpod.http_client import ClientSession from runpod.serverless.modules.rp_logger import RunPodLogger @@ -57,7 +58,19 @@ async def _handle_result( try: session.headers["X-Request-ID"] = job["id"] - serialized_job_data = json.dumps(job_data, ensure_ascii=False) + try: + validate_json(job_data, "job result") + serialized_job_data = json.dumps( + job_data, ensure_ascii=False, allow_nan=False + ) + except (TypeError, ValueError, RecursionError) as err: + error = { + "error_type": "SerializationError", + "error_message": str(err), + } + serialized_job_data = json.dumps({"error": json.dumps(error)}) + url_template = JOB_DONE_URL + log.error(f"Error while returning job result. | {err}", job["id"]) is_stream = "true" if is_stream else "false" url = url_template.replace("$ID", job["id"]) + f"&isStream={is_stream}" @@ -68,7 +81,7 @@ async def _handle_result( except ClientError as err: log.error(f"Failed to return job results. | {err}", job["id"]) - except (TypeError, RuntimeError) as err: + except (TypeError, ValueError, RuntimeError) as err: log.error(f"Error while returning job result. | {err}", job["id"]) finally: diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py index 108efc4a6..fc1d40c0f 100644 --- a/runpod/serverless/modules/rp_job.py +++ b/runpod/serverless/modules/rp_job.py @@ -10,6 +10,7 @@ import aiohttp +from runpod._serialization import validate_json from runpod.http_client import ClientSession, TooManyRequests from runpod.serverless.modules.rp_logger import RunPodLogger @@ -256,12 +257,14 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: with capture(enabled=handler_capture_enabled()) as cap: try: + validate_json(job.get("input"), "job input") handler_return = handler(job) job_output = ( await handler_return if inspect.isawaitable(handler_return) else handler_return ) + validate_json(job_output, "return value") log.debug(f"Handler output: {job_output}", job["id"]) @@ -281,7 +284,9 @@ async def run_job(handler: Callable, job: Dict[str, Any]) -> Dict[str, Any]: else: run_result = {"output": job_output} - if run_result.get("output") == {}: + if run_result.get("output") == {} and ( + "error" in run_result or "stopPod" in run_result + ): run_result.pop("output") check_return_size(run_result) # Checks the size of the return body. @@ -324,14 +329,17 @@ async def run_job_generator( with capture(enabled=handler_capture_enabled()) as cap: try: + validate_json(job.get("input"), "job input") job_output = handler(job) if is_async_gen: async for output_partial in job_output: + validate_json(output_partial, "stream chunk") log.debug(f"Async Generator output: {output_partial}", job["id"]) yield {"output": output_partial} else: for output_partial in job_output: + validate_json(output_partial, "stream chunk") log.debug(f"Generator output: {output_partial}", job["id"]) yield {"output": output_partial} diff --git a/tests/test_apps/test_dispatch.py b/tests/test_apps/test_dispatch.py index 5d0c27d57..b2967a418 100644 --- a/tests/test_apps/test_dispatch.py +++ b/tests/test_apps/test_dispatch.py @@ -552,19 +552,29 @@ def test_whole_module_ships(self, tmp_path): finally: sys.modules.pop("shipmod", None) - def test_missing_module_on_deserialize_is_actionable(self): + def test_pickled_result_is_rejected_without_importing_remote_modules( + self, monkeypatch + ): import base64 + import builtins + import pickle - import cloudpickle - import pytest - - from runpod.apps.errors import RemoteExecutionError - from runpod.apps.serialization import deserialize_result + from runpod.apps.serialization import unwrap_function_response class FakePickle: def __reduce__(self): return (__import__, ("nonexistent_pkg_xyz",)) - payload = base64.b64encode(cloudpickle.dumps(FakePickle())).decode() - with pytest.raises(RemoteExecutionError, match="nonexistent_pkg_xyz"): - deserialize_result(payload) + payload = base64.b64encode(pickle.dumps(FakePickle())).decode() + real_import = builtins.__import__ + imported = [] + + def track_import(name, *args, **kwargs): + if name == "nonexistent_pkg_xyz": + imported.append(name) + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", track_import) + with pytest.raises(RemoteExecutionError, match="must use json_result"): + unwrap_function_response({"success": True, "result": payload}) + assert imported == [] diff --git a/tests/test_apps/test_tasks.py b/tests/test_apps/test_tasks.py index 024d67198..f20934bc3 100644 --- a/tests/test_apps/test_tasks.py +++ b/tests/test_apps/test_tasks.py @@ -3,9 +3,9 @@ import asyncio import base64 import json +import pickle from unittest.mock import AsyncMock, MagicMock, patch -import cloudpickle import pytest from runpod.apps import App @@ -31,14 +31,6 @@ def clean_registry(): _clear_registry() -def _b64(value): - return base64.b64encode(cloudpickle.dumps(value)).decode("utf-8") - - -def _unb64(value): - return cloudpickle.loads(base64.b64decode(value)) - - class TestPodInput: def _spec(self, **kw): defaults = dict(kind=ResourceKind.TASK, name="t") @@ -113,8 +105,10 @@ def test_cpu5_uses_supported_datacenters(self): class TestUnwrapTaskResponse: - def test_success_cloudpickle(self): - assert unwrap_task_response({"success": True, "result": _b64(42)}) == 42 + def test_pickled_result_is_rejected(self): + payload = base64.b64encode(pickle.dumps(42)).decode("utf-8") + with pytest.raises(RemoteExecutionError, match="must use json_result"): + unwrap_task_response({"success": True, "result": payload}) def test_success_json(self): assert unwrap_task_response({"success": True, "json_result": {"a": 1}}) == { @@ -160,7 +154,7 @@ async def poll_result(): if clock[0] == 0: clock[0] = 7200.0 return None - return {"success": True, "result": _b64(10)} + return {"success": True, "json_result": 10} async def delete(pod_id): pods.remove(pod_id) diff --git a/tests/test_serialization.py b/tests/test_serialization.py new file mode 100644 index 000000000..2fd6fe500 --- /dev/null +++ b/tests/test_serialization.py @@ -0,0 +1,231 @@ +"""json normalization and response safety shared with worker runtimes.""" + +import json +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from runpod._serialization import normalize_json, validate_json +from runpod.apps.errors import RemoteExecutionError +from runpod.apps.protocol import FunctionRequest, FunctionResponse +from runpod.apps.targets import LiveTarget +from runpod.apps.tasks import unwrap_task_response +from runpod.serverless.modules import rp_fastapi, rp_http, rp_job + + +@pytest.mark.parametrize( + "value,expected", + [ + ((1, 2), [1, 2]), + ({1: (None, True)}, {"1": [None, True]}), + ({None: False}, {"null": False}), + (None, None), + ], +) +def test_shared_json_conversion_matches_wire(value, expected): + validate_json(value) + assert normalize_json(value) == expected + + +@pytest.mark.parametrize( + "value", + [b"bytes", datetime(2026, 1, 1), object(), float("nan"), float("inf"), {1, 2}], +) +def test_shared_validation_rejects_non_json_values(value): + with pytest.raises(TypeError, match="json-serializable"): + validate_json(value) + with pytest.raises(TypeError, match="json-serializable"): + normalize_json(value) + + +def test_shared_validation_rejects_cycles(): + value = [] + value.append(value) + with pytest.raises(TypeError, match="(?i)circular reference"): + validate_json(value) + + +def test_function_request_defaults_to_json_and_normalizes_values(): + request = FunctionRequest(function_name="echo", args=[(1, 2)]) + assert request.to_input()["serialization_format"] == "json" + assert FunctionRequest.from_input(request.to_input()).args == [[1, 2]] + + +@pytest.mark.parametrize("fmt", ["cloudpickle", "typo", None, False]) +def test_request_rejects_unsupported_formats(fmt): + with pytest.raises(ValueError, match="serialization_format"): + FunctionRequest.from_input( + {"function_name": "echo", "serialization_format": fmt} + ) + + +@pytest.mark.parametrize("value", [None, False, 0, "", [], {}, (1, 2), {1: "key"}]) +def test_task_and_live_responses_share_json_conversion(value): + response = {"success": True, "json_result": value} + expected = normalize_json(value) + assert unwrap_task_response(response) == expected + assert LiveTarget("ep")._unwrap_response(response) == expected + + +@pytest.mark.parametrize( + "response", + [ + {"success": True}, + {"success": True, "result": "encoded-pickle"}, + {"success": True, "result": "encoded-pickle", "json_result": 42}, + {"success": True, "serialization_format": "cloudpickle", "json_result": 42}, + {"success": "true", "json_result": 42}, + ], +) +def test_task_and_live_reject_invalid_response_envelopes(response): + for decode in (unwrap_task_response, LiveTarget("ep")._unwrap_response): + with pytest.raises(RemoteExecutionError, match="invalid remote response"): + decode(response) + + +def test_response_requires_boolean_success(): + with pytest.raises(TypeError, match="boolean success"): + FunctionResponse.from_output({"json_result": 1}) + + +@pytest.mark.parametrize("value", [b"bytes", datetime(2026, 1, 1), float("nan")]) +async def test_serverless_reports_non_json_results(value): + result = await rp_job.run_job(lambda job: value, {"id": "test", "input": {}}) + error = json.loads(result["error"]) + assert "json-serializable" in error["error_message"] + assert "output" not in result + + +async def test_serverless_rejects_input_before_handler(): + called = [] + result = await rp_job.run_job( + lambda job: called.append(job), {"id": "test", "input": {"x": float("nan")}} + ) + assert not called + assert "job input" in json.loads(result["error"])["error_message"] + + +@pytest.mark.parametrize("asynchronous", [True, False]) +async def test_serverless_stream_reports_invalid_chunk(asynchronous): + async def async_stream(job): + yield 1 + yield b"bytes" + + def sync_stream(job): + yield 1 + yield b"bytes" + + handler = async_stream if asynchronous else sync_stream + chunks = [item async for item in rp_job.run_job_generator(handler, {"id": "test"})] + assert chunks[0] == {"output": 1} + assert "json-serializable" in chunks[1]["error"] + assert len(chunks) == 2 + + +@pytest.mark.parametrize("stream", [True, False]) +async def test_http_serialization_failure_sends_terminal_error(monkeypatch, stream): + transmit = AsyncMock() + monkeypatch.setattr(rp_http, "_transmit", transmit) + session = SimpleNamespace(headers={}) + send = rp_http.stream_result if stream else rp_http.send_result + await send(session, {"output": b"bytes"}, {"id": "test"}) + transmit.assert_awaited_once() + _, url, body = transmit.await_args.args + assert url == rp_http.JOB_DONE_URL.replace("$ID", "test") + "&isStream=false" + error = json.loads(json.loads(body)["error"]) + assert error["error_type"] == "SerializationError" + assert "json-serializable" in error["error_message"] + + +@pytest.mark.parametrize("route", ["runsync", "status", "stream"]) +async def test_local_api_reports_invalid_stream_chunk(monkeypatch, route): + def handler(job): + yield "first" + yield datetime(2026, 1, 1) + + removed = [] + monkeypatch.setattr( + rp_fastapi, + "job_list", + SimpleNamespace( + get=lambda job_id: SimpleNamespace(id=job_id, input={}, webhook=None), + remove=removed.append, + ), + ) + worker = object.__new__(rp_fastapi.WorkerAPI) + worker.config = {"handler": handler} + argument = rp_fastapi.DefaultRequest(input={}) if route == "runsync" else "test" + result = await getattr(worker, f"_sim_{route}")(argument) + assert result["status"] == "FAILED" + assert "json-serializable" in result["error"] + assert "output" not in result + if route != "runsync": + assert removed == ["test"] + + +@pytest.mark.parametrize("value", [None, {}, {"items": []}]) +async def test_local_api_preserves_json_results(value): + worker = object.__new__(rp_fastapi.WorkerAPI) + worker.config = {"handler": lambda job: value} + result = await worker._sim_runsync(rp_fastapi.DefaultRequest(input={})) + assert result["status"] == "COMPLETED" + assert result["output"] == value + + +@pytest.mark.parametrize("value", [None, {}, [], False, 0, ""]) +async def test_serverless_preserves_empty_json_results(value): + result = await rp_job.run_job(lambda job: value, {"id": "test", "input": {}}) + assert result == {"output": value} + + +@pytest.mark.parametrize("route", ["runsync", "status/test"]) +@pytest.mark.parametrize("value", [None, {}]) +def test_local_http_preserves_empty_json_results(monkeypatch, route, value): + from fastapi.testclient import TestClient + + monkeypatch.setattr(rp_fastapi.Heartbeat, "start_ping", lambda self, mirror: None) + monkeypatch.setattr(rp_fastapi.rp_capture, "install", lambda **kwargs: None) + monkeypatch.setattr( + rp_fastapi, + "job_list", + SimpleNamespace( + set_mirror=lambda mirror: None, + get=lambda job_id: SimpleNamespace(id=job_id, input={}, webhook=None), + remove=lambda job_id: None, + ), + ) + worker = rp_fastapi.WorkerAPI({"handler": lambda job: value}) + client = TestClient(worker.rp_app) + try: + response = client.post(f"/{route}", json={"input": {}}) + finally: + client.close() + assert response.status_code == 200 + assert response.json()["status"] == "COMPLETED" + assert "output" in response.json() + assert response.json()["output"] == value + + +@pytest.mark.parametrize("stream", [True, False]) +async def test_deployed_job_sends_serialization_failure(monkeypatch, stream): + def generator(job): + yield "first" + yield b"bytes" + + send = AsyncMock() + send_chunk = AsyncMock() + monkeypatch.setattr(rp_job, "send_result", send) + monkeypatch.setattr(rp_job, "stream_result", send_chunk) + handler = generator if stream else lambda job: datetime(2026, 1, 1) + await rp_job.handle_job( + SimpleNamespace(headers={}), + {"handler": handler, "return_aggregate_stream": True}, + {"id": "test", "input": {}}, + ) + send.assert_awaited_once() + result = send.await_args.args[1] + assert "json-serializable" in result["error"] + assert "output" not in result + assert send_chunk.await_count == (1 if stream else 0)