sync: 191 file da Baida98/AI@5943984b (2026-08-29 20:27 UTC)

Dockerfile

@@ -38,4 +38,4 @@ COPY --chown=user . /home/user/app/
 
 EXPOSE 7860
 
-CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port ${PORT:-7860} --workers 1"]
+CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port ${PORT:-7860} --workers ${WEB_CONCURRENCY:-2}"]

api/agent.py

@@ -42,8 +42,21 @@ def _sanitize_for_json(obj: object) -> object:
 from fastapi import APIRouter, Depends, HTTPException, Request, Body
 from fastapi.responses import StreamingResponse
 from .auth_guard import require_role, AuthRole
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, TypeAdapter, ValidationError, field_validator
 from typing import Literal
+
+# P0 JSON contract: expose structured JSON only after schema validation.
+_STRUCTURED_JSON_ADAPTER = TypeAdapter(dict[str, object] | list[object])
+
+def _validated_response_json(output: object) -> dict[str, object] | list[object] | None:
+    if not isinstance(output, str) or not output.strip():
+        return None
+    try:
+        parsed = json.loads(output)
+        return _STRUCTURED_JSON_ADAPTER.validate_python(parsed)
+    except (json.JSONDecodeError, ValidationError, TypeError, ValueError):
+        return None
+
 from .state import (
     _agent_tasks, _task_ai_clients, _task_checkpoints, _loop_registry, _run_stream_tasks,
     _prune_agent_tasks, _prune_checkpoints, _prune_loop_registry,
@@ -89,6 +102,83 @@ except Exception:
 router = APIRouter()
 
 
+class _LLMAdmission:
+    """Bounded in-process admission control for LLM loops."""
+    def __init__(self) -> None:
+        self.max_concurrency = max(1, int(os.getenv('LLM_MAX_CONCURRENCY', '2')))
+        self.max_queue = max(0, int(os.getenv('LLM_MAX_QUEUE', '8')))
+        self._condition = asyncio.Condition()
+        self._active = 0
+        self._queued = 0
+
+    async def reserve(self) -> bool:
+        async with self._condition:
+            if self._active + self._queued >= self.max_concurrency + self.max_queue:
+                return False
+            self._queued += 1
+            return True
+
+    async def acquire(self) -> None:
+        async with self._condition:
+            try:
+                while self._active >= self.max_concurrency:
+                    await self._condition.wait()
+                self._queued = max(0, self._queued - 1)
+                self._active += 1
+            except BaseException:
+                self._queued = max(0, self._queued - 1)
+                self._condition.notify(1)
+                raise
+
+    async def cancel(self) -> None:
+        async with self._condition:
+            self._queued = max(0, self._queued - 1)
+            self._condition.notify(1)
+
+    async def release(self) -> None:
+        async with self._condition:
+            self._active = max(0, self._active - 1)
+            self._condition.notify(1)
+
+    async def snapshot(self) -> dict[str, int]:
+        async with self._condition:
+            return {
+                'active': self._active,
+                'queued': self._queued,
+                'max_concurrency': self.max_concurrency,
+                'max_queue': self.max_queue,
+            }
+
+
+_LLM_ADMISSION = _LLMAdmission()
+
+
+async def _record_phase(phase: str, duration_ms: float = 0.0,
+                       outcome: str = 'ok', error_class: str | None = None) -> None:
+    try:
+        from .agent_telemetry import record_runtime_phase
+        await record_runtime_phase(phase, duration_ms, outcome, error_class)
+    except Exception:
+        pass
+
+
+def _queue_full_error() -> HTTPException:
+    return HTTPException(
+        status_code=429,
+        detail={
+            'error': 'llm_queue_full',
+            'message': 'Coda LLM satura. Riprova tra pochi secondi.',
+            'retry_after_seconds': 5,
+        },
+        headers={'Retry-After': '5'},
+    )
+
+
+@router.get('/api/agent/queue-status')
+async def agent_queue_status(role: AuthRole = Depends(require_role(AuthRole.MACHINE))):
+    return await _LLM_ADMISSION.snapshot()
+
+
 def _attach_byok_client(task_id: str, credentials: object) -> None:
     """Create a task-scoped LLM client without persisting or logging credentials."""
     if credentials is None:
@@ -204,10 +294,17 @@ async def agent_run_stream(
     body: ReasonLoopIn, request: Request,
     role: AuthRole = Depends(require_role(AuthRole.MACHINE)),  # GAP-1-fix: era fail-open
 ):
+    await _record_phase('auth')
+    if not await _LLM_ADMISSION.reserve():
+        await _record_phase('queue', outcome='rejected', error_class='queue_full')
+        raise _queue_full_error()
+    await _record_phase('queue')
     async def generate():
         queue: asyncio.Queue = asyncio.Queue()
 
         async def step_cb(step: dict) -> None:
+            if str(step.get('action', '')).startswith(('tool', 'executor:')):
+                await _record_phase('tool', outcome='ok')
             await queue.put(step)
 
         async def run_loop() -> None:
@@ -247,16 +344,19 @@ async def agent_run_stream(
                 _neg_c = getattr(body, 'negative_constraints', '') or ''
                 if _neg_c:
                     context_str = f"[VINCOLI OPERATIVI APPRESI — NON VIOLARE]\n{_neg_c}\n\n{context_str}".strip()
+                _provider_started = time.perf_counter()
                 result = await loop.run(
                     goal=body.goal, context=context_str,
                     max_steps=body.max_steps, on_step=step_cb,
                     session_id=getattr(body, "session_id", "") or "",
                 )
+                await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000)
                 await queue.put({
                     '__done__': True,
                     'result':   result.get('output',  ''),
                     'engine':   result.get('engine',  'fallback'),
                     'success':  result.get('success', False),
+                    'response_json': _validated_response_json(result.get('output', '')),
                 })
             except Exception as exc:
                 # GAP-A1: log incident in registry (fire-and-forget, non-blocking)
@@ -270,7 +370,13 @@ async def agent_run_stream(
                     _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
                 await queue.put({'__error__': str(exc)})
 
-        task    = asyncio.create_task(run_loop())
+        async def _admitted_run_loop():
+            await _LLM_ADMISSION.acquire()
+            try:
+                await run_loop()
+            finally:
+                await _LLM_ADMISSION.release()
+        task    = asyncio.create_task(_admitted_run_loop())
         task.add_done_callback(_log_task_exc)  # BUG-CB-1
         task_id = str(uuid.uuid4())
         # ABORT-1: registra task + queue per permettere cancellazione via POST /api/agent/abort
@@ -333,7 +439,13 @@ async def agent_run_stream(
                                           _err_detail = _ss(item.get('error', ''))
                                           _final_res = (f"\u26a0\ufe0f {_err_detail}" if _err_detail
                                                         else "\u26a0\ufe0f Tutti i provider AI sono temporaneamente indisponibili (rate limit). Riprova tra qualche minuto.")
-                                      yield f"data: {json.dumps({'type': 'task_done', 'taskId': task_id, 'result': _final_res, 'engine': item.get('engine', 'fallback'), 'success': item.get('success', False)})}\n\n"
+                                      _done_payload = {
+                                          'type': 'task_done', 'taskId': task_id, 'result': _final_res,
+                                          'engine': item.get('engine', 'fallback'), 'success': item.get('success', False),
+                                      }
+                                      if item.get('response_json') is not None:
+                                          _done_payload['response_json'] = _sanitize_for_json(item['response_json'])
+                                      yield f"data: {json.dumps(_done_payload)}\n\n"
                                       break
                 # S393 Priority 1: Narrative Streaming — arricchisce step_done con explanation
                 _NARR_QUICK = {
@@ -456,7 +568,19 @@ async def reason_loop(body: ReasonLoopIn, role: AuthRole = Depends(require_role(
                 'action': step_data.get('action', ''),
                 'output': str(step_data.get('output', ''))[:400],  # S577: 200→400
             })
-        result = await loop.run(goal=body.goal, context=context_str, max_steps=body.max_steps, on_step=_on_step, session_id=getattr(body, "session_id", "") or "")
+        if not await _LLM_ADMISSION.reserve():
+            raise _queue_full_error()
+        await _LLM_ADMISSION.acquire()
+        _provider_started = time.perf_counter()
+        try:
+            result = await loop.run(goal=body.goal, context=context_str, max_steps=body.max_steps, on_step=_on_step, session_id=getattr(body, "session_id", "") or "")
+        except Exception as exc:
+            await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000, 'error', type(exc).__name__)
+            raise
+        else:
+            await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000)
+        finally:
+            await _LLM_ADMISSION.release()
         if isinstance(result, dict):
             output_text = result.get('output', '') or ''
             engine_used = result.get('engine', 'unknown')
@@ -465,6 +589,7 @@ async def reason_loop(body: ReasonLoopIn, role: AuthRole = Depends(require_role(
             output_text = str(result)
             engine_used = 'unknown'
             errors_list = []
+        response_json = _validated_response_json(output_text)
         return {
             'ok':     bool(output_text and output_text.strip()),
             'success': bool(output_text and output_text.strip()),  # alias compat frontend
@@ -474,6 +599,7 @@ async def reason_loop(body: ReasonLoopIn, role: AuthRole = Depends(require_role(
             'engine': engine_used,
             'errors': errors_list,
             'steps':  _steps_log,  # N-2-FIX: step intermedi per debug/telemetria frontend
+            **({'response_json': response_json} if response_json is not None else {}),
         }
     except Exception as e:
         _logger.error("[reason/loop] Error: %s", e)
@@ -605,7 +731,7 @@ async def _create_task_internal(task_id: str, goal: str, job: dict) -> dict:
 
 
 @router.post('/api/agent/tasks')
-async def create_agent_task(body: AgentTaskIn, role: AuthRole = Depends(require_role(AuthRole.MACHINE))):  # GAP-1-fix
+async def create_agent_task(body: AgentTaskIn, request: Request, role: AuthRole = Depends(require_role(AuthRole.MACHINE))):  # GAP-1-fix
     """
     Crea o recupera un task agent.
 
@@ -613,11 +739,27 @@ async def create_agent_task(body: AgentTaskIn, role: AuthRole = Depends(require_
     il task viene ripristinato dallo store persistente invece di essere riavviato.
     Questo preserva lo stato SUCCESS/ERROR precedente senza sprecare token.
     """
+    await _record_phase('auth')
     _prune_agent_tasks()
-    task_id = body.taskId or str(uuid.uuid4())
+    raw_idempotency_key = (body.idempotency_key or request.headers.get('Idempotency-Key') or '').strip()
+    if len(raw_idempotency_key) > 200:
+        raise HTTPException(400, detail={'error': 'invalid_idempotency_key'})
+    task_id = body.taskId or (
+        str(uuid.uuid5(uuid.NAMESPACE_URL, f'baida98-ai:{body.session_id or ""}:{raw_idempotency_key}'))
+        if raw_idempotency_key else str(uuid.uuid4())
+    )
+    _idempotency_claimed = False
+    if raw_idempotency_key and task_id not in _agent_tasks:
+        # Claim before the first await: concurrent retries in this worker see the
+        # same task immediately and cannot create a second provider loop/write.
+        _agent_tasks[task_id] = {
+            'id': task_id, 'status': 'CREATING',
+            'idempotency_key': raw_idempotency_key,
+        }
+        _idempotency_claimed = True
 
     # Already in memory → return immediately (normal path, includes S358 reconnect)
-    if task_id in _agent_tasks:
+    if task_id in _agent_tasks and not _idempotency_claimed:
         if task_id not in _task_ai_clients:
             _attach_byok_client(task_id, body.provider_credentials)
         return {'taskId': task_id, 'status': _agent_tasks[task_id]['status']}
@@ -648,6 +790,7 @@ async def create_agent_task(body: AgentTaskIn, role: AuthRole = Depends(require_
         'resume_from_step':  body.resume_from_step, # P16-F3: passo resume dalla coda
         'persona':           body.persona,           # P17-F5: expertise persona hint
         'session_id':        body.session_id or '',   # P17-F2: BB session key (normalize None→'')
+        'idempotency_key':   raw_idempotency_key,
         'forbid_tools':       _tool_policy.forbid_tools,
         'literal_response':   _tool_policy.literal_response,
         'allow_local_csv_conversion': _tool_policy.allow_local_csv_conversion,
@@ -655,7 +798,6 @@ async def create_agent_task(body: AgentTaskIn, role: AuthRole = Depends(require_
     # Le credenziali BYOK restano in una mappa runtime separata dai metadata task
     # e non raggiungono Supabase, checkpoint o buffer SSE.
     _attach_byok_client(task_id, body.provider_credentials)
-
     # BG-4: restore cross-session handoff context (async, non-blocking)
     if body.session_id:
         _hctx = await sb_restore_handoff_context(body.session_id)
@@ -666,6 +808,7 @@ async def create_agent_task(body: AgentTaskIn, role: AuthRole = Depends(require_
     asyncio.create_task(
         sb_upsert_task(task_id, body.goal, 'QUEUED', body.max_steps, body.context, created_at)
     ).add_done_callback(_log_task_exc)
+    await _record_phase('persistence')
     # S361: gli strumenti speculativi sono consentiti solo quando il messaggio
     # utente non li vieta esplicitamente. La policy è fail-closed per questo task.
     if not _tool_policy.forbid_tools:
@@ -816,6 +959,7 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
       - Task era RUNNING   → replay buffer parziale + evento task_interrupted.
       - Task non trovato   → prova sb_restore_task prima di 404.
     """
+    await _record_phase('auth')
     # S359: se task_id non è in memoria, prova il restore da Supabase
     if task_id not in _agent_tasks:
         restored = await sb_restore_task(task_id)
@@ -952,6 +1096,18 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
                     yield "data: [DONE]\n\n"
                     return
         # ── Case 3: nuova esecuzione ──────────────────────────────────────────────
+        if not await _LLM_ADMISSION.reserve():
+            await _record_phase('queue', outcome='rejected', error_class='queue_full')
+            _agent_tasks[task_id]['status'] = 'RATE_LIMITED'
+            _rate_limited = json.dumps(_sanitize_for_json({
+                'event': 'task_error', 'taskId': task_id,
+                'statusCode': 429, 'retryAfter': 5,
+                'error': 'llm_queue_full',
+            }))
+            yield f'data: {_rate_limited}\n\n'
+            yield 'data: [DONE]\n\n'
+            return
+        await _record_phase('queue')
         _prune_loop_registry()
         reg_entry: dict = {
             'asyncio_task':      None,
@@ -1401,6 +1557,7 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
                     except Exception:
                         pass  # S364: skeleton injection is optional
 
+                _provider_started = time.perf_counter()
                 result = await loop.run(
                     goal=task['goal'],
                     context=context_str,
@@ -1408,9 +1565,9 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
                     on_step=step_cb,
                     session_id=task.get('session_id', '') or '',
                     allow_tools=not bool(task.get('forbid_tools', False)),
-                    allow_local_csv_conversion=bool(task.get('allow_local_csv_conversion', False)),
+                                        allow_local_csv_conversion=bool(task.get('allow_local_csv_conversion', False)),
                 )
-
+                await _record_phase('provider', (time.perf_counter() - _provider_started) * 1000)
                 # ARTIFACT-CONTRACT-FALLBACK: alcuni provider restituiscono codice HTML
                 # nella risposta finale dopo aver narrato una scrittura, senza produrre
                 # il tool event `file_written`. Non committiamo mai una falsa positività:
@@ -1579,7 +1736,13 @@ async def stream_agent_task(task_id: str, request: Request, resume: int = 0, rol
                     except Exception as _exc:
                         _logger.debug("[agent] silenced %s", type(_exc).__name__)  # noqa: BLE001
 
-        reg_entry['asyncio_task'] = asyncio.create_task(run_loop())
+        async def _admitted_run_loop():
+            await _LLM_ADMISSION.acquire()
+            try:
+                await run_loop()
+            finally:
+                await _LLM_ADMISSION.release()
+        reg_entry['asyncio_task'] = asyncio.create_task(_admitted_run_loop())
         reg_entry['asyncio_task'].add_done_callback(_log_task_exc)  # BUG-CB-2
 
         try:

api/agent_telemetry.py

@@ -34,6 +34,52 @@ _MAX_STORE  = 500  # max entry totali prima del pruning
 # lo stesso store e si sovrascriverebbero. asyncio.Lock() è safe a livello di modulo
 # in Python 3.10+ (non richiede event loop attivo all'init del modulo).
 _store_lock = asyncio.Lock()
+_RUNTIME_PHASES = ('auth', 'queue', 'provider', 'tool', 'persistence')
+_RUNTIME_MAX_SAMPLES = 200
+_runtime_lock = asyncio.Lock()
+_runtime_store: dict[str, dict] = {
+    phase: {'samples_ms': [], 'ok': 0, 'errors': {}}
+    for phase in _RUNTIME_PHASES
+}
+
+
+async def record_runtime_phase(phase: str, duration_ms: float = 0.0,
+                               outcome: str = 'ok', error_class: str | None = None) -> None:
+    if phase not in _runtime_store:
+        return
+    async with _runtime_lock:
+        bucket = _runtime_store[phase]
+        samples = bucket['samples_ms']
+        samples.append(max(0.0, round(float(duration_ms), 2)))
+        if len(samples) > _RUNTIME_MAX_SAMPLES:
+            del samples[:-_RUNTIME_MAX_SAMPLES]
+        if outcome == 'ok':
+            bucket['ok'] += 1
+        else:
+            key = error_class or outcome or 'unknown'
+            bucket['errors'][key] = bucket['errors'].get(key, 0) + 1
+
+
+def _percentile(samples: list[float], percentile: float) -> float:
+    if not samples:
+        return 0.0
+    ordered = sorted(samples)
+    index = min(len(ordered) - 1, int(round((percentile / 100) * (len(ordered) - 1))))
+    return ordered[index]
+
+
+async def runtime_snapshot() -> dict[str, dict]:
+    async with _runtime_lock:
+        return {
+            phase: {
+                'count': len(bucket['samples_ms']),
+                'ok': bucket['ok'],
+                'errors': dict(bucket['errors']),
+                'p50_ms': _percentile(bucket['samples_ms'], 50),
+                'p95_ms': _percentile(bucket['samples_ms'], 95),
+            }
+            for phase, bucket in _runtime_store.items()
+        }
 
 # ─── Store helpers ────────────────────────────────────────────────────────────
 
@@ -115,6 +161,11 @@ class TelemetrySyncBody(BaseModel):
 
 # ─── Endpoints ────────────────────────────────────────────────────────────────
 
+@router.get("/api/agent-telemetry/runtime")
+async def get_runtime_telemetry() -> JSONResponse:
+    return JSONResponse({'ok': True, 'phases': await runtime_snapshot(), 'server_ts': int(time.time() * 1000)})
+
+
 @router.post("/api/agent-telemetry/sync")
 async def post_agent_telemetry(body: TelemetrySyncBody) -> JSONResponse:
     """

api/auth_managed.py

@@ -11,7 +11,7 @@ I CLIENT_ID/SECRET vanno nei secret HF Spaces (mai nel codice).
 
 Cifratura token: Fernet(PBKDF2HMAC-SHA256, 260k iter, salt fisso) — richiede cryptography>=42.
 """
-import os, time, secrets, json, logging, asyncio
+import os, time, secrets, json, logging, asyncio, hashlib
 from typing import Optional
 from fastapi import APIRouter, Depends, Request, HTTPException
 from .auth_guard import require_role, AuthRole
@@ -21,6 +21,29 @@ import httpx
 _logger = logging.getLogger('api.auth_managed')
 router  = APIRouter()
 
+_DIAGNOSTIC_SECRET_NAMES = (
+    'ADMIN_DIAGNOSTICS_TOKEN', 'INTERNAL_TOKEN', 'PUBLIC_API_TOKEN',
+    'PRIVATE_STATE_INTERNAL_TOKEN', 'SUPABASE_SERVICE_ROLE_KEY',
+    'SUPABASE_SERVICE_ROLE_KEY_B', 'GROQ_API_KEY', 'HF_TOKEN',
+    'GEMINI_API_KEY', 'CEREBRAS_API_KEY', 'SAMBANOVA_API_KEY',
+    'OPENROUTER_API_KEY', 'OPENROUTER_API_KEY_B', 'OPENROUTER_API_KEY_C',
+    'OPENROUTER_PROFILES_JSON', 'NVIDIA_API_KEY', 'NVIDIA_API_KEY_B',
+)
+
+
+def _secret_fingerprint(name: str) -> dict[str, object]:
+    value = os.getenv(name, '').strip()
+    return {
+        'configured': bool(value),
+        'length': len(value),
+        'sha256': hashlib.sha256(value.encode('utf-8')).hexdigest()[:16] if value else None,
+    }
+
+
+@router.get('/api/admin/secret-fingerprint')
+async def secret_fingerprint(role: AuthRole = Depends(require_role(AuthRole.MACHINE))):
+    return {'secrets': {name: _secret_fingerprint(name) for name in _DIAGNOSTIC_SECRET_NAMES}}
+
 # ── Fernet encryption setup ──────────────────────────────────────────────────
 # Salt statico pubblico: accettabile per chiave macchina (non password utente).
 # Il VAULT_KEY è il segreto; il salt previene rainbow-table cross-application.

api/state.py

@@ -319,6 +319,7 @@ class AgentTaskIn(BaseModel):
     context: list[dict] = []
     max_steps: int = 8
     taskId: Optional[str] = None
+    idempotency_key: Optional[str] = None
     project_context: str = ""
     learning_hints: list[str] = []
     session_id: Optional[str] = None

api/webhook.py

@@ -1,6 +1,27 @@
 """backend/api/webhook.py — Webhook inbound + Public REST API (S354)."""
 import os, asyncio
 from typing import Optional
+
+# Public agent execution is deliberately bounded: each agent run fans out to
+# several provider calls, so unbounded HTTP concurrency can exhaust a 1-worker
+# Space and cause the peer to reset connections. The limit is configurable but
+# capped to keep deployments safe.
+_PUBLIC_AGENT_CONCURRENCY = max(1, min(int(os.getenv("PUBLIC_AGENT_CONCURRENCY", "3")), 8))
+_PUBLIC_AGENT_QUEUE_TIMEOUT = max(5.0, float(os.getenv("PUBLIC_AGENT_QUEUE_TIMEOUT", "90")))
+_PUBLIC_AGENT_TIMEOUT = max(15.0, float(os.getenv("PUBLIC_AGENT_TIMEOUT", "90")))
+_public_agent_slots = asyncio.Semaphore(_PUBLIC_AGENT_CONCURRENCY)
+_public_agent_client = None
+_public_agent_client_lock = asyncio.Lock()
+
+async def _get_public_agent_client():
+    """Create AIClient once per worker, off the event loop, and reuse it."""
+    global _public_agent_client
+    if _public_agent_client is None:
+        async with _public_agent_client_lock:
+            if _public_agent_client is None:
+                from models.ai_client import AIClient
+                _public_agent_client = await asyncio.to_thread(AIClient, None, True)
+    return _public_agent_client
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, ValidationError, field_validator
 from .state import _get_mem_manager, _get_executor, _get_planner
@@ -271,7 +292,7 @@ async def inbound_webhook(webhook_token: str, request: Request):
                 loop.run(goal=body.goal, context=context_str,
                          max_steps=body.max_steps, on_step=lambda _s: None,
                          allow_tools=not _task_policy.forbid_tools),
-                timeout=float(os.getenv('AGENT_STREAM_TIMEOUT', '120')),
+                timeout=min(float(os.getenv('AGENT_STREAM_TIMEOUT', '120')), _PUBLIC_AGENT_TIMEOUT),
             )
         except asyncio.TimeoutError:
             asyncio.ensure_future(_tg_error(_wh_task_id, body.goal, "Timeout: task terminato dopo 120s"))
@@ -306,7 +327,7 @@ async def public_chat(payload: PublicChatPayload, request: Request):
     S292 — API REST pubblica autenticata per integrazioni esterne.
     Auth: Authorization: Bearer <PUBLIC_API_TOKEN>
     """
-    _expected = (os.getenv('PUBLIC_API_TOKEN') or os.getenv('INTERNAL_TOKEN', '')).strip()
+    _expected = (os.getenv('PUBLIC_API_TOKEN') or os.getenv('INTERNAL_TOKEN')).strip()
     if not _expected:
         raise HTTPException(
             status_code=503,
@@ -324,10 +345,16 @@ async def public_chat(payload: PublicChatPayload, request: Request):
             'conversation_id': payload.conversation_id, 'steps': 0,
         }
 
+    acquired = False
     try:
+        try:
+            await asyncio.wait_for(_public_agent_slots.acquire(), timeout=_PUBLIC_AGENT_QUEUE_TIMEOUT)
+            acquired = True
+        except asyncio.TimeoutError as exc:
+            raise HTTPException(429, detail="Server occupato: riprova tra poco.", headers={"Retry-After": "5"}) from exc
+
         from agents.unified_loop import UnifiedAgentLoop
-        from models.ai_client import AIClient
-        client = AIClient()
+        client = await _get_public_agent_client()
         try:
             from agents.critic import Critic
             from agents.response_verifier import ResponseVerifier
@@ -347,7 +374,7 @@ async def public_chat(payload: PublicChatPayload, request: Request):
                 loop.run(goal=payload.message, context='',
                          max_steps=payload.max_steps, on_step=lambda _s: None,
                          allow_tools=not _task_policy.forbid_tools),
-                timeout=float(os.getenv('AGENT_STREAM_TIMEOUT', '120')),
+                timeout=min(float(os.getenv('AGENT_STREAM_TIMEOUT', '120')), _PUBLIC_AGENT_TIMEOUT),
             )
         except asyncio.TimeoutError:
             asyncio.ensure_future(_tg_error(_pc_task_id, payload.message, "Timeout dopo 120s"))
@@ -368,3 +395,6 @@ async def public_chat(payload: PublicChatPayload, request: Request):
         raise
     except Exception as exc:
         raise HTTPException(status_code=500, detail=f'Errore agente: {exc}')
+    finally:
+        if acquired:
+            _public_agent_slots.release()

models/ai_client.py

@@ -74,9 +74,10 @@ _PROVIDER_DEFS = [
 
 
 class AIClient:
-    def __init__(self, byok_credentials: dict[str, list[str]] | None = None) -> None:
+    def __init__(self, byok_credentials: dict[str, list[str]] | None = None, single_provider: bool = False) -> None:
         # Le chiavi BYOK appartengono a un singolo task e vivono solo in questa
         # istanza: non vengono scritte in env, Supabase, cache semantica o log.
+        self._single_provider = single_provider
         self._byok_providers = self._providers_from_byok(byok_credentials or {})
         # I profili BYOK precedono i provider runtime: la stessa API conserva
         # comunque il fallback server-side in caso di quota o errore upstream.
@@ -447,6 +448,11 @@ class AIClient:
         # profili condividano quota e client, mentre provider diversi restano
         # disponibili come ensemble/fallback.
         pool = self._execution_pool(pool, primary_purpose)
+        # Public API requests use one provider at a time. The previous ensemble
+        # fan-out multiplied outbound connections per request and exhausted the
+        # small Space under concurrent load. Fallback remains available below.
+        if self._single_provider:
+            pool = pool[:1]
         results = []
         if pool:
             tasks = [self._fetch_one(p, messages, temperature, max_tokens) for p in pool]