pragent/pilot/feedback_scores.py

#!/usr/bin/env python3
"""pragent pilot — feedback DB to Langfuse scores.

`feedback.db` already records every reaction, thread resolution and reply a
maintainer leaves on a bot comment. That is the only ground truth pragent has
about whether a finding was any good, and until now it went to a markdown report
nobody reads and nowhere else. This ships it to Langfuse as session-level
scores, so "was the reviewer right" sits on the same axis as "what did it cost".

Session, not trace
------------------
`langfuse_trace` sets `sessionId` to `"{repo}#{pr}"` and lets the trace id be a
fresh uuid per review. Feedback arrives days later against a PR, not against one
particular re-run of the reviewer, and nothing in `feedback.db` records which
trace produced which comment. Scoring the session is therefore both the
available join and the honest granularity: this is feedback on the review of
this PR, not on one invocation.

Two scores, deliberately separated
----------------------------------
  * `review_engagement` — the share of a PR's findings that got any human
    response at all. This is a signal about the *feedback loop*, not the
    reviewer: at the time of writing it is 0.0 across all 113 recorded reviews,
    which is exactly the fact that makes an accuracy metric impossible today.
    It must be watched first, because every other quality number is vapour
    until it moves.
  * `review_acceptance` — net verdict over the findings that *did* get a
    response: (upvotes + resolved) - (downvotes + negation replies), normalised
    to -1..1. Computed only over engaged findings, so an ignored review scores
    `None` rather than 0. Zero would read as "humans judged this exactly
    neutral"; the truth is nobody looked.

Fail-open and idempotent. Score ids are derived from (repo, pr, name) so a
re-run overwrites rather than duplicates.
"""
from __future__ import annotations

import argparse
import json
import os
import sqlite3
import sys
import uuid
from datetime import datetime, timezone

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from feedback_harvest import classify_reaction, _is_negation_reply  # noqa: E402

REVIEW_ENGAGEMENT = "review_engagement"
REVIEW_ACCEPTANCE = "review_acceptance"

# Stable namespace so the same (repo, pr, score) always produces the same score
# id — Langfuse treats a repeated id as an update, which is what a backfill of a
# still-accumulating PR should do.
_NS = uuid.UUID("6f1d9c2e-4a77-4f2a-9c1a-0d3b5e8a7c41")


def _score_id(repo: str, pr: int, name: str) -> str:
    return str(uuid.uuid5(_NS, f"{repo}#{pr}#{name}"))


def collect_pr_feedback(conn: sqlite3.Connection, repo: str, pr: int) -> dict:
    """Tally one PR's findings and the human responses attached to them.

    Returns counts only — the scoring maths lives in `score_pr` so it can be
    tested without a database.
    """
    rows = conn.execute(
        "SELECT id, comment_id FROM inline_finding WHERE repo = ? AND pr = ?",
        (repo, pr),
    ).fetchall()
    total = len(rows)
    engaged = 0
    positive = 0
    negative = 0

    for row in rows:
        fid = row["id"] if isinstance(row, sqlite3.Row) else row[0]
        cid = row["comment_id"] if isinstance(row, sqlite3.Row) else row[1]
        pos = neg = 0

        if cid is not None:
            for r in conn.execute(
                "SELECT content FROM reaction WHERE comment_id = ?", (cid,)
            ):
                kind = classify_reaction(r[0])
                if kind == "positive":
                    pos += 1
                elif kind == "negative":
                    neg += 1

        for r in conn.execute(
            "SELECT resolved FROM thread_state WHERE finding_id = ?", (fid,)
        ):
            # A resolved thread means the maintainer acted on the finding.
            if r[0]:
                pos += 1

        # A reply counts as engagement either way; only a negation phrase makes
        # it a vote against. A neutral reply ("done", "good catch, but…") is
        # deliberately not a positive vote — it says someone looked, not that
        # they agreed.
        replied = 0
        for r in conn.execute(
            "SELECT body FROM reply WHERE finding_id = ?", (fid,)
        ):
            replied += 1
            if _is_negation_reply(r[0]):
                neg += 1

        if pos or neg or replied:
            engaged += 1
        positive += pos
        negative += neg

    return {"total": total, "engaged": engaged, "positive": positive, "negative": negative}


