Claude
7a510a926d
Group review, feedback, evaluation, observability, and entrypoint code into packages. Keep thin top-level compatibility shims for existing scripts and imports, and mirror the structure in the tests.
356 lines
13 KiB
Python
356 lines
13 KiB
Python
"""pragent pilot — feedback storage.
|
|
|
|
A thin SQLite layer that records every bot review comment + the reactions /
|
|
thread-state / replies it accumulates over time. Powers the daily analysis
|
|
that produces suggested addenda for `.pr-review.json:instructions` and
|
|
`PRAGENT_ADDITIONAL_CONTEXT_URL` (see `feedback_analyze.py`).
|
|
|
|
Why SQLite: stdlib, no extra deps in the container, single writer (the
|
|
webhook server is one process per pod). Mount at `/data/feedback.db`
|
|
via the `feedback-data` PVC.
|
|
|
|
Schema (idempotent — safe to call `init` at every boot):
|
|
|
|
review(repo, pr, head_sha, body_comment_id, posted_at, review_id_gitea)
|
|
inline_finding(review_id → review.id, repo, pr, path, line,
|
|
severity, problem, fix, suggestion,
|
|
comment_id, posthash UNIQUE, posted_at)
|
|
reaction(comment_id, user, content, created_at,
|
|
PRIMARY KEY (comment_id, user, content))
|
|
thread_state(finding_id → inline_finding.id, resolved, checked_at,
|
|
PRIMARY KEY (finding_id))
|
|
reply(finding_id → inline_finding.id, author, body, created_at,
|
|
PRIMARY KEY (finding_id, created_at))
|
|
|
|
`posthash` is a short hash of (path|line|severity|first 80 chars of problem).
|
|
It survives across reviews of the same finding on the same line — same
|
|
finding on PR #5 and PR #12 of the same file de-duplicate, so the daily
|
|
analyzer can count votes across reviews instead of one-at-a-time.
|
|
|
|
Everything is best-effort. The webhook server never aborts a review
|
|
because the feedback DB had a hiccup — `record_*` functions log and
|
|
swallow.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import logging
|
|
import sqlite3
|
|
import time
|
|
from typing import Iterable, Optional
|
|
|
|
log = logging.getLogger("pragent.feedback")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Schema
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS review (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
repo TEXT NOT NULL,
|
|
pr INTEGER NOT NULL,
|
|
head_sha TEXT NOT NULL,
|
|
review_id_gitea INTEGER,
|
|
body_comment_id INTEGER,
|
|
posted_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS review_repo_pr ON review(repo, pr);
|
|
|
|
CREATE TABLE IF NOT EXISTS inline_finding (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
review_id INTEGER REFERENCES review(id),
|
|
repo TEXT NOT NULL,
|
|
pr INTEGER NOT NULL,
|
|
path TEXT NOT NULL,
|
|
line INTEGER NOT NULL,
|
|
severity TEXT NOT NULL,
|
|
problem TEXT NOT NULL,
|
|
fix TEXT,
|
|
suggestion TEXT,
|
|
comment_id INTEGER,
|
|
posthash TEXT NOT NULL,
|
|
posted_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS inline_finding_posthash_idx ON inline_finding(posthash);
|
|
CREATE INDEX IF NOT EXISTS inline_finding_repo_pr ON inline_finding(repo, pr);
|
|
CREATE INDEX IF NOT EXISTS inline_finding_posthash ON inline_finding(posthash);
|
|
|
|
CREATE TABLE IF NOT EXISTS reaction (
|
|
comment_id INTEGER NOT NULL,
|
|
user TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY (comment_id, user, content)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS reaction_comment ON reaction(comment_id);
|
|
|
|
CREATE TABLE IF NOT EXISTS thread_state (
|
|
finding_id INTEGER NOT NULL REFERENCES inline_finding(id),
|
|
resolved INTEGER NOT NULL,
|
|
checked_at INTEGER NOT NULL,
|
|
PRIMARY KEY (finding_id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS reply (
|
|
finding_id INTEGER NOT NULL REFERENCES inline_finding(id),
|
|
author TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
PRIMARY KEY (finding_id, created_at)
|
|
);
|
|
"""
|
|
|
|
|
|
def init(db_path: str) -> sqlite3.Connection:
|
|
"""Open (or create) the DB, ensure schema. Returns a Connection."""
|
|
conn = sqlite3.connect(db_path)
|
|
conn.row_factory = sqlite3.Row # so callers can use row["name"]
|
|
conn.executescript(_SCHEMA)
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Posthash — cross-review finding dedup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def posthash(path: str, line: int, severity: str, problem: str) -> str:
|
|
"""Short stable hash of the finding's identifying triple + a problem
|
|
fingerprint. Designed so two reviews of the SAME finding (same file,
|
|
same line, same severity, same core complaint) collapse to one row —
|
|
reactions across PRs aggregate.
|
|
|
|
`line` is the post-change (RIGHT-side) line — the agent anchors on it
|
|
and so does this hash. Different lines = different finding, by design.
|
|
`severity` participates because "this is a CRITICAL bug" and "this is a
|
|
LOW nitpick" at the same line on the same problem text are different
|
|
signals to learn from.
|
|
"""
|
|
h = hashlib.sha256()
|
|
h.update(f"{path}\n".encode())
|
|
h.update(f"{line}\n".encode())
|
|
h.update(f"{severity.upper()}\n".encode())
|
|
h.update(problem[:80].strip().lower().encode())
|
|
return h.hexdigest()[:16]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Write helpers — all best-effort. Log + swallow.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def record_review(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
repo: str,
|
|
pr: int,
|
|
head_sha: str,
|
|
review_id_gitea: Optional[int] = None,
|
|
body_comment_id: Optional[int] = None,
|
|
posted_at: Optional[int] = None,
|
|
) -> Optional[int]:
|
|
"""Insert a review row. Returns the new row id, or None on failure."""
|
|
try:
|
|
cur = conn.execute(
|
|
"INSERT INTO review(repo, pr, head_sha, review_id_gitea, body_comment_id, posted_at) "
|
|
"VALUES(?,?,?,?,?,?)",
|
|
(repo, pr, head_sha, review_id_gitea, body_comment_id, posted_at or int(time.time())),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
except Exception as e:
|
|
log.warning("record_review failed: %s", e)
|
|
return None
|
|
|
|
|
|
def record_inline_finding(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
review_id: Optional[int],
|
|
repo: str,
|
|
pr: int,
|
|
path: str,
|
|
line: int,
|
|
severity: str,
|
|
problem: str,
|
|
fix: str = "",
|
|
suggestion: str = "",
|
|
comment_id: Optional[int] = None,
|
|
posted_at: Optional[int] = None,
|
|
) -> Optional[int]:
|
|
"""Insert an inline-finding row, deduped on posthash.
|
|
|
|
`comment_id` is filled in by the harvester when it discovers the
|
|
Gitea-assigned comment id for this finding. The post path returns the
|
|
`review_id` only; the inline ids come from a follow-up fetch.
|
|
"""
|
|
ph = posthash(path, line, severity, problem)
|
|
ts = posted_at or int(time.time())
|
|
# Every call inserts a fresh row. Aggregation by posthash is the
|
|
# caller's job — see `findings_with_votes` which GROUP BYs posthash.
|
|
# Letting each finding be its own row means reactions on different
|
|
# comment_ids across multiple PR reviews are not lost when one of
|
|
# those comment_ids becomes stale.
|
|
try:
|
|
cur = conn.execute(
|
|
"INSERT INTO inline_finding(review_id, repo, pr, path, line, severity, "
|
|
"problem, fix, suggestion, comment_id, posthash, posted_at) "
|
|
"VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
|
|
(review_id, repo, pr, path, line, severity, problem, fix, suggestion,
|
|
comment_id, ph, ts),
|
|
)
|
|
conn.commit()
|
|
return cur.lastrowid
|
|
except Exception as e:
|
|
log.warning("record_inline_finding failed: %s", e)
|
|
return None
|
|
|
|
|
|
def record_reaction(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
comment_id: int,
|
|
user: str,
|
|
content: str,
|
|
created_at: Optional[int] = None,
|
|
) -> bool:
|
|
"""Upsert one reaction. PK = (comment_id, user, content)."""
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO reaction(comment_id, user, content, created_at) "
|
|
"VALUES(?,?,?,?)",
|
|
(comment_id, user, content, created_at or int(time.time())),
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
log.warning("record_reaction failed: %s", e)
|
|
return False
|
|
|
|
|
|
def record_thread_state(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
finding_id: int,
|
|
resolved: bool,
|
|
checked_at: Optional[int] = None,
|
|
) -> bool:
|
|
"""Upsert the latest thread-state check."""
|
|
try:
|
|
conn.execute(
|
|
"INSERT INTO thread_state(finding_id, resolved, checked_at) "
|
|
"VALUES(?,?,?) "
|
|
"ON CONFLICT(finding_id) DO UPDATE SET "
|
|
" resolved = excluded.resolved, checked_at = excluded.checked_at",
|
|
(finding_id, 1 if resolved else 0, checked_at or int(time.time())),
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
log.warning("record_thread_state failed: %s", e)
|
|
return False
|
|
|
|
|
|
def record_reply(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
finding_id: int,
|
|
author: str,
|
|
body: str,
|
|
created_at: int,
|
|
) -> bool:
|
|
"""Insert one reply. PK includes created_at → re-imports are idempotent."""
|
|
try:
|
|
conn.execute(
|
|
"INSERT OR IGNORE INTO reply(finding_id, author, body, created_at) "
|
|
"VALUES(?,?,?,?)",
|
|
(finding_id, author, body, created_at),
|
|
)
|
|
conn.commit()
|
|
return True
|
|
except Exception as e:
|
|
log.warning("record_reply failed: %s", e)
|
|
return False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Read helpers — for the analyzer
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def findings_with_votes(
|
|
conn: sqlite3.Connection,
|
|
*,
|
|
repo: Optional[str] = None,
|
|
since_ts: Optional[int] = None,
|
|
) -> Iterable[sqlite3.Row]:
|
|
"""Stream every inline finding with rolled-up votes attached.
|
|
|
|
Joins:
|
|
inline_finding ◀ reaction (count by content)
|
|
inline_finding ◀ thread_state (latest resolved flag)
|
|
inline_finding ◀ reply (count + concatenation of bodies for negation
|
|
pattern matching)
|
|
|
|
Yielded rows expose:
|
|
id, repo, pr, path, line, severity, problem, fix, suggestion,
|
|
comment_id, posthash, posted_at,
|
|
upvotes INT, downvotes INT,
|
|
resolved INT (0/1/NULL),
|
|
reply_count INT,
|
|
reply_bodies TEXT ('\\n\\n'-joined for substring match),
|
|
review_posted_at INT
|
|
"""
|
|
where = []
|
|
params: list = []
|
|
if repo:
|
|
where.append("f.repo = ?")
|
|
params.append(repo)
|
|
if since_ts is not None:
|
|
where.append("COALESCE(r.posted_at, f.posted_at) >= ?")
|
|
params.append(since_ts)
|
|
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
|
|
|
|
sql = f"""
|
|
SELECT
|
|
f.posthash AS id, -- alias for compat — every row IS an aggregated posthash
|
|
f.repo, MAX(f.pr) AS pr, f.path, f.line, MAX(f.severity) AS severity,
|
|
MAX(f.problem) AS problem, MAX(f.fix) AS fix, MAX(f.suggestion) AS suggestion,
|
|
MAX(f.comment_id) AS comment_id, f.posthash, MAX(f.posted_at) AS posted_at,
|
|
COUNT(*) AS occurrences,
|
|
r.posted_at AS review_posted_at,
|
|
COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes,
|
|
COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes,
|
|
MAX(ts.resolved) AS resolved,
|
|
COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id IN (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), 0) AS reply_count,
|
|
COALESCE((SELECT GROUP_CONCAT(body, char(10)||char(10)) FROM reply WHERE finding_id IN (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), '') AS reply_bodies
|
|
FROM inline_finding f
|
|
LEFT JOIN review r ON r.id = f.review_id
|
|
LEFT JOIN reaction rct ON rct.comment_id = f.comment_id
|
|
LEFT JOIN thread_state ts ON ts.finding_id = f.id
|
|
{where_sql}
|
|
GROUP BY f.posthash, f.repo, f.path, f.line
|
|
ORDER BY posted_at DESC
|
|
"""
|
|
return conn.execute(sql, params)
|
|
|
|
|
|
def known_posthashes_for_repo(conn: sqlite3.Connection, repo: str) -> set[str]:
|
|
"""For the harvester: which findings on this repo have already been
|
|
recorded? Used to skip re-fetching reactions we already harvested this
|
|
round."""
|
|
return {
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT DISTINCT posthash FROM inline_finding WHERE repo = ?", (repo,)
|
|
).fetchall()
|
|
}
|
|
|
|
|
|
def comment_ids_for_finding(conn: sqlite3.Connection, posthash: str) -> Optional[int]:
|
|
"""Return the current Gitea comment_id for an existing finding (used to
|
|
harvest votes for findings the harvester discovers on a brand-new PR that
|
|
ALSO has older bot comments on prior PRs)."""
|
|
row = conn.execute(
|
|
"SELECT comment_id FROM inline_finding WHERE posthash = ?", (posthash,)
|
|
).fetchone()
|
|
return row[0] if row else None |