fix(tags): report server 4xx rejections instead of masking them

pluto/api.py

@@ -161,14 +161,19 @@ 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,
                 'fileType': f._ext[1:],
                 'time': int(f._stat.st_mtime * 1000),
                 'logName': k,
                 'step': step,
+                'sampleIndex': sample_index,
             }
             caption = getattr(f, '_caption', None)
             if caption is not None:

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": "<reason>"}`` 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 (

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}')