Cluster relianility

aisbf/app/middleware.py

@@ -254,6 +254,12 @@ def make_auth_middleware(get_server_config, get_config, get_db, url_for_fn):
                         not (expires_at and int(time.time()) > expires_at)):
                     return await call_next(request)
 
+            if request.url.path.startswith("/api/market"):
+                expires_at = request.session.get('expires_at')
+                if (request.session.get('logged_in') and
+                        not (expires_at and int(time.time()) > expires_at)):
+                    return await call_next(request)
+
             if request.method == "GET" and request.url.path in ["/api/models", "/api/v1/models"]:
                 return await call_next(request)
 

aisbf/coderai_broker.py

 """
-CoderAI broker and session registry for NAT-friendly outbound connections.
+CoderAI broker – cluster-aware WebSocket session manager.
+
+Design contract:
+  - A CoderAI client connects via WSS to whichever AISBF instance the load
+    balancer picks.  That instance owns the WebSocket for the lifetime of the
+    connection.
+  - Requests from ANY instance in the cluster are enqueued in the shared cache
+    under request_queue:{session_id}.  The owning instance drains that queue
+    and forwards each envelope to the CoderAI client.
+  - Replies are stored in the shared cache under reply:{request_id}.  The
+    requesting instance polls until the reply appears (or the session dies).
+  - Streaming chunks follow the same reply-key path when the request originates
+    on a remote instance; when it originates locally the asyncio.Queue fast-path
+    is used.
+  - Session metadata (including broker_node_id and connection_state) is kept in
+    the shared cache with a TTL.  The owning instance refreshes the TTL via a
+    background heartbeat task.  If the instance crashes without a clean
+    disconnect, the TTL expires and the session disappears from the cache –
+    every other instance will then see the provider as offline within
+    SESSION_TTL_SECONDS seconds.
+  - node_id is a UUID generated fresh at process start and is NEVER shared
+    through the cache.  Each instance has its own unique node_id.
 """
 
 from __future__ import annotations
@@ -8,11 +29,9 @@ import asyncio
 from collections import deque
 import json
 import logging
-import os
 import time
 import uuid
 from dataclasses import dataclass, field
-from pathlib import Path
 from typing import Any, Dict, Optional
 
 from fastapi import WebSocket
@@ -22,11 +41,19 @@ from .cache import get_cache_manager
 
 logger = logging.getLogger(__name__)
 
-SESSION_TTL_SECONDS = 120
+# Session metadata TTL in the shared cache.  The owning instance refreshes
+# this every SESSION_HEARTBEAT_SECONDS, so the effective death-detection
+# window is SESSION_TTL_SECONDS after the last heartbeat.
+SESSION_TTL_SECONDS = 90
+SESSION_HEARTBEAT_SECONDS = 30
 REQUEST_TTL_SECONDS = 300
 REPLY_TTL_SECONDS = 300
+# How long broker_blocking_pop blocks waiting for a new queue item before
+# looping.  Keep at 1 s so the drain loop stays responsive to cancellation.
 HEARTBEAT_POLL_SECONDS = 1
 PERFORMANCE_WINDOW_SIZE = 100
+# How often (seconds) the slow-path polling checks session liveness.
+LIVENESS_CHECK_INTERVAL = 5.0
 
 
 class BrokerPlaceholderWebSocket:
@@ -60,14 +87,18 @@ class CoderAISession:
 class CoderAIBroker:
     def __init__(self):
         self._lock = asyncio.Lock()
+        # In-memory local sessions (WebSocket is on this instance).
         self._sessions_by_key: Dict[str, CoderAISession] = {}
         self._sessions_by_id: Dict[str, CoderAISession] = {}
+        # Pending requests whose Future/Queue lives on this instance.
         self._pending: Dict[str, PendingCoderAIRequest] = {}
-        self._state_path = Path.home() / '.aisbf' / 'coderai_broker_sessions.json'
-        self._state_path.parent.mkdir(parents=True, exist_ok=True)
+        # Per-session background heartbeat tasks (only for locally owned sessions).
+        self._heartbeat_tasks: Dict[str, asyncio.Task] = {}
         self._cache = get_cache_manager()
-        self._node_id = self._cache.broker_node_id()
-        self._load_persisted_sessions()
+        # Unique per process restart – never written to the shared cache.
+        self._node_id = uuid.uuid4().hex
+
+    # ------------------------------------------------------------------ keys
 
     @staticmethod
     def _session_key(provider_id: str, client_id: str) -> str:
@@ -89,6 +120,8 @@ class CoderAIBroker:
     def _session_index_key() -> str:
         return "session_index"
 
+    # --------------------------------------------------------- serialisation
+
     def _serialize_session(self, session: CoderAISession) -> Dict[str, Any]:
         return {
             "session_id": session.session_id,
@@ -143,16 +176,10 @@ class CoderAIBroker:
             result["gpus"] = normalized_gpus
             result["gpu_count"] = len(normalized_gpus)
             result["total_vram_mb"] = total_vram_mb
-            # Only overwrite the top-level available_vram_mb when the GPU loop
-            # actually produced data; otherwise keep the value that arrived at
-            # the top level of the metadata (e.g. coderai sends it there but
-            # does not repeat it inside each GPU dict).
             if available_vram_mb > 0:
                 result["available_vram_mb"] = available_vram_mb
             elif result.get("available_vram_mb") is None:
                 result["available_vram_mb"] = 0.0
-            # Propagate top-level free VRAM into the single GPU so the UI can
-            # read it from the per-GPU entry too.
             if len(normalized_gpus) == 1 and normalized_gpus[0].get("available_vram_mb") is None:
                 top_free = result.get("available_vram_mb")
                 if top_free is not None:
@@ -168,17 +195,11 @@ class CoderAIBroker:
         if latency_ms is None:
             latency_ms = max((time.time() - started_at) * 1000.0, 0.0)
         usage = payload.get("usage") or {}
-        prompt_tokens = payload.get("prompt_tokens")
-        completion_tokens = payload.get("completion_tokens")
-        total_tokens = payload.get("total_tokens")
-        if prompt_tokens is None:
-            prompt_tokens = usage.get("prompt_tokens")
-        if completion_tokens is None:
-            completion_tokens = usage.get("completion_tokens")
-        if total_tokens is None:
-            total_tokens = usage.get("total_tokens")
+        prompt_tokens = payload.get("prompt_tokens") or usage.get("prompt_tokens")
+        completion_tokens = payload.get("completion_tokens") or usage.get("completion_tokens")
+        total_tokens = payload.get("total_tokens") or usage.get("total_tokens")
         if total_tokens is None:
-            values = [value for value in (prompt_tokens, completion_tokens) if isinstance(value, (int, float))]
+            values = [v for v in (prompt_tokens, completion_tokens) if isinstance(v, (int, float))]
             total_tokens = sum(values) if values else None
         tokens_per_second = payload.get("tokens_per_second") or payload.get("tok_per_s") or payload.get("throughput_tps")
         if tokens_per_second is None and isinstance(total_tokens, (int, float)) and latency_ms:
@@ -211,10 +232,10 @@ class CoderAIBroker:
                 "last_tokens_per_second": None,
                 "last_total_tokens": None,
             }
-        latencies = [sample["latency_ms"] for sample in samples if isinstance(sample.get("latency_ms"), (int, float))]
-        tps_values = [sample["tokens_per_second"] for sample in samples if isinstance(sample.get("tokens_per_second"), (int, float))]
-        total_tokens = [sample["total_tokens"] for sample in samples if isinstance(sample.get("total_tokens"), (int, float))]
-        success_count = sum(1 for sample in samples if sample.get("success"))
+        latencies = [s["latency_ms"] for s in samples if isinstance(s.get("latency_ms"), (int, float))]
+        tps_values = [s["tokens_per_second"] for s in samples if isinstance(s.get("tokens_per_second"), (int, float))]
+        total_tokens = [s["total_tokens"] for s in samples if isinstance(s.get("total_tokens"), (int, float))]
+        success_count = sum(1 for s in samples if s.get("success"))
         last = samples[-1]
         return {
             "window_size": PERFORMANCE_WINDOW_SIZE,
@@ -229,134 +250,86 @@ class CoderAIBroker:
             "last_total_tokens": last.get("total_tokens"),
         }
 
