ai-memory-backend / api /agent_checkpoint.py
Baida07's picture
ab-p1-sync: ab-unlock-p1 (#8)
b2b3de9
Raw
History Blame Contribute Delete
5.3 kB
"""
backend/api/agent_checkpoint.py β€” Simplified checkpoint endpoints (ARCH-K2.3)
Aggiunge alias /api/agent/checkpoint (senza task_id nella path) per uso diretto dal frontend:
GET /api/agent/checkpoint β€” lista tutti i checkpoint attivi in memoria
POST /api/agent/checkpoint β€” salva checkpoint (taskId opzionale nel body)
GET /api/agent/checkpoint/{task_id} β€” recupera checkpoint specifico
DELETE /api/agent/checkpoint/{task_id} β€” elimina checkpoint
I checkpoint per-task esistono giΓ  su /api/agent/tasks/{id}/checkpoint (agent.py).
Questi alias sono piΓΉ comodi quando il frontend non ha un task_id esplicito
(es. salvataggio periodico dello stato dell'agente, resume dopo refresh).
ROUTING CF PAGES: /api/agent/* β†’ HANDS (Space B) via HANDS_PATTERNS[0].
Nessuna modifica a [[catchall]].ts necessaria.
NOTA: Import da api.agent e api.persistence sono LAZY (dentro le funzioni)
per evitare import circolari β€” agent.py importa giΓ  molti altri moduli.
"""
import time
import asyncio
import logging
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from .auth_guard import require_role, AuthRole
_logger = logging.getLogger("api.agent_checkpoint")
router = APIRouter(dependencies=[Depends(require_role(AuthRole.MACHINE))])
class CheckpointBody(BaseModel):
taskId: Optional[str] = None # se omesso β†’ usa "default"
step: int = 0
goal: str = ""
plan: list = []
logs: list[str] = []
artifacts: list[str] = []
retryCount: int = 0
extra: dict = {}
# ── GET /api/agent/checkpoint ─────────────────────────────────────────────────
@router.get("/api/agent/checkpoint")
async def list_checkpoints_alias():
"""
Lista tutti i checkpoint attivi in memoria.
Alias leggero per /api/agent/checkpoints (agent.py).
"""
# Import lazy β€” evita circolaritΓ 
from api.agent import _task_checkpoints, _prune_checkpoints # type: ignore[import]
_prune_checkpoints()
now = int(time.time() * 1000)
return {
"count": len(_task_checkpoints),
"checkpoints": [
{
"taskId": k,
"step": v.get("step", 0),
"goal": v.get("goal", "")[:300],
"age_ms": now - v.get("savedAt", now),
}
for k, v in _task_checkpoints.items()
],
}
# ── POST /api/agent/checkpoint ────────────────────────────────────────────────
@router.post("/api/agent/checkpoint")
async def save_checkpoint_alias(body: CheckpointBody):
"""
Salva un checkpoint. taskId opzionale: se omesso usa 'default'.
Replica la logica di /api/agent/tasks/{id}/checkpoint con Supabase persist.
"""
from api.agent import _task_checkpoints, _prune_checkpoints # type: ignore[import]
from api.persistence import sb_save_checkpoint # type: ignore[import]
_prune_checkpoints()
task_id = body.taskId or "default"
cp: dict = {
"taskId": task_id,
"step": body.step,
"goal": body.goal,
"plan": body.plan,
"logs": body.logs[-50:], # mantieni solo gli ultimi 50 log
"artifacts": body.artifacts,
"retryCount": body.retryCount,
"extra": body.extra,
"savedAt": int(time.time() * 1000),
}
_task_checkpoints[task_id] = cp
# Persist su Supabase β€” fire-and-forget (stesso pattern di agent.py)
asyncio.create_task(sb_save_checkpoint(task_id, body.step, cp))
return {"saved": True, "taskId": task_id, "step": body.step}
# ── GET /api/agent/checkpoint/{task_id} ──────────────────────────────────────
@router.get("/api/agent/checkpoint/{task_id}")
async def get_checkpoint_alias(task_id: str):
"""
Recupera il checkpoint per un task specifico.
Cerca prima in memoria (_task_checkpoints), poi su Supabase via sb_get_checkpoint.
"""
from api.agent import _task_checkpoints, _prune_checkpoints # type: ignore[import]
from api.persistence import sb_get_checkpoint # type: ignore[import]
_prune_checkpoints()
cp = _task_checkpoints.get(task_id)
if not cp:
cp = await sb_get_checkpoint(task_id)
if not cp:
raise HTTPException(
status_code=404,
detail={"error": "checkpoint_not_found", "taskId": task_id},
)
return cp
# ── DELETE /api/agent/checkpoint/{task_id} ───────────────────────────────────
@router.delete("/api/agent/checkpoint/{task_id}")
async def delete_checkpoint_alias(task_id: str):
"""Rimuove il checkpoint da memoria in-process (non elimina da Supabase)."""
from api.agent import _task_checkpoints # type: ignore[import]
_task_checkpoints.pop(task_id, None)
return {"deleted": task_id}