api/logstore.py

"""Every generation request/response pair, stored in one SQLite file for Stage 2 evaluation.

One row per request: what the tester asked, what was retrieved, every model
attempt with its validation result, the final output, plus two review
channels: the tester's own feedback (did it work when they ran it?) and the
Orcanion reviewer's verdict (``python -m api.review``).
"""

from __future__ import annotations

import datetime as dt
import json
import sqlite3
import uuid
from contextlib import contextmanager
from pathlib import Path

SCHEMA = """
CREATE TABLE IF NOT EXISTS generations (
    id              TEXT PRIMARY KEY,
    created_at      TEXT NOT NULL,
    tester          TEXT NOT NULL,
    kind            TEXT NOT NULL,          -- 'simulate' | 'code'
    request         TEXT NOT NULL,          -- JSON of the request body
    model           TEXT,
    sources         TEXT,                   -- JSON list of retrieved chunk sources
    attempts        TEXT,                   -- JSON list of {raw_output, validation, ...}
    output          TEXT,                   -- final MJCF, or JSON {path: content} for code
    valid           INTEGER,                -- automatic validity check of the final output
    validation      TEXT,                   -- JSON of the final validation result
    error           TEXT,                   -- server-side failure (LLM down, etc.)
    duration_ms     INTEGER,
    tester_rating   TEXT,                   -- 'works' | 'partly' | 'broken'
    tester_note     TEXT,
    review_verdict  TEXT,                   -- 'correct' | 'partial' | 'incorrect'
    review_note     TEXT,
    reviewed_at     TEXT
);
CREATE INDEX IF NOT EXISTS generations_created ON generations(created_at);

-- Every backend job: simulation runs, sweeps, ROS 2 checks, video reviews, dataset exports, Isaac renders
-- (usage visibility: who ran what, when, and what it cost).
CREATE TABLE IF NOT EXISTS runs (
    id              TEXT PRIMARY KEY,
    created_at      TEXT NOT NULL,
    tester          TEXT NOT NULL,
    kind            TEXT NOT NULL DEFAULT 'sim',  -- sim | sweep | ros_check | review | export | isaac
    parent_id       TEXT,                   -- the run a sweep/review/render was made from
    hardware        TEXT,                   -- price key: t4 | l4 | cpu | runpod
    params          TEXT,                   -- JSON: kind-specific request parameters
    cost_usd        REAL,                   -- estimated, set when the job ends
    generation_id   TEXT,                   -- source generation, if the scene came from /simulate/generate
    robot           TEXT,
    duration_s      REAL,
    controller      TEXT,
    backend         TEXT,                   -- modal | runpod
    job_id          TEXT,                   -- the backend's id for the job
    status          TEXT NOT NULL,          -- SUBMITTING | IN_QUEUE | IN_PROGRESS | COMPLETED | FAILED | ...
    error           TEXT,
    output          TEXT,                   -- JSON job output: summary + R2 keys
    execution_ms    INTEGER,                -- GPU execution time (RunPod's figure, else the worker's own)
    delay_ms        INTEGER,                -- queue + cold start (RunPod only)
    updated_at      TEXT
);
CREATE INDEX IF NOT EXISTS runs_tester_created ON runs(tester, created_at);

-- Every AI model call, for the rupee budget (api/budget.py).
CREATE TABLE IF NOT EXISTS usage (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    created_at      TEXT NOT NULL,
    tester          TEXT NOT NULL,
    generation_id   TEXT,
    category        TEXT NOT NULL,          -- coding | simulation_ai | debugging | testing
    model           TEXT,
    tier            TEXT,
    provider        TEXT,                   -- who billed it: openrouter, ...
    input_tokens    INTEGER,
    output_tokens   INTEGER,
    cost_usd        REAL,
    cost_inr        REAL,
    billable        INTEGER NOT NULL,       -- 0: OpenRouter-exempt, recorded but not deducted
    exempt_reason   TEXT,
    estimated       INTEGER NOT NULL DEFAULT 0,  -- 1: cost computed from list prices, not reported by the provider
    estimate_inr    REAL                    -- what the plan expected before running
);
CREATE INDEX IF NOT EXISTS usage_tester ON usage(tester, created_at);

-- Lightweight feedback on anything: generations, runs, checks, routing choices.
CREATE TABLE IF NOT EXISTS feedback (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    created_at      TEXT NOT NULL,
    tester          TEXT NOT NULL,
    target_type     TEXT NOT NULL,          -- generation | job
    target_id       TEXT NOT NULL,
    rating          TEXT NOT NULL,          -- helpful | not_helpful
    note            TEXT,
    context         TEXT                    -- JSON: task, routing tier/model, simulator, result state, ...
);
"""

