[API server][Security] Keep worker and decoder error text out of client responses

mstar/api_server/data_worker.py

@@ -47,6 +47,7 @@
 from mstar.observability.recorders import EdgeTransfers
 from mstar.profile.format import InputInfo, RxInfo, TxInfo
 from mstar.utils import profiler
+from mstar.utils.client_errors import client_error_text
 from mstar.utils.ipc_format import (
     AbortRequest,
     ConductorMessage,
@@ -909,18 +910,16 @@ def _fail_request(
         ``traceback`` False logs one line, for an error already logged where
         it happened.
         """
-        if traceback:
-            logger.error("%s failed for request %s", stage, request_id, exc_info=exc)
-        else:
-            logger.error(
-                "%s failed for request %s: %s: %s", stage, request_id, type(exc).__name__, exc,
-            )
+        text, _ = client_error_text(
+            exc, f"{stage} failed for request {request_id}", log=logger,
+            traceback=traceback,
+        )
         status = 400 if isinstance(exc, (ValueError, TypeError)) else 500
         chunks = [
             ResultChunk(
                 request_id=request_id,
                 modality="error",
-                data=f"{stage} failed: {type(exc).__name__}: {exc}".encode("utf-8"),
+                data=f"{stage} failed: {text}".encode("utf-8"),
                 metadata={"status": status},
             )
             for _ in range(max(count, 1))

mstar/api_server/entrypoint.py

@@ -53,6 +53,7 @@
 from mstar.profile.display import pretty_print_profile
 from mstar.profile.format import OutputInfo, RequestProfile, RequestTiming
 from mstar.utils import profiler
+from mstar.utils.client_errors import client_error_text
 from mstar.utils.exitcode import describe_exitcode
 from mstar.utils.gc_freeze import freeze_after_setup
 from mstar.utils.logging_config import quiet_noisy_loggers
@@ -1710,16 +1711,20 @@ async def serve_one(
         except (WebSocketDisconnect, asyncio.CancelledError):
             raise
         except Exception as exc:  # noqa: BLE001 — reported in-band, the socket stays up
-            reply = {"request_id": request_id, "error": str(exc)}
             if isinstance(exc, HTTPException):
-                reply.update(error=exc.detail, status=exc.status_code)
-            if isinstance(exc, HTTPException) and exc.status_code < 500:
-                logger.info(
-                    "generate/ws request %s refused (%d): %s",
-                    request_id, exc.status_code, exc.detail,
-                )
+                reply = {"request_id": request_id, "error": exc.detail, "status": exc.status_code}
+                if exc.status_code < 500:
+                    logger.info(
+                        "generate/ws request %s refused (%d): %s",
+                        request_id, exc.status_code, exc.detail,
+                    )
+                else:
+                    logger.exception("generate/ws request failed")
             else:
-                logger.exception("generate/ws request failed")
+                text, _ = client_error_text(
+                    exc, f"generate/ws request {request_id} failed", log=logger, type_name=False,
+                )
+                reply = {"request_id": request_id, "error": text}
             try:
                 await send(reply, binary)
             except Exception:  # noqa: BLE001
@@ -1925,7 +1930,10 @@ async def generate(
     except HTTPException:
         raise
     except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e)) from e
+        text, _ = client_error_text(
+            e, f"generate request {request_id} failed", log=logger, type_name=False,
+        )
+        raise HTTPException(status_code=500, detail=text) from e
     finally:
         # Deferred cleanup of uploaded files
         if file_paths:

mstar/api_server/openai/router.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import logging
 
 from fastapi import APIRouter, Request, WebSocket
 from fastapi.responses import JSONResponse, StreamingResponse
@@ -32,6 +33,9 @@
     TranscriptionRequest,
     VideoGenerationRequest,
 )
+from mstar.utils.client_errors import client_error_text
+
+logger = logging.getLogger(__name__)
 
 router = APIRouter()
 
@@ -59,6 +63,26 @@ def _error(status: int, message: str, type_: str | None = None) -> JSONResponse:
     )
 
 
+def _exception_error(e: Exception, route: str) -> JSONResponse:
+    """The error envelope for an exception a route's handler raised.
+
+    An ``HTTPException`` was raised on purpose and keeps its status and
+    detail. Anything else goes through :func:`client_error_text`: a
+    ``ValueError`` or ``TypeError`` is the request's fault, a 400 with server
+    paths cut to base names, and any other error is a 500 that shows the
+    client only an error id, with the full error logged under it. An
+    exception class that names its own ``status_code`` (``AudioTooLongError``
+    is a 413) keeps that status.
+    """
+    status, detail = getattr(e, "status_code", None), getattr(e, "detail", None)
+    if isinstance(status, int) and detail is not None:
+        return _error(status, str(detail))
+    text, _ = client_error_text(e, f"{route} request failed", log=logger, type_name=False)
+    if not isinstance(status, int):
+        status = 400 if isinstance(e, (ValueError, TypeError)) else 500
+    return _error(status, text)
+
+
 def _resolve(require: str):
     """Return (api, model_name, adapter, error_response). ``error`` is non-None
     when the loaded model can't serve ``require`` (e.g. 'supports_chat')."""
@@ -92,8 +116,7 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
     try:
         result = await serving_chat.create_chat_completion(api, model_name, adapter, request, raw_request)
     except Exception as e:  # noqa: BLE001 — surface as an OpenAI error envelope
-        default_status = 400 if isinstance(e, (ValueError, TypeError)) else 500
-        return _error(getattr(e, "status_code", default_status), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/chat/completions")
     if request.stream:
         return StreamingResponse(
             result, media_type="text/event-stream", headers={"Cache-Control": "no-cache"}
@@ -109,7 +132,7 @@ async def audio_speech(request: SpeechRequest, raw_request: Request):
     try:
         return await serving_speech.create_speech(api, model_name, adapter, request, raw_request)
     except Exception as e:  # noqa: BLE001
-        return _error(getattr(e, "status_code", 500), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/audio/speech")
 
 
 @router.post("/v1/audio/transcriptions")
@@ -151,8 +174,7 @@ async def audio_transcriptions(request: Request):
             getattr(upload, "filename", None), raw_request=request,
         )
     except Exception as e:  # noqa: BLE001
-        default_status = 400 if isinstance(e, (ValueError, TypeError)) else 500
-        return _error(getattr(e, "status_code", default_status), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/audio/transcriptions")
 
 
 @router.websocket("/v1/realtime")
@@ -192,9 +214,7 @@ async def images_generations(request: ImageGenerationRequest, raw_request: Reque
     try:
         result = await serving_images.create_images(api, model_name, adapter, request, raw_request)
     except Exception as e:  # noqa: BLE001
-        # a malformed request (adapter / model validation) is the client's error, not the server's
-        default_status = 400 if isinstance(e, (ValueError, TypeError)) else 500
-        return _error(getattr(e, "status_code", default_status), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/images/generations")
     return JSONResponse(result)
 
 
@@ -206,8 +226,7 @@ async def videos_generations(request: VideoGenerationRequest, raw_request: Reque
     try:
         result = await serving_videos.create_videos(api, model_name, adapter, request, raw_request)
     except Exception as e:  # noqa: BLE001
-        default_status = 400 if isinstance(e, (ValueError, TypeError)) else 500
-        return _error(getattr(e, "status_code", default_status), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/videos/generations")
     if (request.model_extra or {}).get("stream_video"):
         # A windowed request delivering each window as it lands: NDJSON lines
         # (see serving_videos._stream_ndjson) instead of one JSON body.
@@ -256,6 +275,5 @@ async def images_edits(request: Request):
             raw_request=request,
         )
     except Exception as e:  # noqa: BLE001
-        default_status = 400 if isinstance(e, (ValueError, TypeError)) else 500
-        return _error(getattr(e, "status_code", default_status), str(getattr(e, "detail", e)))
+        return _exception_error(e, "/v1/images/edits")
     return JSONResponse(result)

mstar/api_server/rust_frontend.py

@@ -60,6 +60,7 @@
     RustZMQCommunicator,
 )
 from mstar.model.multimodal import PromptPart
+from mstar.utils.client_errors import client_error_text
 
 logger = logging.getLogger(__name__)
 
@@ -185,8 +186,10 @@ def _submit(self, msg: dict) -> None:
                     "data": str(detail).encode(), "metadata": {"status": status},
                 })
                 return
-            logger.warning("submit %s failed: %r", rid, e)
-            self._err(rid, sub, repr(e))
+            text, _ = client_error_text(
+                e, f"submit {rid} failed", log=logger, type_name=False,
+            )
+            self._err(rid, sub, text)
             return
         # Bound to this submission now, not when the task first runs: a later
         # submit in the same receive batch may reuse the rid.
@@ -204,9 +207,15 @@ async def _relay(self, rid: str, sub: int | None, chunks) -> None:
                 })
             self._send({"t": "done", "rid": rid, "sub": sub})
         except Exception as e:  # noqa: BLE001 — includes the per-request timeout
-            detail = getattr(e, "detail", None) or repr(e)
-            logger.warning("request %s failed: %s", rid, detail)
-            self._err(rid, sub, str(detail))
+            detail = getattr(e, "detail", None)
+            if detail:
+                detail = str(detail)
+                logger.warning("request %s failed: %s", rid, detail)
+            else:
+                detail, _ = client_error_text(
+                    e, f"request {rid} failed", log=logger, type_name=False,
+                )
+            self._err(rid, sub, detail)
         finally:
             self._keys.pop(sub, None)
 

mstar/conductor/conductor.py

@@ -42,6 +42,7 @@
 from mstar.profile.format import RxInfo, TxInfo
 from mstar.profile.worker import GraphTimings
 from mstar.utils import knobs
+from mstar.utils.client_errors import client_error_text, scrub_paths
 from mstar.utils.exitcode import describe_exitcode
 from mstar.utils.gc_freeze import freeze_after_setup
 from mstar.utils.generation_kwargs import check_seed
@@ -1159,18 +1160,19 @@ def _admit_request(self, body: NewRequestConductor):
             if request_data is None or not request_data.dispatched:
                 self.requests.pop(body.request_id, None)
                 if isinstance(exc, ValueError):
-                    self._reject_request(body.request_id, str(exc), status=400)
+                    self._reject_request(body.request_id, scrub_paths(str(exc)), status=400)
                     return
-                logger.exception("Ingest failed for request %s", body.request_id)
-                self._reject_request(
-                    body.request_id, f"Ingest failed: {type(exc).__name__}: {exc}", status=500,
+                text, _ = client_error_text(
+                    exc, f"Ingest failed for request {body.request_id}", log=logger,
                 )
+                self._reject_request(body.request_id, f"Ingest failed: {text}", status=500)
             elif body.request_id not in self.draining:
-                logger.exception("Ingest failed for request %s", body.request_id)
+                text, _ = client_error_text(
+                    exc, f"Ingest failed for request {body.request_id}", log=logger,
+                )
                 # Some workers may already hold it: drain them like a worker failure.
                 self._remove_request(
-                    body.request_id, request_data,
-                    failure_error=f"Ingest failed: {type(exc).__name__}: {exc}",
+                    body.request_id, request_data, failure_error=f"Ingest failed: {text}",
                 )
 
     def _reject_request(self, request_id: str, error: str, status: int):
@@ -2008,7 +2010,9 @@ def _fail_after_error(
 
         A request whose ``request_complete`` already went out is only torn
         down: its client has the result."""
-        logger.exception("Conductor error %s for request %s", step, request_id)
+        text, _ = client_error_text(
+            exc, f"Conductor error {step} for request {request_id}", log=logger,
+        )
         request_data = self.requests.get(request_id)
         if request_data is None or request_id in self.draining:
             return  # unknown, never admitted, or already being torn down
@@ -2017,7 +2021,7 @@ def _fail_after_error(
                 request_id, request_data,
                 failure_error=(
                     None if request_data.completion_sent
-                    else f"Conductor error {step}: {type(exc).__name__}: {exc}"
+                    else f"Conductor error {step}: {text}"
                 ),
             )
         except Exception:

mstar/engine/engine.py

@@ -59,6 +59,7 @@
 )
 from mstar.platforms.compile import with_inductor_overrides
 from mstar.profile.worker import ExecTimings
+from mstar.utils.client_errors import client_error_text
 from mstar.utils.device import DeviceCapability
 from mstar.utils.ipc_format import OffloadDelta, ScheduleTPNode
 from mstar.utils.profiler import (
@@ -406,8 +407,12 @@ def release_waiters(self):
         self.outputs_ready.set()
         self.commit_done.set()
 
-    def register_failure(self, rid: int, error: Exception):
-        self.failed_requests[rid] = f"{type(error).__name__}: {error}"
+    def register_failure(self, rid: int, error: Exception, walk: str | None = None):
+        self.failed_requests[rid], _ = client_error_text(
+            error,
+            f"request {rid} failed (node={self.node_name}, walk={walk or self.graph_walk})",
+            log=logger,
+        )
 
     def drop_rids(self, rids: set[int]):
         """Take rids out of the step, leaving the rest of the batch intact."""
@@ -1013,11 +1018,7 @@ def _prepare_inputs(self, batch: ExecutingBatch) -> None:
                     # the submodule reads a mixed step's rows by their own walk
                     req_inputs.graph_walk = row_walk
             except Exception as error:
-                logger.exception(
-                    "prepare_inputs failed for request %s (node=%s, walk=%s)",
-                    rid, batch.node_name, row_walk,
-                )
-                batch.register_failure(rid, error)
+                batch.register_failure(rid, error, walk=row_walk)
                 continue
             if req_inputs is None:
                 batch.skipped_rids.add(rid)
@@ -1766,10 +1767,6 @@ def _postprocess_batch(
                     inputs=node_inputs,
                 )
             except Exception as error:
-                logger.exception(
-                    "postprocess failed for request %s (node=%s, walk=%s)",
-                    rid, batch.node_name, batch.step_context.graph_walk,
-                )
                 batch.register_failure(rid, error)
         if type(submodule).batch_postprocess is not NodeSubmodule.batch_postprocess:
             live = [
@@ -1868,10 +1865,6 @@ def check_stop_for_batch(
                     rid, batch.per_request_info_wrapped[rid], rid_outputs
                 )
             except Exception as error:
-                logger.exception(
-                    "check_stop failed for request %s (node=%s, walk=%s)",
-                    rid, batch.node_name, batch.step_context.graph_walk,
-                )
                 batch.register_failure(rid, error)
                 continue
             if rid_stops:

mstar/engine/resources/kv/manager.py

@@ -55,6 +55,7 @@
     Segment,
     StepContext,
 )
+from mstar.utils.client_errors import client_error_text
 from mstar.utils.h2d import PinnedStager
 from mstar.utils.knobs import kv_chain_lazy_steps, kv_pinned_indptrs, kv_plan_cache
 from mstar.utils.profiler import PhaseLaps
@@ -236,6 +237,8 @@ class CacheStream:
     # a failed retrieve, latched: the future is consumed once, but every later
     # readiness check has to keep reporting the stream as unusable
     read_error: BaseException | None = None
+    # what a client may see of it (no server paths or internal text)
+    read_error_text: str = ""
     # None while the stream is unkeyed, and once its keys stop describing it
     chain: PrefixChain | None = None
     # pages the index matched for this stream and is holding for it, from the
@@ -3470,16 +3473,17 @@ def _check_ready(
                 future.result()
             except Exception as e:
                 stream.read_error = e
+                stream.read_error_text, _ = client_error_text(
+                    e, f"KV retrieve for request {rid} label {label!r} failed", log=logger,
+                )
             else:
                 stream.read_pending = False
         if stream.host_copy is not None and stream.host_copy.landed:
             stream.host_copy = None
             stream.read_pending = stream.read_future is not None
         if stream.read_error is not None:
-            err = stream.read_error
             return False, AdmitRuntimeError(
-                f"KV retrieve for request {rid} label {label!r} failed: "
-                f"{type(err).__name__}: {err}"
+                f"KV retrieve for request {rid} label {label!r} failed: {stream.read_error_text}"
             )
         return not stream.read_pending, None
 

mstar/model/cosmos3/tests/test_dit_step_guard.py

@@ -139,7 +139,7 @@ def test_finished_request_re_queued_fails_alone_instead_of_asserting():
     batch.register_prepare_batch = lambda inputs: setattr(batch, "inputs", inputs)
     batch.drop_rids = lambda rids: setattr(
         batch, "request_ids", [r for r in batch.request_ids if r not in rids])
-    batch.register_failure = lambda rid, err: batch.failed_requests.__setitem__(rid, str(err))
+    batch.register_failure = lambda rid, err, walk=None: batch.failed_requests.__setitem__(rid, str(err))
     eng = Engine.__new__(Engine)
     eng._submodules = {"dit": SimpleNamespace(submodule=dit, resources={})}
     eng._enable_nvtx = False