-    async def _record_request_metric(self, request_id: str, message: Dict[str, Any]) -> None:
-        async with self._lock:
-            pending = self._pending.get(request_id)
-            if not pending:
-                return
-            session_id = (pending.request_snapshot or {}).get("session_id")
-            session = self._sessions_by_id.get(session_id) if session_id else None
-            if not session:
-                return
-            session.recent_requests.append(self._estimate_performance_sample(message, pending.created_at))
-            self._store_session_cache(session)
-            self._persist_sessions_locked()
-
-    def _persist_sessions_locked(self) -> None:
-        payload = {
-            "sessions": [
-                self._serialize_session(session)
-                for session in self._sessions_by_id.values()
-            ],
-            "updated_at": time.time(),
-        }
-        temp_path = self._state_path.with_suffix(self._state_path.suffix + '.tmp')
-        with open(temp_path, 'w') as f:
-            json.dump(payload, f, indent=2)
-            f.flush()
-            os.fsync(f.fileno())
-        os.replace(temp_path, self._state_path)
-
-    def _quarantine_invalid_state_file(self) -> Optional[Path]:
-        if not self._state_path.exists():
-            return None
-        timestamp = int(time.time())
-        quarantine_path = self._state_path.with_name(f"{self._state_path.name}.corrupt.{timestamp}")
-        counter = 1
-        while quarantine_path.exists():
-            quarantine_path = self._state_path.with_name(f"{self._state_path.name}.corrupt.{timestamp}.{counter}")
-            counter += 1
-        try:
-            self._state_path.replace(quarantine_path)
-            return quarantine_path
-        except Exception:
-            logger.exception("Failed to quarantine invalid CoderAI broker session state file")
-            return None
-
-    def _load_persisted_sessions(self) -> None:
-        if not self._state_path.exists():
-            return
-        try:
-            with open(self._state_path) as f:
-                payload = json.load(f)
-        except Exception as e:
-            quarantine_path = self._quarantine_invalid_state_file()
-            if quarantine_path is not None:
-                logger.warning(
-                    "Failed to load persisted CoderAI broker sessions: %s. Moved invalid state file to %s",
-                    e,
-                    quarantine_path,
-                )
-            else:
-                logger.warning(f"Failed to load persisted CoderAI broker sessions: {e}")
-            return
-
-        for raw_session in payload.get("sessions") or []:
-            if not isinstance(raw_session, dict):
-                continue
-            provider_id = raw_session.get("provider_id")
-            client_id = raw_session.get("client_id")
-            session_id = raw_session.get("session_id") or f"coderai_{uuid.uuid4().hex}"
-            if not provider_id or not client_id:
-                continue
-            session = CoderAISession(
-                session_id=session_id,
-                provider_id=provider_id,
-                client_id=client_id,
-                websocket=BrokerPlaceholderWebSocket(),
-                connected_at=float(raw_session.get("connected_at") or time.time()),
-                metadata=dict(raw_session.get("metadata") or {}),
-                capabilities=dict(raw_session.get("capabilities") or {}),
-                last_seen=float(raw_session.get("last_seen") or time.time()),
-                closed=bool(raw_session.get("closed", True)),
-            )
-            perf = raw_session.get("performance") or {}
-            if isinstance(perf, dict) and int(perf.get("sample_count") or 0) > 0:
-                sample_count = min(int(perf.get("sample_count") or 0), PERFORMANCE_WINDOW_SIZE)
-                for _ in range(sample_count):
-                    session.recent_requests.append({
-                        "latency_ms": float(perf.get("avg_latency_ms") or 0.0),
-                        "tokens_per_second": float(perf.get("avg_tokens_per_second")) if isinstance(perf.get("avg_tokens_per_second"), (int, float)) else None,
-                        "total_tokens": float(perf.get("avg_total_tokens")) if isinstance(perf.get("avg_total_tokens"), (int, float)) else None,
-                        "success": True,
-                        "recorded_at": session.last_seen,
-                    })
-            session.metadata.setdefault("persisted", True)
-            session.metadata.setdefault("connection_state", "disconnected")
-            key = self._session_key(provider_id, client_id)
-            self._sessions_by_key[key] = session
-            self._sessions_by_id[session_id] = session
+    # -------------------------------------------------- shared cache helpers
 
     def _store_session_cache(self, session: CoderAISession) -> None:
         payload = self._serialize_session(session)
         payload["metadata"] = dict(payload.get("metadata") or {})
-        payload["metadata"]["broker_node_id"] = session.metadata.get("broker_node_id") or self._node_id
-        session.metadata.setdefault("broker_node_id", self._node_id)
+        payload["metadata"]["broker_node_id"] = self._node_id
         payload["metadata"]["connection_state"] = "connected" if not session.closed else "disconnected"
-        self._cache.broker_set(self._session_meta_key(session.provider_id, session.client_id), payload, ttl=SESSION_TTL_SECONDS)
-
+        meta_key = self._session_meta_key(session.provider_id, session.client_id)
+        self._cache.broker_set(meta_key, payload, ttl=SESSION_TTL_SECONDS)
+        # Keep a global index of all live session meta keys so list_sessions
+        # can enumerate them without a Redis SCAN.
         index = self._cache.broker_get(self._session_index_key()) or []
         if not isinstance(index, list):
             index = []
-        key = self._session_meta_key(session.provider_id, session.client_id)
-        if key not in index:
-            index.append(key)
+        if meta_key not in index:
+            index.append(meta_key)
             self._cache.broker_set(self._session_index_key(), index, ttl=SESSION_TTL_SECONDS * 10)
 
-    def _delete_session_cache(self, provider_id: str, client_id: str) -> None:
-        key = self._session_meta_key(provider_id, client_id)
-        self._cache.broker_delete(key)
+    def _mark_session_offline_cache(self, provider_id: str, client_id: str) -> None:
+        """Write a closed=True tombstone so remote nodes see the disconnect immediately."""
+        meta_key = self._session_meta_key(provider_id, client_id)
+        existing = self._cache.broker_get(meta_key) or {}
+        existing["closed"] = True
+        existing["metadata"] = dict(existing.get("metadata") or {})
+        existing["metadata"]["connection_state"] = "disconnected"
+        existing["last_seen"] = time.time()
+        # Keep tombstone visible for a short window so pollers detect it fast.
+        self._cache.broker_set(meta_key, existing, ttl=SESSION_TTL_SECONDS)
+
+    def _remove_session_from_index(self, provider_id: str, client_id: str) -> None:
+        meta_key = self._session_meta_key(provider_id, client_id)
         index = self._cache.broker_get(self._session_index_key()) or []
-        if isinstance(index, list) and key in index:
-            index = [item for item in index if item != key]
+        if isinstance(index, list) and meta_key in index:
+            index = [k for k in index if k != meta_key]
             self._cache.broker_set(self._session_index_key(), index, ttl=SESSION_TTL_SECONDS * 10)
 
-    async def register(self, websocket: WebSocket, provider_id: str, client_id: str, metadata: Optional[Dict[str, Any]] = None, capabilities: Optional[Dict[str, Any]] = None, session_id: Optional[str] = None) -> CoderAISession:
+    # -------------------------------------------------- heartbeat management
+
+    def _start_heartbeat(self, session_id: str) -> None:
+        task = asyncio.create_task(self._heartbeat_loop(session_id))
+        self._heartbeat_tasks[session_id] = task
+
+    def _stop_heartbeat(self, session_id: str) -> None:
+        task = self._heartbeat_tasks.pop(session_id, None)
+        if task:
+            task.cancel()
+
+    async def _heartbeat_loop(self, session_id: str) -> None:
+        """Refresh the shared-cache TTL for a locally owned session."""
+        try:
+            while True:
+                await asyncio.sleep(SESSION_HEARTBEAT_SECONDS)
+                async with self._lock:
+                    session = self._sessions_by_id.get(session_id)
+                    if not session or session.closed:
+                        return
+                    session.last_seen = time.time()
+                    self._store_session_cache(session)
+        except asyncio.CancelledError:
+            pass
+
+    # -------------------------------------------------- session lifecycle
+
+    async def register(
+        self,
+        websocket: WebSocket,
+        provider_id: str,
+        client_id: str,
+        metadata: Optional[Dict[str, Any]] = None,
+        capabilities: Optional[Dict[str, Any]] = None,
+        session_id: Optional[str] = None,
+    ) -> CoderAISession:
         async with self._lock:
             key = self._session_key(provider_id, client_id)
             existing = self._sessions_by_key.get(key)
             if existing:
                 existing.closed = True
