Return 422 for malformed JSON bodies and surface invalid resolver timestamps (#46, #47)

Malformed JSON request bodies previously produced a generic Starlette 500
because the endpoints called request.json() directly without guarding
json.JSONDecodeError. Now all four POST endpoints (/v1/events,
/v1/events:batch, /v1/search, /v1/query) use a shared _json_body() helper
that raises ValidationError (→ 422) on decode failure, consistent with
the existing error-to-status mapping.

Invalid resolver begin/end timestamps were silently dropped by
_valid_optional_timestamp, widening the search to unbounded instead of
surfacing the error. Now a malformed timestamp raises QueryError, which
the query error handler catches and returns as status='error'.

Tests:
- Parametrized test covering all four POST endpoints with malformed JSON.
- Updated test_malformed_optional_timestamp to assert the new error behavior.

Author: randileeharper

historian/http.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import contextvars
+import json
 import time
 import uuid
 from dataclasses import asdict
@@ -67,6 +68,13 @@ def _bearer_token(request: Request) -> str:
     return token.strip()
 
 
+async def _json_body(request: Request) -> Any:
+    try:
+        return await request.json()
+    except json.JSONDecodeError as exc:
+        raise ValidationError("Request body must be valid JSON.") from exc
+
+
 def build_agent_card(base_url: str) -> AgentCard:
     bearer = SecurityScheme(
         http_auth_security_scheme=HTTPAuthSecurityScheme(
@@ -262,7 +270,7 @@ async def agent_card_alias() -> JSONResponse:
 
     @app.post("/v1/events")
     async def ingest_event(request: Request) -> dict[str, Any]:
-        event, duplicate = context.service.ingest(request.state.principal, await request.json())
+        event, duplicate = context.service.ingest(request.state.principal, await _json_body(request))
         _LOG.info(
             "event_id=%s producer_app=%s type=%s duplicate=%s event_ingested",
             event.event_id,
@@ -274,7 +282,7 @@ async def ingest_event(request: Request) -> dict[str, Any]:
 
     @app.post("/v1/events:batch")
     async def ingest_batch(request: Request) -> dict[str, Any]:
-        body = await request.json()
+        body = await _json_body(request)
         events = body.get("events") if isinstance(body, dict) else None
         if not isinstance(events, list):
             raise ValidationError("Batch body must contain an events array.")
@@ -294,7 +302,7 @@ async def ingest_batch(request: Request) -> dict[str, Any]:
 
     @app.post("/v1/search")
     async def search_events(request: Request) -> dict[str, Any]:
-        body = await request.json()
+        body = await _json_body(request)
         try:
             spec = SearchSpec(**body)
         except TypeError as exc:
@@ -333,7 +341,7 @@ async def get_event(event_id: str, request: Request) -> dict[str, Any]:
 
     @app.post("/v1/query")
     async def structured_query(request: Request) -> dict[str, Any]:
-        body = await request.json()
+        body = await _json_body(request)
         question = str(body.get("question", "")) if isinstance(body, dict) else ""
         return to_jsonable(context.service.query(request.state.principal, question))
 

historian/service.py

@@ -391,8 +391,8 @@ def _valid_optional_timestamp(cls, value: Any) -> str | None:
             return None
         try:
             datetime.fromisoformat(text.replace("Z", "+00:00"))
-        except ValueError:
-            return None
+        except ValueError as exc:
+            raise QueryError(f"Resolver returned an invalid timestamp: {text!r}") from exc
         return text
 
     def _evidence_chunks(

tests/test_http.py

@@ -3,6 +3,7 @@
 import asyncio
 
 import httpx
+import pytest
 from a2a.helpers import new_text_part
 from a2a.types import Message, Role, SendMessageRequest
 from a2a.utils.constants import PROTOCOL_VERSION_1_0, VERSION_HEADER
@@ -180,3 +181,28 @@ def test_search_endpoint_rejects_invalid_spec(context, vesper_token) -> None:
     )
     assert response.status_code == 422
     assert response.json()["status"] == "error"
+
+
+@pytest.mark.parametrize(
+    "path,method",
+    [
+        ("/v1/events", "POST"),
+        ("/v1/events:batch", "POST"),
+        ("/v1/search", "POST"),
+        ("/v1/query", "POST"),
+    ],
+)
+def test_malformed_json_returns_422(context, vesper_token, path, method) -> None:
+    """POST endpoints return 422 for malformed JSON bodies, not 500."""
+    app = create_http_app(context)
+    response = asyncio.run(
+        _request(
+            app,
+            method,
+            path,
+            headers={"Authorization": f"Bearer {vesper_token}", "Content-Type": "application/json"},
+            content=b"{not valid json",
+        )
+    )
+    assert response.status_code == 422
+    assert response.json()["status"] == "error"

tests/test_query.py

@@ -112,9 +112,10 @@ def test_search_timestamps_and_text_are_optional(context, resolver, vesper_token
     ]
 
 
-def test_malformed_optional_timestamp_does_not_abort_search(
+def test_malformed_optional_timestamp_raises_query_error(
     context, resolver, vesper_token
 ) -> None:
+    """A malformed begin/end from the resolver surfaces as a QueryError, not a silent unbounded search."""
     principal = context.store.authenticate(vesper_token)
     context.service.ingest(principal, event())
     resolver.plans.append(
@@ -133,9 +134,8 @@ def test_malformed_optional_timestamp_does_not_abort_search(
     )
     resolver.answers.append(_answer())
     result = context.service.query(principal, "What has Vesper played?")
-    assert result.status == "ok"
-    assert result.searches[0]["begin"] is None
-    assert result.searches[0]["end"] is None
+    assert result.status == "error"
+    assert "invalid timestamp" in result.message
 
 
 def test_model_plan_is_executed_faithfully(