JSON_FIELDS = ("request", "sources", "attempts", "validation")


def _decode_run(d: dict) -> dict:
    d["output"] = json.loads(d["output"]) if d.get("output") else None
    d["params"] = json.loads(d["params"]) if d.get("params") else {}
    return d


def _now() -> str:
    return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")


class LogStore:
    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self._conn() as c:
            c.executescript(SCHEMA)
            self._migrate(c)

    @staticmethod
    def _migrate(c) -> None:
        """Bring run tables created by earlier versions up to the current schema."""
        cols = {r[1] for r in c.execute("PRAGMA table_info(runs)")}
        if "runpod_job_id" in cols:  # before Modal support
            c.execute("ALTER TABLE runs RENAME COLUMN runpod_job_id TO job_id")
        added = {
            "backend": "TEXT",
            "kind": "TEXT NOT NULL DEFAULT 'sim'",
            "parent_id": "TEXT",
            "hardware": "TEXT",
            "params": "TEXT",
            "cost_usd": "REAL",
        }
        for name, decl in added.items():
            if name not in cols:
                c.execute(f"ALTER TABLE runs ADD COLUMN {name} {decl}")
        if "runpod_job_id" in cols and "backend" not in cols:
            c.execute("UPDATE runs SET backend = 'runpod' WHERE job_id IS NOT NULL")
        c.execute("CREATE INDEX IF NOT EXISTS runs_created ON runs(created_at)")

    @contextmanager
    def _conn(self):
        conn = sqlite3.connect(self.path, timeout=30)
        conn.row_factory = sqlite3.Row
        try:
            conn.execute("PRAGMA journal_mode=WAL")
            yield conn
            conn.commit()
        finally:
            conn.close()

    def record(
        self,
        *,
        tester: str,
        kind: str,
        request: dict,
        model: str | None = None,
        sources: list | None = None,
        attempts: list | None = None,
        output: str | None = None,
        valid: bool | None = None,
        validation: dict | None = None,
        error: str | None = None,
        duration_ms: int | None = None,
    ) -> str:
        gen_id = uuid.uuid4().hex[:12]
        with self._conn() as c:
            c.execute(
                "INSERT INTO generations (id, created_at, tester, kind, request, model, sources, attempts, output, "
                "valid, validation, error, duration_ms) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (
                    gen_id,
                    _now(),
                    tester,
                    kind,
                    json.dumps(request),
                    model,
                    json.dumps(sources or []),
                    json.dumps(attempts or []),
                    output,
                    None if valid is None else int(valid),
                    json.dumps(validation or {}),
                    error,
                    duration_ms,
                ),
            )
        return gen_id

    def get(self, gen_id: str) -> dict | None:
        with self._conn() as c:
            row = c.execute("SELECT * FROM generations WHERE id = ?", (gen_id,)).fetchone()
        return self._decode(row) if row else None

    def list(
        self, kind: str | None = None, tester: str | None = None, unreviewed: bool = False, limit: int = 50
    ) -> list[dict]:
        sql, args = "SELECT * FROM generations WHERE 1=1", []
        if kind:
            sql += " AND kind = ?"
            args.append(kind)
        if tester:
            sql += " AND tester = ?"
            args.append(tester)
        if unreviewed:
            sql += " AND review_verdict IS NULL"
        sql += " ORDER BY created_at DESC LIMIT ?"
        args.append(limit)
        with self._conn() as c:
            return [self._decode(r) for r in c.execute(sql, args).fetchall()]

    def all(self) -> list[dict]:
        with self._conn() as c:
            return [self._decode(r) for r in c.execute("SELECT * FROM generations ORDER BY created_at").fetchall()]

    def set_feedback(self, gen_id: str, tester: str, rating: str, note: str) -> bool:
        with self._conn() as c:
            cur = c.execute(
                "UPDATE generations SET tester_rating = ?, tester_note = ? WHERE id = ? AND tester = ?",
                (rating, note, gen_id, tester),
            )
        return cur.rowcount == 1

    def set_review(self, gen_id: str, verdict: str, note: str) -> bool:
        with self._conn() as c:
            cur = c.execute(
                "UPDATE generations SET review_verdict = ?, review_note = ?, reviewed_at = ? WHERE id = ?",
                (verdict, note, _now(), gen_id),
            )
        return cur.rowcount == 1

    def summary(self) -> list[dict]:
        sql = """SELECT kind, COUNT(*) AS total, SUM(valid = 1) AS auto_valid,
                        SUM(tester_rating = 'works') AS tester_works, SUM(tester_rating IS NOT NULL) AS tester_rated,
                        SUM(review_verdict = 'correct') AS reviewed_correct,
                        SUM(review_verdict IS NOT NULL) AS reviewed
                 FROM generations GROUP BY kind ORDER BY kind"""
        with self._conn() as c:
            return [dict(r) for r in c.execute(sql).fetchall()]

    # ------------------------------------------------------------------ AI usage and feedback

    def record_usage(self, **row) -> None:
        allowed = {"created_at", "tester", "generation_id", "category", "model", "tier",
                   "provider", "input_tokens", "output_tokens", "cost_usd", "cost_inr",
                   "billable", "exempt_reason", "estimated", "estimate_inr"}
        row = {k: v for k, v in row.items() if k in allowed}
        row = {"created_at": _now(), **row}
        cols = ", ".join(row)
        with self._conn() as c:
            c.execute(f"INSERT INTO usage ({cols}) VALUES ({', '.join('?' * len(row))})", tuple(row.values()))

    def usage_rows(self, tester: str | None = None, since_iso: str | None = None) -> list[dict]:
        sql, args = "SELECT * FROM usage WHERE 1=1", []
        if tester:
            sql += " AND tester = ?"
            args.append(tester)
        if since_iso:
            sql += " AND created_at >= ?"
            args.append(since_iso)
        with self._conn() as c:
            return [dict(r) for r in c.execute(sql + " ORDER BY id", args).fetchall()]

    def record_feedback(self, tester: str, target_type: str, target_id: str, rating: str, note: str, context: dict):
        with self._conn() as c:
            c.execute(
                "INSERT INTO feedback (created_at, tester, target_type, target_id, rating, note, context) "
                "VALUES (?,?,?,?,?,?,?)",
                (_now(), tester, target_type, target_id, rating, note, json.dumps(context)),
            )

    def feedback_rows(self) -> list[dict]:
        with self._conn() as c:
            rows = [dict(r) for r in c.execute("SELECT * FROM feedback ORDER BY id").fetchall()]
        for r in rows:
            r["context"] = json.loads(r["context"]) if r.get("context") else {}
        return rows

    # ------------------------------------------------------------------ runs

    def record_run(
        self,
        *,
        run_id: str,
        tester: str,
        generation_id: str | None,
        robot: str | None,
        duration_s: float | None,
        controller: str | None,
        kind: str = "sim",
        parent_id: str | None = None,
        hardware: str | None = None,
        params: dict | None = None,
    ) -> None:
        now = _now()
        with self._conn() as c:
            c.execute(
                "INSERT INTO runs (id, created_at, tester, kind, parent_id, hardware, params, generation_id, robot, "
                "duration_s, controller, status, updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
                (
                    run_id,
                    now,
                    tester,
                    kind,
                    parent_id,
                    hardware,
                    json.dumps(params) if params is not None else None,
                    generation_id,
                    robot,
                    duration_s,
                    controller,
                    "SUBMITTING",
                    now,
                ),
            )

    def update_run(self, run_id: str, **fields) -> None:
        allowed = {"backend", "job_id", "status", "error", "output", "execution_ms", "delay_ms", "cost_usd"}
        if set(fields) - allowed:
            raise ValueError(f"unknown run fields {set(fields) - allowed}")
        if "output" in fields and not isinstance(fields["output"], (str, type(None))):
            fields["output"] = json.dumps(fields["output"])
        fields["updated_at"] = _now()
        cols = ", ".join(f"{k} = ?" for k in fields)
        with self._conn() as c:
            c.execute(f"UPDATE runs SET {cols} WHERE id = ?", (*fields.values(), run_id))

    def get_run(self, run_id: str) -> dict | None:
        with self._conn() as c:
            row = c.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone()
        if not row:
            return None
        return _decode_run(dict(row))

    def list_runs(self, tester: str | None = None, limit: int = 50) -> list[dict]:
        sql, args = "SELECT * FROM runs", []
        if tester:
            sql += " WHERE tester = ?"
            args.append(tester)
        sql += " ORDER BY created_at DESC LIMIT ?"
        args.append(limit)
        with self._conn() as c:
            rows = [dict(r) for r in c.execute(sql, args).fetchall()]
        return [_decode_run(r) for r in rows]

    def jobs_since(self, since_iso: str, kind: str | None = None) -> list[dict]:
        """Every job created at or after `since_iso` (for spend and daily-limit checks)."""
        sql, args = "SELECT * FROM runs WHERE created_at >= ?", [since_iso]
        if kind:
            sql += " AND kind = ?"
            args.append(kind)
        with self._conn() as c:
            return [_decode_run(dict(r)) for r in c.execute(sql, args).fetchall()]

    def children(self, parent_id: str, kind: str | None = None) -> list[dict]:
        sql, args = "SELECT * FROM runs WHERE parent_id = ?", [parent_id]
        if kind:
            sql += " AND kind = ?"
            args.append(kind)
        sql += " ORDER BY created_at"
        with self._conn() as c:
            return [_decode_run(dict(r)) for r in c.execute(sql, args).fetchall()]

    def runs_since(self, tester: str, since_iso: str) -> int:
        with self._conn() as c:
            return c.execute(
                "SELECT COUNT(*) FROM runs WHERE tester = ? AND created_at >= ?", (tester, since_iso)
            ).fetchone()[0]

    def usage(self, window_minutes: int = 10) -> list[dict]:
        """Per tester: total jobs, jobs in the last 24 h, busiest window, compute seconds, estimated cost."""
        with self._conn() as c:
            rows = c.execute(
                "SELECT tester, created_at, execution_ms, cost_usd FROM runs ORDER BY tester, created_at"
            ).fetchall()
        day_ago = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(days=1)).isoformat(timespec="seconds")
        out: dict[str, dict] = {}
        times: dict[str, list[dt.datetime]] = {}
        for tester, created, exec_ms, cost in rows:
            u = out.setdefault(
                tester,
                {
                    "tester": tester,
                    "runs": 0,
                    "last_24h": 0,
                    "gpu_seconds": 0.0,
                    "cost_usd": 0.0,
                    "first": created,
                    "last": created,
                },
            )
            u["runs"] += 1
            u["last"] = created
            u["last_24h"] += created >= day_ago
            u["gpu_seconds"] += (exec_ms or 0) / 1000
            u["cost_usd"] += cost or 0.0
            times.setdefault(tester, []).append(dt.datetime.fromisoformat(created))
        window = dt.timedelta(minutes=window_minutes)
        for tester, ts in times.items():
            best, j = 0, 0
            for i, t in enumerate(ts):
                while t - ts[j] > window:
                    j += 1
                best = max(best, i - j + 1)
            out[tester][f"max_in_{window_minutes}min"] = best
            out[tester]["gpu_seconds"] = round(out[tester]["gpu_seconds"], 1)
            out[tester]["cost_usd"] = round(out[tester]["cost_usd"], 3)
        return sorted(out.values(), key=lambda u: -u["runs"])

    @staticmethod
    def _decode(row: sqlite3.Row) -> dict:
        d = dict(row)
        for f in JSON_FIELDS:
            if d.get(f):
                d[f] = json.loads(d[f])
        if d.get("valid") is not None:
            d["valid"] = bool(d["valid"])
        return d