diff --git a/README.md b/README.md index 63c8fee..a8e65c3 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,35 @@ async def evaluate(response, answer, params: Params) -> Result: ... ``` +## Reporting Errors + +Raise one of these from a handler to report an error with its own code. The message is passed to the caller verbatim, without the "Error calling user handler" prefix. + +| Raise | JSON-RPC code (stdio/IPC) | File IO `error.code` | Use when | +|---|---|---|---| +| `InvalidSubmissionError` | 422 | `INVALID_SUBMISSION` | the submitted response can't be processed | +| `UpstreamTimeoutError` | 424 | `UPSTREAM_TIMEOUT` | a service the function depends on did not answer in time (retryable) | +| `UpstreamUnavailableError` | 424 | `UPSTREAM_UNAVAILABLE` | a service the function depends on refused or failed (retryable) | +| `EvaluationFunctionError` | 500 | `INTERNAL_ERROR` | any other error the function chooses to report | + +Upstream failures use 424 (Failed Dependency) rather than 5xx, because the evaluation function itself is working; 5xx is kept for errors in the function. Over JSON-RPC the File IO code is also sent as the error's `data` (`{"code": "UPSTREAM_TIMEOUT"}`), which tells the two 424s apart. + +```python +from lf_toolkit import UpstreamTimeoutError + +@server.eval +def evaluate(response, answer, params: Params) -> Result: + try: + feedback = call_model(response, timeout=20) + except TimeoutError as e: + raise UpstreamTimeoutError("The feedback service did not answer in time") from e + ... +``` + +Any other `ValueError` is reported as `InvalidSubmissionError` (422). Any other exception is reported with code `0` (and no file IO `code`), with its message prefixed by "Error calling user handler for ''". + +The message reaches the caller, so keep secrets out of it. The toolkit does not log requests, responses or error messages, since they can contain student submissions. + ## Result Types ### `evaluation.Result` diff --git a/lf_toolkit/__init__.py b/lf_toolkit/__init__.py index 8b8ccad..2030df5 100644 --- a/lf_toolkit/__init__.py +++ b/lf_toolkit/__init__.py @@ -1,8 +1,12 @@ import importlib.metadata as importlib_metadata +from .io import EvaluationFunctionError from .io import FileServer +from .io import InvalidSubmissionError from .io import IPCServer from .io import StdioServer +from .io import UpstreamTimeoutError +from .io import UpstreamUnavailableError from .io import create_server from .io import run from .io import serve diff --git a/lf_toolkit/io/__init__.py b/lf_toolkit/io/__init__.py index 9873a9f..07d73a4 100644 --- a/lf_toolkit/io/__init__.py +++ b/lf_toolkit/io/__init__.py @@ -1,4 +1,8 @@ from .file_server import FileServer +from .handler import EvaluationFunctionError +from .handler import InvalidSubmissionError +from .handler import UpstreamTimeoutError +from .handler import UpstreamUnavailableError from .ipc_server import IPCServer from .serve import create_server from .serve import run diff --git a/lf_toolkit/io/file_server.py b/lf_toolkit/io/file_server.py index 048b03f..e3cba19 100644 --- a/lf_toolkit/io/file_server.py +++ b/lf_toolkit/io/file_server.py @@ -6,7 +6,7 @@ from .base_server import BaseServer from .handler import Handler -from .handler import InvalidSubmissionError +from .handler import EvaluationFunctionError class FileHandler(Handler): @@ -17,10 +17,10 @@ async def dispatch(self, req: str) -> str: try: result = await self.handle(command, request) response = {"command": command, "result": result} - except InvalidSubmissionError as e: + except EvaluationFunctionError as e: response = { "command": command, - "error": {"message": str(e), "code": "INVALID_SUBMISSION"}, + "error": {"message": str(e), "code": e.file_code}, } except Exception as e: response = {"command": command, "error": {"message": str(e)}} diff --git a/lf_toolkit/io/handler.py b/lf_toolkit/io/handler.py index 2fc0152..d4877e8 100644 --- a/lf_toolkit/io/handler.py +++ b/lf_toolkit/io/handler.py @@ -17,10 +17,45 @@ # JSON-RPC error code for submissions the user handler cannot process. INVALID_SUBMISSION_CODE = 422 +# JSON-RPC error code for failures of a service the user handler depends on. +# These are not failures of the evaluation function itself, so they are +# reported as 424 (Failed Dependency) rather than 5xx, matching shimmy. +UPSTREAM_ERROR_CODE = 424 -class InvalidSubmissionError(Exception): + +class EvaluationFunctionError(Exception): + """An error the evaluation function chose to report. + + The message is passed to the caller verbatim, and the class decides the + JSON-RPC code (stdio/IPC) and the error code (file IO) it is reported with. + Over JSON-RPC the error code is also sent as the error's `data`, since + several classes share a JSON-RPC code. + """ + + rpc_code = 500 + file_code = "INTERNAL_ERROR" + + +class InvalidSubmissionError(EvaluationFunctionError): """Raised when the user handler cannot process the submitted response.""" + rpc_code = INVALID_SUBMISSION_CODE + file_code = "INVALID_SUBMISSION" + + +class UpstreamTimeoutError(EvaluationFunctionError): + """A service the evaluation function depends on did not answer in time.""" + + rpc_code = UPSTREAM_ERROR_CODE + file_code = "UPSTREAM_TIMEOUT" + + +class UpstreamUnavailableError(EvaluationFunctionError): + """A service the evaluation function depends on refused or failed.""" + + rpc_code = UPSTREAM_ERROR_CODE + file_code = "UPSTREAM_UNAVAILABLE" + class Handler(ABC): @@ -47,6 +82,8 @@ async def _call_user_handler(self, req: str, *args, **kwargs): return await handler(*args, **kwargs) else: return await anyio.to_thread.run_sync(handler, *args, **kwargs) + except EvaluationFunctionError: + raise except ValueError as e: raise InvalidSubmissionError(str(e)) from e except Exception as e: diff --git a/lf_toolkit/io/rpc_handler.py b/lf_toolkit/io/rpc_handler.py index d5b8241..c32dfc5 100644 --- a/lf_toolkit/io/rpc_handler.py +++ b/lf_toolkit/io/rpc_handler.py @@ -5,9 +5,8 @@ from jsonrpcserver import async_dispatch from ..shared import Command -from .handler import INVALID_SUBMISSION_CODE +from .handler import EvaluationFunctionError from .handler import Handler -from .handler import InvalidSubmissionError class JsonRpcHandler(Handler): @@ -33,12 +32,12 @@ async def wrapped(req: dict): try: result = await handler.handle(name, {"params": req}) return Success(result) - except InvalidSubmissionError as e: - return Error(INVALID_SUBMISSION_CODE, str(e)) + # Pass only strings: the exception object is not JSON + # serializable, so sending it as `data` makes serializing the + # error response raise, which tears down the serve loop. + except EvaluationFunctionError as e: + return Error(e.rpc_code, str(e), {"code": e.file_code}) except Exception as e: - # Pass only the message: the exception object is not JSON - # serializable, so sending it as `data` makes serializing the - # error response raise, which tears down the serve loop. return Error(0, str(e)) return wrapped diff --git a/lf_toolkit/io/stream_io.py b/lf_toolkit/io/stream_io.py index 2f0cb86..fdeab06 100644 --- a/lf_toolkit/io/stream_io.py +++ b/lf_toolkit/io/stream_io.py @@ -9,6 +9,13 @@ from .base_server import BaseServer +def _print_exception(e: BaseException): + # Requests carry student submissions, and exception messages often quote + # them, so log only the exception type and where it was raised. + print(f"{type(e).__name__} raised:", file=sys.stderr, flush=True) + traceback.print_tb(e.__traceback__, file=sys.stderr) + + class StreamIO(ABC): @abstractmethod @@ -100,13 +107,13 @@ async def _serve_once(self, io: StreamIO) -> bool: try: print("waiting for data...", file=sys.stderr, flush=True) data = await io.read(4096) - print(f"got data: {data[:80]}", file=sys.stderr, flush=True) + print(f"got data ({len(data)} bytes)", file=sys.stderr, flush=True) except (anyio.EndOfStream, anyio.ClosedResourceError): return False - except Exception: + except Exception as e: # A read failure may have consumed part of a frame, so there is # no point in the stream we can safely resume from. - traceback.print_exc(file=sys.stderr) + _print_exception(e) return False if not data: @@ -115,13 +122,13 @@ async def _serve_once(self, io: StreamIO) -> bool: try: print("dispatching...", file=sys.stderr, flush=True) response = await self.dispatch(data.decode("utf-8")) - print(f"got response: {str(response)[:80]}", file=sys.stderr, flush=True) - except Exception: + print(f"got response ({len(response)} chars)", file=sys.stderr, flush=True) + except Exception as e: # One bad request must not end the session: the frame was read in # full, so the stream is still aligned and the next request can be # served. The caller gets no reply for this one and will time out, # which beats every later request on this worker timing out too. - traceback.print_exc(file=sys.stderr) + _print_exception(e) return True try: @@ -129,8 +136,8 @@ async def _serve_once(self, io: StreamIO) -> bool: print("wrote response", file=sys.stderr, flush=True) except (anyio.EndOfStream, anyio.ClosedResourceError): return False - except Exception: - traceback.print_exc(file=sys.stderr) + except Exception as e: + _print_exception(e) return False return True diff --git a/tests/io/file_server_test.py b/tests/io/file_server_test.py index 007f306..1958577 100644 --- a/tests/io/file_server_test.py +++ b/tests/io/file_server_test.py @@ -4,6 +4,10 @@ from unittest.mock import Mock, AsyncMock, patch, MagicMock from typing import Any +from lf_toolkit import EvaluationFunctionError +from lf_toolkit import InvalidSubmissionError +from lf_toolkit import UpstreamTimeoutError +from lf_toolkit import UpstreamUnavailableError from lf_toolkit.io.file_server import FileHandler, FileServer from lf_toolkit.io.handler import Handler @@ -307,3 +311,22 @@ def preview_fn(response, params): assert "code" not in response["error"] assert "boom" in response["error"]["message"] + + @pytest.mark.parametrize("error_class, code", [ + (EvaluationFunctionError, "INTERNAL_ERROR"), + (InvalidSubmissionError, "INVALID_SUBMISSION"), + (UpstreamTimeoutError, "UPSTREAM_TIMEOUT"), + (UpstreamUnavailableError, "UPSTREAM_UNAVAILABLE"), + ]) + @pytest.mark.asyncio + async def test_evaluation_function_error_returns_its_file_code(self, error_class, code): + def eval_fn(response, answer, params): + raise error_class("OpenRouter is unavailable") + + handler = FileHandler() + handler.register("eval", eval_fn) + + request = ujson.dumps({"command": "eval", "params": {"response": "x"}}) + response = ujson.loads(await handler.dispatch(request)) + + assert response["error"] == {"message": "OpenRouter is unavailable", "code": code} diff --git a/tests/io/handler_test.py b/tests/io/handler_test.py index 6dc48c8..be50386 100644 --- a/tests/io/handler_test.py +++ b/tests/io/handler_test.py @@ -5,6 +5,10 @@ from lf_toolkit.chat import ChatRequest from lf_toolkit.chat import ChatResponse from lf_toolkit.chat import Message +from lf_toolkit import EvaluationFunctionError +from lf_toolkit import InvalidSubmissionError +from lf_toolkit import UpstreamTimeoutError +from lf_toolkit import UpstreamUnavailableError from lf_toolkit.io.file_server import FileHandler from lf_toolkit.shared.mued_api_v0_1_0 import DataPolicySupport from lf_toolkit.shared.mued_api_v0_1_0 import HealthStatus @@ -90,6 +94,48 @@ async def test_passes_through_non_chat_health_response_result(self, handler): assert result == {"status": "OK"} +class TestCallUserHandlerErrors: + + @pytest.fixture + def handler(self): + return FileHandler() + + @pytest.mark.parametrize("error_class", [ + EvaluationFunctionError, + InvalidSubmissionError, + UpstreamTimeoutError, + UpstreamUnavailableError, + ]) + @pytest.mark.asyncio + async def test_evaluation_function_error_is_not_rewrapped(self, handler, error_class): + raised = error_class("upstream failed") + + def eval_fn(response, answer, params): + raise raised + + handler.register("eval", eval_fn) + + with pytest.raises(error_class) as exc_info: + await handler.handle_eval({"params": {"response": "x"}}) + + assert exc_info.value is raised + + @pytest.mark.asyncio + async def test_value_error_becomes_invalid_submission(self, handler): + def eval_fn(response, answer, params): + raise ValueError("bad input") + + handler.register("eval", eval_fn) + + with pytest.raises(InvalidSubmissionError, match="^bad input$"): + await handler.handle_eval({"params": {"response": "x"}}) + + def test_upstream_errors_are_not_value_errors(self): + # Otherwise `except ValueError` in user code would swallow them. + assert not issubclass(UpstreamTimeoutError, ValueError) + assert not issubclass(UpstreamUnavailableError, ValueError) + + class TestHandleDispatch: """Covers the name -> method lookup in Handler.handle, including the 'chat/health' slash normalisation.""" diff --git a/tests/io/rpc_handler_test.py b/tests/io/rpc_handler_test.py index 4a902e2..55424c0 100644 --- a/tests/io/rpc_handler_test.py +++ b/tests/io/rpc_handler_test.py @@ -6,6 +6,10 @@ from lf_toolkit.chat import ChatRequest from lf_toolkit.chat import ChatResponse from lf_toolkit.chat import Message +from lf_toolkit import EvaluationFunctionError +from lf_toolkit import InvalidSubmissionError +from lf_toolkit import UpstreamTimeoutError +from lf_toolkit import UpstreamUnavailableError from lf_toolkit.io.rpc_handler import JsonRpcHandler from lf_toolkit.shared.mued_api_v0_1_0 import DataPolicySupport from lf_toolkit.shared.mued_api_v0_1_0 import HealthStatus @@ -108,4 +112,46 @@ def eval_fn(response, answer, params): response = ujson.loads(await handler.dispatch(self._eval_request("x+1"))) assert response["error"]["code"] == 0 - assert "boom" in response["error"]["message"] + assert response["error"]["message"] == "Error calling user handler for 'eval': boom" + + @pytest.mark.parametrize("error_class, code, file_code", [ + (EvaluationFunctionError, 500, "INTERNAL_ERROR"), + (InvalidSubmissionError, 422, "INVALID_SUBMISSION"), + (UpstreamTimeoutError, 424, "UPSTREAM_TIMEOUT"), + (UpstreamUnavailableError, 424, "UPSTREAM_UNAVAILABLE"), + ]) + @pytest.mark.asyncio + async def test_evaluation_function_error_returns_its_own_code(self, handler, error_class, code, file_code): + def eval_fn(response, answer, params): + raise error_class("OpenRouter did not answer in time") + + handler.register("eval", eval_fn) + + response = ujson.loads(await handler.dispatch(self._eval_request("x+1"))) + + assert response["id"] == 1 + assert response["error"] == { + "code": code, + "message": "OpenRouter did not answer in time", + "data": {"code": file_code}, + } + + @pytest.mark.asyncio + async def test_error_response_serialises_after_an_error(self, handler): + """Regression for v1.1.2: building the error response must not raise, + and the handler must keep serving the next request.""" + calls = [] + + def eval_fn(response, answer, params): + calls.append(response) + if len(calls) == 1: + raise UpstreamTimeoutError("timed out") + return {"is_correct": True} + + handler.register("eval", eval_fn) + + first = ujson.loads(await handler.dispatch(self._eval_request("x+1"))) + second = ujson.loads(await handler.dispatch(self._eval_request("x+2"))) + + assert first["error"]["code"] == 424 + assert second["result"] == {"is_correct": True} diff --git a/tests/io/stream_io_test.py b/tests/io/stream_io_test.py index 80020fc..ba626d3 100644 --- a/tests/io/stream_io_test.py +++ b/tests/io/stream_io_test.py @@ -354,6 +354,65 @@ async def test_oversized_message_does_not_corrupt_subsequent_message(self, strea ) +class TestNoStudentInputLogged: + """Requests carry student submissions, which must never reach the logs.""" + + SUBMISSION = "student-submission-7f3a" + + @pytest.fixture + def stream(self): + return FakeStreamIO() + + @pytest.fixture + def server(self): + return IPCServer() + + def _feed_eval(self, stream): + stream.feed(make_newline_message(ujson.dumps({ + "jsonrpc": "2.0", + "method": "eval", + "params": [{"response": self.SUBMISSION, "answer": "x", "params": {}}], + "id": 1, + }))) + + @pytest.mark.anyio + async def test_successful_request(self, stream, server, capsys): + server.eval(lambda response, answer, params: {"response_latex": response}) + self._feed_eval(stream) + + await server._handle_client(stream) + + assert self.SUBMISSION.encode() in stream.responses[0] + assert self.SUBMISSION not in capsys.readouterr().err + + @pytest.mark.anyio + async def test_handler_error_quoting_the_submission(self, stream, server, capsys): + def eval_fn(response, answer, params): + raise RuntimeError(f"could not parse {response}") + + server.eval(eval_fn) + self._feed_eval(stream) + + await server._handle_client(stream) + + assert self.SUBMISSION.encode() in stream.responses[0] + assert self.SUBMISSION not in capsys.readouterr().err + + @pytest.mark.anyio + async def test_dispatch_error_quoting_the_submission(self, stream, server, capsys, monkeypatch): + async def failing_dispatch(req): + raise RuntimeError(f"failed on {req}") + + monkeypatch.setattr(server, "dispatch", failing_dispatch) + self._feed_eval(stream) + + await server._handle_client(stream) + + err = capsys.readouterr().err + assert "RuntimeError" in err + assert self.SUBMISSION not in err + + class TestStdioServerSubprocess: def test_binary_pipe_roundtrip(self):