Spaces:
Running
Running
File size: 4,922 Bytes
047d1bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """Writer idempotente per lo snapshot pubblico operativo.
Lo snapshot è un DTO sanificato e singleton: viene aggiornato dal backend con
service_role e letto dall'endpoint pubblico senza esporre task, sessioni o log.
Un errore di persistenza non deve impedire l'avvio del servizio.
"""
from __future__ import annotations
import asyncio
import logging
import os
from typing import Any
from .state import _agent_tasks, _loop_registry
try:
from supabase import create_client
except ImportError: # pragma: no cover - deployment dependency guard
create_client = None
_snapshot_client: Any | None = None
_snapshot_client_initialized = False
def get_snapshot_client() -> Any | None:
"""Return one stable Supabase client for the canonical snapshot project.
Unlike the general round-robin pool, this client never changes project
between a write and a read. Dedicated variables can be used when the
snapshot lives in a separate project; otherwise project A is canonical.
"""
global _snapshot_client, _snapshot_client_initialized
if _snapshot_client_initialized:
return _snapshot_client
_snapshot_client_initialized = True
if create_client is None:
_logger.warning("BOOT: public snapshot skipped — supabase package unavailable")
return None
url = os.getenv("PUBLIC_SNAPSHOT_SUPABASE_URL") or os.getenv("SUPABASE_URL")
key = (
os.getenv("PUBLIC_SNAPSHOT_SUPABASE_SERVICE_ROLE_KEY")
or os.getenv("SUPABASE_SERVICE_ROLE_KEY")
or os.getenv("SUPABASE_KEY")
)
if not url or not key:
_logger.warning("BOOT: public snapshot skipped — canonical Supabase URL/key unavailable")
return None
try:
_snapshot_client = create_client(url, key)
_logger.info("BOOT: canonical Supabase snapshot client initialized")
except Exception as exc:
_logger.warning("BOOT: canonical Supabase snapshot client init failed type=%s", type(exc).__name__)
return _snapshot_client
from .version import RUNTIME_VERSION
_logger = logging.getLogger("agente_ai.public_snapshot")
_SNAPSHOT_ATTEMPTS = 3
_SNAPSHOT_ATTEMPT_TIMEOUT_SECONDS = 4.0
_SNAPSHOT_RETRY_DELAY_SECONDS = 1.0
_SNAPSHOT_TOTAL_BUDGET_SECONDS = (
_SNAPSHOT_ATTEMPTS * _SNAPSHOT_ATTEMPT_TIMEOUT_SECONDS
+ (_SNAPSHOT_ATTEMPTS - 1) * _SNAPSHOT_RETRY_DELAY_SECONDS
)
_ACTIVE_STATUSES = {"RUNNING", "IN_PROGRESS", "EXECUTING", "PROCESSING"}
def _snapshot_row() -> dict[str, Any]:
tasks = list(_agent_tasks.values())
return {
"singleton": True,
"service_status": "operational",
"active_sessions": len(_loop_registry),
"queued_tasks": sum(1 for task in tasks if str(task.get("status", "")).upper() == "QUEUED"),
"in_progress_tasks": sum(1 for task in tasks if str(task.get("status", "")).upper() in _ACTIVE_STATUSES),
"app_version": RUNTIME_VERSION,
}
async def write_public_dashboard_snapshot() -> bool:
"""Upsert the singleton public snapshot without blocking application startup.
Returns ``False`` on missing Supabase or a schema/permission error. The
public status route remains available with its degraded response while the
service continues booting; the exact exception is kept in backend logs.
"""
client = get_snapshot_client()
if client is None:
_logger.warning("BOOT: public snapshot skipped — Supabase client unavailable")
return False
row = _snapshot_row()
def operation() -> None:
client.table("public_dashboard_snapshot").upsert(row, on_conflict="singleton").execute()
last_error: Exception | None = None
for attempt in range(1, _SNAPSHOT_ATTEMPTS + 1):
try:
await asyncio.wait_for(
asyncio.to_thread(operation),
timeout=_SNAPSHOT_ATTEMPT_TIMEOUT_SECONDS,
)
_logger.info(
"BOOT: public snapshot upserted status=%s sessions=%d queued=%d in_progress=%d version=%s",
row["service_status"], row["active_sessions"], row["queued_tasks"],
row["in_progress_tasks"], row["app_version"],
)
return True
except Exception as exc:
last_error = exc
if attempt < _SNAPSHOT_ATTEMPTS:
await asyncio.sleep(_SNAPSHOT_RETRY_DELAY_SECONDS)
error_code = getattr(last_error, "code", None) or getattr(last_error, "status_code", None)
_logger.warning(
"BOOT: public snapshot upsert failed after retries type=%s code=%s; public status remains degraded",
type(last_error).__name__ if last_error else "unknown",
str(error_code)[:32] if error_code is not None else "unknown",
)
return False
async def refresh_public_dashboard_snapshot() -> bool:
"""Alias for future heartbeat refreshes; intentionally idempotent."""
return await write_public_dashboard_snapshot()
|