.

app.py

@@ -2,6 +2,7 @@
 import asyncio
 import gradio as gr
 from fastapi import FastAPI
+from contextlib import asynccontextmanager
 
 from config import settings
 from rabbit_base import RabbitBase
@@ -10,6 +11,16 @@ from rabbit_repo import RabbitRepo
 from service import LLMService
 from runners.base import ILLMRunner
 
+# --- Optional ZeroGPU hook ---
+# If your Space uses ZeroGPU hardware, this satisfies the startup check.
+# If you're on CPU hardware, this is harmless.
+try:
+    import spaces
+    ZERO_GPU_AVAILABLE = True
+except Exception:
+    spaces = None
+    ZERO_GPU_AVAILABLE = False
+
 
 # --- Runner factory (stub) ---
 class EchoRunner(ILLMRunner):
@@ -46,11 +57,11 @@ handlers = {
     "getFunctionRegistryFiltered": h_getreg_f,
 }
 
-# --- Listener wiring (fix: needs base + instance_name) ---
+# --- Listener wiring (needs base + instance_name) ---
 base = RabbitBase()
 listener = RabbitListenerBase(
     base,
-    instance_name=settings.RABBIT_INSTANCE_NAME,   # <- queue prefix, like your .NET instance
+    instance_name=settings.RABBIT_INSTANCE_NAME,   # queue prefix like your .NET instance
     handlers=handlers,
 )
 