+                old_task = self._heartbeat_tasks.pop(existing.session_id, None)
+                if old_task:
+                    old_task.cancel()
             resolved_session_id = session_id or f"coderai_{uuid.uuid4().hex}"
             session = CoderAISession(
                 session_id=resolved_session_id,
@@ -367,16 +340,23 @@ class CoderAIBroker:
                 capabilities=dict(capabilities or {}),
             )
             session.closed = False
-            session.metadata.setdefault("connection_state", "connected")
-            session.metadata["persisted"] = True
+            session.metadata["connection_state"] = "connected"
+            session.metadata["broker_node_id"] = self._node_id
             self._sessions_by_key[key] = session
             self._sessions_by_id[resolved_session_id] = session
             self._store_session_cache(session)
-            self._persist_sessions_locked()
-            logger.info(f"CoderAI broker registered session provider={provider_id} client={client_id} session_id={resolved_session_id}")
+
+        # Start heartbeat outside the lock so the task is independent.
+        self._start_heartbeat(resolved_session_id)
+        logger.info(
+            "CoderAI broker registered provider=%s client=%s session_id=%s node=%s",
+            provider_id, client_id, resolved_session_id, self._node_id,
+        )
         return session
 
     async def unregister(self, session_id: str) -> None:
+        # Cancel heartbeat first – no more TTL refreshes after this point.
+        self._stop_heartbeat(session_id)
         async with self._lock:
             session = self._sessions_by_id.get(session_id)
             if not session:
@@ -387,11 +367,19 @@ class CoderAIBroker:
             session.metadata["connection_state"] = "disconnected"
             session.last_seen = time.time()
             self._sessions_by_key[key] = session
-            self._store_session_cache(session)
-            self._persist_sessions_locked()
-            logger.info(f"CoderAI broker unregistered session provider={session.provider_id} client={session.client_id} session_id={session.session_id}")
+            # Write closed tombstone so all cluster nodes see offline immediately.
+            self._mark_session_offline_cache(session.provider_id, session.client_id)
+            logger.info(
+                "CoderAI broker unregistered provider=%s client=%s session_id=%s",
+                session.provider_id, session.client_id, session.session_id,
+            )
 
-    async def touch(self, session_id: str, metadata: Optional[Dict[str, Any]] = None, capabilities: Optional[Dict[str, Any]] = None) -> Optional[CoderAISession]:
+    async def touch(
+        self,
+        session_id: str,
+        metadata: Optional[Dict[str, Any]] = None,
+        capabilities: Optional[Dict[str, Any]] = None,
+    ) -> Optional[CoderAISession]:
         async with self._lock:
             session = self._sessions_by_id.get(session_id)
             if not session:
@@ -402,54 +390,62 @@ class CoderAIBroker:
             if capabilities:
                 session.capabilities = dict(capabilities)
             self._store_session_cache(session)
-            self._persist_sessions_locked()
             return session
 
-    async def get_session(self, provider_id: str, client_id: Optional[str] = None) -> Optional[CoderAISession]:
+    # ------------------------------------------------------- session queries
+
+    async def get_session(
+        self, provider_id: str, client_id: Optional[str] = None
+    ) -> Optional[CoderAISession]:
+        """Return the local CoderAISession if this node owns the WebSocket, else None."""
         async with self._lock:
             if client_id:
                 session = self._sessions_by_key.get(self._session_key(provider_id, client_id))
                 if session and not session.closed:
                     return session
             else:
-                candidates = [session for session in self._sessions_by_key.values() if session.provider_id == provider_id and not session.closed]
+                candidates = [
+                    s for s in self._sessions_by_key.values()
+                    if s.provider_id == provider_id and not s.closed
+                ]
                 if candidates:
-                    candidates.sort(key=lambda session: session.last_seen, reverse=True)
+                    candidates.sort(key=lambda s: s.last_seen, reverse=True)
                     return candidates[0]
 
-        # Session exists on another broker node: we have no local WebSocket handle,
-        # so we cannot deliver requests directly. Return None; send_request will
-        # route through the Redis queue to the owning node.
+        # Not local – check the shared cache to report remote status to callers.
         snapshot = await self.get_session_snapshot(provider_id, client_id)
         if snapshot and snapshot.get("connected"):
-            logger.debug(f"CoderAI session for provider={provider_id} is connected on a remote broker node")
+            node = (snapshot.get("metadata") or {}).get("broker_node_id", "unknown")
+            logger.debug(
+                "CoderAI session provider=%s is connected on remote node %s",
+                provider_id, node,
+            )
         return None
 
     async def list_sessions(self) -> list[Dict[str, Any]]:
-        cache_sessions = []
+        """Return all known sessions across the cluster (from shared cache + local)."""
+        merged: Dict[str, Dict[str, Any]] = {}
+
+        # Pull all sessions written by any cluster node via the shared index.
         index = self._cache.broker_get(self._session_index_key()) or []
         if isinstance(index, list):
-            for key in index:
-                meta = self._cache.broker_get(key.replace(self._cache.broker_key(''), '')) if False else None
-                stored = self._cache.get(key) if key.startswith(self._cache.config.get('redis_key_prefix', 'aisbf:')) else self._cache.broker_get(key)
-                if stored:
-                    cache_sessions.append(stored)
-
-        merged: Dict[str, Dict[str, Any]] = {}
-        for item in cache_sessions:
-            if not isinstance(item, dict):
+            for meta_key in index:
+                stored = self._cache.broker_get(meta_key)
+                if not stored or not isinstance(stored, dict):
                     continue
