[Fix] Share OneJev's HTTP deadline across retries

CHANGELOG.md

@@ -4,6 +4,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); ver
 
 ## Unreleased
 
+### Fixed
+
+- Served OneJev uses one HTTP decision deadline across its permitted retry. A retry receives only the remaining
+  `ONEJEV_TIMEOUT_S` budget, and an exhausted budget fails without sending another request.
+
 ### Changed
 
 - The direct TypeSafe path pins `jev-1.13.0` instead of the moving `jev-latest` alias (which moved to Jev 1.14.0

s1a/decision_models/onejev.py

@@ -135,12 +135,16 @@ async def warm(self) -> Json:
     async def decide(self, body: Json, request_id: str) -> tuple[Json, int, dict[str, str]]:
         """One decision: the payload, its round trip in ms and the response headers."""
         headers = {"X-Request-Id": request_id}
+        deadline = self._clock() + self._timeout_s
         retried = False
         while True:
+            remaining = deadline - self._clock()
+            if remaining <= 0:
+                raise self._no_answer(None)
             started = time.perf_counter()
-            post = self._client.post(f"{self.url}/v1/systemone", json=body, headers=headers, timeout=self._timeout_s)
+            post = self._client.post(f"{self.url}/v1/systemone", json=body, headers=headers, timeout=remaining)
             try:
-                response = await asyncio.wait_for(post, self._timeout_s)
+                response = await asyncio.wait_for(post, remaining)
             except TimeoutError as exc:
                 raise self._no_answer(exc) from exc
             except _RETRIED_TRANSPORT as exc:
@@ -183,7 +187,7 @@ async def decide(self, body: Json, request_id: str) -> tuple[Json, int, dict[str
                 raise build_error(StatusCode.MODEL_CALL_FAILED, error_msg="served OneJev returned no answers object")
             return payload, ms, dict(response.headers)
 
-    def _no_answer(self, cause: Exception) -> Exception:
+    def _no_answer(self, cause: Exception | None) -> Exception:
         return build_error(
             StatusCode.MODEL_CALL_FAILED,
             cause=cause,

tests/test_decision_models_onejev.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import base64
 import json
 import os
@@ -176,6 +177,83 @@ async def test_http_error_is_a_call_failure(self) -> None:
         self.assertIn("500", str(caught.exception))
 
 
+class TestDecisionDeadline(IsolatedAsyncioTestCase):
+    async def test_retry_uses_remaining_budget_for_httpx_and_wait_for(self) -> None:
+        for failure in (httpx.ConnectError("refused"), httpx.Response(502), httpx.Response(504)):
+            with self.subTest(failure=failure):
+                now = 0.0
+                requests = []
+
+                def handler(request: httpx.Request) -> httpx.Response:
+                    nonlocal now
+                    requests.append(request)
+                    if len(requests) == 1:
+                        now += 1.25
+                        if isinstance(failure, Exception):
+                            raise failure
+                        return failure
+                    return httpx.Response(200, json={"answers": {}})
+
+                client = OneJevClient(url=URL, timeout_s=2.0, transport=httpx.MockTransport(handler), clock=lambda: now)
+                self.addAsyncCleanup(client.close)
+                with patch("s1a.decision_models.onejev.asyncio.wait_for", wraps=asyncio.wait_for) as wait_for:
+                    await client.decide({"questions": {}}, "same-request")
+                self.assertEqual([call.args[1] for call in wait_for.call_args_list], [2.0, 0.75])
+                for request, timeout in zip(requests, (2.0, 0.75), strict=True):
+                    self.assertEqual(set(request.extensions["timeout"].values()), {timeout})
+                    self.assertEqual(request.headers["X-Request-Id"], "same-request")
+
+    async def test_exhausted_budget_prevents_retry(self) -> None:
+        for failure in (httpx.ConnectError("refused"), httpx.Response(502), httpx.Response(504)):
+            with self.subTest(failure=failure):
+                now = 0.0
+                requests = []
+
+                def handler(request: httpx.Request) -> httpx.Response:
+                    nonlocal now
+                    requests.append(request)
+                    now += 2.0
+                    if isinstance(failure, Exception):
+                        raise failure
+                    return failure
+
+                client = OneJevClient(url=URL, timeout_s=2.0, transport=httpx.MockTransport(handler), clock=lambda: now)
+                self.addAsyncCleanup(client.close)
+                with self.assertRaises(BaseError) as caught:
+                    await client.decide({"questions": {}}, "exhausted")
+                self.assertEqual(caught.exception.status, StatusCode.MODEL_CALL_FAILED)
+                self.assertIn("within 2 s", str(caught.exception))
+                self.assertEqual(len(requests), 1)
+
+    async def test_stalled_retry_is_cancelled_at_remaining_deadline(self) -> None:
+        now = 0.0
+        attempts = 0
+        stopped = asyncio.Event()
+
+        async def handler(request: httpx.Request) -> httpx.Response:
+            nonlocal now, attempts
+            attempts += 1
+            if attempts == 1:
+                now = 0.99
+                raise httpx.ConnectError("refused", request=request)
+            try:
+                await asyncio.Event().wait()
+            finally:
+                stopped.set()
+            raise AssertionError("stalled retry was not cancelled")
+
+        client = OneJevClient(url=URL, timeout_s=1.0, transport=httpx.MockTransport(handler), clock=lambda: now)
+        self.addAsyncCleanup(client.close)
+        with patch("s1a.decision_models.onejev.asyncio.wait_for", wraps=asyncio.wait_for) as wait_for:
+            with self.assertRaises(BaseError) as caught:
+                await client.decide({"questions": {}}, "stalled")
+        self.assertEqual(caught.exception.status, StatusCode.MODEL_CALL_FAILED)
+        self.assertIsInstance(caught.exception.__cause__, TimeoutError)
+        self.assertAlmostEqual(wait_for.call_args_list[1].args[1], 0.01)
+        self.assertEqual(attempts, 2)
+        self.assertTrue(stopped.is_set())
+
+
 class TestMappingHelpers(TestCase):
     def test_state_wraps_text_and_marks_one_image(self) -> None:
         self.assertEqual(