added rabbitmq user id to the outbound rabbitmq message

app.py

@@ -10,8 +10,7 @@ from oa_server import OpenAIServers
 #from vllm_backend import VLLMChatBackend, StubImagesBackend
 #from transformers_backend import TransformersChatBackend, StubImagesBackend
 #from hf_backend import HFChatBackend, StubImagesBackend
-from hf_backend import StubImagesBackend
-from timesfm_backend import TimesFMBackend
+from timesfm_backend import TimesFMBackend, StubImagesBackend
 
 
 logging.basicConfig(
@@ -62,7 +61,7 @@ async def _startup_init():
     try:
         await base.connect()          # connect to RabbitMQ
         await listener.start(DECLS)   # start queue listeners
-        return "OpenAI MQ + vLLM: ready"
+        return "OpenAI MQ + TimesFM: ready"
     except Exception as e:
         log.exception("Startup init failed")
         return f"ERROR: {e}"

oa_server.py

@@ -10,6 +10,28 @@ logger = logging.getLogger(__name__)
 
 # ------------------ helpers ------------------
 def _now() -> int: return int(time.time())
+def _extract_user_id(data: Dict[str, Any]) -> Optional[str]:
+    if not isinstance(data, dict):
+        return None
+    for k in ("userId", "UserID", "user_id"):
+        v = data.get(k)
+        if isinstance(v, str) and v.strip():
+            return v.strip()
+    ui = data.get("UserInfo")
+    if isinstance(ui, dict):
+        v = ui.get("UserID") or ui.get("userId") or ui.get("user_id")
+        if isinstance(v, str) and v.strip():
+            return v.strip()
+    return None
+
+def _with_user_id(payload: Dict[str, Any], user_id: Optional[str]) -> Dict[str, Any]:
+    if not user_id:
+        return payload
+    if isinstance(payload, dict):
+        if not payload.get("userId"):
+            payload = {**payload, "userId": user_id}
+    return payload
+
 def _chunk_text(s: str, sz: int = 140) -> List[str]:
     return [s[i:i+sz] for i in range(0, len(s or ""), sz)] if s else []
 def _last_user_text(messages: List[Dict[str, Any]]) -> str:
@@ -55,17 +77,26 @@ class OpenAIServers:
             logger.error("oaChatCreate: missing reply_key")
             return
 
+        user_id = _extract_user_id(data)
         try:
             async for chunk in self._chat.stream(data):
                 try:
-                    await self._pub.publish("oa.chat.reply", chunk, routing_key=reply_key)
+                    await self._pub.publish(
+                        "oa.chat.reply",
+                        _with_user_id(chunk, user_id),
+                        routing_key=reply_key,
+                    )
                 except Exception:
                     logger.exception("oaChatCreate: publish failed")
                     break  # stop streaming on publish failure
 
             # Optional sentinel
             try:
-                await self._pub.publish("oa.chat.reply", {"object": "stream.end"}, routing_key=reply_key)
+                await self._pub.publish(
+                    "oa.chat.reply",
+                    _with_user_id({"object": "stream.end"}, user_id),
+                    routing_key=reply_key,
+                )
             except Exception:
                 logger.exception("oaChatCreate: publish sentinel failed")
         except Exception:
@@ -84,11 +115,16 @@ class OpenAIServers:
             logger.error("oaImagesGenerate: missing reply_key")
             return
 
+        user_id = _extract_user_id(data)
         try:
             b64 = await self._img.generate_b64(data)
             resp = {"created": _now(), "data":[{"b64_json": b64}]}
             try:
-                await self._pub.publish("oa.images.reply", resp, routing_key=reply_key)
+                await self._pub.publish(
+                    "oa.images.reply",
+                    _with_user_id(resp, user_id),
+                    routing_key=reply_key,
+                )
             except Exception:
                 logger.exception("oaImagesGenerate: publish failed")
         except Exception:

rabbit_repo.py

@@ -3,6 +3,7 @@ import uuid
 import asyncio
 import logging
 from typing import Any, Optional
+from urllib.parse import urlsplit, unquote
 
 import aiormq
 import aio_pika
@@ -30,14 +31,30 @@ class RabbitRepo(RabbitBase):
                 return settings.EXCHANGE_TYPES[max(matches, key=len)]
         return "fanout"
 
+    def _publisher_user_id(self) -> Optional[str]:
+        user = getattr(settings, "RABBIT_USER_NAME", None)
+        if isinstance(user, str) and user.strip():
+            return user.strip()
+        amqp_url = getattr(settings, "AMQP_URL", None)
+        if isinstance(amqp_url, str) and amqp_url.strip():
+            try:
+                parsed = urlsplit(amqp_url)
+                if parsed.username:
+                    return unquote(parsed.username)
+            except Exception:
+                pass
+        return None
+
     async def _publish_with_retry(self, exchange: str, body: bytes, routing_key: str = "") -> None:
         attempts, delay = 0, 0.5
+        publisher_user_id = self._publisher_user_id()
         while True:
             try:
                 ex = await self.ensure_exchange(exchange)
                 msg = aio_pika.Message(
                     body=body,
                     delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+                    user_id=publisher_user_id,
                 )
                 await ex.publish(msg, routing_key=routing_key)
                 return

requirements.txt

@@ -11,7 +11,6 @@ pydantic>=2.11.7
 pydantic-settings>=2.10.1
 
 # Hugging Face / LLM
-spaces
 accelerate
 autoawq
 huggingface_hub