ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/session_store.py

"""Starea sesiunilor: state.json, scriere atomica, lock per fir, PID reuse, recovery.

Autoritatea pe schema din INTERFACES.md o are acest modul (T7 + partea de nucleu a T5).
"""

from __future__ import annotations

import asyncio
import contextlib
import datetime as _dt
import fcntl
import json
import logging
import os
import time
from typing import Any, Callable

import config

log = logging.getLogger("discord-bridge.state")

# Lane C poate lipsi cand rulam nucleul singur; import tolerant, fara varianta proprie.
try:  # pragma: no cover - depinde de ordinea de merge a lane-urilor
    import alerts  # type: ignore
except ImportError:  # pragma: no cover
    class _NoAlerts:
        @staticmethod
        def alert(level: str, subject: str, body: str, dedup_key: str | None = None) -> None:
            log.warning("alerta (%s) %s: %s", level, subject, body)

    alerts = _NoAlerts()  # type: ignore

VERSION = 1
_CLOCK_TICKS = os.sysconf("SC_CLK_TCK") if hasattr(os, "sysconf") else 100.0


# ------------------------------------------------------------------ procese
def pid_start_time(pid: int) -> float | None:
    """Campul 22 din /proc/<pid>/stat (starttime, in secunde de la boot).

    Il folosim ca semnatura a procesului: un PID reciclat are alt starttime.
    """
    try:
        with open(f"/proc/{int(pid)}/stat", "rb") as fh:
            data = fh.read().decode("utf-8", "replace")
    except (OSError, ValueError):
        return None
    # comm poate contine spatii si paranteze -> taiem dupa ultima ')'
    close = data.rfind(")")
    if close < 0:
        return None
    fields = data[close + 2:].split()
    # dupa comm, fields[0] este campul 3 (state); campul 22 e fields[19]
    if len(fields) < 20:
        return None
    try:
        return int(fields[19]) / float(_CLOCK_TICKS)
    except (TypeError, ValueError):
        return None


def pid_alive(pid: int | None, start_time: float | None = None) -> bool:
    """True doar daca PID-ul traieste SI (optional) are acelasi starttime."""
    if not pid:
        return False
    cur = pid_start_time(int(pid))
    if cur is None:
        return False
    if start_time is None:
        return True
    return abs(cur - float(start_time)) < 0.05


def today() -> str:
    return _dt.date.today().isoformat()


def new_thread_record(cwd: str | None = None, model: str | None = None) -> dict[str, Any]:
    return {
        "sid": None,
        "cwd": cwd or config.get("DEFAULT_CWD", "/workspace"),
        "model": model or config.get("MODEL_DEFAULT", "sonnet"),
        "pid": None,
        "pid_start_time": None,
        "inflight": None,
        "cost_usd_total": 0.0,
        "last_active": 0.0,
    }


def empty_state() -> dict[str, Any]:
    return {"version": VERSION, "threads": {}, "cost": {"day": today(), "usd": 0.0}}


