feat(feedback): move feedback storage layer from WIP into pilot/

This commit is contained in:
Marcos
2026-08-22 14:46:04 +00:00
parent cdf116ece9
commit 6cdccb48ad
2 changed files with 673 additions and 0 deletions
+356
View File
@@ -0,0 +1,356 @@
"""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
+317
View File
@@ -0,0 +1,317 @@
"""Tests for pilot/feedback.py — SQLite storage for review feedback signals.
Covers: schema bootstrap, posthash stability, dedup-on-insert, reaction /
thread-state / reply upserts, the analyzer-side `findings_with_votes` join,
and graceful failure on bad inputs.
"""
import sqlite3
import tempfile
import unittest
from pilot import feedback
class TestPosthash(unittest.TestCase):
def test_stable_across_calls(self):
a = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
b = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
self.assertEqual(a, b)
def test_length_is_short(self):
h = feedback.posthash("a", 1, "low", "x")
self.assertEqual(len(h), 16)
def test_different_path_different_hash(self):
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("b", 1, "LOW", "x"),
)
def test_different_line_different_hash(self):
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("a", 2, "LOW", "x"),
)
def test_different_severity_different_hash(self):
# Same line, same problem, different severity → different signal.
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("a", 1, "CRITICAL", "x"),
)
def test_problem_prefix_used_only(self):
# First 80 chars participate; rest is ignored.
self.assertEqual(
feedback.posthash("a", 1, "LOW", "x" * 80 + "tail"),
feedback.posthash("a", 1, "LOW", "x" * 80),
)
def test_case_and_whitespace_normalized_in_problem(self):
# Lowercased + stripped → same hash.
self.assertEqual(
feedback.posthash("a", 1, "LOW", " Same Finding "),
feedback.posthash("a", 1, "LOW", "same finding"),
)
class TestInit(unittest.TestCase):
def test_init_creates_db(self):
with tempfile.TemporaryDirectory() as d:
db = f"{d}/f.db"
conn = feedback.init(db)
# Application tables exist (sqlite_sequence is a bookkeeping table
# created by AUTOINCREMENT — not part of the contract).
tables = {r[0] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()}
self.assertTrue(
{"review", "inline_finding", "reaction", "thread_state", "reply"}.issubset(tables),
f"missing tables: got {tables}",
)
conn.close()
def test_init_is_idempotent(self):
with tempfile.TemporaryDirectory() as d:
db = f"{d}/f.db"
feedback.init(db)
# Second call must not raise.
feedback.init(db)
class TestRecordReview(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_returns_id_and_row(self):
rid = feedback.record_review(
self.conn, repo="o/r", pr=1, head_sha="abc",
review_id_gitea=99, body_comment_id=42,
)
self.assertIsNotNone(rid)
row = self.conn.execute("SELECT * FROM review WHERE id = ?", (rid,)).fetchone()
self.assertEqual(row[1], "o/r")
self.assertEqual(row[4], 99)
self.assertEqual(row[5], 42)
class TestRecordInlineFinding(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def _new_review(self):
return feedback.record_review(
self.conn, repo="o/r", pr=1, head_sha="x",
)
def test_insert_returns_id(self):
rid = self._new_review()
fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH",
problem="bug", fix="patch", suggestion="code",
comment_id=555,
)
self.assertIsNotNone(fid)
def test_dedup_on_posthash(self):
# Two reviews of the SAME finding on different PRs insert two
# rows — deduplication by posthash is the *analyzer's* job
# (findings_with_votes GROUP BY posthash). Storing one row per
# review preserves per-comment reactions across PRs.
rid1 = self._new_review()
feedback.record_inline_finding(
self.conn, review_id=rid1, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH", problem="race",
comment_id=100,
)
rid2 = feedback.record_review(self.conn, repo="o/r", pr=2, head_sha="y")
feedback.record_inline_finding(
self.conn, review_id=rid2, repo="o/r", pr=2,
path="a/b.ts", line=10, severity="HIGH", problem="race",
comment_id=200,
)
rows = self.conn.execute(
"SELECT id, comment_id FROM inline_finding WHERE path='a/b.ts' AND line=10 ORDER BY id"
).fetchall()
self.assertEqual(len(rows), 2)
# Both comment_ids preserved (PK dedup is the *reaction* table's job).
self.assertEqual([r[1] for r in rows], [100, 200])
def test_posthash_set(self):
rid = self._new_review()
feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="nit",
)
ph = self.conn.execute(
"SELECT posthash FROM inline_finding LIMIT 1"
).fetchone()[0]
expected = feedback.posthash("a", 1, "LOW", "nit")
self.assertEqual(ph, expected)
class TestRecordReaction(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_insert_upsert(self):
ok = feedback.record_reaction(
self.conn, comment_id=10, user="alice", content="+1",
)
self.assertTrue(ok)
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 1)
# Re-insert same PK → no duplicate.
feedback.record_reaction(self.conn, comment_id=10, user="alice", content="+1")
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 1)
def test_distinct_users_can_react(self):
feedback.record_reaction(self.conn, comment_id=10, user="a", content="+1")
feedback.record_reaction(self.conn, comment_id=10, user="b", content="-1")
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 2)
class TestRecordThreadState(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
self.fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_upsert_overwrites(self):
feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=True)
row = self.conn.execute(
"SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
).fetchone()
self.assertEqual(row[0], 1)
feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=False)
row = self.conn.execute(
"SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
).fetchone()
self.assertEqual(row[0], 0)
class TestRecordReply(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
self.fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_insert_idempotent(self):
feedback.record_reply(
self.conn, finding_id=self.fid, author="a",
body="hi", created_at=1000,
)
feedback.record_reply(
self.conn, finding_id=self.fid, author="a",
body="hi", created_at=1000, # same PK
)
n = self.conn.execute("SELECT COUNT(*) FROM reply").fetchone()[0]
self.assertEqual(n, 1)
class TestFindingsWithVotes(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def _seed(self):
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH",
problem="race", comment_id=500,
)
feedback.record_reaction(self.conn, comment_id=500, user="u1", content="+1")
feedback.record_reaction(self.conn, comment_id=500, user="u2", content="-1")
feedback.record_thread_state(self.conn, finding_id=fid, resolved=True)
feedback.record_reply(
self.conn, finding_id=fid, author="u3",
body="this is fine because of X", created_at=2000,
)
return fid
def test_join_rolls_up_votes(self):
self._seed()
rows = list(feedback.findings_with_votes(self.conn))
self.assertEqual(len(rows), 1)
r = rows[0]
self.assertEqual(r["upvotes"], 1)
self.assertEqual(r["downvotes"], 1)
self.assertEqual(r["resolved"], 1)
self.assertEqual(r["reply_count"], 1)
self.assertIn("this is fine", r["reply_bodies"])
def test_repo_filter(self):
self._seed()
# Add a finding under a different repo.
feedback.record_inline_finding(
self.conn, review_id=None, repo="other/r", pr=99,
path="x", line=1, severity="LOW", problem="y",
)
rows = list(feedback.findings_with_votes(self.conn, repo="o/r"))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["repo"], "o/r")
def test_findings_with_no_signals_return_zero_votes(self):
feedback.record_inline_finding(
self.conn, review_id=None, repo="x/y", pr=1,
path="p", line=1, severity="LOW", problem="z",
)
rows = list(feedback.findings_with_votes(self.conn))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["upvotes"], 0)
self.assertEqual(rows[0]["downvotes"], 0)
self.assertIsNone(rows[0]["resolved"])
class TestKnownPosthashes(unittest.TestCase):
def test_returns_distinct_set(self):
with tempfile.TemporaryDirectory() as d:
conn = feedback.init(f"{d}/f.db")
feedback.record_inline_finding(
conn, review_id=None, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
feedback.record_inline_finding(
conn, review_id=None, repo="o/r", pr=1,
path="a", line=2, severity="LOW", problem="y",
)
phs = feedback.known_posthashes_for_repo(conn, "o/r")
self.assertEqual(len(phs), 2)
conn.close()
if __name__ == "__main__":
unittest.main()