""" Replay API — start/step/seek/reset replay sessions against the timeline journal. """ from fastapi import APIRouter from pydantic import BaseModel from typing import Optional, Any, Dict from backend.analysis.timeline_store import STORE as TIMELINE_STORE from backend.session.db import SessionDB router = APIRouter(prefix="/replay", tags=["replay"]) # Process-wide session DB (now SQLite-backed) SESSION_DB = SessionDB() class StartReplayBody(BaseModel): journal: Optional[str] = None # session id or "latest" session_id: Optional[str] = None # ------------------------- # START REPLAY SESSION # ------------------------- @router.post("/start") def start_replay(body: StartReplayBody): """Start a new replay session. Snapshots the current live timeline into a new session id, so the session has its own immutable event list. F-07 remediation: the original code created the session row first and then logged events one-by-one in a separate loop. If the process crashed mid-loop the session row was left behind with a partial event log (and the QA noted "no rollback guard"). This version snapshots the events *before* creating the session row, creates the session, and rolls back the session if any log_event call fails — so the database is never left with a half-populated session. """ import logging import uuid log = logging.getLogger(__name__) # Step 1: snapshot the live timeline BEFORE creating the session row. # This is the "patch the source" the original comment referred to — # done up-front so the session is never half-populated. try: events = TIMELINE_STORE.all() except Exception as e: log.exception("timeline snapshot failed") return { "session_id": None, "event_count": 0, "events_available": 0, "error": f"timeline_snapshot_failed:{type(e).__name__}", } # Step 2: create the session row. sid = SESSION_DB.create_session(body.journal or "live") # Step 3: log each event. If any log_event call fails, roll back # the session so we don't leave a half-populated row in storage. logged = 0 try: for event in events: SESSION_DB.log_event(sid, event.get("type", "event"), event) logged += 1 except Exception as e: log.exception("log_event failed mid-session; rolling back %s", sid) # Best-effort rollback — delete the session row and its # partially-written events from the in-memory cache + SQLite. try: SESSION_DB.sessions.pop(sid, None) SESSION_DB.events.pop(sid, None) SESSION_DB.snapshots.pop(sid, None) except Exception: pass return { "session_id": sid, "event_count": logged, "events_available": len(events), "error": f"session_rolled_back:{type(e).__name__}", } return { "session_id": sid, "event_count": logged, "events_available": len(events), } # ------------------------- # GET EVENTS (full journal snapshot) # ------------------------- @router.get("/events") def get_events(): """Return the full current journal. Used by the replay UI on load.""" return {"events": TIMELINE_STORE.all()} # ------------------------- # GET EVENTS FOR A SPECIFIC SESSION # ------------------------- @router.get("/events/{session_id}") def get_session_events(session_id: str): return {"session_id": session_id, "events": SESSION_DB.get_replay_stream(session_id)} # ------------------------- # STEP FORWARD # ------------------------- @router.post("/step") def replay_step(session_id: str): events = SESSION_DB.get_replay_stream(session_id) return { "session_id": session_id, "event_count": len(events), "events": events[:1], } # ------------------------- # SEEK # ------------------------- @router.post("/seek") def replay_seek(session_id: str, timestamp: float): events = SESSION_DB.get_replay_stream(session_id) best = None for e in events: ts = e.get("ts") or e.get("timestamp") or 0 if ts <= timestamp: best = e else: break return {"event": best, "session_id": session_id} # ------------------------- # RESET # ------------------------- @router.post("/reset") def replay_reset(session_id: str): events = SESSION_DB.get_replay_stream(session_id) return { "session_id": session_id, "event": events[0] if events else None, "event_count": len(events), } # ------------------------- # Convenience: replay_endpoint wrapper for api.py facade # ------------------------- def replay_endpoint(session_id, registry=None): """Legacy wrapper — returns the session events for a given id.""" return { "session_id": session_id, "events": SESSION_DB.get_replay_stream(session_id), }