sync: 193 file da Baida98/AI@8845b81e (2026-08-29 21:08 UTC)

api/public_snapshot.py

@@ -0,0 +1,74 @@
+"""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
+from typing import Any
+
+from .state import _agent_tasks, _loop_registry, sb
+from .version import RUNTIME_VERSION
+
+_logger = logging.getLogger("agente_ai.public_snapshot")
+
+_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 = sb()
+    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, 4):
+        try:
+            await asyncio.to_thread(operation)
+            _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 < 3:
+                await asyncio.sleep(2)
+
+    _logger.warning(
+        "BOOT: public snapshot upsert failed after retries (%s); public status remains degraded",
+        type(last_error).__name__ if last_error else "unknown",
+    )
+    return False
+
+
+async def refresh_public_dashboard_snapshot() -> bool:
+    """Alias for future heartbeat refreshes; intentionally idempotent."""
+    return await write_public_dashboard_snapshot()

main.py

@@ -88,7 +88,24 @@ async def _run_auto_migration():
         TO service_role
         USING (true)
         WITH CHECK (true);
-    -- 5. Healthcheck function
+    -- 5. Public dashboard snapshot used by /api/public/status
+    CREATE TABLE IF NOT EXISTS public.public_dashboard_snapshot (
+        singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
+        service_status text NOT NULL DEFAULT 'operational',
+        active_sessions integer NOT NULL DEFAULT 0,
+        queued_tasks integer NOT NULL DEFAULT 0,
+        in_progress_tasks integer NOT NULL DEFAULT 0,
+        app_version text,
+        updated_at timestamptz NOT NULL DEFAULT now()
+    );
+    ALTER TABLE public.public_dashboard_snapshot ENABLE ROW LEVEL SECURITY;
+    GRANT SELECT ON public.public_dashboard_snapshot TO anon, authenticated;
+    GRANT SELECT, INSERT, UPDATE, DELETE ON public.public_dashboard_snapshot TO service_role;
+    DROP POLICY IF EXISTS "public_dashboard_snapshot_read" ON public.public_dashboard_snapshot;
+    CREATE POLICY "public_dashboard_snapshot_read" ON public.public_dashboard_snapshot
+        FOR SELECT TO anon, authenticated USING (true);
+
+    -- 6. Healthcheck function
     CREATE OR REPLACE FUNCTION public.health_check()
     RETURNS jsonb AS $$
     BEGIN
@@ -236,6 +253,12 @@ async def startup_event():
     except Exception as e:
         _logger.warning(f"⚠️ BOOT: apply_rls_fix_sync() fallito (non bloccante): {e}")
     asyncio.create_task(_run_auto_migration())
+    try:
+        from api.public_snapshot import write_public_dashboard_snapshot
+        asyncio.create_task(write_public_dashboard_snapshot())
+        _logger.info("✅ BOOT: public dashboard snapshot writer avviato.")
+    except Exception as e:
+        _logger.warning(f"⚠️ BOOT: avvio public snapshot writer fallito (non bloccante): {e}")
     try:
         from api.providers import start_heartbeat
         start_heartbeat()

tests/test_public_snapshot_writer.py

@@ -0,0 +1,58 @@
+import asyncio
+
+from api import public_snapshot
+
+
+class _FakeTable:
+    def __init__(self, calls):
+        self.calls = calls
+
+    def upsert(self, row, on_conflict=None):
+        self.calls.append((row, on_conflict))
+        return self
+
+    def execute(self):
+        return object()
+
+
+class _FakeClient:
+    def __init__(self):
+        self.calls = []
+
+    def table(self, name):
+        assert name == "public_dashboard_snapshot"
+        return _FakeTable(self.calls)
+
+
+def test_snapshot_writer_is_idempotent_and_counts_runtime_state(monkeypatch):
+    fake = _FakeClient()
+    monkeypatch.setattr(public_snapshot, "sb", lambda: fake)
+    monkeypatch.setattr(
+        public_snapshot,
+        "_agent_tasks",
+        {
+            "queued": {"status": "QUEUED"},
+            "running": {"status": "RUNNING"},
+            "done": {"status": "SUCCESS"},
+        },
+    )
+    monkeypatch.setattr(public_snapshot, "_loop_registry", {"session-1": object()})
+
+    assert asyncio.run(public_snapshot.write_public_dashboard_snapshot()) is True
+    assert asyncio.run(public_snapshot.write_public_dashboard_snapshot()) is True
+
+    assert len(fake.calls) == 2
+    first, second = fake.calls
+    assert first[1] == "singleton"
+    assert second[1] == "singleton"
+    assert first[0] == second[0]
+    assert first[0]["singleton"] is True
+    assert first[0]["active_sessions"] == 1
+    assert first[0]["queued_tasks"] == 1
+    assert first[0]["in_progress_tasks"] == 1
+    assert first[0]["service_status"] == "operational"
+
+
+def test_snapshot_writer_is_non_blocking_when_supabase_is_unavailable(monkeypatch):
+    monkeypatch.setattr(public_snapshot, "sb", lambda: None)
+    assert asyncio.run(public_snapshot.write_public_dashboard_snapshot()) is False