def score_pr(tally: dict) -> dict:
    """Turn one PR's tally into score values.

    `review_acceptance` is `None` when nothing was engaged — see the module
    docstring on why that is not 0.
    """
    total = int(tally.get("total") or 0)
    engaged = int(tally.get("engaged") or 0)
    pos = int(tally.get("positive") or 0)
    neg = int(tally.get("negative") or 0)

    engagement = round(engaged / total, 4) if total else None
    acceptance = None
    if pos or neg:
        acceptance = round((pos - neg) / (pos + neg), 4)
    return {REVIEW_ENGAGEMENT: engagement, REVIEW_ACCEPTANCE: acceptance}


def build_score_events(
    repo: str, pr: int, values: dict, environment: str = "default",
    timestamp: str | None = None,
) -> list[dict]:
    """`score-create` events for one PR's feedback.

    Every event carries a timestamp: the ingestion endpoint rejects those that
    do not, and it reports the rejection as a per-event 400 inside an HTTP 207,
    which reads as success to a caller that only checks the status code.
    """
    ts = timestamp or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
    events = []
    for name, value in values.items():
        if value is None:
            continue
        events.append(
            {
                "id": str(uuid.uuid4()),
                "type": "score-create",
                "timestamp": ts,
                "body": {
                    "id": _score_id(repo, pr, name),
                    "sessionId": f"{repo}#{pr}",
                    "name": name,
                    "value": float(value),
                    "dataType": "NUMERIC",
                    "environment": environment,
                    "comment": f"from feedback.db · {repo}#{pr}",
                },
            }
        )
    return events


SCORE_CONFIGS = [
    {
        "name": REVIEW_ENGAGEMENT,
        "dataType": "NUMERIC",
        "minValue": 0,
        "maxValue": 1,
        "description": "Share of a PR's findings that drew any human reaction, resolution or reply. 0 = nobody engaged with the review.",
    },
    {
        "name": REVIEW_ACCEPTANCE,
        "dataType": "NUMERIC",
        "minValue": -1,
        "maxValue": 1,
        "description": "Net human verdict over engaged findings: +1 all accepted, -1 all rejected. Absent when nothing was engaged.",
    },
]


def iter_prs(conn: sqlite3.Connection):
    for row in conn.execute(
        "SELECT DISTINCT repo, pr FROM inline_finding ORDER BY repo, pr"
    ):
        yield row[0], int(row[1])


def backfill(db_path: str, *, environment: str = "default", dry_run: bool = False) -> dict:
    """Score every PR in the feedback DB. Returns a summary dict."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    events: list[dict] = []
    scanned = 0
    engaged_prs = 0
    try:
        for repo, pr in iter_prs(conn):
            scanned += 1
            tally = collect_pr_feedback(conn, repo, pr)
            values = score_pr(tally)
            if (values.get(REVIEW_ENGAGEMENT) or 0) > 0:
                engaged_prs += 1
            events.extend(build_score_events(repo, pr, values, environment))
    finally:
        conn.close()

    summary = {"prs_scanned": scanned, "prs_with_engagement": engaged_prs, "scores": len(events)}
    if dry_run or not events:
        summary["posted"] = False
        return summary

    import langfuse_trace

    conf = langfuse_trace._enabled()
    if conf is None:
        summary["posted"] = False
        summary["error"] = "Langfuse not configured (LANGFUSE_HOST / keys unset)"
        return summary
    host, pk, sk = conf
    status = langfuse_trace._post(host, pk, sk, events, 15.0)
    summary["posted"] = status in (200, 201, 207)
    summary["http_status"] = status
    return summary


def main() -> int:
    ap = argparse.ArgumentParser(description="Ship feedback.db verdicts to Langfuse as scores")
    ap.add_argument("--db", default=os.environ.get("PRAGENT_FEEDBACK_DB", "/data/feedback.db"))
    ap.add_argument("--environment", default="default")
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()

    summary = backfill(args.db, environment=args.environment, dry_run=args.dry_run)
    print(json.dumps(summary, indent=2))
    return 0 if summary.get("posted") or args.dry_run else 1


if __name__ == "__main__":
    raise SystemExit(main())