# ------------------------------------------------------------------- store
class SessionStore:
    """state.json cu scriere atomica, lock de fisier intre procese si lock per fir."""

    def __init__(self, path: os.PathLike | str | None = None, alerter: Callable | None = None):
        self.path = os.fspath(path) if path else os.fspath(config.STATE_FILE)
        self.lock_path = self.path + ".lock"
        self._state: dict[str, Any] = empty_state()
        self._thread_locks: dict[str, asyncio.Lock] = {}
        self._write_lock = asyncio.Lock()
        self._alert = alerter or alerts.alert
        self.loaded = False

    # ------------------------------------------------------------ interne
    def _dir(self) -> str:
        return os.path.dirname(os.path.abspath(self.path)) or "."

    @contextlib.contextmanager
    def _file_lock(self):
        os.makedirs(self._dir(), exist_ok=True)
        fh = open(self.lock_path, "a+")
        try:
            fcntl.flock(fh.fileno(), fcntl.LOCK_EX)
            yield
        finally:
            with contextlib.suppress(OSError):
                fcntl.flock(fh.fileno(), fcntl.LOCK_UN)
            fh.close()

    def _quarantine(self, reason: str) -> None:
        stamp = time.strftime("%Y%m%d-%H%M%S")
        dest = f"{self.path}.corrupt-{stamp}"
        try:
            os.replace(self.path, dest)
        except OSError as exc:
            dest = f"(nu am putut muta: {exc})"
        log.error("state.json corupt (%s), salvat ca %s", reason, dest)
        try:
            self._alert(
                "CRITICAL",
                "state.json corupt",
                f"Motiv: {reason}\nFisierul stricat: {dest}\nPuntea reporneste de la stare goala.",
                "state-corrupt",
            )
        except Exception:  # o alerta esuata nu are voie sa doboare botul
            log.exception("alerta pentru state corupt a esuat")

    def _normalize(self, data: Any) -> dict[str, Any]:
        if not isinstance(data, dict) or not isinstance(data.get("threads"), dict):
            raise ValueError("structura neasteptata")
        state = empty_state()
        state["version"] = data.get("version", VERSION)
        cost = data.get("cost")
        if isinstance(cost, dict):
            state["cost"] = {
                "day": str(cost.get("day") or today()),
                "usd": float(cost.get("usd") or 0.0),
            }
        for tid, rec in data["threads"].items():
            base = new_thread_record()
            if isinstance(rec, dict):
                base.update({k: v for k, v in rec.items() if k in base})
            state["threads"][str(tid)] = base
        return state

    # ------------------------------------------------------------- publice
    def load(self) -> dict[str, Any]:
        """Incarca starea. La fisier corupt: quarantine + alerta + stare goala."""
        with self._file_lock():
            try:
                raw = open(self.path, "rb").read()
            except FileNotFoundError:
                self._state = empty_state()
                self.loaded = True
                return self._state
            except OSError as exc:
                log.error("nu pot citi %s: %s", self.path, exc)
                self._state = empty_state()
                self.loaded = True
                return self._state
            try:
                self._state = self._normalize(json.loads(raw.decode("utf-8")))
            except Exception as exc:
                self._quarantine(str(exc))
                self._state = empty_state()
                self._write_locked()
        self.loaded = True
        return self._state

    def _write_locked(self) -> None:
        """Scriere atomica: tmp in acelasi director + fsync + os.replace + fsync dir."""
        d = self._dir()
        os.makedirs(d, exist_ok=True)
        tmp = os.path.join(d, f".state.{os.getpid()}.tmp")
        payload = json.dumps(self._state, indent=2, sort_keys=True).encode("utf-8")
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            os.write(fd, payload)
            os.fsync(fd)
        finally:
            os.close(fd)
        os.replace(tmp, self.path)
        dfd = os.open(d, os.O_RDONLY)
        try:
            os.fsync(dfd)
        finally:
            os.close(dfd)

    def save(self) -> None:
        with self._file_lock():
            self._write_locked()

    @property
    def state(self) -> dict[str, Any]:
        return self._state

    def lock_for(self, thread_id: str) -> asyncio.Lock:
        """Lock per fir (in proces). Intre procese lucreaza flock-ul de fisier."""
        key = str(thread_id)
        lock = self._thread_locks.get(key)
        if lock is None:
            lock = self._thread_locks[key] = asyncio.Lock()
        return lock

    def thread(self, thread_id: str, create: bool = True) -> dict[str, Any] | None:
        key = str(thread_id)
        rec = self._state["threads"].get(key)
        if rec is None and create:
            rec = self._state["threads"][key] = new_thread_record()
        return rec

    def update_thread(self, thread_id: str, **fields) -> dict[str, Any]:
        """Modifica un fir si scrie atomic. Apeleaza-l sub `lock_for(thread_id)`."""
        rec = self.thread(thread_id)
        assert rec is not None
        rec.update(fields)
        rec["last_active"] = fields.get("last_active", time.time())
        self.save()
        return rec

    def set_pid(self, thread_id: str, pid: int | None) -> dict[str, Any]:
        st = pid_start_time(pid) if pid else None
        return self.update_thread(thread_id, pid=pid, pid_start_time=st)

    def thread_process_alive(self, thread_id: str) -> bool:
        rec = self.thread(thread_id, create=False)
        if not rec:
            return False
        return pid_alive(rec.get("pid"), rec.get("pid_start_time"))

    # ------------------------------------------------------------ inflight
    def set_inflight(self, thread_id: str, turn_id: str, user_id: str, message_id: str) -> dict:
        return self.update_thread(
            thread_id,
            inflight={
                "turn_id": turn_id,
                "started_at": time.time(),
                "user_id": str(user_id),
                "message_id": str(message_id),
            },
        )

    def clear_inflight(self, thread_id: str) -> dict:
        return self.update_thread(thread_id, inflight=None)

    def is_inflight(self, thread_id: str) -> bool:
        rec = self.thread(thread_id, create=False)
        return bool(rec and rec.get("inflight"))

    def sweep_lost_turns(self) -> list[dict[str, Any]]:
        """T5: la pornire, orice fir cu `inflight` al carui proces nu mai e al nostru
        primeste turul marcat drept PIERDUT. Fara reluare automata (dubla executie).
        """
        lost: list[dict[str, Any]] = []
        changed = False
        for tid, rec in self._state["threads"].items():
            inflight = rec.get("inflight")
            if not inflight:
                continue
            if pid_alive(rec.get("pid"), rec.get("pid_start_time")):
                continue  # procesul nostru inca traieste: turul e viu
            lost.append(
                {
                    "thread_id": tid,
                    "turn_id": inflight.get("turn_id"),
                    "user_id": inflight.get("user_id"),
                    "message_id": inflight.get("message_id"),
                    "started_at": inflight.get("started_at"),
                    "warning": (
                        "Turul anterior s-a pierdut (botul a fost repornit sau procesul a murit). "
                        "NU il reiau automat, ca sa nu se execute de doua ori. Trimite-l din nou daca mai e nevoie."
                    ),
                }
            )
            rec["inflight"] = None
            rec["pid"] = None
            rec["pid_start_time"] = None
            changed = True
        if changed:
            self.save()
            try:
                self._alert(
                    "WARN",
                    "tururi pierdute la pornire",
                    "Fire afectate: " + ", ".join(x["thread_id"] for x in lost),
                    "sweep-lost",
                )
            except Exception:
                log.exception("alerta pentru sweep a esuat")
        return lost

    # ---------------------------------------------------------------- cost
    def roll_day(self) -> None:
        cost = self._state.setdefault("cost", {"day": today(), "usd": 0.0})
        if cost.get("day") != today():
            cost["day"] = today()
            cost["usd"] = 0.0

    def add_cost(self, thread_id: str | None, usd: float) -> float:
        """Adauga costul unui tur; returneaza totalul pe ziua curenta."""
        try:
            usd = float(usd)
        except (TypeError, ValueError):
            usd = 0.0
        self.roll_day()
        self._state["cost"]["usd"] = round(self._state["cost"]["usd"] + usd, 6)
        if thread_id is not None:
            rec = self.thread(thread_id)
            rec["cost_usd_total"] = round(float(rec.get("cost_usd_total") or 0.0) + usd, 6)
        self.save()
        return self._state["cost"]["usd"]

    def cost_today(self) -> float:
        self.roll_day()
        return float(self._state["cost"]["usd"])