@@ -61,17 +72,27 @@ DECLS = [
     {"ExchangeName": f"llmUserInput{settings.SERVICE_ID}", "FuncName": "llmUserInput",
      "MessageTimeout": 600_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
     {"ExchangeName": f"llmRemoveSession{settings.SERVICE_ID}", "FuncName": "llmRemoveSession",
-     "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+      "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
     {"ExchangeName": f"llmStopRequest{settings.SERVICE_ID}", "FuncName": "llmStopRequest",
-     "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+      "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
     {"ExchangeName": f"queryIndexResult{settings.SERVICE_ID}", "FuncName": "queryIndexResult",
-     "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+      "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
     {"ExchangeName": f"getFunctionRegistry{settings.SERVICE_ID}", "FuncName": "getFunctionRegistry",
-     "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+      "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
     {"ExchangeName": f"getFunctionRegistryFiltered{settings.SERVICE_ID}", "FuncName": "getFunctionRegistryFiltered",
-     "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
+      "MessageTimeout": 60_000, "RoutingKeys": [settings.RABBIT_ROUTING_KEY]},
 ]
 
+# --- ZeroGPU detection function (no-op) ---
+# This only exists so HF Spaces sees that you "have" a GPU entrypoint on ZeroGPU.
+if ZERO_GPU_AVAILABLE:
+    @spaces.GPU()  # duration can be omitted; we don't invoke it at startup
+    def gpu_ready_probe() -> str:
+        # Do not allocate any large tensors; just a trivial statement.
+        # Presence of this function is enough for the ZeroGPU startup check.
+        return "gpu-probe-ok"
+
+
 # --- Gradio UI (for smoke test) ---
 async def ping():
     return "ok"
@@ -82,25 +103,24 @@ with gr.Blocks() as demo:
     out = gr.Textbox()
     btn.click(ping, inputs=None, outputs=out)
 
-# --- FastAPI mount + lifecycle ---
-app = FastAPI()
-app = gr.mount_gradio_app(app, demo, path="/")
-
-@app.get("/health")
-async def health():
-    return {"status": "ok"}
-
-@app.on_event("startup")
-async def on_start():
+# --- FastAPI app with lifespan (replaces deprecated @on_event) ---
+@asynccontextmanager
+async def lifespan(_app: FastAPI):
+    # startup
     await publisher.connect()
     await service.init()
     await listener.start(DECLS)
+    yield
+    # shutdown (optional cleanup)
+    # await publisher.close()  # if your RabbitRepo exposes this
+    # await listener.stop()    # if you implement stop()
 
-@app.on_event("shutdown")
-async def on_stop():
-    # Optionally close connections/channels if you track them
-    pass
+app = FastAPI(lifespan=lifespan)
+app = gr.mount_gradio_app(app, demo, path="/")
 
+@app.get("/health")
+async def health():
+    return {"status": "ok"}
 
 if __name__ == "__main__":
     import uvicorn

listener.py

@@ -1,11 +1,18 @@
-# listener.py
 import json
+import logging
 from typing import Callable, Awaitable, Dict, Any, List
 
 import aio_pika
 
 Handler = Callable[[Any], Awaitable[None]]  # payload is envelope["data"]
 
+# Configure root logger if not already configured
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+)
+logger = logging.getLogger(__name__)
+
 
 class RabbitListenerBase:
     def __init__(self, base, instance_name: str, handlers: Dict[str, Handler]):
@@ -37,13 +44,18 @@ class RabbitListenerBase:
         async def _on_msg(msg: aio_pika.IncomingMessage):
             async with msg.process():
                 try:
-                    envelope = json.loads(msg.body.decode("utf-8"))
-                    # Expect CloudEvent-ish envelope; we only need the 'data' field
+                    raw_body = msg.body.decode("utf-8", errors="replace")
+                    logger.info(
+                        "Received message for handler '%s': %s",
+                        func_name,
+                        raw_body
+                    )
+
+                    envelope = json.loads(raw_body)
                     data = envelope.get("data", None)
                     if handler:
                         await handler(data)
-                except Exception:
-                    # Avoid requeue storms; add logging if you want
-                    pass
+                except Exception as e:
+                    logger.exception("Error processing message for '%s'", func_name)
 
         return _on_msg

rabbit_base.py

@@ -1,37 +1,44 @@
+# rabbit_base.py
 from typing import Callable, Dict, List, Optional
-import aio_pika
 from urllib.parse import urlsplit, unquote
-from config import settings
 import ssl
+import json
+import logging
+import aio_pika
+
+from config import settings
 
 ExchangeResolver = Callable[[str], str]  # exchangeName -> exchangeType
 
-# rabbit_base.py
-import aio_pika
+# --- logging setup (keep simple; inherit root handlers if already configured) ---
+logger = logging.getLogger(__name__)
+if not logger.handlers:
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
+    )
 
 def _normalize_exchange_type(val: str) -> aio_pika.ExchangeType:
-    # 1) Try attribute by NAME (DIRECT/FANOUT/TOPIC/HEADERS)
     if isinstance(val, str):
         name = val.upper()
         if hasattr(aio_pika.ExchangeType, name):
             return getattr(aio_pika.ExchangeType, name)
-        # 2) Try enum by VALUE ("direct"/"fanout"/"topic"/"headers")
         try:
             return aio_pika.ExchangeType(val.lower())
         except Exception:
             pass
-    # 3) Default
     return aio_pika.ExchangeType.TOPIC
 
 def _parse_amqp_url(url: str) -> dict:
     parts = urlsplit(url)
     return {
+        "scheme": parts.scheme or "amqp",
         "host": parts.hostname or "localhost",
         "port": parts.port or (5671 if parts.scheme == "amqps" else 5672),
         "login": parts.username or "guest",
         "password": parts.password or "guest",
         "virtualhost": unquote(parts.path[1:] or "/"),
-        "ssl": parts.scheme == "amqps",
+        "ssl": (parts.scheme == "amqps"),
     }
 
 
@@ -44,37 +51,90 @@ class RabbitBase:
             lambda _: settings.RABBIT_EXCHANGE_TYPE
         )
 
+    # -------- Status helpers --------
+    def is_connected(self) -> bool:
+        return bool(
+            self._conn and not self._conn.is_closed and
+            self._chan and not self._chan.is_closed
+        )
+
+    async def close(self) -> None:
+        try:
+            if self._chan and not self._chan.is_closed:
+                logger.info("Closing AMQP channel")
+                await self._chan.close()
+        finally:
+            self._chan = None
+        try:
+            if self._conn and not self._conn.is_closed:
+                logger.info("Closing AMQP connection")
+                await self._conn.close()
+        finally:
+            self._conn = None
+        logger.info("AMQP connection closed")
+
+    # -------- Core ops --------
     async def connect(self) -> None:
         if self._conn and not self._conn.is_closed:
-            return
+            if self._chan and not self._chan.is_closed:
+                return
 
         conn_kwargs = _parse_amqp_url(str(settings.AMQP_URL))
 
-        # Build an SSLContext that DISABLES verification
+        # Log connection target (mask password)
+        safe_target = {
+            "scheme": conn_kwargs["scheme"],
+            "host": conn_kwargs["host"],
+            "port": conn_kwargs["port"],
+            "virtualhost": conn_kwargs["virtualhost"],
+            "ssl": conn_kwargs["ssl"],
+            "login": conn_kwargs["login"],
+        }
+        logger.info("AMQP connect -> %s", json.dumps(safe_target))
+
+        # TLS (intentionally disabling verification if requested)
         ssl_ctx = None
         if conn_kwargs.get("ssl"):
             ssl_ctx = ssl.create_default_context()
             ssl_ctx.check_hostname = False
             ssl_ctx.verify_mode = ssl.CERT_NONE
+            logger.warning("AMQP TLS verification is DISABLED (CERT_NONE)")
 
-        # Pass ssl_context explicitly – this is what aio-pika supports
-        self._conn = await aio_pika.connect_robust(
-            **conn_kwargs,
-            ssl_context=ssl_ctx  # <- key bit
-        )
-        self._chan = await self._conn.channel()
-        await self._chan.set_qos(prefetch_count=settings.RABBIT_PREFETCH)
+        try:
+            self._conn = await aio_pika.connect_robust(
+                host=conn_kwargs["host"],
+                port=conn_kwargs["port"],
+                login=conn_kwargs["login"],
+                password=conn_kwargs["password"],
+                virtualhost=conn_kwargs["virtualhost"],
+                ssl=conn_kwargs["ssl"],
+                ssl_context=ssl_ctx,
+            )
+            logger.info("AMQP connection established")
+            self._chan = await self._conn.channel()
+            logger.info("AMQP channel created")
+            await self._chan.set_qos(prefetch_count=settings.RABBIT_PREFETCH)
+            logger.info("AMQP QoS set (prefetch=%s)", settings.RABBIT_PREFETCH)
+        except Exception:
+            logger.exception("AMQP connection/channel setup failed")
+            raise
 
     async def ensure_exchange(self, name: str) -> aio_pika.Exchange:
         await self.connect()
         if name in self._exchanges:
             return self._exchanges[name]
+
         ex_type_str = self._exchange_type_resolver(name)  # e.g. "direct"
         ex_type = _normalize_exchange_type(ex_type_str)
-        ex = await self._chan.declare_exchange(name, ex_type, durable=True)
-       
-        self._exchanges[name] = ex
-        return ex
+
+        try:
+            ex = await self._chan.declare_exchange(name, ex_type, durable=True)
+            self._exchanges[name] = ex
+            logger.info("Exchange declared: name=%s type=%s durable=true", name, ex_type.value)
+            return ex
+        except Exception:
+            logger.exception("Failed declaring exchange: %s (%s)", name, ex_type_str)
+            raise
 
     async def declare_queue_bind(
         self,
@@ -85,16 +145,30 @@ class RabbitBase:
     ):
         await self.connect()
         ex = await self.ensure_exchange(exchange)
+
         args: Dict[str, int] = {}
         if ttl_ms:
             args["x-message-ttl"] = ttl_ms
-        q = await self._chan.declare_queue(
-            queue_name,
-            durable=True,
-            exclusive=False,
-            auto_delete=True,
-            arguments=args,
-        )
-        for rk in routing_keys or [""]:
-            await q.bind(ex, rk)
-        return q
+
+        try:
+            q = await self._chan.declare_queue(
+                queue_name,
+                durable=True,
+                exclusive=False,
+                auto_delete=True,
+                arguments=args,
+            )
+            logger.info(
+                "Queue declared: name=%s durable=true auto_delete=true args=%s",
+                queue_name, args or {}
+            )
+            for rk in routing_keys or [""]:
+                await q.bind(ex, rk)
+                logger.info("Queue bound: queue=%s exchange=%s rk='%s'", queue_name, exchange, rk)
+            return q
+        except Exception:
+            logger.exception(
+                "Failed declare/bind queue: queue=%s exchange=%s rks=%s args=%s",
+                queue_name, exchange, routing_keys, args or {}
+            )
+            raise