-            merged[item.get("session_id") or f"missing:{uuid.uuid4().hex}"] = {
-                "session_id": item.get("session_id"),
-                "provider_id": item.get("provider_id"),
-                "client_id": item.get("client_id"),
-                "connected_at": item.get("connected_at"),
-                "last_seen": item.get("last_seen"),
-                "metadata": item.get("metadata") or {},
-                "capabilities": item.get("capabilities") or {},
-                "connected": not bool(item.get("closed")),
+                sid = stored.get("session_id") or meta_key
+                merged[sid] = {
+                    "session_id": stored.get("session_id"),
+                    "provider_id": stored.get("provider_id"),
+                    "client_id": stored.get("client_id"),
+                    "connected_at": stored.get("connected_at"),
+                    "last_seen": stored.get("last_seen"),
+                    "metadata": stored.get("metadata") or {},
+                    "capabilities": stored.get("capabilities") or {},
+                    "connected": not bool(stored.get("closed")),
                 }
 
+        # Local sessions are always authoritative – they have the live WebSocket.
         async with self._lock:
             for session in self._sessions_by_id.values():
                 merged[session.session_id] = {
@@ -465,53 +461,81 @@ class CoderAIBroker:
 
         return list(merged.values())
 
-    async def get_session_snapshot(self, provider_id: str, client_id: Optional[str] = None) -> Optional[Dict[str, Any]]:
+    async def get_session_snapshot(
+        self, provider_id: str, client_id: Optional[str] = None
+    ) -> Optional[Dict[str, Any]]:
         if client_id:
+            # Shared cache is the first source of truth for status queries.
             payload = self._cache.broker_get(self._session_meta_key(provider_id, client_id))
-            if payload:
+            if payload and isinstance(payload, dict):
                 payload["connected"] = not bool(payload.get("closed"))
                 return payload
+            # Fall back to local in-memory (e.g. cache miss on first heartbeat).
             async with self._lock:
-                local_session = self._sessions_by_key.get(self._session_key(provider_id, client_id))
-                if local_session:
+                local = self._sessions_by_key.get(self._session_key(provider_id, client_id))
+                if local:
                     return {
-                        "session_id": local_session.session_id,
-                        "provider_id": local_session.provider_id,
-                        "client_id": local_session.client_id,
-                        "connected_at": local_session.connected_at,
-                        "last_seen": local_session.last_seen,
-                        "closed": local_session.closed,
-                        "metadata": dict(local_session.metadata),
-                        "capabilities": dict(local_session.capabilities),
-                        "connected": not local_session.closed,
+                        "session_id": local.session_id,
+                        "provider_id": local.provider_id,
+                        "client_id": local.client_id,
+                        "connected_at": local.connected_at,
+                        "last_seen": local.last_seen,
+                        "closed": local.closed,
+                        "metadata": dict(local.metadata),
+                        "capabilities": dict(local.capabilities),
+                        "connected": not local.closed,
                     }
+            return None
 
-        candidates = []
-        for session in await self.list_sessions():
-            if session.get("provider_id") == provider_id:
-                candidates.append(session)
+        candidates = [s for s in await self.list_sessions() if s.get("provider_id") == provider_id]
         if not candidates:
             return None
-        candidates.sort(key=lambda item: item.get("last_seen") or 0, reverse=True)
+        candidates.sort(key=lambda s: s.get("last_seen") or 0, reverse=True)
         return candidates[0]
 
-    async def send_request(self, provider_id: str, op: str, payload: Dict[str, Any], timeout: float = 300.0, client_id: Optional[str] = None, owner_user_id: Optional[int] = None, extra: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    # -------------------------------------------------- request / reply flow
+
+    async def _record_request_metric(self, request_id: str, message: Dict[str, Any]) -> None:
+        async with self._lock:
+            pending = self._pending.get(request_id)
+            if not pending:
+                return
+            session_id = (pending.request_snapshot or {}).get("session_id")
+            session = self._sessions_by_id.get(session_id) if session_id else None
+            if not session:
+                return
+            session.recent_requests.append(self._estimate_performance_sample(message, pending.created_at))
+            self._store_session_cache(session)
+
+    async def send_request(
+        self,
+        provider_id: str,
+        op: str,
+        payload: Dict[str, Any],
+        timeout: float = 300.0,
+        client_id: Optional[str] = None,
+        owner_user_id: Optional[int] = None,
+        extra: Optional[Dict[str, Any]] = None,
+    ) -> Dict[str, Any]:
         snapshot = await self.get_session_snapshot(provider_id, client_id)
+        # Allow a fallback where the client_id itself is the provider_id.
         if (not snapshot or not snapshot.get("connected")) and client_id:
-            fallback_snapshot = await self.get_session_snapshot(client_id, client_id)
-            fallback_metadata = (fallback_snapshot or {}).get("metadata") or {}
-            if fallback_snapshot and fallback_snapshot.get("connected") and fallback_metadata.get("owner_user_id") == owner_user_id:
-                snapshot = fallback_snapshot
+            fallback = await self.get_session_snapshot(client_id, client_id)
+            fb_meta = (fallback or {}).get("metadata") or {}
+            if fallback and fallback.get("connected") and fb_meta.get("owner_user_id") == owner_user_id:
+                snapshot = fallback
                 provider_id = snapshot.get("provider_id") or provider_id
         if not snapshot or not snapshot.get("connected"):
             raise RuntimeError(f"No active CoderAI broker session for provider '{provider_id}'")
-        if owner_user_id != ((snapshot.get('metadata') or {}).get('owner_user_id')):
-            raise RuntimeError(f"No active CoderAI broker session for provider '{provider_id}' owned by the current principal")
+        if owner_user_id != ((snapshot.get("metadata") or {}).get("owner_user_id")):
+            raise RuntimeError(
+                f"No active CoderAI broker session for provider '{provider_id}' owned by the current principal"
+            )
 
         request_id = f"broker_{uuid.uuid4().hex}"
         loop = asyncio.get_running_loop()
         future = loop.create_future()
-        stream_queue = asyncio.Queue() if payload.get("stream") else None
+        stream_queue: asyncio.Queue | None = asyncio.Queue() if payload.get("stream") else None
         envelope = {
             "v": 1,
             "op": op,
@@ -538,34 +562,44 @@ class CoderAIBroker:
                 },
             )
 
-        local_session = None
-        target_node_id = None
+        # Fast path: the WebSocket lives on this node.
+        # Slow path: enqueue in the shared cache; the owning node drains it.
+        target_node_id = (snapshot.get("metadata") or {}).get("broker_node_id")
         async with self._lock:
             local_session = self._sessions_by_id.get(snapshot["session_id"])
-            local_node_id = local_session.metadata.get("broker_node_id") if local_session else None
-            target_node_id = ((snapshot.get("metadata") or {}).get("broker_node_id") if snapshot else None)
-            if target_node_id is None:
-                target_node_id = local_node_id
-
         use_local_fast_path = bool(
             local_session
             and not local_session.closed
-            and (target_node_id is None or target_node_id == self._node_id)
+            and target_node_id == self._node_id
         )
 
+        resolved_client_id = snapshot.get("client_id") or client_id
+
         try:
             if use_local_fast_path:
                 await local_session.websocket.send_text(json.dumps(envelope))
-                logger.debug(f"CoderAI broker fast-pathed op={op} provider={provider_id} client={snapshot.get('client_id')} request_id={request_id}")
+                logger.debug(
+                    "CoderAI broker fast-path op=%s provider=%s request_id=%s",
+                    op, provider_id, request_id,
+                )
             else:
-                self._cache.broker_push(self._request_queue_key(snapshot["session_id"]), envelope, ttl=REQUEST_TTL_SECONDS)
-                logger.debug(f"CoderAI broker enqueued op={op} provider={provider_id} client={snapshot.get('client_id')} request_id={request_id}")
+                self._cache.broker_push(
+                    self._request_queue_key(snapshot["session_id"]),
+                    envelope,
+                    ttl=REQUEST_TTL_SECONDS,
+                )
+                logger.debug(
+                    "CoderAI broker queued op=%s provider=%s target_node=%s request_id=%s",
+                    op, provider_id, target_node_id, request_id,
+                )
 
             async def wait_reply() -> Dict[str, Any]:
                 if use_local_fast_path:
                     return await asyncio.wait_for(future, timeout=timeout)
 
+                # Slow path: poll the shared cache for the reply.
                 deadline = time.time() + timeout
+                last_liveness = time.time()
                 while time.time() < deadline:
                     reply = self._cache.broker_get(self._reply_key(request_id))
                     if reply is not None:
@@ -573,6 +607,21 @@ class CoderAIBroker:
                         return reply
                     if future.done():
                         return future.result()
+                    # Periodically verify the session is still alive on the remote node.
+                    now = time.time()
+                    if now - last_liveness >= LIVENESS_CHECK_INTERVAL:
+                        live = self._cache.broker_get(
+                            self._session_meta_key(provider_id, resolved_client_id)
+                        )
+                        if live is None:
+                            raise RuntimeError(
+                                f"CoderAI session for provider '{provider_id}' expired while waiting for reply"
+                            )
+                        if live.get("closed"):
+                            raise RuntimeError(
+                                f"CoderAI session for provider '{provider_id}' went offline while waiting for reply"
+                            )
+                        last_liveness = now
                     await asyncio.sleep(0.1)
                 raise asyncio.TimeoutError()
 
@@ -585,7 +634,12 @@ class CoderAIBroker:
                 self._pending.pop(request_id, None)
 
     async def consume_request(self, session_id: str, timeout: int = HEARTBEAT_POLL_SECONDS) -> Optional[Dict[str, Any]]:
-        return await asyncio.to_thread(self._cache.broker_blocking_pop, self._request_queue_key(session_id), timeout)
+        """Block-pop the next request for this session from the shared queue."""
+        return await asyncio.to_thread(
+            self._cache.broker_blocking_pop,
+            self._request_queue_key(session_id),
+            timeout,
+        )
 
     async def publish_response(self, message: Dict[str, Any]) -> None:
         request_id = message.get("request_id")
@@ -596,7 +650,9 @@ class CoderAIBroker:
             await self._publish_stream_response(message)
             return
         await self._record_request_metric(request_id, message)
+        # Write to shared cache so the requesting node (possibly remote) can poll it.
         self._cache.broker_set(self._reply_key(request_id), message, ttl=REPLY_TTL_SECONDS)
+        # Also resolve any local Future for the fast-path case.
         await self.resolve_response(message)
 
     async def _publish_stream_response(self, message: Dict[str, Any]) -> None:
@@ -604,7 +660,9 @@ class CoderAIBroker:
         async with self._lock:
             pending = self._pending.get(request_id)
             queue = pending.stream_queue if pending else None
+
         if queue is not None:
+            # Fast path: local asyncio.Queue for this instance's pending request.
             pending.event_log.append(message)
             if message.get("event") in {"done", "completed"} and not pending.future.done():
                 await self._record_request_metric(request_id, message)
@@ -612,6 +670,8 @@ class CoderAIBroker:
             else:
                 await queue.put(message)
             return
+
+        # Slow path: push chunk to the shared cache list for the remote requester.
         self._cache.broker_push(self._reply_key(request_id), message, ttl=REPLY_TTL_SECONDS)
 
     async def wait_for_stream_event(self, request_id: str, timeout: float = 300.0) -> Dict[str, Any]:
@@ -623,9 +683,11 @@ class CoderAIBroker:
                     if event.get("event") not in {"done", "completed"}:
                         pending.event_log.pop(idx)
                         return event
+
         if queue is not None:
             return await asyncio.wait_for(queue.get(), timeout=timeout)
 
+        # Slow path: poll the shared cache list.
         deadline = time.time() + timeout
         while time.time() < deadline:
             reply = self._cache.broker_pop_nowait(self._reply_key(request_id))
@@ -650,6 +712,9 @@ class CoderAIBroker:
         async with self._lock:
             pending_items = list(self._pending.items())
         for request_id, pending in pending_items:
+            snap_sid = (pending.request_snapshot or {}).get("session_id")
+            if snap_sid and snap_sid != session_id:
+                continue
             if not pending.future.done():
                 pending.future.set_exception(RuntimeError(error))
             async with self._lock:

aisbf/database.py

@@ -1049,7 +1049,7 @@ class DatabaseManager:
             cursor = conn.cursor()
             placeholder = '?' if self.db_type == 'sqlite' else '%s'
             cursor.execute(f'''
-                SELECT id, username, email, display_name, role, is_active, email_verified, created_at, last_verification_email_sent, profile_pic
+                SELECT id, username, email, display_name, role, is_active, email_verified, created_at, last_verification_email_sent, profile_pic, can_publish_market
                 FROM users
                 WHERE id = {placeholder}
             ''', (user_id,))
@@ -1066,7 +1066,8 @@ class DatabaseManager:
                     'email_verified': row[6],
                     'created_at': row[7],
                     'last_verification_email_sent': row[8],
-                    'profile_pic': row[9] or None
+                    'profile_pic': row[9] or None,
+                    'can_publish_market': bool(row[10]) if row[10] is not None else True
                 }
             return None
 
@@ -1628,7 +1629,8 @@ class DatabaseManager:
                     u.is_active,
                     u.tier_id,
                     u.display_name,
-                    t.name as tier_name
+                    t.name as tier_name,
+                    u.can_publish_market
                 FROM users u
                 LEFT JOIN account_tiers t ON u.tier_id = t.id
                 {where_clause}
@@ -1645,8 +1647,8 @@ class DatabaseManager:
             users = []
             for row in cursor.fetchall():
                 user = dict(zip(columns, row))
-                # Normalize boolean fields
                 user['is_active'] = bool(user['is_active']) if user['is_active'] is not None else True
+                user['can_publish_market'] = bool(user['can_publish_market']) if user.get('can_publish_market') is not None else True
                 users.append(user)
 
             return {
@@ -1720,6 +1722,16 @@ class DatabaseManager:
             cursor.execute(query, params)
             conn.commit()
 
+    def set_user_market_publish(self, user_id: int, can_publish: bool):
+        with self._get_connection() as conn:
+            cursor = conn.cursor()
+            placeholder = '?' if self.db_type == 'sqlite' else '%s'
+            cursor.execute(
+                f'UPDATE users SET can_publish_market = {placeholder} WHERE id = {placeholder}',
+                (1 if can_publish else 0, user_id)
+            )
+            conn.commit()
+
     def create_notification(self, user_id: int, title: str, message: str, notification_type: str = 'message') -> int:
         """Create a notification for a user. Returns the new notification id."""
         with self._get_connection() as conn:
@@ -6117,6 +6129,7 @@ def DatabaseManager__run_config_migrations(self, cursor, auto_increment, timesta
             ('reset_password_token', 'VARCHAR(255)'),
             ('reset_password_token_expires', 'TIMESTAMP NULL'),
             ('profile_pic', 'MEDIUMTEXT'),
+            ('can_publish_market', f'{boolean_type} DEFAULT 1'),
         ]
         if self.db_type == 'sqlite':
             cursor.execute("PRAGMA table_info(users)")

aisbf/routes/dashboard/market.py

@@ -4,6 +4,7 @@ from aisbf.routes.auth import require_dashboard_auth, require_api_auth, require_
 from aisbf.database import DatabaseRegistry
 from aisbf.analytics import get_analytics
 from aisbf.coderai_broker import broker as coderai_broker
+from aisbf.studio import serialize_studio_capability_choices
 import json
 import logging
 
@@ -405,6 +406,41 @@ async def dashboard_market(request: Request):
         await _attach_hardware_snapshot(listing_copy)
         enriched.append(listing_copy)
 
+    is_admin = request.session.get('role') == 'admin'
+    user_resources = {'providers': [], 'rotations': [], 'autoselects': [], 'models_by_provider': {}}
+
+    def _add_provider(pid, cfg):
+        user_resources['providers'].append({'id': pid, 'label': cfg.get('name') or pid})
+        models = cfg.get('models') or []
+        if models:
+            user_resources['models_by_provider'][pid] = [
+                {'name': (m.get('name') if isinstance(m, dict) else str(m))}
+                for m in models if (m.get('name') if isinstance(m, dict) else m)
+            ]
+
+    if current_user_id:
+        for p in db.get_user_providers(current_user_id):
+            _add_provider(p['provider_id'], p.get('config') or {})
+        for r in db.get_user_rotations(current_user_id):
+            cfg = r.get('config') or {}
+            rid = r['rotation_id']
+            user_resources['rotations'].append({'id': rid, 'label': cfg.get('name') or rid})
+        for a in db.get_user_autoselects(current_user_id):
+            cfg = a.get('config') or {}
+            aid = a['autoselect_id']
+            user_resources['autoselects'].append({'id': aid, 'label': cfg.get('name') or aid})
+
+    if is_admin and _config:
+        for pid, pobj in (getattr(_config, 'providers', None) or {}).items():
+            cfg = pobj.model_dump() if hasattr(pobj, 'model_dump') else (vars(pobj) if not isinstance(pobj, dict) else pobj)
+            _add_provider(pid, cfg)
+        for rid, robj in (getattr(_config, 'rotations', None) or {}).items():
+            cfg = robj.model_dump() if hasattr(robj, 'model_dump') else (vars(robj) if not isinstance(robj, dict) else robj)
+            user_resources['rotations'].append({'id': rid, 'label': cfg.get('name') or rid})
+        for aid, aobj in (getattr(_config, 'autoselects', None) or {}).items():
+            cfg = aobj.model_dump() if hasattr(aobj, 'model_dump') else (vars(aobj) if not isinstance(aobj, dict) else aobj)
+            user_resources['autoselects'].append({'id': aid, 'label': cfg.get('name') or aid})
+
     return _templates.TemplateResponse(
         request=request,
         name='dashboard/market.html',
@@ -415,6 +451,8 @@ async def dashboard_market(request: Request):
             'currency_symbol': db.get_currency_settings().get('currency_symbol', '$'),
             'current_user_id': current_user_id,
             'market_settings_json': json.dumps(market_settings),
+            'user_resources_json': json.dumps(user_resources),
+            'all_capabilities_json': json.dumps(serialize_studio_capability_choices()),
         },
     )
 
@@ -525,6 +563,8 @@ async def api_publish_market_listing(request: Request):
         if not market_settings.get('allow_user_publish', True):
             return JSONResponse({'error': 'User publishing is disabled'}, status_code=403)
         user = db.get_user_by_id(user_id)
+        if not user.get('can_publish_market', True):
+            return JSONResponse({'error': 'Your account is not allowed to publish to the market'}, status_code=403)
         owner_user_id = user_id
         owner_username = user['username']
         source_scope = 'user'

aisbf/routes/dashboard/settings.py

@@ -985,6 +985,24 @@ async def dashboard_users_toggle(request: Request, user_id: int):
     except Exception as e:
         return JSONResponse({"success": False, "error": str(e)}, status_code=500)
 
+@router.post("/dashboard/users/{user_id}/toggle-market-publish")
+async def dashboard_users_toggle_market_publish(request: Request, user_id: int):
+    """Toggle a user's ability to publish to the market"""
+    auth_check = require_admin(request)
+    if auth_check:
+        return auth_check
+
+    db = DatabaseRegistry.get_config_database()
+    try:
+        user = db.get_user_by_id(user_id)
+        if not user:
+            return JSONResponse({"success": False, "error": "User not found"}, status_code=404)
+        new_value = not user.get('can_publish_market', True)
+        db.set_user_market_publish(user_id, new_value)
+        return JSONResponse({"success": True, "can_publish_market": new_value})
+    except Exception as e:
+        return JSONResponse({"success": False, "error": str(e)}, status_code=500)
+
 @router.post("/dashboard/users/{user_id}/delete")
 async def dashboard_users_delete(request: Request, user_id: int):
     """Delete a user"""

minio-archive-plan.md

+# MinIO Archive for Studio-Generated Files
+
+## Overview
+
+Replace the local `~/.aisbf/studio/archive/` filesystem writes with an optional MinIO-backed object store. When MinIO is configured, all instances in a cluster write to and read from the same bucket. Local filesystem remains the fallback when MinIO is not configured, so single-node installs need no changes.
+
+---
+
+## Step 1 — Dependency
+
+Add `boto3` to requirements (the AWS SDK, which is the standard MinIO Python client since MinIO exposes an S3-compatible API):
+
+```
+boto3>=1.34
+```
+
+No `aioboto3` needed — archive writes happen in response to generation requests and a synchronous upload is acceptable; the generation itself already takes seconds.
+
+---
+
+## Step 2 — Admin Settings (DB + API + UI)
+
+### Database
+
+Add `get_archive_storage_settings` / `save_archive_storage_settings` in `database.py`, following the exact pattern of `get_market_settings` / `save_market_settings`. Stored under key `archive_storage` in the `admin_settings` table as a JSON object:
+
+```json
+{
+  "backend": "local",
+  "endpoint_url": "http://minio:9000",
+  "access_key": "",
+  "secret_key": "",
+  "bucket": "aisbf-archive",
+  "region": "us-east-1",
+  "presign_ttl_seconds": 3600,
+  "public_url_base": ""
+}
+```
+
+`backend` is either `"local"` or `"minio"`. `public_url_base` is optional — if set (e.g. `https://cdn.example.com/aisbf-archive`), presigned URLs are skipped and a plain public URL is constructed instead (useful when the bucket is public behind a CDN).
+
+### API
+
+Two endpoints in `routes/dashboard/admin.py`:
+
+- `GET /api/admin/settings/archive-storage` — returns current settings with `secret_key` masked
+- `POST /api/admin/settings/archive-storage` — saves and immediately validates the connection by calling `list_buckets()` on the client; returns an error if unreachable
+
+### UI
+
+New section in `admin_payment_settings.html` (or a dedicated `admin_storage_settings.html` linked from the settings nav). Fields:
+
+- Backend toggle: local / MinIO
+- Endpoint URL
+- Access key
+- Secret key
+- Bucket name
+- Region
+- Presign TTL (seconds)
+- Public URL base (optional)
+
+Save button calls the POST endpoint and shows a connection test result inline.
+
+---
+
+## Step 3 — Storage Abstraction (`aisbf/archive_storage.py`)
+
+New module with a simple interface:
+
+```python
+class ArchiveStorage:
+    def put(self, object_key: str, data: bytes, content_type: str) -> str:
+        """Store bytes, return a URL (presigned or public)."""
+
+    def get_url(self, object_key: str) -> str:
+        """Return a URL for an existing object."""
+
+    def delete(self, object_key: str) -> None: ...
+
+    def list_prefix(self, prefix: str) -> list[dict]:
+        """List objects under a prefix, return dicts with key/size/last_modified."""
+```
+
+### `LocalArchiveStorage`
+
+Wraps the current `Path.write_bytes` / `iterdir` / `unlink` logic extracted from `studio_services.py`. Returns `/dashboard/static/studio-archive/{key}` URLs. No new logic — this is a refactor of what exists today.
+
+### `MinioArchiveStorage`
+
+Uses `boto3.client('s3', endpoint_url=..., aws_access_key_id=..., aws_secret_access_key=...)`.
+
+- `put`: `client.put_object(Bucket=bucket, Key=object_key, Body=data, ContentType=content_type)`
+- `get_url`: if `public_url_base` is set, returns `f"{public_url_base}/{object_key}"`, otherwise calls `client.generate_presigned_url('get_object', Params={...}, ExpiresIn=ttl)`
+- `delete`: `client.delete_object(Bucket=bucket, Key=object_key)`
+- `list_prefix`: `client.list_objects_v2(Bucket=bucket, Prefix=prefix)`
+
+### Factory
+
+`get_archive_storage() -> ArchiveStorage` reads settings from the DB and returns the right implementation. Result is cached in a module-level variable and invalidated when settings are saved via the admin POST endpoint.
+
+### Object key convention
+
+`{scope}/{filename}` where scope is `admin` or `user_{owner_id}` — matches the existing local directory structure exactly, so the scope-path logic in `studio_services.py` maps 1:1 to object key prefixes.
+
+---
+
+## Step 4 — Integration into `studio_services.py`
+
+Three touch points:
+
+### `_scope_dir`
+
+Currently creates and returns a `Path`. Extract the scope name logic into a separate `_scope_prefix(scope, owner_id) -> str` method used by both backends:
+
+```python
+def _scope_prefix(self, scope: str, owner_id: Optional[int]) -> str:
+    return "admin" if scope == "admin" or owner_id is None else f"user_{owner_id}"
+```
+
+`_scope_dir` continues to use this internally for local storage; MinIO code uses `_scope_prefix` directly.
+
+### Output file saving
+
+Currently done inline wherever generation results are written (e.g. `target.write_bytes(base64.b64decode(encoded))`). These become:
+
+```python
+url = storage.put(f"{scope_prefix}/{filename}", data, content_type)
+```
+
+The returned URL is stored in the result metadata instead of constructing a static path string.
+
+### `list_archive`
+
+Replace `scoped.iterdir()` with `storage.list_prefix(scope_prefix)`. The returned dicts have `key`, `size`, `last_modified` — map to the existing response shape (`filename`, `url`, `size`, `created`, `type`). URL comes from `storage.get_url(key)`.
+
+### File deletion
+
+Replace `file_path.unlink()` with `storage.delete(object_key)`.
+
+---
+
+## Step 5 — Presigned URL Expiry Handling
+
+Presigned URLs expire (default 1 hour, configurable in admin settings). The frontend calls the archive listing endpoint fresh on each page load, which regenerates presigned URLs at listing time — no special handling needed.
+
+If a user keeps the archive tab open past the TTL and clicks a stale link, it will 403. This is acceptable standard S3 presigned URL behavior. The TTL can be set higher (e.g. 24h) for lower-churn archives.
+
+---
+
+## Step 6 — Migration Tool (optional)
+
+A one-shot admin action:
+
+`POST /api/admin/settings/archive-storage/migrate`
+
+Iterates `~/.aisbf/studio/archive/` on the local node, uploads each file to MinIO using the same key convention, then optionally deletes local copies. Returns a progress count in the response.
+
+Only needs to run once on one node after MinIO is configured. Not strictly required — new files go to MinIO immediately after the backend is switched; old local files simply stop being accessible via the new URLs. Whether to migrate historical files is the admin's choice.
+
+---
+
+## Step 7 — Bucket Bootstrapping
+
+On `MinioArchiveStorage.__init__`, check if the configured bucket exists and create it if not:
+
+```python
+client.create_bucket(Bucket=bucket)
+```
+
+This means the admin only needs to provide credentials — the bucket is auto-provisioned on first use. If creation fails (permissions issue), surface the error in the connection test response from the admin POST endpoint.
+
+---
+
+## Files to Create / Modify
+
+| File | Action | Notes |
+|---|---|---|
+| `aisbf/archive_storage.py` | **Create** | `ArchiveStorage` base, `LocalArchiveStorage`, `MinioArchiveStorage`, `get_archive_storage()` factory |
+| `aisbf/database.py` | **Modify** | Add `get_archive_storage_settings`, `save_archive_storage_settings` |
+| `aisbf/studio_services.py` | **Modify** | Extract `_scope_prefix`, replace all `Path` write/read/list/delete calls in archive methods with `get_archive_storage()` calls |
+| `aisbf/routes/dashboard/admin.py` | **Modify** | Add GET/POST `/api/admin/settings/archive-storage` endpoints; optional `/migrate` action |
+| `templates/dashboard/admin_payment_settings.html` | **Modify** | Add MinIO settings section with connection test feedback |
+| `requirements.txt` / `pyproject.toml` | **Modify** | Add `boto3>=1.34` |
+
+---
+
+## Current State Reference
+
+- Local archive path: `~/.aisbf/studio/archive/{scope}/`
+- Current serving: FastAPI `StaticFiles` mounted at `/dashboard/static/studio-archive/`
+- Current URL pattern: `/dashboard/static/studio-archive/admin/{filename}` or `/dashboard/static/studio-archive/user_{owner_id}/{filename}`
+- Write entry point: `StudioService._store_uploads()` in `studio_services.py:566`
+- List entry point: `StudioService.list_archive()` in `studio_services.py:696`
+- Settings storage pattern: `admin_settings` table, key/JSON-value pairs — see `get_market_settings` / `save_market_settings` in `database.py` as the reference implementation

templates/dashboard/market.html

@@ -6,8 +6,10 @@
 <h2 style="margin-bottom: 20px;"><i class="fas fa-store" style="color: var(--color-link);"></i> Market</h2>
 <p style="color: var(--color-muted); margin-bottom: 20px;">Browse shared providers and models, review pricing and reputation, and import market resources into your account.</p>
 
-<div style="background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 16px; margin-bottom: 18px; display:grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px;">
-    <input id="market-search" placeholder="Search name, type, capability" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
+<div style="background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 16px; margin-bottom: 18px;">
+    <datalist id="market-search-suggestions"></datalist>
+    <div style="display:grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; margin-bottom: 12px;">
+        <input id="market-search" list="market-search-suggestions" placeholder="Search name, type, capability" autocomplete="off" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
         <select id="market-source-type" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
             <option value="">All resource types</option>
             <option value="provider">Provider</option>
@@ -15,7 +17,6 @@
             <option value="rotation">Rotation</option>
             <option value="autoselect">Autoselect</option>
         </select>
-    <input id="market-capabilities" placeholder="Capabilities comma-separated" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
         <input id="market-model-type" placeholder="Model type / family" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
         <input id="market-model-name" placeholder="Model name" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
         <input id="market-size" placeholder="Size / params" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
@@ -29,7 +30,15 @@
             <option value="requests">Most used</option>
             <option value="revenue">Top grossing</option>
         </select>
-    <button class="btn btn-secondary" onclick="reloadMarket()">Apply Filters</button>
+    </div>
+    <div style="display:flex; flex-wrap:wrap; gap:8px; align-items:center; margin-bottom:10px;">
+        <div id="capabilities-filter" style="display:flex; flex-wrap:wrap; gap:8px; align-items:center; flex:1;">
+            <span style="font-size:13px; color:var(--color-muted); margin-right:4px;">Capabilities:</span>
+            <button type="button" onclick="setAllCaps(true)" style="padding:3px 10px; border-radius:999px; border:1px solid var(--color-border); background:var(--bg-page); color:var(--color-muted); font-size:12px; cursor:pointer;">all</button>
+            <button type="button" onclick="setAllCaps(false)" style="padding:3px 10px; border-radius:999px; border:1px solid var(--color-border); background:var(--bg-page); color:var(--color-muted); font-size:12px; cursor:pointer;">none</button>
+        </div>
+        <button class="btn btn-secondary" onclick="reloadMarket()" style="white-space:nowrap;">Search</button>
+    </div>
 </div>
 
 <div style="display:grid; grid-template-columns: repeat(auto-fit, minmax(320px, 1fr)); gap: 16px;" id="market-grid"></div>
@@ -37,15 +46,17 @@
 <div style="margin-top: 30px; background: var(--bg-panel); border: 1px solid var(--color-border); border-radius: 10px; padding: 18px;">
     <h3 style="margin-top:0;">Publish one of your resources</h3>
     <div style="display:grid; grid-template-columns: 1fr 1fr; gap: 12px;">
-        <select id="publish-source-type" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
+        <select id="publish-source-type" onchange="onPublishTypeChange()" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
             <option value="provider">Provider</option>
             <option value="model">Model</option>
             <option value="rotation">Rotation</option>
             <option value="autoselect">Autoselect</option>
         </select>
-        <input id="publish-source-id" placeholder="Provider / model / rotation / autoselect id" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
+        <select id="publish-source-select" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
+        </select>
         <input id="publish-title" placeholder="Listing title" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
-        <input id="publish-provider-id" placeholder="Provider id for model listings" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
+        <select id="publish-provider-select" onchange="onPublishProviderChange()" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text); display:none;">
+        </select>
         <input id="publish-price-tokens" type="number" step="0.0001" min="0" placeholder="Price / 1M tokens" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
         <input id="publish-price-requests" type="number" step="0.0001" min="0" placeholder="Price / 1000 requests" style="padding:10px; border:1px solid var(--color-border); border-radius:6px; background:var(--bg-page); color:var(--color-text);">
     </div>
