From 63bc5a2659123d183ca28f26b1f844fe26a5e8fb Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Jul 2026 00:25:38 +0000 Subject: [PATCH 1/5] fix(tags): report server 4xx rejections instead of masking them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The backend (server-private, staging-v2) now rejects with HTTP 400 any create / tag-update payload whose tags array carries more than one group:* tag ({"error":"A run can have at most one group:* tag."}). The SDK handled the rejection badly: the write client _try retried the 400 with backoff, then either silently no-op'd (tags/update) or raised a misleading ConnectionError("...Check connection...") (create). The real reason was only ever in a buried warning log. This makes the SDK surface server validation errors loudly and honestly, without second-guessing or rewriting the caller's tags: - iface.py: add PlutoRequestError + _server_error_message(). _try now treats 4xx as terminal (no retry, no wasted backoff) and, when raise_on_error=True, raises PlutoRequestError carrying the server's reason (parsed from the JSON "error" field). Only 5xx and network/timeout errors are retried — mirroring query.py. - op.py: create/resume pass raise_on_error=True. A server rejection now raises RuntimeError("Failed to create run: "); a genuine unreachable server still raises the ConnectionError. add_tags/ remove_tags surface a rejection at WARNING instead of burying it at DEBUG. - sync/process.py: the sync-subprocess uploader (_post_with_retry) — the primary path for run.add_tags(), which sends the full tags array — also treats 4xx as terminal and raises PlutoRequestError with the server's reason, so the sync loop logs the actual message rather than a bare "400 Bad Request". No client-side tag rewriting: if a caller sends two group:* tags, the request fails and the server's reason is reported, rather than the SDK quietly dropping one. Tests (tests/test_iface_errors.py): _try does not retry a 400 and raises PlutoRequestError with the server message (not ConnectionError), retries 5xx, and returns None on network errors; the sync uploader does not retry a 400 and surfaces the reason, and still retries 5xx. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/iface.py | 69 ++++++++++++++-- pluto/op.py | 68 +++++++++------ pluto/sync/process.py | 35 ++++++-- tests/test_iface_errors.py | 164 +++++++++++++++++++++++++++++++++++++ 4 files changed, 297 insertions(+), 39 deletions(-) create mode 100644 tests/test_iface_errors.py diff --git a/pluto/iface.py b/pluto/iface.py index 485830d..836c6e1 100644 --- a/pluto/iface.py +++ b/pluto/iface.py @@ -20,6 +20,35 @@ tag = 'Interface' +class PlutoRequestError(Exception): + """Raised when a Pluto write request fails with a server validation error. + + Carries the server-provided reason (parsed from the JSON ``error`` field + when present) so callers can surface *why* the request was rejected — e.g. + ``"A run can have at most one group:* tag."`` — instead of a generic + connection error. Mirrors ``query.PlutoQueryError`` for the read path. + """ + + def __init__(self, message: str, status_code: Optional[int] = None): + self.status_code = status_code + super().__init__(message) + + +def _server_error_message(r: httpx.Response) -> str: + """Best-effort extraction of the human-readable reason from a response. + + The backend returns ``{"error": ""}`` on validation failures; fall + back to the raw (truncated) body when the payload isn't the expected shape. + """ + try: + body = r.json() + if isinstance(body, dict) and body.get('error'): + return str(body['error']) + except Exception: + pass + return r.text[:500] + + @contextmanager def _suppress_sentry_breadcrumbs(): """Prevent the host app's Sentry from capturing Pluto's internal HTTP traffic. @@ -143,12 +172,18 @@ def update_status(self, trace: Union[Any, None] = None) -> None: ) def update_tags(self, tags: List[str]) -> None: - """Update tags on the server via HTTP API.""" + """Update tags on the server via HTTP API. + + Raises ``PlutoRequestError`` if the server rejects the update (e.g. a + validation error), so a failed ``run.update_tags(...)`` surfaces the + reason instead of silently no-op'ing. + """ self._post_v1( self.settings.url_update_tags, self.headers, make_compat_update_tags_v1(self.settings, tags), client=self.client_api, + raise_on_error=True, ) def update_config(self, config: Dict[str, Any]) -> None: @@ -253,6 +288,7 @@ def _try( max_retries: Optional[int] = None, timeout: Optional[float] = None, suppress_httpx_logs: bool = False, + raise_on_error: bool = False, ): effective_max_retries = ( max_retries @@ -283,6 +319,13 @@ def _try( retry_count=retry, ) + # Distinguish a persistent server error (we got HTTP responses but + # they never succeeded) from an unreachable server (network + # exceptions → error_info doesn't start with "HTTP"). Only the + # former carries a server-provided reason worth raising. + if raise_on_error and error_info.startswith('HTTP '): + raise PlutoRequestError(error_info, status_code=None) + return None try: @@ -299,11 +342,10 @@ def _try( return r # Capture error info for potential failure logging - error_info = f'HTTP {r.status_code}: {r.text[:100]}' + server_msg = _server_error_message(r) + error_info = f'HTTP {r.status_code}: {server_msg[:200]}' - status_code = r.status_code if r else 'N/A' target = len(drained) if drained else 'request' - response = r.text if r else 'N/A' # High-frequency endpoints (the trigger/heartbeat that fires # every ~4 s) set suppress_httpx_logs; route their non-200 # responses to DEBUG so a flaky server doesn't spam WARNING. @@ -314,11 +356,23 @@ def _try( name, retry + 1, effective_max_retries + 1, - status_code, + r.status_code, target, url, - response, + server_msg, ) + + # 4xx is a client/validation error — retrying the identical payload + # will never succeed, so stop immediately (no wasted backoff) and, + # when asked, surface the server's reason to the caller. + if 400 <= r.status_code < 500: + if raise_on_error: + raise PlutoRequestError(server_msg, status_code=r.status_code) + return None + except PlutoRequestError: + # A deliberate terminal error (4xx) — propagate, don't treat it as + # a transient network failure to be retried by the except below. + raise except ( BrokenPipeError, ConnectionResetError, @@ -369,6 +423,7 @@ def _try( max_retries=effective_max_retries, timeout=timeout, suppress_httpx_logs=suppress_httpx_logs, + raise_on_error=raise_on_error, ) def _put_v1( @@ -403,6 +458,7 @@ def _post_v1( max_retries: Optional[int] = None, timeout: Optional[float] = None, suppress_httpx_logs: bool = False, + raise_on_error: bool = False, ): # Support both queue and direct content if isinstance(q, queue.Queue): @@ -424,6 +480,7 @@ def _post_v1( max_retries=max_retries, timeout=timeout, suppress_httpx_logs=suppress_httpx_logs, + raise_on_error=raise_on_error, ) if ( diff --git a/pluto/op.py b/pluto/op.py index b37870c..363abee 100644 --- a/pluto/op.py +++ b/pluto/op.py @@ -27,7 +27,7 @@ from .auth import login from .data import Data from .file import Artifact, Audio, File, Image, Text, Video -from .iface import ServerInterface +from .iface import PlutoRequestError, ServerInterface from .log import setup_logger, teardown_logger from .store import DataStore from .sync import SyncProcessManager @@ -319,31 +319,41 @@ def __init__(self, config, settings, tags=None, resume=False) -> None: if self.settings._sys == {}: self.settings._sys = System(self.settings) tmp_iface = ServerInterface(config=config, settings=settings) - if ( - settings._resume_run_id is not None - or settings._resume_display_id is not None - ): - # Resume existing run via /api/runs/resume - r = tmp_iface._post_v1( - self.settings.url_resume, - tmp_iface.headers, - make_compat_resume_v1(self.settings), - client=tmp_iface.client_api, - ) - else: - # Create new run (or resume via externalId) - r = tmp_iface._post_v1( - self.settings.url_start, # create-run - tmp_iface.headers, - make_compat_start_v1( - self.config, - self.settings, - self.settings._sys.get_info(), - self.tags, - ), - client=tmp_iface.client_api, - ) + try: + if ( + settings._resume_run_id is not None + or settings._resume_display_id is not None + ): + # Resume existing run via /api/runs/resume + r = tmp_iface._post_v1( + self.settings.url_resume, + tmp_iface.headers, + make_compat_resume_v1(self.settings), + client=tmp_iface.client_api, + raise_on_error=True, + ) + else: + # Create new run (or resume via externalId) + r = tmp_iface._post_v1( + self.settings.url_start, # create-run + tmp_iface.headers, + make_compat_start_v1( + self.config, + self.settings, + self.settings._sys.get_info(), + self.tags, + ), + client=tmp_iface.client_api, + raise_on_error=True, + ) + except PlutoRequestError as e: + # The server rejected the request (e.g. a validation error such + # as "A run can have at most one group:* tag."). Surface the + # server's reason rather than a misleading connection error. + raise RuntimeError(f'Failed to create run: {e}') from e if not r: + # No response after retries → the server was unreachable (a + # 4xx/5xx would have raised PlutoRequestError above). raise ConnectionError( 'Failed to create or resume run. Check connection to Pluto server.' ) @@ -948,6 +958,10 @@ def add_tags(self, tags: Union[str, List[str]]) -> None: elif self._iface: try: self._iface._update_tags(self.tags) + except PlutoRequestError as e: + # Server rejected the tags update (validation error) — this is + # not a transient failure, so surface it rather than hide it. + logger.warning(f'{tag}: server rejected tags update: {e}') except Exception as e: logger.debug(f'{tag}: failed to sync tags to server: {e}') @@ -978,6 +992,10 @@ def remove_tags(self, tags: Union[str, List[str]]) -> None: elif self._iface: try: self._iface._update_tags(self.tags) + except PlutoRequestError as e: + # Server rejected the tags update (validation error) — this is + # not a transient failure, so surface it rather than hide it. + logger.warning(f'{tag}: server rejected tags update: {e}') except Exception as e: logger.debug(f'{tag}: failed to sync tags to server: {e}') diff --git a/pluto/sync/process.py b/pluto/sync/process.py index dfa8c4b..3d7bfe2 100644 --- a/pluto/sync/process.py +++ b/pluto/sync/process.py @@ -30,6 +30,7 @@ # Fallback for environments without filelock FileLock = None # type: ignore[misc, assignment] +from ..iface import PlutoRequestError, _server_error_message from .store import FileRecord, RecordType, SyncRecord, SyncStore # Type alias for subprocess @@ -1423,17 +1424,35 @@ def _post_with_retry( headers=headers, timeout=timeout, ) - response.raise_for_status() - return except Exception as e: + # Network/timeout error — retryable. last_error = e - if attempt < max_retries - 1: - wait = min(self.retry_backoff**attempt, max_backoff) - self.log.debug( - f'Request failed (attempt {attempt + 1}/{max_retries}), ' - f'retrying in {wait}s: {e}' + else: + if response.status_code < 400: + return + # 4xx is a client/validation error (e.g. "A run can have at + # most one group:* tag.") — retrying the identical payload will + # never succeed, so stop now and surface the server's reason. + # The plain raise_for_status() message omits the response body, + # which is exactly where that reason lives. + if 400 <= response.status_code < 500: + raise PlutoRequestError( + _server_error_message(response), + status_code=response.status_code, ) - time.sleep(wait) + # 5xx — retryable. + last_error = Exception( + f'HTTP {response.status_code}: ' + f'{_server_error_message(response)}' + ) + + if attempt < max_retries - 1: + wait = min(self.retry_backoff**attempt, max_backoff) + self.log.debug( + f'Request failed (attempt {attempt + 1}/{max_retries}), ' + f'retrying in {wait}s: {last_error}' + ) + time.sleep(wait) raise last_error or Exception('Request failed after retries') diff --git a/tests/test_iface_errors.py b/tests/test_iface_errors.py new file mode 100644 index 0000000..ef0e6ce --- /dev/null +++ b/tests/test_iface_errors.py @@ -0,0 +1,164 @@ +"""Unit tests for ServerInterface write-error handling (pluto/iface.py). + +These exercise the retry/raise policy of ``_try`` without a real server: +- 4xx responses are terminal (no retry) and surface the server's message. +- 5xx responses are retried. +- Network exceptions never raise PlutoRequestError (caller sees None). +""" + +import httpx +import pytest + +from pluto.iface import PlutoRequestError, ServerInterface, _server_error_message +from pluto.sets import Settings + + +def _make_iface(): + settings = Settings() + settings.mode = 'noop' + settings.update_host() + # Keep retry backoff instant so the 5xx test doesn't actually sleep long. + settings.x_file_stream_retry_max = 2 + settings.x_file_stream_retry_wait_min_seconds = 0 + settings.x_file_stream_retry_wait_max_seconds = 0 + return ServerInterface(config={}, settings=settings) + + +def _resp(status_code, json_body=None, text=''): + if json_body is not None: + return httpx.Response(status_code, json=json_body) + return httpx.Response(status_code, text=text) + + +def test_server_error_message_prefers_error_field(): + r = _resp(400, json_body={'error': 'A run can have at most one group:* tag.'}) + assert _server_error_message(r) == 'A run can have at most one group:* tag.' + # Falls back to raw body when not the expected shape. + assert _server_error_message(_resp(400, text='plain boom')) == 'plain boom' + + +def test_try_400_does_not_retry_and_raises_server_message(): + iface = _make_iface() + calls = {'n': 0} + + def fake_method(url, content=None, headers=None, **kwargs): + calls['n'] += 1 + return _resp( + 400, json_body={'error': 'A run can have at most one group:* tag.'} + ) + + with pytest.raises(PlutoRequestError) as excinfo: + iface._try( + fake_method, + 'https://example/api/runs/create', + {}, + b'{}', + name='create', + raise_on_error=True, + ) + + assert calls['n'] == 1, 'a 400 must not be retried' + assert 'at most one group' in str(excinfo.value) + assert not isinstance(excinfo.value, ConnectionError) + assert excinfo.value.status_code == 400 + + +def test_try_400_without_raise_returns_none_no_retry(): + iface = _make_iface() + calls = {'n': 0} + + def fake_method(url, content=None, headers=None, **kwargs): + calls['n'] += 1 + return _resp(400, json_body={'error': 'bad'}) + + r = iface._try(fake_method, 'https://x', {}, b'{}', name='tags') + assert r is None + assert calls['n'] == 1, 'a 400 must not be retried even without raise_on_error' + + +def test_try_500_is_retried_then_gives_up(): + iface = _make_iface() # x_file_stream_retry_max = 2 + calls = {'n': 0} + + def fake_method(url, content=None, headers=None, **kwargs): + calls['n'] += 1 + return _resp(500, text='boom') + + # raise_on_error → persistent 5xx raises after exhausting retries. + with pytest.raises(PlutoRequestError): + iface._try( + fake_method, 'https://x', {}, b'{}', name='create', raise_on_error=True + ) + # initial attempt + 2 retries = 3 calls + assert calls['n'] == 3, calls['n'] + + +def test_try_network_error_returns_none_not_request_error(): + iface = _make_iface() + + def fake_method(url, content=None, headers=None, **kwargs): + raise httpx.ConnectError('no route') + + # Even with raise_on_error, a pure network failure is NOT a server + # validation error — caller gets None and decides (ConnectionError). + r = iface._try( + fake_method, 'https://x', {}, b'{}', name='create', raise_on_error=True + ) + assert r is None + + +# --- sync-process uploader (pluto/sync/process.py) ------------------------- + + +class _FakeClient: + """Stand-in httpx client that returns a canned response and counts posts.""" + + def __init__(self, response): + self._response = response + self.calls = 0 + + def post(self, url, content=None, headers=None, timeout=None): + self.calls += 1 + return self._response + + +def _make_uploader(response, retry_max=4): + import logging + + from pluto.sync.process import _SyncUploader + + uploader = _SyncUploader( + {'sync_process_retry_max': retry_max, 'sync_process_retry_backoff': 0.0}, + logging.getLogger('test'), + ) + uploader._client = _FakeClient(response) + return uploader + + +def test_sync_post_400_is_terminal_and_surfaces_reason(): + """The sync uploader must not retry a 400 and must raise the server's + reason (which the plain raise_for_status message would omit).""" + resp = httpx.Response( + 400, json={'error': 'A run can have at most one group:* tag.'} + ) + uploader = _make_uploader(resp) + + with pytest.raises(PlutoRequestError) as excinfo: + uploader._post_with_retry('https://x/api/runs/tags/update', b'{}', {}) + + assert uploader.client.calls == 1, 'a 400 must not be retried' + assert 'at most one group' in str(excinfo.value) + assert excinfo.value.status_code == 400 + + +def test_sync_post_500_is_retried(): + """5xx stays retryable in the sync uploader.""" + resp = httpx.Response(500, text='boom') + uploader = _make_uploader(resp, retry_max=2) + + with pytest.raises(Exception) as excinfo: + uploader._post_with_retry('https://x', b'{}', {}) + + # retry_max = 2 → 2 attempts total; never a PlutoRequestError (that's 4xx). + assert uploader.client.calls == 2 + assert not isinstance(excinfo.value, PlutoRequestError) From 80ad8c10c3182625081b250dc6e5dc0afbaff325 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Jul 2026 00:53:06 +0000 Subject: [PATCH 2/5] fix(iface): populate PlutoRequestError.status_code on retry-exhaustion path Addresses PR review: on the persistent-5xx exhaustion path _try raised PlutoRequestError with status_code=None, inconsistent with the 4xx path which sets the real code. Thread the last HTTP status through the retry recursion (last_status) so the exception reports the actual code (e.g. 500) instead of None. Prefer threading over parsing it back out of the formatted error_info string, which would couple the value to the log format. Assert the code is populated in the 5xx test. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/iface.py | 11 +++++++++-- tests/test_iface_errors.py | 4 +++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/pluto/iface.py b/pluto/iface.py index 836c6e1..efb4319 100644 --- a/pluto/iface.py +++ b/pluto/iface.py @@ -285,6 +285,7 @@ def _try( drained: Optional[List[Any]] = None, retry: int = 0, error_info: str = '', + last_status: Optional[int] = None, max_retries: Optional[int] = None, timeout: Optional[float] = None, suppress_httpx_logs: bool = False, @@ -322,9 +323,11 @@ def _try( # Distinguish a persistent server error (we got HTTP responses but # they never succeeded) from an unreachable server (network # exceptions → error_info doesn't start with "HTTP"). Only the - # former carries a server-provided reason worth raising. + # former carries a server-provided reason worth raising. last_status + # is threaded down from the last HTTP response so the exception + # reports the real code (e.g. 500) rather than None. if raise_on_error and error_info.startswith('HTTP '): - raise PlutoRequestError(error_info, status_code=None) + raise PlutoRequestError(error_info, status_code=last_status) return None @@ -344,6 +347,9 @@ def _try( # Capture error info for potential failure logging server_msg = _server_error_message(r) error_info = f'HTTP {r.status_code}: {server_msg[:200]}' + # Remember the status so a later retry-exhaustion raise carries the + # real code, not None (network errors below leave this untouched). + last_status = r.status_code target = len(drained) if drained else 'request' # High-frequency endpoints (the trigger/heartbeat that fires @@ -420,6 +426,7 @@ def _try( drained=drained, retry=retry + 1, error_info=error_info, + last_status=last_status, max_retries=effective_max_retries, timeout=timeout, suppress_httpx_logs=suppress_httpx_logs, diff --git a/tests/test_iface_errors.py b/tests/test_iface_errors.py index ef0e6ce..3beba8f 100644 --- a/tests/test_iface_errors.py +++ b/tests/test_iface_errors.py @@ -85,12 +85,14 @@ def fake_method(url, content=None, headers=None, **kwargs): return _resp(500, text='boom') # raise_on_error → persistent 5xx raises after exhausting retries. - with pytest.raises(PlutoRequestError): + with pytest.raises(PlutoRequestError) as excinfo: iface._try( fake_method, 'https://x', {}, b'{}', name='create', raise_on_error=True ) # initial attempt + 2 retries = 3 calls assert calls['n'] == 3, calls['n'] + # The raised error carries the real status code, not None. + assert excinfo.value.status_code == 500 def test_try_network_error_returns_none_not_request_error(): From d3cec26b4da1bed9d8d937d3e138494321a763eb Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Jul 2026 01:23:19 +0000 Subject: [PATCH 3/5] fix(sync): detect upload errors via raise_for_status, not response.status_code MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous rewrite of _post_with_retry read response.status_code directly, which broke ~11 existing sync-uploader tests: they mock the response with a no-op raise_for_status() and never set status_code, so `MagicMock() < 400` raised TypeError. Restore the original contract — call raise_for_status(); success if it doesn't raise — and branch on the httpx.HTTPStatusError it raises: 4xx is terminal and raises PlutoRequestError with the server's reason, 5xx and network/timeout errors stay retryable. Same external behavior as intended, compatible with the existing mocks. Update the two new sync tests to attach a request to their canned httpx.Response objects (raise_for_status needs one to build its error). Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/sync/process.py | 38 +++++++++++++++++++------------------- tests/test_iface_errors.py | 11 +++++++++-- 2 files changed, 28 insertions(+), 21 deletions(-) diff --git a/pluto/sync/process.py b/pluto/sync/process.py index 3d7bfe2..a3b6549 100644 --- a/pluto/sync/process.py +++ b/pluto/sync/process.py @@ -1404,7 +1404,9 @@ def _post_with_retry( headers: Dict[str, str], ) -> None: """POST with exponential backoff retry. Respects urgent mode settings.""" - last_error = None + import httpx + + last_error: Optional[Exception] = None # Use urgent mode settings if enabled if self._urgent_mode: @@ -1424,27 +1426,25 @@ def _post_with_retry( headers=headers, timeout=timeout, ) + response.raise_for_status() + return + except httpx.HTTPStatusError as e: + status = e.response.status_code + # 4xx is a client/validation error (e.g. "A run can have at most + # one group:* tag.") — retrying the identical payload will never + # succeed, so stop now and surface the server's reason. The plain + # HTTPStatusError message omits the response body, which is + # exactly where that reason lives. + if 400 <= status < 500: + raise PlutoRequestError( + _server_error_message(e.response), + status_code=status, + ) from e + # 5xx — retryable. + last_error = e except Exception as e: # Network/timeout error — retryable. last_error = e - else: - if response.status_code < 400: - return - # 4xx is a client/validation error (e.g. "A run can have at - # most one group:* tag.") — retrying the identical payload will - # never succeed, so stop now and surface the server's reason. - # The plain raise_for_status() message omits the response body, - # which is exactly where that reason lives. - if 400 <= response.status_code < 500: - raise PlutoRequestError( - _server_error_message(response), - status_code=response.status_code, - ) - # 5xx — retryable. - last_error = Exception( - f'HTTP {response.status_code}: ' - f'{_server_error_message(response)}' - ) if attempt < max_retries - 1: wait = min(self.retry_backoff**attempt, max_backoff) diff --git a/tests/test_iface_errors.py b/tests/test_iface_errors.py index 3beba8f..012bca6 100644 --- a/tests/test_iface_errors.py +++ b/tests/test_iface_errors.py @@ -137,11 +137,18 @@ def _make_uploader(response, retry_max=4): return uploader +# raise_for_status() needs a request attached to build its error, so give the +# canned responses one (real httpx.Client always sets it). +_REQ = httpx.Request('POST', 'https://x') + + def test_sync_post_400_is_terminal_and_surfaces_reason(): """The sync uploader must not retry a 400 and must raise the server's reason (which the plain raise_for_status message would omit).""" resp = httpx.Response( - 400, json={'error': 'A run can have at most one group:* tag.'} + 400, + json={'error': 'A run can have at most one group:* tag.'}, + request=_REQ, ) uploader = _make_uploader(resp) @@ -155,7 +162,7 @@ def test_sync_post_400_is_terminal_and_surfaces_reason(): def test_sync_post_500_is_retried(): """5xx stays retryable in the sync uploader.""" - resp = httpx.Response(500, text='boom') + resp = httpx.Response(500, text='boom', request=_REQ) uploader = _make_uploader(resp, retry_max=2) with pytest.raises(Exception) as excinfo: From 42ab5b5b8aa194d6dc0642334e758278d212479d Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Jul 2026 01:56:10 +0000 Subject: [PATCH 4/5] fix(iface): keep transient 4xx (401/408/429) retryable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Making *all* 4xx terminal regressed resilience against transient auth failures. In CI, the auth/token-validation step intermittently times out and the create endpoint returns a transient 401; previously _try retried all non-2xx up to 5x so it recovered, but the "4xx is terminal" change hard-failed on the first 401 (RuntimeError: Failed to create run: Unauthorized). Introduce RETRYABLE_STATUS_CODES = {401, 408, 429} — auth races, server timeouts, rate limits — and treat only the *other* 4xx (400, 403, 404, 409, 422, ...) as terminal. Genuine client/validation errors like 400 "A run can have at most one group:* tag." still fail fast with the server's reason; transient auth/timeout/rate-limit 4xx retry like 5xx and, if they never clear, still raise PlutoRequestError with the real code after exhausting retries. Applied to both the write client (_try) and the sync uploader (_post_with_retry). Tests: transient 401 that clears on retry succeeds; persistent 401 is retried then raises with status_code 401; 400 stays terminal/no-retry. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/iface.py | 23 +++++++++++++++++++---- pluto/sync/process.py | 20 ++++++++++++-------- tests/test_iface_errors.py | 38 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 69 insertions(+), 12 deletions(-) diff --git a/pluto/iface.py b/pluto/iface.py index efb4319..7026df9 100644 --- a/pluto/iface.py +++ b/pluto/iface.py @@ -19,6 +19,16 @@ logger = logging.getLogger(f'{__name__.split(".")[0]}') tag = 'Interface' +# 4xx status codes that are commonly *transient* and worth retrying like 5xx: +# 401 Unauthorized — auth/token validation races (e.g. the token service +# timing out mid-request), which clear on retry. +# 408 Request Timeout — server-side timeout, not a bad request. +# 429 Too Many Requests — rate limited; back off and retry. +# Every other 4xx (400, 403, 404, 409, 422, ...) is a permanent client/ +# validation error — e.g. 400 "A run can have at most one group:* tag." — +# where retrying the identical request would just fail again, so it's terminal. +RETRYABLE_STATUS_CODES = frozenset({401, 408, 429}) + class PlutoRequestError(Exception): """Raised when a Pluto write request fails with a server validation error. @@ -368,10 +378,15 @@ def _try( server_msg, ) - # 4xx is a client/validation error — retrying the identical payload - # will never succeed, so stop immediately (no wasted backoff) and, - # when asked, surface the server's reason to the caller. - if 400 <= r.status_code < 500: + # A permanent 4xx client/validation error — retrying the identical + # payload will never succeed, so stop immediately (no wasted backoff) + # and, when asked, surface the server's reason to the caller. + # Transient 4xx (RETRYABLE_STATUS_CODES: 401/408/429) fall through to + # the retry/backoff path below, same as 5xx. + if ( + 400 <= r.status_code < 500 + and r.status_code not in RETRYABLE_STATUS_CODES + ): if raise_on_error: raise PlutoRequestError(server_msg, status_code=r.status_code) return None diff --git a/pluto/sync/process.py b/pluto/sync/process.py index a3b6549..731200a 100644 --- a/pluto/sync/process.py +++ b/pluto/sync/process.py @@ -30,7 +30,11 @@ # Fallback for environments without filelock FileLock = None # type: ignore[misc, assignment] -from ..iface import PlutoRequestError, _server_error_message +from ..iface import ( + RETRYABLE_STATUS_CODES, + PlutoRequestError, + _server_error_message, +) from .store import FileRecord, RecordType, SyncRecord, SyncStore # Type alias for subprocess @@ -1430,17 +1434,17 @@ def _post_with_retry( return except httpx.HTTPStatusError as e: status = e.response.status_code - # 4xx is a client/validation error (e.g. "A run can have at most - # one group:* tag.") — retrying the identical payload will never - # succeed, so stop now and surface the server's reason. The plain - # HTTPStatusError message omits the response body, which is - # exactly where that reason lives. - if 400 <= status < 500: + # A permanent 4xx client/validation error (e.g. "A run can have + # at most one group:* tag.") — retrying the identical payload + # will never succeed, so stop now and surface the server's + # reason. The plain HTTPStatusError message omits the response + # body, which is exactly where that reason lives. + if 400 <= status < 500 and status not in RETRYABLE_STATUS_CODES: raise PlutoRequestError( _server_error_message(e.response), status_code=status, ) from e - # 5xx — retryable. + # 5xx or transient 4xx (401/408/429) — retryable. last_error = e except Exception as e: # Network/timeout error — retryable. diff --git a/tests/test_iface_errors.py b/tests/test_iface_errors.py index 012bca6..efdaca3 100644 --- a/tests/test_iface_errors.py +++ b/tests/test_iface_errors.py @@ -95,6 +95,44 @@ def fake_method(url, content=None, headers=None, **kwargs): assert excinfo.value.status_code == 500 +def test_try_transient_4xx_is_retried_then_recovers(): + """401/408/429 are transient (auth blips, timeouts, rate limits) and must + be retried — a 401 that clears on retry should succeed, not hard-fail.""" + iface = _make_iface() + calls = {'n': 0} + + def fake_method(url, content=None, headers=None, **kwargs): + calls['n'] += 1 + # First two attempts flake with 401, third succeeds. + if calls['n'] < 3: + return _resp(401, text='Unauthorized') + return _resp(200, json_body={'ok': True}) + + r = iface._try( + fake_method, 'https://x', {}, b'{}', name='create', raise_on_error=True + ) + assert r is not None and r.status_code == 200 + assert calls['n'] == 3, 'a transient 401 must be retried, not terminal' + + +def test_try_persistent_401_raises_after_retries(): + """A 401 that never clears still raises (after exhausting retries), carrying + the real status code — not a first-attempt hard fail, not None.""" + iface = _make_iface() # x_file_stream_retry_max = 2 + calls = {'n': 0} + + def fake_method(url, content=None, headers=None, **kwargs): + calls['n'] += 1 + return _resp(401, text='Unauthorized') + + with pytest.raises(PlutoRequestError) as excinfo: + iface._try( + fake_method, 'https://x', {}, b'{}', name='create', raise_on_error=True + ) + assert calls['n'] == 3, 'a 401 is retried (initial + 2 retries)' + assert excinfo.value.status_code == 401 + + def test_try_network_error_returns_none_not_request_error(): iface = _make_iface() From c55ebd49661d06deda00ec4fbac05071440c6ab8 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Wed, 8 Jul 2026 22:29:23 +0000 Subject: [PATCH 5/5] feat(files): send sampleIndex so media lists keep logged order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a user logs a list of media at one step — pluto.log({"eval/img": [img0, img1, img2]}, step=s) — the server sorted a step's files by fileName, losing the logged order. Send an explicit 0-based sampleIndex per file so the server can restore it. Contract (matches the server): each object in the POST /files `files: [...]` array gets a non-negative integer `sampleIndex` = the file's position within a single (logName, step) log() call. A list of N media → 0..N-1; a scalar (single, non-list) media → 0. Missing sampleIndex is treated as 0 server-side (older SDKs keep working), so no version gate is needed. Transparent to users — no new public API. The index is derived purely from list position via enumerate(), captured at log() time (op.py, where order is known) and threaded through the sync DB to the upload payload: - op.py: enumerate items in the log loop → _process_log_item_sync → _enqueue_file_sync → SyncProcessManager.enqueue_file. - sync/store.py: new `sample_index` column (additive migration, defaults 0), persisted on enqueue and read back into FileRecord. - sync/process.py: `_get_presigned_urls` adds `sampleIndex` to each entry. - api.py: make_compat_file_v1 (legacy builder) enumerates each logName list. Deriving from stored position (not from the batch) is deliberate — sync batches can split or mix steps, so numbering the batch would be wrong. Tests: store round-trip (list → 0..N-1, default 0); sync presign payload carries sampleIndex; make_compat_file_v1 numbers a list 0..N-1 and a single media 0. Co-Authored-By: Claude Opus 4.8 (1M context) --- pluto/api.py | 7 ++- pluto/op.py | 15 ++++- pluto/sync/process.py | 5 ++ pluto/sync/store.py | 20 ++++++- tests/test_sync_process.py | 120 ++++++++++++++++++++++++++++++++++++- 5 files changed, 158 insertions(+), 9 deletions(-) diff --git a/pluto/api.py b/pluto/api.py index 626788c..b75b41b 100644 --- a/pluto/api.py +++ b/pluto/api.py @@ -161,7 +161,11 @@ def make_compat_data_v1(data, timestamp, step): def make_compat_file_v1(file, timestamp, step): batch = [] for k, fl in file.items(): - for f in fl: + # enumerate() gives each file its 0-based position within this + # (logName, step) list so the server can restore logged order instead + # of sorting by fileName. A single (non-list) media logs as a 1-element + # list here, so it naturally gets sampleIndex 0. + for sample_index, f in enumerate(fl): i = { 'fileName': f'{f._name}{f._ext}', 'fileSize': f._stat.st_size, @@ -169,6 +173,7 @@ def make_compat_file_v1(file, timestamp, step): 'time': int(f._stat.st_mtime * 1000), 'logName': k, 'step': step, + 'sampleIndex': sample_index, } caption = getattr(f, '_caption', None) if caption is not None: diff --git a/pluto/op.py b/pluto/op.py index 363abee..544258b 100644 --- a/pluto/op.py +++ b/pluto/op.py @@ -611,9 +611,15 @@ def _log_via_sync( if items: self._register_meta_sync(k, items[0], new_metric_names, new_file_meta) - for item in items: + # enumerate() gives each item its 0-based position within this + # (key, step) log() call — the sampleIndex the server uses to + # restore logged order for media lists. A scalar is wrapped in a + # 1-element list above, so it naturally gets sampleIndex 0. + for sample_index, item in enumerate(items): try: - self._process_log_item_sync(k, item, metrics, timestamp_ms) + self._process_log_item_sync( + k, item, metrics, timestamp_ms, sample_index + ) except Exception as e: # One bad item (e.g. media that can't be encoded/staged) # must not abort the rest of the batch or crash the caller's @@ -692,6 +698,7 @@ def _process_log_item_sync( value: Any, metrics: Dict[str, Any], timestamp_ms: int, + sample_index: int = 0, ) -> None: """Process a single log item for sync process mode.""" if self._sync_manager is None: @@ -703,7 +710,7 @@ def _process_log_item_sync( metrics[key] = value.item() elif isinstance(value, File): # Handle file types (Image, Audio, Video, Text, Artifact) - self._enqueue_file_sync(key, value, timestamp_ms) + self._enqueue_file_sync(key, value, timestamp_ms, sample_index) elif isinstance(value, Data): # Handle structured data types (Graph, Histogram, Table) self._sync_manager.enqueue_data( @@ -719,6 +726,7 @@ def _enqueue_file_sync( log_name: str, file_obj: File, timestamp_ms: int, + sample_index: int = 0, ) -> None: """ Process and enqueue a file for upload via sync process. @@ -747,6 +755,7 @@ def _enqueue_file_sync( timestamp_ms=timestamp_ms, step=self._step, caption=file_obj._caption, + sample_index=sample_index, ) logger.debug( f'{tag}: enqueued file {file_obj._name}{file_obj._ext} for sync' diff --git a/pluto/sync/process.py b/pluto/sync/process.py index 731200a..be7da3d 100644 --- a/pluto/sync/process.py +++ b/pluto/sync/process.py @@ -352,6 +352,7 @@ def enqueue_file( timestamp_ms: int, step: Optional[int] = None, caption: Optional[str] = None, + sample_index: int = 0, ) -> None: """ Enqueue a file for upload. @@ -370,6 +371,7 @@ def enqueue_file( timestamp_ms=timestamp_ms, step=step, caption=caption, + sample_index=sample_index, ) # Update heartbeat to show we're alive self.store.heartbeat(self.run_id) @@ -1301,6 +1303,9 @@ def _get_presigned_urls( 'time': f.timestamp_ms, 'logName': f.log_name, 'step': f.step, + # 0-based position within a single (logName, step) log() call so + # the server can restore logged order instead of sorting by name. + 'sampleIndex': f.sample_index, } # Only include caption when set — keeps the payload identical to # before for un-captioned files and older servers (which ignore diff --git a/pluto/sync/store.py b/pluto/sync/store.py index 1e549ee..d00e175 100644 --- a/pluto/sync/store.py +++ b/pluto/sync/store.py @@ -163,6 +163,10 @@ class FileRecord: error_message: Optional[str] presigned_url: Optional[str] caption: Optional[str] = None + # 0-based position of this file within a single (log_name, step) log() call + # (e.g. pluto.log({"k": [a, b, c]}) → 0, 1, 2). Lets the server restore the + # logged order instead of sorting by fileName. Scalars get 0. + sample_index: int = 0 @dataclass @@ -323,6 +327,7 @@ def _init_schema(self) -> None: file_ext TEXT, log_name TEXT, caption TEXT, + sample_index INTEGER DEFAULT 0, timestamp_ms INTEGER NOT NULL, step INTEGER, status INTEGER DEFAULT 0, @@ -338,6 +343,9 @@ def _init_schema(self) -> None: # CREATE TABLE IF NOT EXISTS above is a no-op on an existing table, # so backfill new columns here. Idempotent (skips if present). self._add_column_if_missing(cursor, 'file_uploads', 'caption', 'TEXT') + self._add_column_if_missing( + cursor, 'file_uploads', 'sample_index', 'INTEGER DEFAULT 0' + ) self.conn.commit() @@ -694,6 +702,7 @@ def enqueue_file( timestamp_ms: int, step: Optional[int] = None, caption: Optional[str] = None, + sample_index: int = 0, ) -> int: """Add a file to the upload queue. Returns file record ID.""" now = time.time() @@ -704,9 +713,9 @@ def enqueue_file( INSERT INTO file_uploads ( run_id, local_path, file_type, file_size, timestamp_ms, step, created_at, file_name, file_ext, - log_name, caption + log_name, caption, sample_index ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( run_id, @@ -720,6 +729,7 @@ def enqueue_file( file_ext, log_name, caption, + sample_index, ), ) return cursor.lastrowid or 0 @@ -763,6 +773,12 @@ def get_pending_files( error_message=row['error_message'], presigned_url=row['remote_url'], caption=(row['caption'] if 'caption' in row.keys() else None), + sample_index=( + row['sample_index'] + if 'sample_index' in row.keys() + and row['sample_index'] is not None + else 0 + ), ) ) return records diff --git a/tests/test_sync_process.py b/tests/test_sync_process.py index 6a7d05f..a8b157c 100644 --- a/tests/test_sync_process.py +++ b/tests/test_sync_process.py @@ -123,6 +123,43 @@ def test_enqueue_file_caption_roundtrip(self, store): assert by_name['cat'].caption == 'a fluffy orange cat' assert by_name['dog'].caption is None + def test_enqueue_file_sample_index_roundtrip(self, store): + """sample_index set on enqueue survives read-back; the default is 0.""" + store.register_run('test-run-1', 'test-project') + + # A 3-image list: img0/img1/img2 → sample_index 0/1/2. + for i in range(3): + store.enqueue_file( + run_id='test-run-1', + local_path=f'/tmp/img{i}.png', + file_name=f'img{i}', + file_ext='.png', + file_type='image/png', + file_size=1024, + log_name='eval/img', + timestamp_ms=int(time.time() * 1000), + step=5, + sample_index=i, + ) + # A file enqueued without sample_index defaults to 0. + store.enqueue_file( + run_id='test-run-1', + local_path='/tmp/solo.png', + file_name='solo', + file_ext='.png', + file_type='image/png', + file_size=1024, + log_name='eval/solo', + timestamp_ms=int(time.time() * 1000), + step=5, + ) + + by_name = {f.file_name: f for f in store.get_pending_files(limit=10)} + assert by_name['img0'].sample_index == 0 + assert by_name['img1'].sample_index == 1 + assert by_name['img2'].sample_index == 2 + assert by_name['solo'].sample_index == 0 + def test_caption_migration_on_legacy_db(self, tmp_path): """A DB created without the caption column gets it added on open (additive migration), so enqueue/select with caption works.""" @@ -436,9 +473,9 @@ def test_sync_process_registers_metric_names(self): assert 'test_metric_gamma' in run.settings.meta # Verify _iface exists and would have been used for metadata - assert ( - run._iface is not None - ), 'ServerInterface must exist to register metric names with server' + assert run._iface is not None, ( + 'ServerInterface must exist to register metric names with server' + ) run.finish() @@ -686,6 +723,56 @@ def test_file_payload_includes_caption(self, uploader): assert by_name['cat.png']['caption'] == 'a fluffy orange cat' assert 'caption' not in by_name['dog.png'] + def test_file_payload_includes_sample_index(self, uploader): + """The /files presign payload carries `sampleIndex` for each file so the + server can restore logged order for a media list.""" + # Three images logged as one list at (eval/img, step=5). + records = [ + FileRecord( + id=i + 1, + run_id='test-run', + local_path=f'/tmp/img{i}.png', + file_name=f'img{i}', + file_ext='.png', + file_type='image/png', + file_size=10, + log_name='eval/img', + timestamp_ms=1705600000000, + step=5, + status=SyncStatus.PENDING, + retry_count=0, + created_at=time.time(), + last_attempt_at=None, + error_message=None, + presigned_url=None, + sample_index=i, + ) + for i in range(3) + ] + + mock_response = MagicMock() + mock_response.raise_for_status = MagicMock() + mock_response.json.return_value = {} + + with patch('httpx.Client') as MockClient: + mock_client = MagicMock() + mock_client.post.return_value = mock_response + MockClient.return_value = mock_client + uploader._client = None + + uploader._get_presigned_urls(records) + + call_args = mock_client.post.call_args + body = call_args.kwargs.get('content') or call_args[1].get('content') + if isinstance(body, bytes): + body = body.decode('utf-8') + files = json.loads(body)['files'] + by_name = {f['fileName']: f for f in files} + + assert by_name['img0.png']['sampleIndex'] == 0 + assert by_name['img1.png']['sampleIndex'] == 1 + assert by_name['img2.png']['sampleIndex'] == 2 + def test_metrics_payload_filters_non_numeric(self, uploader): """Test that non-numeric values are filtered from metrics.""" records = [ @@ -1554,3 +1641,30 @@ def test_bad_item_dropped_loudly_batch_survives(self, tmp_path, caplog): ) assert not [r for r in caplog.records if r.levelno >= logging.WARNING] assert cap.call_count == 1 + + +def test_make_compat_file_v1_assigns_sample_index(): + """make_compat_file_v1 numbers files 0..N-1 by their position within a + (logName, step) list; a single (1-element) media gets sampleIndex 0.""" + import types + + from pluto.api import make_compat_file_v1 + + def _f(name): + return types.SimpleNamespace( + _name=name, + _ext='.png', + _stat=types.SimpleNamespace(st_size=10, st_mtime=1.0), + _caption=None, + ) + + # A list of 3 media under one logName → sampleIndex 0, 1, 2 in order. + payload = json.loads( + make_compat_file_v1({'eval/img': [_f('a'), _f('b'), _f('c')]}, 0, 5).decode() + ) + by_name = {e['fileName']: e['sampleIndex'] for e in payload['files']} + assert by_name == {'a.png': 0, 'b.png': 1, 'c.png': 2} + + # A single (non-list) media logs as a 1-element list → sampleIndex 0. + solo = json.loads(make_compat_file_v1({'eval/solo': [_f('s')]}, 0, 5).decode()) + assert solo['files'][0]['sampleIndex'] == 0