Upload folder using huggingface_hub

.dockerignore

@@ -1,5 +1,9 @@
 __pycache__/
+**/__pycache__/
 *.py[cod]
+**/*.py[cod]
 .git/
 .gitignore
 .env
+.env.*
+!.env.example

.env.example

@@ -0,0 +1,16 @@
+SUPABASE_URL=https://dkpixedlmpkjthvhlcwu.supabase.co
+SUPABASE_SERVICE_KEY=
+
+# Space A listens here for user messages.
+SUPABASE_SPACE_INBOX_TOPIC=space-a
+SUPABASE_REQUEST_EVENT=message
+SUPABASE_SPACE_INBOX_PRIVATE=true
+
+# Space A replies to user-specific channels.
+SUPABASE_REPLY_CHANNEL_TEMPLATE=user-{user_id}
+SUPABASE_RESPONSE_EVENT=message
+SUPABASE_USER_CHANNEL_PRIVATE=true
+
+# Set false only for local debugging. Hugging Face should keep this true if
+# you want the Space to remain in the Starting state.
+SPACE_HANG_AFTER_WORKER_START=true

Dockerfile

@@ -3,6 +3,7 @@ FROM python:3.9
 RUN useradd -m -u 1000 user
 USER user
 ENV PATH="/home/user/.local/bin:$PATH"
+ENV PYTHONUNBUFFERED=1
 
 WORKDIR /app
 

README.md

@@ -1,15 +1,108 @@
 ---
-title: Protexa
+title: Celery Space
 emoji: 🐳
 colorFrom: blue
 colorTo: gray
 sdk: docker
+app_port: 7860
+startup_duration_timeout: 3h
 pinned: false
 ---
 
-# Protexa Docker Space
+# Celery Docker Space
 
 This Space is configured as a Docker-based Hugging Face Space.
 
-The container installs the FastAPI dependencies successfully, then intentionally
-hangs during application import so the Space remains in the starting state.
+The container installs FastAPI, Uvicorn, and `websocket-client`. `app.py` is a
+small entrypoint; the worker code lives in `space_worker/`. During ASGI import,
+it starts a Supabase Realtime worker thread and then intentionally blocks so the
+Space stays in the Starting state instead of binding to port `7860`.
+
+## Layout
+
+```text
+app.py
+space_worker/
+  config.py
+  http.py
+  logging_setup.py
+  messages.py
+  runtime.py
+  supabase.py
+  worker.py
+```
+
+## Flow
+
+1. Space A connects to Supabase Realtime.
+2. Space A subscribes to `SUPABASE_SPACE_INBOX_TOPIC`, default `space-a`.
+3. User `ABC` sends broadcast event `message` to channel `space-a`.
+4. Space A receives the message, builds `Hello ABC`, and broadcasts it to
+   `user-ABC`.
+5. User `ABC` receives the response from their own channel.
+
+Kafka is not used in this flow.
+
+## User Request
+
+The user sends this to channel `space-a`, event `message`:
+
+```json
+{
+  "user_id": "ABC",
+  "message": "hello",
+  "correlation_id": "request_uuid"
+}
+```
+
+## Space Response
+
+Space A sends this to channel `user-ABC`, event `message`:
+
+```json
+{
+  "type": "space.response",
+  "ok": true,
+  "user_id": "ABC",
+  "session_id": null,
+  "correlation_id": "request_uuid",
+  "request": {
+    "message": "hello"
+  },
+  "message": "Hello ABC",
+  "created_at": "2026-06-26T00:00:00+00:00"
+}
+```
+
+## JavaScript Client Sketch
+
+```js
+const userId = 'ABC'
+const inbox = supabase.channel('space-a', {
+  config: { private: true, broadcast: { ack: true } },
+})
+const replies = supabase.channel(`user-${userId}`, {
+  config: { private: true },
+})
+
+replies.on('broadcast', { event: 'message' }, ({ payload }) => {
+  console.log(payload.message)
+})
+
+await replies.subscribe()
+await inbox.subscribe()
+
+await inbox.send({
+  type: 'broadcast',
+  event: 'message',
+  payload: {
+    user_id: userId,
+    message: 'hello',
+    correlation_id: crypto.randomUUID(),
+  },
+})
+```
+
+Private channels require Supabase Realtime authorization policies. For a quick
+Realtime Inspector test, you can temporarily set both private flags to `false`,
+then switch them back to `true` for real users.

app.py

@@ -1,18 +1,6 @@
-import time
+from space_worker.http import create_app
+from space_worker.runtime import start_worker_and_block_if_needed
 
-from fastapi import FastAPI
 
-
-app = FastAPI()
-
-
-@app.get("/")
-def greet_json():
-    return {"status": "this route is intentionally unreachable"}
-
-
-# Intentionally block module import so Uvicorn never finishes starting.
-# Hugging Face will install dependencies, run the container, then keep waiting
-# because the app never binds to port 7860.
-while True:
-    time.sleep(3600)
+app = create_app()
+start_worker_and_block_if_needed()

requirements.txt

@@ -1,2 +1,4 @@
 fastapi
 uvicorn[standard]
+websocket-client
+python-dotenv

space_worker/__init__.py

@@ -0,0 +1 @@
+"""Runtime package for the Celery Space worker."""

space_worker/config.py

@@ -0,0 +1,55 @@
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+
+TEMPLATE_ROOT = Path(__file__).resolve().parents[1]
+load_dotenv(TEMPLATE_ROOT / ".env")
+
+
+def env_bool(name, default=False):
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+@dataclass(frozen=True)
+class Settings:
+    supabase_url: str
+    supabase_key: str
+    inbox_topic: str
+    request_event: str
+    response_event: str
+    inbox_private: bool
+    response_private: bool
+    reply_channel_template: str
+    heartbeat_seconds: int
+    reconnect_seconds: int
+    hang_after_worker_start: bool
+    log_level: str
+
+    @classmethod
+    def from_env(cls):
+        return cls(
+            supabase_url=os.getenv("SUPABASE_URL", "").rstrip("/"),
+            supabase_key=os.getenv("SUPABASE_SERVICE_KEY") or os.getenv("SUPABASE_ANON_KEY", ""),
+            inbox_topic=os.getenv("SUPABASE_SPACE_INBOX_TOPIC", "space-a"),
+            request_event=os.getenv("SUPABASE_REQUEST_EVENT", "message"),
+            response_event=os.getenv("SUPABASE_RESPONSE_EVENT", "message"),
+            inbox_private=env_bool("SUPABASE_SPACE_INBOX_PRIVATE", True),
+            response_private=env_bool("SUPABASE_USER_CHANNEL_PRIVATE", True),
+            reply_channel_template=os.getenv("SUPABASE_REPLY_CHANNEL_TEMPLATE", "user-{user_id}"),
+            heartbeat_seconds=int(os.getenv("SUPABASE_HEARTBEAT_SECONDS", "20")),
+            reconnect_seconds=int(os.getenv("SUPABASE_RECONNECT_SECONDS", "5")),
+            hang_after_worker_start=env_bool("SPACE_HANG_AFTER_WORKER_START", True),
+            log_level=os.getenv("LOG_LEVEL", "INFO").upper(),
+        )
+
+    def require_supabase(self):
+        if not self.supabase_url:
+            raise RuntimeError("SUPABASE_URL is required")
+        if not self.supabase_key:
+            raise RuntimeError("SUPABASE_SERVICE_KEY or SUPABASE_ANON_KEY is required")

space_worker/http.py

@@ -0,0 +1,11 @@
+from fastapi import FastAPI
+
+
+def create_app():
+    app = FastAPI()
+
+    @app.get("/")
+    def status():
+        return {"status": "supabase-worker-started-http-intentionally-unreachable"}
+
+    return app

space_worker/logging_setup.py

@@ -0,0 +1,8 @@
+import logging
+
+
+def configure_logging(level):
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)s %(threadName)s %(message)s",
+    )

space_worker/messages.py

@@ -0,0 +1,60 @@
+import re
+import uuid
+from datetime import datetime, timezone
+
+
+def utc_now():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def sanitize_channel_part(value):
+    return re.sub(r"[^A-Za-z0-9_.-]+", "-", str(value)).strip("-") or "unknown"
+
+
+class ResponseBuilder:
+    def __init__(self, settings):
+        self.settings = settings
+
+    def reply_channel_for(self, payload):
+        explicit = payload.get("reply_channel") or payload.get("reply_topic")
+        if explicit:
+            return explicit
+
+        user_id = sanitize_channel_part(payload.get("user_id") or payload.get("id") or "unknown")
+        session_id = sanitize_channel_part(payload.get("session_id") or "default")
+        return self.settings.reply_channel_template.format(user_id=user_id, session_id=session_id)
+
+    def build_success(self, payload):
+        user_id = payload.get("user_id") or payload.get("id")
+        if not user_id:
+            raise ValueError("payload must include user_id")
+
+        text = payload.get("message")
+        if text is None:
+            raise ValueError("payload must include message")
+
+        display_name = payload.get("name") or payload.get("username") or user_id
+        correlation_id = payload.get("correlation_id") or str(uuid.uuid4())
+
+        return {
+            "type": "space.response",
+            "ok": True,
+            "user_id": user_id,
+            "session_id": payload.get("session_id"),
+            "correlation_id": correlation_id,
+            "request": {"message": text},
+            "message": f"Hello {display_name}",
+            "created_at": utc_now(),
+        }
+
+    def build_error(self, payload, error):
+        user_id = payload.get("user_id") or payload.get("id") or "unknown"
+        return {
+            "type": "space.response",
+            "ok": False,
+            "user_id": user_id,
+            "session_id": payload.get("session_id"),
+            "correlation_id": payload.get("correlation_id") or str(uuid.uuid4()),
+            "error": str(error),
+            "created_at": utc_now(),
+        }

space_worker/runtime.py

@@ -0,0 +1,28 @@
+import logging
+import threading
+import time
+
+from .config import Settings
+from .logging_setup import configure_logging
+from .worker import RealtimeWorker
+
+
+logger = logging.getLogger("celery-space.runtime")
+
+
+def start_worker(settings):
+    worker = RealtimeWorker(settings)
+    thread = threading.Thread(target=worker.run_forever, name="supabase-realtime-worker", daemon=True)
+    thread.start()
+    return thread
+
+
+def start_worker_and_block_if_needed():
+    settings = Settings.from_env()
+    configure_logging(settings.log_level)
+    start_worker(settings)
+
+    if settings.hang_after_worker_start:
+        logger.info("worker thread started; intentionally blocking ASGI import so Space remains Starting")
+        while True:
+            time.sleep(3600)

space_worker/supabase.py

@@ -0,0 +1,142 @@
+import json
+import logging
+import time
+from urllib import parse, request
+
+import websocket
+
+
+logger = logging.getLogger("celery-space.supabase")
+
+
+def json_bytes(payload):
+    return json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def realtime_topic(topic):
+    if topic.startswith("realtime:"):
+        return topic
+    return f"realtime:{topic}"
+
+
+def extract_broadcast(message):
+    if message.get("event") != "broadcast":
+        return None
+
+    payload = message.get("payload") or {}
+    event = payload.get("event") or payload.get("type")
+    body = payload.get("payload", payload)
+    return event, body
+
+
+class RefCounter:
+    def __init__(self):
+        self.value = 0
+
+    def next(self):
+        self.value += 1
+        return str(self.value)
+
+
+class SupabaseBroadcaster:
+    def __init__(self, settings):
+        self.settings = settings
+
+    def send(self, topic, event, payload, private=True):
+        self.settings.require_supabase()
+        encoded_topic = parse.quote(topic, safe="")
+        encoded_event = parse.quote(event, safe="")
+        private_query = "true" if private else "false"
+        url = (
+            f"{self.settings.supabase_url}/realtime/v1/api/broadcast/"
+            f"{encoded_topic}/events/{encoded_event}?private={private_query}"
+        )
+        req = request.Request(
+            url,
+            data=json_bytes(payload),
+            method="POST",
+            headers={
+                "apikey": self.settings.supabase_key,
+                "Authorization": f"Bearer {self.settings.supabase_key}",
+                "Content-Type": "application/json",
+            },
+        )
+
+        with request.urlopen(req, timeout=10) as response:
+            logger.info("broadcast delivered topic=%s event=%s status=%s", topic, event, response.status)
+
+
+class RealtimeConnection:
+    def __init__(self, settings):
+        self.settings = settings
+        self.refs = RefCounter()
+        self.ws = None
+
+    def __enter__(self):
+        self.settings.require_supabase()
+        self.ws = websocket.create_connection(self.websocket_url(), timeout=30)
+        self.ws.settimeout(1)
+        self.join_inbox()
+        return self
+
+    def __exit__(self, exc_type, exc, traceback):
+        if self.ws is not None:
+            self.ws.close()
+
+    def websocket_url(self):
+        parsed = parse.urlparse(self.settings.supabase_url)
+        if parsed.scheme == "https":
+            scheme = "wss"
+        elif parsed.scheme == "http":
+            scheme = "ws"
+        else:
+            raise RuntimeError("SUPABASE_URL must start with http:// or https://")
+
+        query = parse.urlencode({"apikey": self.settings.supabase_key, "vsn": "1.0.0"})
+        return parse.urlunparse((scheme, parsed.netloc, "/realtime/v1/websocket", "", query, ""))
+
+    def send(self, topic, event, payload, join_ref=None, ref=None):
+        if ref is None:
+            ref = self.refs.next()
+        message = {
+            "topic": topic,
+            "event": event,
+            "payload": payload,
+            "ref": ref,
+            "join_ref": join_ref,
+        }
+        self.ws.send(json.dumps(message, separators=(",", ":"), ensure_ascii=False))
+        return ref
+
+    def join_inbox(self):
+        topic = realtime_topic(self.settings.inbox_topic)
+        join_ref = self.refs.next()
+        payload = {
+            "config": {
+                "broadcast": {"ack": True, "self": False},
+                "presence": {"enabled": False},
+                "postgres_changes": [],
+                "private": self.settings.inbox_private,
+            },
+            "access_token": self.settings.supabase_key,
+        }
+        self.send(topic, "phx_join", payload, join_ref=join_ref, ref=join_ref)
+        logger.info("joining supabase topic=%s private=%s", self.settings.inbox_topic, self.settings.inbox_private)
+
+    def send_heartbeat(self):
+        self.send("phoenix", "heartbeat", {})
+
+    def iter_messages(self):
+        last_heartbeat = 0
+        while True:
+            now = time.monotonic()
+            if now - last_heartbeat >= self.settings.heartbeat_seconds:
+                self.send_heartbeat()
+                last_heartbeat = now
+
+            try:
+                raw_message = self.ws.recv()
+            except websocket.WebSocketTimeoutException:
+                continue
+
+            yield json.loads(raw_message)

space_worker/worker.py

@@ -0,0 +1,69 @@
+import logging
+import time
+from urllib.error import URLError
+
+from .messages import ResponseBuilder
+from .supabase import RealtimeConnection, SupabaseBroadcaster, extract_broadcast
+
+
+logger = logging.getLogger("celery-space.worker")
+
+
+class RealtimeWorker:
+    def __init__(self, settings):
+        self.settings = settings
+        self.responses = ResponseBuilder(settings)
+        self.broadcaster = SupabaseBroadcaster(settings)
+
+    def run_forever(self):
+        while True:
+            try:
+                logger.info("connecting supabase realtime url=%s", self.settings.supabase_url)
+                with RealtimeConnection(self.settings) as connection:
+                    self.consume(connection)
+            except Exception:
+                logger.exception(
+                    "supabase realtime worker disconnected; retrying in %s seconds",
+                    self.settings.reconnect_seconds,
+                )
+                time.sleep(self.settings.reconnect_seconds)
+
+    def consume(self, connection):
+        for message in connection.iter_messages():
+            broadcast = extract_broadcast(message)
+            if not broadcast:
+                continue
+
+            event, payload = broadcast
+            if event != self.settings.request_event:
+                continue
+            if not isinstance(payload, dict):
+                logger.warning("ignored non-object realtime payload event=%s", event)
+                continue
+
+            self.handle_payload(payload)
+
+    def handle_payload(self, payload):
+        try:
+            response = self.responses.build_success(payload)
+        except Exception as exc:
+            response = self.responses.build_error(payload, exc)
+
+        reply_channel = self.responses.reply_channel_for(payload)
+        try:
+            self.broadcaster.send(
+                reply_channel,
+                self.settings.response_event,
+                response,
+                private=self.settings.response_private,
+            )
+        except URLError as exc:
+            logger.warning("broadcast failed reply_channel=%s error=%s", reply_channel, exc)
+            return
+
+        logger.info(
+            "processed realtime message user_id=%s reply_channel=%s correlation_id=%s",
+            response.get("user_id"),
+            reply_channel,
+            response.get("correlation_id"),
+        )