api/persistence.py

"""backend/api/persistence.py — Supabase persistence for agent task state (S359).

Design principles:
  - ALL writes are fire-and-forget: never block the SSE hot path.
  - Reads happen ONLY on lazy restore (when in-memory is empty, i.e. backend restarted).
  - Supabase unavailable → silent fallback to in-memory-only (no exception propagation).
  - Event buffer capped at MAX_EVENTS per task to avoid runaway storage.

GAP-P40D-FIX: aggiunto retry (max 2 tentativi, 0.3s sleep) per sb_upsert_task e
sb_append_event — stesse operazioni già protette in linter.py (P40-D pattern).
Il gap: su Railway cold-start le prime scritture Supabase cadono nel vuoto senza retry.
Il pattern è identico a quello in linter.py (lint_and_store) già confermato stabile.

Required Supabase tables (run backend/migrations/s359_task_persistence.sql once):
  agent_tasks        — task metadata (task_id, goal, status, max_steps, created_at, updated_at)
  agent_task_events  — SSE event buffer (task_id, event_index, event_data, created_at)
"""
import asyncio, time, json
from .state import safe_json_dumps as _sjd  # B11-FIX: surrogate-safe drop-in
from typing import Optional

import logging
_logger = logging.getLogger("api.persistence")

MAX_EVENTS = 500  # max SSE frames persisted per task

_MAX_RETRY   = 2    # GAP-P40D-FIX: tentativi massimi per write Supabase
_RETRY_SLEEP = 0.3  # GAP-P40D-FIX: sleep tra tentativi (secondi)


# ── Write helpers (fire-and-forget, never raise) ───────────────────────────────

async def sb_upsert_task(
    task_id: str, goal: str, status: str, max_steps: int,
    context: list, created_at: int,
) -> None:
    """Persist or update task metadata. Silent if Supabase not configured.
    GAP-P40D-FIX: retry fino a _MAX_RETRY tentativi su errore transitorio."""
    from .state import _sb
    if not _sb:
        return
    now = int(time.time() * 1000)
    payload = {
        'task_id':    task_id,
        'goal':       goal[:1000],
        'status':     status,
        'max_steps':  max_steps,
        'context':    _sjd(context)[:8000],
        'created_at': created_at,
        'updated_at': now,
    }
    # ─── Mirroring su Hugging Face Dataset (Zero-Cost Backup) ─────────────
    try:
        from .hf_storage import hf_fire_and_forget
        hf_fire_and_forget("tasks.jsonl", payload)
    except Exception as hf_exc:
        _logger.debug("HF Mirroring (task) silenced: %s", hf_exc)

    for _attempt in range(_MAX_RETRY):
        try:
            await asyncio.to_thread(
                lambda: _sb.table('agent_tasks').upsert(payload).execute()
            )
            return  # successo
        except Exception as e:
            if _attempt < _MAX_RETRY - 1:
                await asyncio.sleep(_RETRY_SLEEP)
            else:
                _logger.warning('[persist] upsert_task %s failed after %d attempts: %s',
                                task_id, _MAX_RETRY, e)


async def sb_update_status(task_id: str, status: str) -> None:
    """Update only the status column of an existing task."""
    from .state import _sb
    if not _sb:
        return
    now = int(time.time() * 1000)
    try:
        await asyncio.to_thread(
            lambda: _sb.table('agent_tasks')
                .update({'status': status, 'updated_at': now})
                .eq('task_id', task_id)
                .execute()
        )
    except Exception as e:
        _logger.warning('[persist] update_status %s→%s: %s', task_id, status, e)


async def sb_append_event(task_id: str, event_index: int, event_data: str) -> None:
    """Append one SSE event string to the persistent buffer. Capped at MAX_EVENTS.
    GAP-P40D-FIX: retry fino a _MAX_RETRY tentativi su errore transitorio."""
    from .state import _sb
    if not _sb or event_index > MAX_EVENTS:
        return
    now = int(time.time() * 1000)
    payload = {
        'task_id':     task_id,
        'event_index': event_index,
        'event_data':  event_data,
        'created_at':  now,
    }
    # ─── Mirroring su Hugging Face Dataset (Zero-Cost Backup) ─────────────
    try:
        from .hf_storage import hf_fire_and_forget
        # Sharding: i log degli eventi vanno su un dataset potenzialmente diverso (Account B)
        hf_fire_and_forget("task_events.jsonl", payload, repo_id=os.getenv("HF_DATASET_REPO_LOGS"))
    except Exception as hf_exc:
        _logger.debug("HF Mirroring (event) silenced: %s", hf_exc)

    for _attempt in range(_MAX_RETRY):
        try:
            await asyncio.to_thread(
                lambda: _sb.table('agent_task_events').upsert(payload).execute()
            )
            return  # successo
        except Exception as e:
            if _attempt < _MAX_RETRY - 1:
                await asyncio.sleep(_RETRY_SLEEP)
            else:
                _logger.warning('[persist] append_event %s#%d failed after %d attempts: %s',
                                task_id, event_index, _MAX_RETRY, e)


