feat(pi): techo de capacidad para los guests QEMU (multiusuario)

Cada guest es un proceso QEMU con su propia RAM (1-2 GB segun placa) y
sus hilos de vCPU, asi que el limite lo pone la maquina, no el codigo.
No habia ningun tope: el usuario N simplemente empujaba la caja a swap y
la sesion de TODOS se volvia lenta, que es peor que decirle al usuario N
que espere un minuto.

  VELXIO_PI_MAX_INSTANCES   (6)  guests simultaneos en la maquina
  VELXIO_PI_MAX_PER_OWNER   (2)  guests por persona

El "owner" es el hash de la cookie de sesion: sirve solo para contar, no
se guarda ni se lee de vuelta, y cae al host del cliente cuando no hay
cookie (sidecar de escritorio, tests). Sin identidad solo aplica el tope
global.

El rechazo llega como un mensaje que dice que hacer ("prueba en un
minuto" / "para una de tus sesiones"), no como un fallo mudo, y se
comprueba ANTES de lanzar el proceso.

backend/app/api/routes/simulation.py

@@ -1,3 +1,4 @@
+import hashlib
 import json
 import logging
 import socket
@@ -10,6 +11,24 @@ from app.services.stm32_lib_manager import stm32_lib_manager
 from app.core.hooks import dispatch_ws_sim_message
 
 
+def _owner_key(websocket: WebSocket) -> str | None:
+    """Stable, opaque id for "the same person" across tabs.
+
+    The session cookie is hashed rather than stored: this is only used to
+    count concurrent guests per user, so the value never needs to be read
+    back. Falls back to the client host when there is no cookie (desktop
+    sidecar, tests), and to None when there is neither.
+    """
+    try:
+        token = websocket.cookies.get('access_token')
+    except Exception:
+        token = None
+    if token:
+        return 'u:' + hashlib.sha256(token.encode()).hexdigest()[:16]
+    host = getattr(getattr(websocket, 'client', None), 'host', None)
+    return f'h:{host}' if host else None
+
+
 def _find_free_port() -> int:
     """Allocate a free TCP port for WiFi hostfwd."""
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -76,9 +95,19 @@ async def simulation_websocket(websocket: WebSocket, client_id: str):
                 if not await board_allowed(websocket, board):
                     await qemu_callback('error', {'message': PRO_BOARD_MESSAGE})
                 else:
-                    # msg_data carries whatever the client declared for this
+                    # Capacity: a guest is a real QEMU process with its own
-                    # session (an overlay may materialise extra drives from it).
+                    # GBs, so the box is the limit. Refuse with words the
-                    qemu_manager.start_instance(client_id, board, qemu_callback, msg_data)
+                    # user can act on rather than letting the machine swap.
+                    owner = _owner_key(websocket)
+                    full = qemu_manager.capacity_error(owner)
+                    if full:
+                        await qemu_callback('error', {'message': full})
+                    else:
+                        # msg_data carries whatever the client declared for this
+                        # session (an overlay may materialise extra drives from it).
+                        qemu_manager.start_instance(
+                            client_id, board, qemu_callback, msg_data, owner=owner,
+                        )
 
             elif msg_type == 'stop_pi':
                 qemu_manager.stop_instance(client_id)

backend/app/services/qemu_manager.py

@@ -155,6 +155,15 @@ DEFAULT_PI_BOARD = 'raspberry-pi-3'
 # client is told plainly when it trips. Override with VELXIO_PI_MAX_SESSION_S.
 MAX_SESSION_SECONDS = int(os.environ.get('VELXIO_PI_MAX_SESSION_S', '7200'))
 
+# Capacity. Every guest is a real QEMU process holding its own RAM (1-2 GB
+# per board profile) and vCPU threads, so the machine — not the code — is
+# the limit. Without a ceiling the Nth user simply pushes the box into
+# swap and everyone's session gets slow, which is worse than telling the
+# Nth user to wait a minute. Per-owner keeps one person's tabs from
+# eating the pool: opening the same project in six tabs is six guests.
+MAX_INSTANCES = int(os.environ.get('VELXIO_PI_MAX_INSTANCES', '6'))
+MAX_INSTANCES_PER_OWNER = int(os.environ.get('VELXIO_PI_MAX_PER_OWNER', '2'))
+
 # Every key a profile must carry — the boot path reads exactly these.
 _PI_PROFILE_KEYS = frozenset(
     {'qemu', 'cpu', 'smp', 'memory', 'image_set', 'kernel', 'initramfs',
@@ -302,6 +311,9 @@ class PiInstance:
         # Raw `start_pi` payload — carries whatever the client declared for
         # this session (e.g. the packages an overlay must materialise).
         self.start_payload: dict = {}
+        # Who this guest belongs to (opaque key from the route), for the
+        # per-owner capacity check. None when the caller has no identity.
+        self.owner: str | None = None
 
     async def emit(self, event_type: str, data: dict) -> None:
         try:
@@ -321,9 +333,29 @@ class QemuManager:
 
     # ── Public API ────────────────────────────────────────────────────────────
 
+    def capacity_error(self, owner: str | None) -> str | None:
+        """Why this start must be refused, or None when there is room.
+
+        Returned as a message the user reads, so it says what to do next
+        instead of just failing."""
+        if len(self._instances) >= MAX_INSTANCES:
+            return (
+                'All the Linux machines are busy right now. '
+                'Try again in a minute — sessions free up as people stop them.'
+            )
+        if owner:
+            mine = sum(1 for i in self._instances.values() if i.owner == owner)
+            if mine >= MAX_INSTANCES_PER_OWNER:
+                return (
+                    f'You already have {mine} Linux sessions running. '
+                    'Stop one before starting another.'
+                )
+        return None
+
     def start_instance(self, client_id: str, board_type: str,
                        callback: EventCallback,
-                       payload: dict | None = None) -> None:
+                       payload: dict | None = None,
+                       owner: str | None = None) -> None:
         if client_id in self._instances:
             logger.warning('start_instance: %s already running', client_id)
             return
@@ -335,7 +367,12 @@ class QemuManager:
             board_type = DEFAULT_PI_BOARD
         inst = PiInstance(client_id, callback, board_type=board_type)
         inst.start_payload = dict(payload or {})
+        inst.owner = owner
         self._instances[client_id] = inst
+        logger.info(
+            'pi capacity: %d/%d guests running (starting %s)',
+            len(self._instances), MAX_INSTANCES, client_id,
+        )
         asyncio.create_task(self._boot(inst))
 
     def stop_instance(self, client_id: str) -> None:

backend/tests/test_pi_capacity.py

@@ -0,0 +1,55 @@
+"""Capacity limits for the QEMU-Linux guests.
+
+Every guest is a real process holding 1-2 GB and its own vCPU threads, so
+the machine is the ceiling, not the code. These check that the ceiling is
+enforced BEFORE a process is spawned and that the refusal says what the
+user should do next.
+"""
+import app.services.qemu_manager as qm
+
+
+class _Mgr(qm.QemuManager):
+    """Manager with a fake instance table (no processes involved)."""
+
+    def add(self, client_id: str, owner: str | None) -> None:
+        inst = qm.PiInstance(client_id, lambda *_a, **_k: None,
+                             board_type='raspberry-pi-3')
+        inst.owner = owner
+        self._instances[client_id] = inst
+
+
+def test_room_when_empty():
+    assert _Mgr().capacity_error('u:alice') is None
+
+
+def test_global_ceiling(monkeypatch):
+    monkeypatch.setattr(qm, 'MAX_INSTANCES', 2)
+    monkeypatch.setattr(qm, 'MAX_INSTANCES_PER_OWNER', 99)
+    mgr = _Mgr()
+    mgr.add('a', 'u:alice')
+    mgr.add('b', 'u:bob')
+    msg = mgr.capacity_error('u:carol')
+    assert msg and 'busy' in msg.lower()
+
+
+def test_per_owner_ceiling(monkeypatch):
+    monkeypatch.setattr(qm, 'MAX_INSTANCES', 99)
+    monkeypatch.setattr(qm, 'MAX_INSTANCES_PER_OWNER', 2)
+    mgr = _Mgr()
+    mgr.add('tab1', 'u:alice')
+    mgr.add('tab2', 'u:alice')
+    # Alice's third tab is refused...
+    msg = mgr.capacity_error('u:alice')
+    assert msg and 'already have' in msg.lower()
+    # ...while somebody else still gets in.
+    assert mgr.capacity_error('u:bob') is None
+
+
+def test_anonymous_owner_is_not_pooled(monkeypatch):
+    """No identity (desktop sidecar, tests) only hits the global ceiling."""
+    monkeypatch.setattr(qm, 'MAX_INSTANCES', 99)
+    monkeypatch.setattr(qm, 'MAX_INSTANCES_PER_OWNER', 1)
+    mgr = _Mgr()
+    mgr.add('a', None)
+    mgr.add('b', None)
+    assert mgr.capacity_error(None) is None