@@ -60,6 +71,8 @@ const BASE_PATH = {{ (request.scope.get('root_path', '') or '') | tojson }};
 const listings = {{ market_listings_json | safe }};
 const currencySymbol = {{ currency_symbol | tojson }};
 const marketSettings = {{ market_settings_json | safe }};
+const userResources = {{ user_resources_json | safe }};
+const allCapabilities = {{ all_capabilities_json | safe }};
 
 function voteSummary(votes, target) {
     const item = (votes || {})[target] || {upvotes:0, downvotes:0, score:0};
@@ -165,12 +178,58 @@ async function voteListing(id, targetType, vote) {
     renderMarket();
 }
 
+function _populateSelect(selectEl, items, valueFn, labelFn, placeholder) {
+    selectEl.innerHTML = '';
+    if (placeholder) {
+        const opt = document.createElement('option');
+        opt.value = '';
+        opt.textContent = placeholder;
+        selectEl.appendChild(opt);
+    }
+    for (const item of items) {
+        const opt = document.createElement('option');
+        opt.value = valueFn(item);
+        opt.textContent = labelFn(item);
+        selectEl.appendChild(opt);
+    }
+}
+
+function onPublishTypeChange() {
+    const type = document.getElementById('publish-source-type').value;
+    const sourceSelect = document.getElementById('publish-source-select');
+    const providerSelect = document.getElementById('publish-provider-select');
+
+    providerSelect.style.display = 'none';
+
+    if (type === 'provider') {
+        _populateSelect(sourceSelect, userResources.providers, i => i.id, i => i.label, '— select provider —');
+    } else if (type === 'rotation') {
+        _populateSelect(sourceSelect, userResources.rotations, i => i.id, i => i.label, '— select rotation —');
+    } else if (type === 'autoselect') {
+        _populateSelect(sourceSelect, userResources.autoselects, i => i.id, i => i.label, '— select autoselect —');
+    } else if (type === 'model') {
+        _populateSelect(providerSelect, userResources.providers, i => i.id, i => i.label, '— select provider —');
+        providerSelect.style.display = '';
+        onPublishProviderChange();
+    }
+}
+
+function onPublishProviderChange() {
+    const pid = document.getElementById('publish-provider-select').value;
+    const sourceSelect = document.getElementById('publish-source-select');
+    const models = (userResources.models_by_provider[pid] || []);
+    _populateSelect(sourceSelect, models, m => m.name, m => m.name, models.length ? '— select model —' : '(no models configured)');
+}
+
 async function publishListing() {
+    const sourceType = document.getElementById('publish-source-type').value;
+    const sourceId = document.getElementById('publish-source-select').value.trim();
+    const providerId = document.getElementById('publish-provider-select').value.trim();
     const body = {
-        source_type: document.getElementById('publish-source-type').value,
-        provider_id: document.getElementById('publish-provider-id').value.trim(),
-        source_id: document.getElementById('publish-source-id').value.trim(),
-        model_id: document.getElementById('publish-source-type').value === 'model' ? document.getElementById('publish-source-id').value.trim() : undefined,
+        source_type: sourceType,
+        source_id: sourceId,
+        provider_id: sourceType === 'model' ? providerId : undefined,
+        model_id: sourceType === 'model' ? sourceId : undefined,
         title: document.getElementById('publish-title').value.trim(),
         description: document.getElementById('publish-description').value.trim(),
         price_per_million_tokens: parseFloat(document.getElementById('publish-price-tokens').value || '0'),
@@ -189,13 +248,88 @@ async function publishListing() {
     window.location.reload();
 }
 
+function updateSearchSuggestions() {
+    const suggestions = new Set();
+    for (const l of listings) {
+        if (l.title) suggestions.add(l.title);
+        const meta = l.metadata || {};
+        const mn = l.model_id || meta.model_name;
+        if (mn) suggestions.add(mn);
+        const pt = meta.provider_type || meta.type;
+        if (pt) suggestions.add(pt);
+    }
+    const dl = document.getElementById('market-search-suggestions');
+    dl.innerHTML = '';
+    for (const s of suggestions) {
+        const opt = document.createElement('option');
+        opt.value = s;
+        dl.appendChild(opt);
+    }
+}
+
+function _setPillActive(btn, active) {
+    btn.dataset.active = active ? '1' : '0';
+    btn.style.background = active ? 'var(--color-link)' : 'var(--bg-page)';
+    btn.style.color = active ? '#fff' : 'var(--color-text)';
+    btn.style.borderColor = active ? 'var(--color-link)' : 'var(--color-border)';
+}
+
+function setAllCaps(active) {
+    document.querySelectorAll('#capabilities-filter button[data-cap]').forEach(btn => _setPillActive(btn, active));
+    reloadMarket();
+}
+
+function initCapabilityPills() {
+    const container = document.getElementById('capabilities-filter');
+    while (container.children.length > 1) container.removeChild(container.lastChild);
+    for (const cap of allCapabilities) {
+        const btn = document.createElement('button');
+        btn.type = 'button';
+        btn.dataset.cap = cap;
+        btn.textContent = cap.replace(/_/g, ' ');
+        btn.style.cssText = 'padding:4px 12px; border-radius:999px; border:1px solid var(--color-border); background:var(--bg-page); color:var(--color-text); font-size:13px; cursor:pointer; transition: background 0.15s, color 0.15s;';
+        _setPillActive(btn, true);
+        btn.addEventListener('click', () => {
+            _setPillActive(btn, btn.dataset.active !== '1');
+            reloadMarket();
+        });
+        container.appendChild(btn);
+    }
+}
+
+let _searchDebounce = null;
+document.addEventListener('DOMContentLoaded', () => {
+    const searchEl = document.getElementById('market-search');
+    searchEl.addEventListener('input', () => {
+        if (searchEl.value.length === 0 || searchEl.value.length >= 2) {
+            clearTimeout(_searchDebounce);
+            _searchDebounce = setTimeout(reloadMarket, 300);
+        }
+    });
+    for (const id of ['market-source-type', 'market-sort', 'market-online-only', 'market-model-type', 'market-model-name', 'market-size']) {
+        const el = document.getElementById(id);
+        if (el) el.addEventListener('change', reloadMarket);
+    }
+    for (const id of ['market-model-type', 'market-model-name', 'market-size']) {
+        const el = document.getElementById(id);
+        if (el) el.addEventListener('input', () => {
+            clearTimeout(_searchDebounce);
+            _searchDebounce = setTimeout(reloadMarket, 300);
+        });
+    }
+});
+
 renderMarket();
+updateSearchSuggestions();
+initCapabilityPills();
+onPublishTypeChange();
 
 async function reloadMarket() {
     const params = new URLSearchParams();
     const q = document.getElementById('market-search').value.trim();
     const sourceType = document.getElementById('market-source-type').value;
-    const capabilities = document.getElementById('market-capabilities').value.trim();
+    const activePills = [...document.querySelectorAll('#capabilities-filter button[data-active="1"]')];
+    const selectedCaps = activePills.length === allCapabilities.length ? [] : activePills.map(b => b.dataset.cap);
     const modelType = document.getElementById('market-model-type').value.trim();
     const modelName = document.getElementById('market-model-name').value.trim();
     const size = document.getElementById('market-size').value.trim();
@@ -203,7 +337,7 @@ async function reloadMarket() {
     const sortBy = document.getElementById('market-sort').value;
     if (q) params.set('q', q);
     if (sourceType) params.set('source_type', sourceType);
-    if (capabilities) params.set('capabilities', capabilities);
+    if (selectedCaps.length) params.set('capabilities', selectedCaps.join(','));
     if (modelType) params.set('model_type', modelType);
     if (modelName) params.set('model_name', modelName);
     if (size) params.set('size', size);
@@ -218,6 +352,7 @@ async function reloadMarket() {
     listings.length = 0;
     for (const listing of result.listings || []) listings.push(listing);
     renderMarket();
+    updateSearchSuggestions();
 }
 </script>
 {% endblock %}

templates/dashboard/users.html

@@ -211,6 +211,9 @@ along with this program.  If not, see <https://www.gnu.org/licenses/>.
                         <button onclick="toggleUserStatus({{ user.id }}, {{ user.is_active|lower }})" class="btn btn-warning" style="padding: 5px 10px; font-size: 12px; margin: 0;">
                             {% if user.is_active %}Disable{% else %}Enable{% endif %}
                         </button>
+                        <button onclick="toggleMarketPublish({{ user.id }}, {{ (user.can_publish_market if user.can_publish_market is defined else true)|lower }})" class="btn" style="padding: 5px 10px; font-size: 12px; margin: 0; background: {% if user.can_publish_market is defined and not user.can_publish_market %}#6b7280{% else %}#0ea5e9{% endif %};">
+                            {% if user.can_publish_market is defined and not user.can_publish_market %}Allow Publish{% else %}Block Publish{% endif %}
+                        </button>
                         <button onclick="impersonateUser({{ user.id }}, '{{ user.username }}')" class="btn" style="padding: 5px 10px; font-size: 12px; margin: 0; background: #7c3aed;">Impersonate</button>
                         <button onclick="deleteUser({{ user.id }}, '{{ user.username }}')" class="btn btn-danger" style="padding: 5px 10px; font-size: 12px; margin: 0;" data-i18n="users_page.delete_btn">Delete</button>
                     </div>
@@ -836,6 +839,28 @@ async function toggleUserStatus(userId, currentStatus) {
     }
 }
 
+async function toggleMarketPublish(userId, canPublish) {
+    const action = canPublish ? 'block this user from publishing to the market' : 'allow this user to publish to the market';
+    const ok = await showConfirm('Are you sure you want to ' + action + '?', 'Confirm');
+    if (ok) {
+        fetch(baseUrl + userId + '/toggle-market-publish', {
+            method: 'POST',
+            headers: {'Content-Type': 'application/json'}
+        })
+        .then(response => response.json())
+        .then(data => {
+            if (data.success) {
+                location.reload();
+            } else {
+                showAlert(data.error || 'Failed to update market publish permission', 'Error', '❌', 'danger');
+            }
+        })
+        .catch(error => {
+            showAlert('Error: ' + error, 'Error', '❌', 'danger');
+        });
+    }
+}
+
 async function deleteUser(userId, username) {
     const ok = await showDangerConfirm('Are you sure you want to delete user "' + username + '"? This action cannot be undone.', 'Delete User');
     if (ok) {