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/iface.py b/pluto/iface.py index 485830d..7026df9 100644 --- a/pluto/iface.py +++ b/pluto/iface.py @@ -19,6 +19,45 @@ 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. + + 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(): @@ -143,12 +182,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: @@ -250,9 +295,11 @@ 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, + raise_on_error: bool = False, ): effective_max_retries = ( max_retries @@ -283,6 +330,15 @@ 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. 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=last_status) + return None try: @@ -299,11 +355,13 @@ 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]}' + # 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 - 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 +372,28 @@ def _try( name, retry + 1, effective_max_retries + 1, - status_code, + r.status_code, target, url, - response, + server_msg, ) + + # 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 + 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, @@ -366,9 +441,11 @@ 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, + raise_on_error=raise_on_error, ) def _put_v1( @@ -403,6 +480,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 +502,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..544258b 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.' ) @@ -601,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 @@ -682,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: @@ -693,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( @@ -709,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. @@ -737,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' @@ -948,6 +967,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 +1001,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..be7da3d 100644 --- a/pluto/sync/process.py +++ b/pluto/sync/process.py @@ -30,6 +30,11 @@ # Fallback for environments without filelock FileLock = None # type: ignore[misc, assignment] +from ..iface import ( + RETRYABLE_STATUS_CODES, + PlutoRequestError, + _server_error_message, +) from .store import FileRecord, RecordType, SyncRecord, SyncStore # Type alias for subprocess @@ -347,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. @@ -365,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) @@ -1296,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 @@ -1403,7 +1413,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: @@ -1425,15 +1437,31 @@ def _post_with_retry( ) response.raise_for_status() return + except httpx.HTTPStatusError as e: + status = e.response.status_code + # 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 or transient 4xx (401/408/429) — retryable. + last_error = e 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}' - ) - time.sleep(wait) + + 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/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_iface_errors.py b/tests/test_iface_errors.py new file mode 100644 index 0000000..efdaca3 --- /dev/null +++ b/tests/test_iface_errors.py @@ -0,0 +1,211 @@ +"""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) 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_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() + + 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 + + +# 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.'}, + request=_REQ, + ) + 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', request=_REQ) + 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) 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