async def sb_restore_task(task_id: str) -> Optional[dict]:
    """Restore task metadata from Supabase on backend restart."""
    from .state import _sb
    if not _sb:
        return None
    try:
        res = await asyncio.to_thread(
            lambda: _sb.table('agent_tasks').select('*').eq('task_id', task_id).limit(1).execute()
        )
        return res.data[0] if res.data else None
    except Exception as e:
        _logger.warning('[persist] restore_task %s: %s', task_id, e)
        return None


async def sb_get_events(task_id: str) -> list[str]:
    """Retrieve all persisted SSE events for a task, ordered by index."""
    from .state import _sb
    if not _sb:
        return []
    try:
        res = await asyncio.to_thread(
            lambda: _sb.table('agent_task_events')
                .select('event_index, event_data')
                .eq('task_id', task_id)
                .order('event_index')
                .limit(MAX_EVENTS)
                .execute()
        )
        return [r['event_data'] for r in (res.data or [])]
    except Exception as e:
        _logger.warning('[persist] get_events %s: %s', task_id, e)
        return []


async def sb_delete_task_events(task_id: str) -> None:
    """Delete all events for a task (post-replay cleanup)."""
    from .state import _sb
    if not _sb:
        return
    try:
        await asyncio.to_thread(
            lambda: _sb.table('agent_task_events').delete().eq('task_id', task_id).execute()
        )
    except Exception as e:
        _logger.debug('[persist] delete_events %s: %s', task_id, e)


async def sb_list_tasks(limit: int = 50) -> list[dict]:
    """List recent tasks from Supabase."""
    from .state import _sb
    if not _sb:
        return []
    try:
        res = await asyncio.to_thread(
            lambda: _sb.table('agent_tasks')
                .select('task_id, goal, status, created_at, updated_at')
                .order('created_at', desc=True)
                .limit(limit)
                .execute()
        )
        return res.data or []
    except Exception as e:
        _logger.warning('[persist] list_tasks: %s', e)
        return []


# ── Checkpoint helpers (S359: task state snapshots) ───────────────────────────

async def sb_save_checkpoint(task_id: str, step: int, checkpoint_data: dict) -> None:
    """Save a mid-task checkpoint for potential resume."""
    from .state import _sb
    if not _sb:
        return
    now = int(time.time() * 1000)
    try:
        await asyncio.to_thread(
            lambda: _sb.table('agent_tasks')
                .update({'checkpoint': _sjd(checkpoint_data)[:16000], 'updated_at': now})
                .eq('task_id', task_id)
                .execute()
        )
    except Exception as e:
        _logger.debug('[persist] save_checkpoint %s#%d: %s', task_id, step, e)


async def sb_get_checkpoint(task_id: str) -> Optional[dict]:
    """Retrieve latest checkpoint for a task."""
    from .state import _sb
    if not _sb:
        return None
    try:
        res = await asyncio.to_thread(
            lambda: _sb.table('agent_tasks')
                .select('checkpoint')
                .eq('task_id', task_id)
                .limit(1)
                .execute()
        )
        if res.data and res.data[0].get('checkpoint'):
            raw = res.data[0]['checkpoint']
            return json.loads(raw) if isinstance(raw, str) else raw
        return None
    except Exception as e:
        _logger.debug('[persist] get_checkpoint %s: %s', task_id, e)
        return None


# ── Handoff helpers (BG-4: cross-session context handoff) ─────────────────────

async def sb_restore_handoff_context(session_id: str) -> Optional[dict]:
    """Restore cross-session handoff context."""
    from .state import _sb
    if not _sb:
        return None
    try:
        res = await asyncio.to_thread(
            lambda: _sb.table('handoff_contexts')
                .select('*')
                .eq('session_id', session_id)
                .limit(1)
                .execute()
        )
        return res.data[0] if res.data else None
    except Exception as e:
        _logger.debug('[persist] restore_handoff %s: %s', session_id, e)
        return None


async def sb_upsert_handoff(session_id: str, context: dict) -> None:
    """Save handoff context for cross-session resume."""
    from .state import _sb
    if not _sb:
        return
    now = int(time.time() * 1000)
    try:
        await asyncio.to_thread(
            lambda: _sb.table('handoff_contexts').upsert({
                'session_id': session_id,
                'context':    _sjd(context)[:16000],
                'updated_at': now,
            }, on_conflict='session_id').execute()
        )
    except Exception as e:
        _logger.debug('[persist] upsert_handoff %s: %s', session_id, e)


async def sb_delete_handoff(session_id: str) -> None:
    """Remove handoff context after successful resume."""
    from .state import _sb
    if not _sb:
        return
    try:
        await asyncio.to_thread(
            lambda: _sb.table('handoff_contexts').delete().eq('session_id', session_id).execute()
        )
    except Exception as e:
        _logger.debug('[persist] delete_handoff %s: %s', session_id, e)