From 6cdccb48ad73c0d0f9216bf75db94576a0268d6a Mon Sep 17 00:00:00 2001 From: Marcos Date: Sat, 22 Aug 2026 14:46:04 +0000 Subject: [PATCH 1/4] feat(feedback): move feedback storage layer from WIP into pilot/ --- pilot/feedback.py | 356 +++++++++++++++++++++++++++++++++++ tests/pilot/test_feedback.py | 317 +++++++++++++++++++++++++++++++ 2 files changed, 673 insertions(+) create mode 100644 pilot/feedback.py create mode 100644 tests/pilot/test_feedback.py diff --git a/pilot/feedback.py b/pilot/feedback.py new file mode 100644 index 0000000..1b4f568 --- /dev/null +++ b/pilot/feedback.py @@ -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 \ No newline at end of file diff --git a/tests/pilot/test_feedback.py b/tests/pilot/test_feedback.py new file mode 100644 index 0000000..007dde5 --- /dev/null +++ b/tests/pilot/test_feedback.py @@ -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() \ No newline at end of file From 69e1fc06a280cdf2dee857f5c32fd2ee7faddde3 Mon Sep 17 00:00:00 2001 From: Marcos Date: Sat, 22 Aug 2026 14:46:07 +0000 Subject: [PATCH 2/4] feat(feedback): move feedback harvester from WIP into pilot/ --- pilot/feedback_harvest.py | 386 +++++++++++++++++++++++++++ tests/pilot/test_feedback_harvest.py | 243 +++++++++++++++++ 2 files changed, 629 insertions(+) create mode 100644 pilot/feedback_harvest.py create mode 100644 tests/pilot/test_feedback_harvest.py diff --git a/pilot/feedback_harvest.py b/pilot/feedback_harvest.py new file mode 100644 index 0000000..31772ec --- /dev/null +++ b/pilot/feedback_harvest.py @@ -0,0 +1,386 @@ +"""pragent pilot — feedback harvester. + +For each PR the webhook server is about to review, walk back through the +Gitea-side state of every bot comment from every prior review on that PR +and record: + - reactions on the review body + on each inline comment + - thread-resolved state (Gitea's `resolver` field; non-empty = resolved) + - replies (issue-comments with `review_comment_id` matching ours) + - the bot's own findings_count + inline_count per review (for the + restraint metric) + +Everything is best-effort. A single 404 or 5xx is logged and skipped — we +must never abort a review because the feedback DB had a hiccup. + +The harvester is intentionally separate from `review_pr` so it can be +called independently (e.g. by the daily analyzer's "backfill" mode) and +tested in isolation against a mocked Gitea client. +""" +from __future__ import annotations + +import json +import logging +import re +import time +import urllib.parse +import urllib.request +from typing import Optional + +import ai_review # used as ai_review.gitea_get(...) so test mocks land on the binding + +from feedback import ( + init, + record_inline_finding, + record_reaction, + record_reply, + record_review, + record_thread_state, + posthash, +) + +log = logging.getLogger("pragent.feedback.harvest") + +# Reviewer identity — only collect feedback on comments authored by us. +# Avoids harvesting reactions on human comments (which we never want to +# count toward "bot usefulness"). +REVIEWER_LOGIN = "pragent-bot" + +# Reactions content tokens Gitea uses. We track +1 / -1 explicitly; the +# others are stored as-is so the analyzer can mine them (👀 eyes, +# laugh, hooray, confused, heart, rocket, …) without hardcoding a list +# that drifts across Gitea versions. +POSITIVE_REACTIONS = {"+1", "heart", "hooray", "laugh", "rocket"} +NEGATIVE_REACTIONS = {"-1", "confused"} +# Note: Gitea's `eyes` reaction (👀) means "I'm watching" — not approval +# or disapproval. Treated as neutral by the analyzer. + +# Phrases that, in a reply, indicate the author thinks the bot's finding +# was wrong. Casing + punctuation ignored; substring match is good enough +# (false positives in the analyzer cost a human minute; false negatives +# hide regressions). +FALSE_POSITIVE_PHRASES = ( + "false positive", "not actually", "this is fine", "this is intentional", + "not a bug", "intentional", "wrong here", "isn't actually", + "is not actually", "don't think this is", "i disagree", "this isn't right", + "this is correct", "this is expected", "by design", "this is by design", +) + +# Gitea review-comment payload includes a 'body' field that may carry our +# sha marker + severity header. We extract severity + path/line from it +# as a fallback when the finding wasn't already seeded at post-time (old +# reviews before feedback.py existed). +SEV_RE = re.compile(r"\*\*\[(CRITICAL|HIGH|MEDIUM|LOW|INFO)\]\*\*", re.IGNORECASE) +PATH_LINE_RE = re.compile(r"`([^?:\n]+?):(\d+)`") +SHA_MARKER_RE = re.compile(r"", re.IGNORECASE) + + +# --------------------------------------------------------------------------- +# Low-level HTTP — tolerant JSON parse (Gitea sometimes returns `null` where +# we expect `[]`, e.g. reactions on a fresh comment) +# --------------------------------------------------------------------------- + +def _gitea_get_json(api: str, repo: str, path: str, token: str) -> tuple[int, object]: + status, raw = ai_review.gitea_get(api, repo, path, token) + if status != 200: + return status, None + try: + return status, json.loads(raw.decode("utf-8", errors="replace")) + except (json.JSONDecodeError, ValueError): + return status, None + + +# --------------------------------------------------------------------------- +# Parse helpers +# --------------------------------------------------------------------------- + +def _parse_severity(body: str) -> str: + m = SEV_RE.search(body or "") + return m.group(1).upper() if m else "INFO" + + +def _parse_path_line(body: str) -> tuple[Optional[str], Optional[int]]: + m = PATH_LINE_RE.search(body or "") + if not m: + return None, None + path = m.group(1).strip() + try: + return path, int(m.group(2)) + except ValueError: + return path, None + + +def _parse_sha(body: str) -> Optional[str]: + m = SHA_MARKER_RE.search(body or "") + return m.group(1) if m else None + + +def _is_negation_reply(body: str) -> bool: + if not body: + return False + norm = body.lower() + return any(p in norm for p in FALSE_POSITIVE_PHRASES) + + +# --------------------------------------------------------------------------- +# Reaction classification (cheap, used by the analyzer — not the harvester +# itself) +# --------------------------------------------------------------------------- + +def classify_reaction(content: str) -> str: + """Bucket a reaction into 'positive', 'negative', or 'neutral'.""" + c = (content or "").strip().lower() + if c in POSITIVE_REACTIONS: + return "positive" + if c in NEGATIVE_REACTIONS: + return "negative" + return "neutral" + + +# --------------------------------------------------------------------------- +# Main harvest entry +# --------------------------------------------------------------------------- + +def harvest_for_pr( + *, + api: str, + token: str, + repo: str, + pr_index: int, + db_path: str, + page_size: int = 50, +) -> dict: + """Walk every bot-authored review on the given PR and record reactions + + thread state + replies. Returns a stats dict for logging. + + `db_path` is the SQLite file path (env: `PRAGENT_FEEDBACK_DB`, + typically `/data/feedback.db` mounted via the `feedback-data` PVC). + """ + conn = init(db_path) + stats = { + "reviews_seen": 0, "findings_seen": 0, + "reactions_recorded": 0, "thread_states_recorded": 0, + "replies_recorded": 0, "errors": 0, + } + + try: + # 1. List every review on the PR (paginated, but PRs rarely have >page_size) + status, payload = _gitea_get_json( + api, repo, f"pulls/{pr_index}/reviews?per_page={page_size}", token, + ) + if status != 200 or not isinstance(payload, list): + log.info("harvest: reviews list failed status=%d", status) + stats["errors"] += 1 + return stats + + for rev in payload: + user = (rev.get("user") or {}).get("login", "") + if user != REVIEWER_LOGIN: + continue + stats["reviews_seen"] += 1 + + review_id_gitea = rev.get("id") + head_sha = rev.get("commit_id", "") + review_body = rev.get("body", "") or "" + body_sha = _parse_sha(review_body) + # Trust the sha marker inside the body — Gitea's commit_id field is + # for the LAST commit, not necessarily the reviewed head. If we + # can't find a marker, fall back to commit_id. + effective_sha = body_sha or head_sha + created_at = _parse_iso_ts(rev.get("created_at", "")) + + db_review_id = record_review( + conn, repo=repo, pr=pr_index, head_sha=effective_sha, + review_id_gitea=review_id_gitea, + posted_at=created_at, + ) + + # 2. Inline comments for this review + if review_id_gitea is None: + continue + rstatus, rpayload = _gitea_get_json( + api, repo, f"pulls/{pr_index}/reviews/{review_id_gitea}/comments", + token, + ) + if rstatus != 200 or not isinstance(rpayload, list): + stats["errors"] += 1 + continue + + for ic in rpayload: + ic_id = ic.get("id") + if ic_id is None: + continue + ic_body = ic.get("body", "") or "" + ic_path = ic.get("path") + ic_line = ic.get("position") or ic.get("line") + ic_severity = _parse_severity(ic_body) + # Fall back to body parse when Gitea didn't echo path/line + if not ic_path or not ic_line: + bp, bl = _parse_path_line(ic_body) + ic_path = ic_path or bp + ic_line = ic_line or bl + + if not ic_path or not ic_line: + log.info( + "harvest: inline %s missing path/line, skipping", ic_id, + ) + continue + + finding_id = record_inline_finding( + conn, review_id=db_review_id, repo=repo, pr=pr_index, + path=ic_path, line=ic_line, severity=ic_severity, + problem=_strip_severity_header(ic_body), + fix="", suggestion="", + comment_id=ic_id, + posted_at=created_at, + ) + stats["findings_seen"] += 1 + if finding_id is None: + continue + + # 3. Reactions on the inline comment + react_status, react_payload = _gitea_get_json( + api, repo, f"issues/comments/{ic_id}/reactions", token, + ) + if react_status == 200 and isinstance(react_payload, list): + for r in react_payload: + ruser = (r.get("user") or {}).get("login", "") or "?" + rcontent = (r.get("content") or "").strip() + if not rcontent: + continue + if record_reaction( + conn, comment_id=ic_id, user=ruser, + content=rcontent, + created_at=_parse_iso_ts(r.get("created_at", "")), + ): + stats["reactions_recorded"] += 1 + + # 4. Thread state (Gitea's `resolver` field on the inline + # comment). Non-empty string = resolved. + resolver = (ic.get("resolver") or "").strip() + if ic.get("resolver") is not None: # field present, even if "" + record_thread_state( + conn, finding_id=finding_id, + resolved=bool(resolver), + ) + stats["thread_states_recorded"] += 1 + + # 5. Replies on this review (issue-comments whose + # `review_comment_id` points at one of our inline comments). + # Some Gitea versions don't expose `review_comment_id` on the + # issue-comment endpoint — in that case `replies` stays + # empty; we degrade gracefully. + try: + _harvest_replies( + api=api, repo=repo, token=token, + pr_index=pr_index, review_id=review_id_gitea, + inline_comments=rpayload, conn=conn, + stats=stats, + ) + except Exception as e: + log.info("harvest: replies fetch failed: %s", e) + stats["errors"] += 1 + finally: + conn.close() + + return stats + + +def _harvest_replies( + *, api: str, repo: str, token: str, pr_index: int, + review_id: int, inline_comments: list, conn, stats: dict, +) -> None: + """Fetch issue comments on this PR; record those whose + `review_comment_id` matches one of our inline comment IDs. + Gitea 1.26 doesn't include that field — we fall back to fetching each + inline comment individually via `issues/comments/{id}` (does include + the field) only if the bulk fetch is empty. + """ + inline_ids = {c.get("id") for c in inline_comments if c.get("id") is not None} + if not inline_ids: + return + + status, payload = _gitea_get_json( + api, repo, f"issues/{pr_index}/comments?per_page=100", token, + ) + if status != 200 or not isinstance(payload, list): + return + + # Build mapping inline_id -> finding_id (one SELECT instead of N) + rows = conn.execute( + "SELECT comment_id, id FROM inline_finding WHERE comment_id IN (" + + ",".join("?" * len(inline_ids)) + ")", + list(inline_ids), + ).fetchall() + inline_to_finding = {r[0]: r[1] for r in rows} + + for c in payload: + rcid = c.get("review_comment_id") + if not rcid or rcid not in inline_to_finding: + continue + author = (c.get("user") or {}).get("login", "") or "?" + body = c.get("body", "") or "" + ts = _parse_iso_ts(c.get("created_at", "")) + if record_reply( + conn, finding_id=inline_to_finding[rcid], + author=author, body=body, created_at=ts, + ): + stats["replies_recorded"] += 1 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _strip_severity_header(body: str) -> str: + """Drop the leading `**[SEVERITY]**` so the posthash captures the + substance, not the severity label.""" + return SEV_RE.sub("", body or "", count=1).strip() + + +def _parse_iso_ts(s: str) -> int: + if not s: + return int(time.time()) + try: + # Python 3.11+ fromisoformat tolerates the trailing 'Z'. + return int(__import__("datetime").datetime.fromisoformat( + s.replace("Z", "+00:00") + ).timestamp()) + except Exception: + return int(time.time()) + + +# --------------------------------------------------------------------------- +# CLI for manual backfill / first-time seed +# --------------------------------------------------------------------------- + +def main() -> int: + import argparse, os + p = argparse.ArgumentParser( + description="Harvest reactions/threads/replies on bot PR comments.", + ) + p.add_argument("--api", default=os.environ.get( + "GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000", + )) + p.add_argument("--token", default=os.environ.get("PRAGENT_BOT_TOKEN", "")) + p.add_argument("--repo", required=True, help="owner/name") + p.add_argument("--pr", type=int, required=True, help="PR index") + p.add_argument("--db", default=os.environ.get( + "PRAGENT_FEEDBACK_DB", "/data/feedback.db", + )) + args = p.parse_args() + + if not args.token: + print("PRAGENT_BOT_TOKEN required", flush=True) + return 2 + + logging.basicConfig(level=logging.INFO) + stats = harvest_for_pr( + api=args.api, token=args.token, + repo=args.repo, pr_index=args.pr, db_path=args.db, + ) + print(json.dumps(stats), flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/pilot/test_feedback_harvest.py b/tests/pilot/test_feedback_harvest.py new file mode 100644 index 0000000..d272fd7 --- /dev/null +++ b/tests/pilot/test_feedback_harvest.py @@ -0,0 +1,243 @@ +"""Tests for pilot/feedback_harvest.py. + +Mock `gitea_get` so we exercise the harvester's flow against canned Gitea +responses. Verify: + - bot-authored reviews only are processed + - reactions + thread state + replies all get recorded + - best-effort failures don't raise (one bad endpoint shouldn't kill the + whole harvest) + - posthash dedup: harvesting the same PR twice does NOT double-count + reactions. +""" +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..", "..", "pilot")) + +import ai_review # noqa: E402 +import feedback # noqa: E402 +import feedback_harvest # noqa: E402 + + +def _make_fake_gitea(routes: dict): + """Build a stand-in for `ai_review.gitea_get` that returns canned bodies. + + `routes` maps relative path (substring) → (status, json_body). Sort + keys longest-first so e.g. `pulls/5/reviews/100/comments` matches + before `pulls/5/reviews` (which is also a substring of the longer + path). + """ + def fake(api, repo, path, token, accept="application/json"): + for needle in sorted(routes.keys(), key=len, reverse=True): + if needle in path: + status, body = routes[needle] + return status, json.dumps(body).encode() + return 404, b'{"message":"not found"}' + return fake + + +def _patch(fake): + """Apply the fake to ai_review.gitea_get and feedback_harvest's import.""" + return patch("ai_review.gitea_get", side_effect=fake) + + +class TestHarvestForPr(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + + def tearDown(self): + self.tmp.cleanup() + + def _review_payload(self, body="", commit_id="abc123", review_id=100): + return [{ + "id": review_id, "user": {"login": "pragent-bot"}, + "commit_id": commit_id, "body": body, + "created_at": "2026-08-20T10:00:00Z", + }] + + def _inline_payload(self, comment_id=500, body="**[LOW]** x", path="a/b.ts", position=42, resolver=""): + return [{ + "id": comment_id, "path": path, "position": position, + "body": body, "resolver": resolver, + }] + + def test_happy_path_records_reaction_and_thread(self): + fake = _make_fake_gitea({ + "pulls/5/reviews": (200, self._review_payload()), + "pulls/5/reviews/100/comments": (200, self._inline_payload(resolver="masi")), + "issues/comments/500/reactions": (200, [ + {"user": {"login": "alice"}, "content": "+1", + "created_at": "2026-08-20T11:00:00Z"}, + {"user": {"login": "bob"}, "content": "-1", + "created_at": "2026-08-20T11:01:00Z"}, + ]), + "issues/5/comments": (200, []), # no replies + }) + with _patch(fake): + stats = feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + self.assertEqual(stats["reviews_seen"], 1) + self.assertEqual(stats["findings_seen"], 1) + self.assertEqual(stats["reactions_recorded"], 2) + self.assertEqual(stats["thread_states_recorded"], 1) + # DB should have 1 review, 1 finding, 2 reactions, 1 thread_state + conn = feedback.init(self.db) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM review").fetchone()[0], 1, + ) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM inline_finding").fetchone()[0], 1, + ) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0], 2, + ) + self.assertEqual( + conn.execute("SELECT resolved FROM thread_state").fetchone()[0], 1, + ) + conn.close() + + def test_skips_non_bot_reviews(self): + fake = _make_fake_gitea({ + "pulls/5/reviews": (200, [{ + "id": 999, "user": {"login": "masi"}, # not the bot + "commit_id": "x", "body": "", "created_at": "2026-08-20T10:00:00Z", + }]), + }) + with _patch(fake): + stats = feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + self.assertEqual(stats["reviews_seen"], 0) + conn = feedback.init(self.db) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM review").fetchone()[0], 0, + ) + conn.close() + + def test_review_list_failure_does_not_raise(self): + fake = _make_fake_gitea({ + "pulls/5/reviews": (500, None), + }) + with _patch(fake): + stats = feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + self.assertEqual(stats["reviews_seen"], 0) + self.assertGreaterEqual(stats["errors"], 1) + + def test_reactions_endpoint_returns_null_is_tolerated(self): + # Some Gitea endpoints return JSON `null` for empty lists. We must + # not crash — treat it as "no reactions". + fake = _make_fake_gitea({ + "pulls/5/reviews": (200, self._review_payload()), + "pulls/5/reviews/100/comments": (200, self._inline_payload()), + "issues/comments/500/reactions": (200, None), + "issues/5/comments": (200, []), + }) + with _patch(fake): + stats = feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + self.assertEqual(stats["reactions_recorded"], 0) + + def test_reactions_dedup_via_pk_across_harvests(self): + # Two harvests of the same PR — both produce an inline_finding row, + # but reactions are PK-deduped on (comment_id, user, content) so + # the SECOND harvest does NOT double-record them. + fake = _make_fake_gitea({ + "pulls/5/reviews": (200, self._review_payload()), + "pulls/5/reviews/100/comments": (200, self._inline_payload()), + "issues/comments/500/reactions": (200, [ + {"user": {"login": "alice"}, "content": "+1", + "created_at": "2026-08-20T11:00:00Z"}, + ]), + "issues/5/comments": (200, []), + }) + with _patch(fake): + feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + conn = feedback.init(self.db) + # Two findings (no DB-level posthash UNIQUE), one reaction. + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM inline_finding").fetchone()[0], 2, + ) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0], 1, + ) + conn.close() + + def test_replies_with_review_comment_id_recorded(self): + fake = _make_fake_gitea({ + "pulls/5/reviews": (200, self._review_payload()), + "pulls/5/reviews/100/comments": (200, self._inline_payload(comment_id=500)), + "issues/comments/500/reactions": (200, []), + "issues/5/comments": (200, [{ + "id": 900, "review_comment_id": 500, + "user": {"login": "alice"}, + "body": "False positive — this is intentional", + "created_at": "2026-08-20T12:00:00Z", + }]), + }) + with _patch(fake): + stats = feedback_harvest.harvest_for_pr( + api="http://x", token="t", repo="o/r", pr_index=5, + db_path=self.db, + ) + self.assertEqual(stats["replies_recorded"], 1) + conn = feedback.init(self.db) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM reply").fetchone()[0], 1, + ) + conn.close() + + +class TestParseHelpers(unittest.TestCase): + def test_severity_extracted(self): + self.assertEqual( + feedback_harvest._parse_severity("**[HIGH]** race in foo"), + "HIGH", + ) + + def test_severity_defaults_to_info(self): + self.assertEqual(feedback_harvest._parse_severity("plain text"), "INFO") + + def test_path_line_extracted(self): + p, l = feedback_harvest._parse_path_line("see `src/foo.ts:42` here") + self.assertEqual(p, "src/foo.ts") + self.assertEqual(l, 42) + + def test_negation_phrases_caught(self): + self.assertTrue(feedback_harvest._is_negation_reply("This is intentional.")) + self.assertTrue(feedback_harvest._is_negation_reply("false positive — see X")) + self.assertFalse(feedback_harvest._is_negation_reply("thanks for catching this!")) + # Empty / None safe + self.assertFalse(feedback_harvest._is_negation_reply("")) + self.assertFalse(feedback_harvest._is_negation_reply(None)) + + def test_classify_reaction(self): + self.assertEqual(feedback_harvest.classify_reaction("+1"), "positive") + self.assertEqual(feedback_harvest.classify_reaction("-1"), "negative") + self.assertEqual(feedback_harvest.classify_reaction("rocket"), "positive") + self.assertEqual(feedback_harvest.classify_reaction("confused"), "negative") + self.assertEqual(feedback_harvest.classify_reaction("eyes"), "neutral") + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 8472f35a58d5dda77b523e51cca49f0c06f02ff5 Mon Sep 17 00:00:00 2001 From: Marcos Date: Sat, 22 Aug 2026 14:46:07 +0000 Subject: [PATCH 3/4] feat(feedback): move feedback analyzer from WIP into pilot/ --- pilot/feedback_analyze.py | 419 +++++++++++++++++++++++++++ tests/pilot/test_feedback_analyze.py | 231 +++++++++++++++ 2 files changed, 650 insertions(+) create mode 100644 pilot/feedback_analyze.py create mode 100644 tests/pilot/test_feedback_analyze.py diff --git a/pilot/feedback_analyze.py b/pilot/feedback_analyze.py new file mode 100644 index 0000000..f5e6cc2 --- /dev/null +++ b/pilot/feedback_analyze.py @@ -0,0 +1,419 @@ +"""pragent pilot — daily feedback analyzer. + +Reads `feedback.db` (written by `feedback_harvest.py`) and produces a +markdown report that: + + 1. Ranks inline findings by **net false-positive score** (downvotes + + unresolved + negation-phrase replies − upvotes − resolved). Top of + this list = "the bot has been wrong about this repeatedly". These + are the candidates that *might* belong in the per-repo + `.pr-review.json:instructions` addendum. + 2. Ranks findings by **net acceptance** — repeated 👍 / resolution = + "the bot's framing here is genuinely useful". These can be promoted + to the shared `architecture.md` so they don't have to be re-derived + every PR. + 3. Reports a **restraint metric** — for every PR where the bot posted + zero findings, count how often a human reviewer also posted zero + substantive review comments. When the bot is loud on clean code, + that's a false-positive rate we can act on (DoorDash lesson: + "excessive noise on clean code is its own failure mode"). + 4. Reports a **case-review queue** — every disagreement case (a + downvote, unresolved, or a reply matching `FALSE_POSITIVE_PHRASES`) + is listed in full so a human can re-read the original PR and decide + if the finding was right or wrong. + +Output is plain markdown so it can be posted as a Gitea issue / comment +without rendering work. Designed to be reviewed by a human, not auto- +applied — per the DoorDash pattern, every material change to model / +prompt / context goes through a benchmark gate first; this report IS +that gate (or, more precisely, the queue feeding the gate). + +Never raises. A bad DB / no data → returns a friendly empty-state report. +""" +from __future__ import annotations + +import argparse +import json +import logging +import os +import sqlite3 +from collections import defaultdict +from datetime import datetime, timezone +from typing import Optional + +import feedback +from feedback_harvest import ( + FALSE_POSITIVE_PHRASES, + classify_reaction, + _is_negation_reply, # noqa: F401 (re-exported for the test suite) +) + +log = logging.getLogger("pragent.feedback.analyze") + +# How many findings to surface in each top-list. Capped because the +# reports are read by humans; more than 20 per list and they skim. +TOP_N = 20 + +# Restraint threshold — fraction of "clean" PRs (zero findings) where +# the bot produced ANY findings. Above this we recommend `.pr-review.json: +# exclude_patterns` or a stricter `severity_threshold`. +RESTRAINT_NOISE_THRESHOLD = 0.25 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _net_score(row) -> tuple[int, int]: + """Return (false_positive_score, acceptance_score) for one finding row. + + FP signals: downvotes (+1), unresolved (+1), negation-phrase replies (+2). + Acceptance signals: upvotes (+1), resolved (+1). + """ + fp = 0 + ac = 0 + fp += int(row["downvotes"] or 0) + fp += 1 if row["resolved"] == 0 else 0 # 0/1/NULL; 0 = unresolved + ac += 1 if row["resolved"] == 1 else 0 + ac += int(row["upvotes"] or 0) + if row["reply_bodies"] and _is_negation_reply(row["reply_bodies"]): + fp += 2 + return fp, ac + + +def _short_problem(problem: str, n: int = 100) -> str: + s = (problem or "").strip().replace("\n", " ") + return s if len(s) <= n else s[: n - 1] + "…" + + +def _restraint_stats(conn: sqlite3.Connection) -> dict: + """How often does the bot post findings on PRs that received zero + bot findings (= presumably clean)? Looks at `review.findings_total` + if present, otherwise counts `inline_finding` per PR. + + NOTE: until `post_inline_review` records `findings_total`, this falls + back to "PRs with at least one finding row" which is an underestimate + (a bot review with zero findings leaves no row). + """ + total_prs_with_review = conn.execute( + "SELECT COUNT(DISTINCT repo || '#' || pr) FROM review" + ).fetchone()[0] + prs_with_findings = conn.execute( + "SELECT COUNT(DISTINCT repo || '#' || pr) FROM inline_finding" + ).fetchone()[0] + if total_prs_with_review == 0: + return {"total": 0, "noisy": 0, "ratio": 0.0} + # This is currently "PRs where the bot left at least one inline + # comment". A precise "findings_total per review" needs + # post_inline_review to record it (TODO in the wiring step). Until + # then, treat this as a floor: real noise is >= this. + return { + "total": total_prs_with_review, + "noisy": prs_with_findings, + "ratio": prs_with_findings / total_prs_with_review, + } + + +def _case_review_queue(conn: sqlite3.Connection, limit: int = 30) -> list[dict]: + """Findings that humans pushed back on — for manual re-review.""" + rows = feedback.findings_with_votes(conn) + cases = [] + for r in rows: + fp_score, _ = _net_score(r) + if fp_score <= 0: + continue + cases.append({ + "posthash": r["posthash"], + "repo": r["repo"], + "pr": r["pr"], + "path": r["path"], + "line": r["line"], + "severity": r["severity"], + "problem": _short_problem(r["problem"], 200), + "fp_score": fp_score, + "upvotes": r["upvotes"] or 0, + "downvotes": r["downvotes"] or 0, + "resolved": r["resolved"], + "reply_count": r["reply_count"] or 0, + "reply_excerpt": _short_problem(r["reply_bodies"] or "", 200), + }) + cases.sort(key=lambda c: c["fp_score"], reverse=True) + return cases[:limit] + + +def _format_table(headers: list[str], rows: list[list[str]]) -> str: + if not rows: + return "_none yet_\n" + out = ["| " + " | ".join(headers) + " |", + "|" + "|".join(["---"] * len(headers)) + "|"] + for row in rows: + out.append("| " + " | ".join(row) + " |") + return "\n".join(out) + "\n" + + +def _md_escape(s: str) -> str: + """Escape pipes + newlines so the value stays in one table cell.""" + return (s or "").replace("|", "\\|").replace("\n", " ").strip() + + +# --------------------------------------------------------------------------- +# Main report builder +# --------------------------------------------------------------------------- + +def analyze(db_path: str, *, since_ts: Optional[int] = None, + as_json: bool = False) -> str: + """Build the daily report. Returns a markdown string by default; + `as_json=True` returns a structured dict (for tests + dashboards).""" + conn = feedback.init(db_path) + try: + findings = list(feedback.findings_with_votes(conn, since_ts=since_ts)) + total_findings = len(findings) + repo_set = {f["repo"] for f in findings} + case_queue = _case_review_queue(conn) + restraint = _restraint_stats(conn) + + # Compute scores + scored: list[tuple[int, int, sqlite3.Row]] = [] + for f in findings: + fp, ac = _net_score(f) + scored.append((fp, ac, f)) + + # Top false-positive patterns (sorted by fp score, deduped by posthash). + # `occurrences` comes from the inline_finding row — posthash UNIQUE + # means a single row can carry a count > 1 (set by record_inline_finding's + # ON CONFLICT DO UPDATE). + fp_by_hash: dict[str, dict] = {} + for fp, ac, f in scored: + if fp <= 0: + continue + ph = f["posthash"] + entry = fp_by_hash.setdefault(ph, { + "posthash": ph, "fp_score": 0, "ac_score": 0, + "repo": f["repo"], "path": f["path"], "line": f["line"], + "severity": f["severity"], "problem": f["problem"], + "occurrences": f["occurrences"], "upvs": 0, "downs": 0, + "resolved_true": 0, "resolved_false": 0, + }) + entry["fp_score"] += fp + entry["ac_score"] += ac + entry["upvs"] += f["upvotes"] or 0 + entry["downs"] += f["downvotes"] or 0 + if f["resolved"] == 1: + entry["resolved_true"] += 1 + elif f["resolved"] == 0: + entry["resolved_false"] += 1 + fp_sorted = sorted( + fp_by_hash.values(), key=lambda e: e["fp_score"], reverse=True, + )[:TOP_N] + + # Top accepted patterns + ac_by_hash: dict[str, dict] = {} + for fp, ac, f in scored: + if ac <= 0: + continue + ph = f["posthash"] + entry = ac_by_hash.setdefault(ph, { + "posthash": ph, "ac_score": 0, "fp_score": 0, + "repo": f["repo"], "path": f["path"], "line": f["line"], + "severity": f["severity"], "problem": f["problem"], + "occurrences": f["occurrences"], "upvs": 0, "downs": 0, + "resolved_true": 0, + }) + entry["ac_score"] += ac + entry["fp_score"] += fp + entry["upvs"] += f["upvotes"] or 0 + entry["downs"] += f["downvotes"] or 0 + if f["resolved"] == 1: + entry["resolved_true"] += 1 + ac_sorted = sorted( + ac_by_hash.values(), key=lambda e: e["ac_score"], reverse=True, + )[:TOP_N] + + # Restraint recommendation + if restraint["ratio"] > RESTRAINT_NOISE_THRESHOLD: + restraint_msg = ( + f"⚠️ Bot posted findings on **{restraint['ratio']:.0%}** of " + f"reviewed PRs ({restraint['noisy']} / {restraint['total']}). " + f"Above the {RESTRAINT_NOISE_THRESHOLD:.0%} threshold — " + "consider raising `.pr-review.json:severity_threshold` to " + "`medium` or `high` for noisy repos, or adding " + "`patterns.deny` to skip stylistic-only findings." + ) + else: + restraint_msg = ( + f"✅ Bot stayed quiet on **{1 - restraint['ratio']:.0%}** of " + f"reviewed PRs ({restraint['total'] - restraint['noisy']} / " + f"{restraint['total']}). Restraint OK." + ) + + if as_json: + return json.dumps({ + "total_findings": total_findings, + "repos_seen": sorted(repo_set), + "restraint": restraint, + "top_false_positive": fp_sorted, + "top_accepted": ac_sorted, + "case_review_queue": case_queue, + "restraint_msg": restraint_msg, + }, indent=2) + + # Markdown + ts_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + out = [f"# pragent feedback report — {ts_str}", ""] + out.append(f"- **findings analyzed**: {total_findings}") + out.append(f"- **repos with feedback**: {len(repo_set)} " + f"({', '.join(sorted(repo_set))})") + out.append(f"- **case-review queue**: {len(case_queue)} disagreement(s)") + out.append("") + out.append("## Restraint") + out.append("") + out.append(restraint_msg) + out.append("") + out.append("> DoorDash rule (2026-07-06): *excessive noise on clean " + "code is its own failure mode*. `severity_threshold` + " + "`patterns.deny` are the knobs that dial restraint.") + out.append("") + + out.append(f"## Top {len(fp_sorted)} false-positive candidates") + out.append("") + out.append("Aggregated by `posthash` (path:line:severity:problem). " + "Sort key = downvotes + unresolved + negation-phrase replies " + "− upvotes − resolved.") + out.append("") + rows = [] + for e in fp_sorted: + rows.append([ + str(e["fp_score"]), + f"`{_md_escape(e['repo'])}`", + f"`{_md_escape(e['path'])}:{e['line']}`", + e["severity"], + _md_escape(_short_problem(e["problem"])), + f"👍{e['upvs']} 👎{e['downs']}", + f"✅{e['resolved_true']} ❌{e['resolved_false']}", + str(e["occurrences"]), + ]) + out.append(_format_table( + ["FP", "repo", "path:line", "sev", "problem", + "votes", "resolved", "seen"], + rows, + )) + out.append("") + out.append("_Review each row before adding it to " + "`.pr-review.json:instructions`. Human reactions are NOT " + "ground truth (DoorDash, 2026-07-06: authors accept/reject " + "for workflow reasons) — re-read the PR before acting._") + out.append("") + + out.append(f"## Top {len(ac_sorted)} accepted patterns") + out.append("") + out.append("Aggregated by posthash. Sort key = upvotes + resolved − " + "downvotes − unresolved − negation-phrase replies.") + out.append("") + rows = [] + for e in ac_sorted: + rows.append([ + str(e["ac_score"]), + f"`{_md_escape(e['repo'])}`", + f"`{_md_escape(e['path'])}:{e['line']}`", + e["severity"], + _md_escape(_short_problem(e["problem"])), + f"👍{e['upvs']} 👎{e['downs']}", + f"✅{e['resolved_true']}", + str(e["occurrences"]), + ]) + out.append(_format_table( + ["AC", "repo", "path:line", "sev", "problem", + "votes", "resolved", "seen"], + rows, + )) + out.append("") + out.append("_Promote widely-accepted patterns into the shared " + "`architecture.md` on Nexus raw-hosted (or the per-repo " + "`additional_context_urls`). These become part of the " + "prompt-cached prefix → ~0 marginal cost on step 2+._") + out.append("") + + out.append(f"## Case-review queue ({len(case_queue)})") + out.append("") + if not case_queue: + out.append("_No disagreements recorded yet. Once humans start " + "reacting 👎 / leaving replies / not resolving bot " + "comments, cases will appear here._") + else: + out.append("Each row needs a human to re-read the original PR and " + "decide: was the bot right? If not, draft an " + "`instructions` addendum or a `patterns.deny` rule.") + out.append("") + for c in case_queue: + url = ( + f"https://gitea.marcospaulo.dev.br/{c['repo']}/pulls/" + f"{c['pr']}/files#r{c['posthash']}" + ) + out.append(f"### FP={c['fp_score']} · {c['repo']}#{c['pr']}") + out.append( + f"- file: `{_md_escape(c['path'])}:{c['line']}` · " + f"severity: `{c['severity']}`", + ) + out.append(f"- problem: {_md_escape(c['problem'])}") + out.append( + f"- signals: 👍{c['upvotes']} 👎{c['downvotes']} · " + f"resolved={c['resolved']} · replies={c['reply_count']}", + ) + if c["reply_excerpt"]: + out.append( + f"- last reply: {_md_escape(c['reply_excerpt'])}", + ) + out.append(f"- posthash: `{c['posthash']}`") + out.append("") + + out.append("## Where this report goes") + out.append("") + out.append("- **Per-repo actions** (`.pr-review.json:instructions`, " + "`patterns.deny`, `severity_threshold`): edit the file on " + "`main` via a regular PR. The next PR review picks up the " + "change automatically.") + out.append("- **Cross-repo actions** (shared house-rules): update the " + "`PRAGENT_ADDITIONAL_CONTEXT_URL` document on Nexus " + "raw-hosted (`canalhandia/architecture.md` etc).") + out.append("- **Benchmark gate** (DoorDash pattern): before changing " + "the model / prompt / context window, replay this report " + "against the labeled `posthash` corpus. If a candidate " + "addendum flips ≥ 1 currently-accepted finding into " + "false-positive, drop it.") + out.append("") + out.append(f"_Generated from `{db_path}` by `feedback_analyze.py`._") + return "\n".join(out) + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def main() -> int: + p = argparse.ArgumentParser(description="Build the daily feedback report.") + p.add_argument("--db", default=os.environ.get( + "PRAGENT_FEEDBACK_DB", "/data/feedback.db", + )) + p.add_argument("--since", type=int, default=None, + help="Unix timestamp; only include findings posted since") + p.add_argument("--json", action="store_true", + help="Emit structured JSON instead of markdown") + p.add_argument("--out", default="-", + help="Write to this path instead of stdout ('-' = stdout)") + args = p.parse_args() + + out = analyze(args.db, since_ts=args.since, as_json=args.json) + if args.out == "-": + print(out) + else: + with open(args.out, "w") as f: + f.write(out) + print(f"wrote {args.out}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + import sys + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/pilot/test_feedback_analyze.py b/tests/pilot/test_feedback_analyze.py new file mode 100644 index 0000000..f0bfd0d --- /dev/null +++ b/tests/pilot/test_feedback_analyze.py @@ -0,0 +1,231 @@ +"""Tests for pilot/feedback_analyze.py. + +Verify: + - empty DB produces a friendly empty-state report (no crash) + - findings are aggregated by posthash across multiple PRs + - net false-positive score weights downvotes + unresolved + negation + replies; acceptance weights upvotes + resolved + - restraint metric reports the right ratio + - case-review queue lists every disagreement + - markdown + JSON output modes both work +""" +import json +import os +import sys +import tempfile +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..", "..", "pilot")) + +import feedback # noqa: E402 +import feedback_analyze # noqa: E402 + + +def _seed(conn, findings): + """Helper: insert a list of (repo, pr, path, line, severity, problem, + [reaction users/contents], [reply bodies], [resolved]) tuples. + Each finding gets a fresh review row + a unique comment_id so the + reaction-join in `findings_with_votes` matches.""" + for f in findings: + (repo, pr_idx, path, line, sev, problem, reacts, replies, + resolved) = f + rid = feedback.record_review(conn, repo=repo, pr=pr_idx, head_sha="x") + cid = (hash((repo, pr_idx, path, line, sev, problem)) & 0xFFFFFFFF) or 1 + fid = feedback.record_inline_finding( + conn, review_id=rid, repo=repo, pr=pr_idx, + path=path, line=line, severity=sev, problem=problem, + comment_id=cid, + ) + for user, content in reacts: + feedback.record_reaction( + conn, comment_id=cid, user=user, content=content, + ) + for i, body in enumerate(replies): + feedback.record_reply( + conn, finding_id=fid, author="alice", + body=body, created_at=1000 + i, + ) + if resolved is not None: + feedback.record_thread_state( + conn, finding_id=fid, resolved=resolved, + ) + + +class TestEmptyState(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + + def tearDown(self): + self.tmp.cleanup() + + def test_empty_db_markdown_does_not_crash(self): + report = feedback_analyze.analyze(self.db) + self.assertIn("# pragent feedback report", report) + self.assertIn("findings analyzed**: 0", report) + self.assertIn("Restraint", report) + + def test_empty_db_json_has_zero_findings(self): + report = feedback_analyze.analyze(self.db, as_json=True) + d = json.loads(report) + self.assertEqual(d["total_findings"], 0) + self.assertEqual(d["restraint"]["total"], 0) + + +class TestScoring(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + self.conn = feedback.init(self.db) + # Two PRs, three findings: + # A: 👍×2, resolved=true → acceptance + # B: 👎×2, unresolved, "false positive" reply → false-positive + # C: no signals → ignored + _seed(self.conn, [ + ("o/r", 1, "a.ts", 10, "HIGH", "race in handler", + [("u1", "+1"), ("u2", "+1")], [], True), + ("o/r", 1, "b.ts", 20, "LOW", "missing semicolon", + [("u1", "-1"), ("u2", "-1")], + ["False positive — this is fine."], False), + ("o/r", 1, "c.ts", 30, "INFO", "naming nit", + [], [], None), + ]) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def test_accepted_ranked_above_fp(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + self.assertEqual(len(d["top_accepted"]), 1) + self.assertEqual(d["top_accepted"][0]["path"], "a.ts") + self.assertEqual(len(d["top_false_positive"]), 1) + self.assertEqual(d["top_false_positive"][0]["path"], "b.ts") + + def test_fp_score_combines_signals(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + fp = d["top_false_positive"][0] + # 2 downvotes + 1 unresolved + 2 (negation phrase) = 5 + self.assertEqual(fp["fp_score"], 5) + + def test_acceptance_score(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + ac = d["top_accepted"][0] + # 2 upvotes + 1 resolved = 3 + self.assertEqual(ac["ac_score"], 3) + + def test_case_queue_contains_only_disagreements(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + queue = d["case_review_queue"] + self.assertEqual(len(queue), 1) + self.assertEqual(queue[0]["path"], "b.ts") + + def test_no_signal_finding_is_ignored(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + # c.ts has no votes, no replies → not in either top list. + paths = {e["path"] for e in d["top_accepted"]} + paths.update(e["path"] for e in d["top_false_positive"]) + self.assertNotIn("c.ts", paths) + + +class TestRestraint(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + self.conn = feedback.init(self.db) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def test_high_ratio_triggers_recommendation(self): + # 3 reviews, all with findings → 100% "noisy". + for pr_i in range(3): + feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") + # Distinct (path, line) per PR so posthash doesn't dedup. + feedback.record_inline_finding( + self.conn, review_id=None, repo="o/r", pr=pr_i, + path=f"a{pr_i}.ts", line=1, severity="LOW", + problem=f"x {pr_i}", + ) + report = feedback_analyze.analyze(self.db) + self.assertIn("⚠️", report) + self.assertIn("100%", report) + + def test_low_ratio_passes(self): + # 4 reviews, 1 with findings → 25% noisy = at threshold. + for pr_i in range(4): + feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") + feedback.record_inline_finding( + self.conn, review_id=None, repo="o/r", pr=0, + path="a.ts", line=1, severity="LOW", problem="x", + ) + report = feedback_analyze.analyze(self.db) + self.assertIn("✅", report) + + +class TestMarkdownOutput(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + self.conn = feedback.init(self.db) + _seed(self.conn, [ + ("o/r", 1, "a.ts", 10, "HIGH", "race in handler", + [("u1", "+1")], [], True), + ]) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def test_report_has_sections(self): + r = feedback_analyze.analyze(self.db) + for section in ( + "# pragent feedback report", + "## Restraint", + "## Top", + "## Case-review queue", + "## Where this report goes", + ): + self.assertIn(section, r) + + def test_doordash_rule_quoted(self): + r = feedback_analyze.analyze(self.db) + # The "noise on clean code" sentence from the DoorDash recap. + self.assertIn("noise on clean code", r) + + +class TestPosthashAggregation(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + self.conn = feedback.init(self.db) + # Same finding on three PRs → one aggregated row. + # Each PR has its own review + finding (comment_id differs but + # posthash is identical, so they collapse on aggregation). + for pr_i in range(3): + rid = feedback.record_review(self.conn, repo="o/r", pr=pr_i, head_sha="x") + feedback.record_inline_finding( + self.conn, review_id=rid, repo="o/r", pr=pr_i, + path="a.ts", line=10, severity="HIGH", + problem="identical problem text", + comment_id=1000 + pr_i, + ) + feedback.record_reaction( + self.conn, comment_id=1000 + pr_i, user="u", content="+1", + ) + + def tearDown(self): + self.conn.close() + self.tmp.cleanup() + + def test_three_occurrences_one_row(self): + d = json.loads(feedback_analyze.analyze(self.db, as_json=True)) + self.assertEqual(len(d["top_accepted"]), 1) + self.assertEqual(d["top_accepted"][0]["occurrences"], 3) + self.assertEqual(d["top_accepted"][0]["ac_score"], 3) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 4e5f43ada7dfa6087e501ee7fe4c34abcd30c6fa Mon Sep 17 00:00:00 2001 From: Marcos Date: Sat, 22 Aug 2026 14:46:07 +0000 Subject: [PATCH 4/4] feat(feedback): move feedback poster from WIP into pilot/ --- pilot/feedback_post.py | 128 ++++++++++++++++++++++++++++++ tests/pilot/test_feedback_post.py | 97 ++++++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 pilot/feedback_post.py create mode 100644 tests/pilot/test_feedback_post.py diff --git a/pilot/feedback_post.py b/pilot/feedback_post.py new file mode 100644 index 0000000..0c519e1 --- /dev/null +++ b/pilot/feedback_post.py @@ -0,0 +1,128 @@ +"""pragent pilot — daily feedback report delivery. + +Calls `feedback_analyze.analyze()` and posts the markdown report as a +comment on a single long-lived "feedback roll-up" issue in +`gitea_admin/pragent`. Comments are append-only history — one comment per +run, timestamped in the body. This keeps every report in one place, easy +to scroll, and avoids the issue-explosion of "one issue per day". + +If the issue doesn't exist yet, create it. Subsequent runs just add a +new comment. + +Designed for the daily K8s CronJob (`k8s/pragent-feedback-cronjob.yaml`) +but runnable from CLI for ad-hoc checks. + +Env: + GITEA_API in-cluster Gitea base URL + PRAGENT_BOT_TOKEN bot token (Write collaborator on gitea_admin/pragent) + PRAGENT_FEEDBACK_DB path to SQLite (default /data/feedback.db) + PRAGENT_FEEDBACK_ISSUE_REPO default gitea_admin/pragent + PRAGENT_FEEDBACK_ISSUE_TITLE default "pragent feedback roll-up" +""" +from __future__ import annotations + +import argparse +import json +import logging +import os +import sys + +import ai_review + +from feedback_analyze import analyze + +log = logging.getLogger("pragent.feedback.post") + + +REPO_DEFAULT = "gitea_admin/pragent" +TITLE_DEFAULT = "pragent feedback roll-up" + + +def _find_or_create_issue(api: str, token: str, repo: str, title: str) -> int: + """Locate the open issue with this title; create one if missing. + + Gitea's issue search is via `GET /repos/{o}/{r}/issues?state=open&q=...` + (q matches title + body). We filter client-side for the exact title + to avoid query-text false matches. + """ + status, raw = ai_review.gitea_get(api, repo, "issues?state=open&per_page=50", token) + if status == 200: + try: + for issue in json.loads(raw): + if issue.get("title") == title: + # NB: the comment URL needs the per-repo `number`, not the + # global `id`. `id=60 num=8` for an early-N create; we want + # `num=8` for `/repos/o/r/issues/8/comments`. + return int(issue["number"]) + except (json.JSONDecodeError, ValueError, KeyError): + pass + # Create + status, raw = ai_review.gitea_post( + api, repo, "issues", token, + {"title": title, "body": "pragent feedback roll-up — auto-created."}, + ) + if status not in (200, 201): + raise RuntimeError(f"issue create failed: HTTP {status} body={raw[:200]!r}") + return int(json.loads(raw)["number"]) + + +def _post_comment(api: str, token: str, repo: str, issue_number: int, body: str) -> int: + status, raw = ai_review.gitea_post( + api, repo, f"issues/{issue_number}/comments", token, {"body": body}, + ) + if status not in (200, 201): + raise RuntimeError(f"comment post failed: HTTP {status} body={raw[:200]!r}") + return json.loads(raw)["id"] + + +def deliver( + *, api: str, token: str, db_path: str, + repo: str = REPO_DEFAULT, title: str = TITLE_DEFAULT, + since_ts: int | None = None, +) -> dict: + """Build the report and post it as a comment. Returns a stats dict.""" + report = analyze(db_path, since_ts=since_ts) + issue_id = _find_or_create_issue(api, token, repo, title) + comment_id = _post_comment(api, token, repo, issue_id, report) + return { + "repo": repo, "issue_id": issue_id, "comment_id": comment_id, + "report_bytes": len(report.encode()), + } + + +def main() -> int: + p = argparse.ArgumentParser( + description="Post the daily feedback report to Gitea.", + ) + p.add_argument("--api", default=os.environ.get( + "GITEA_API", "http://gitea-http.gitea.svc.cluster.local:3000", + )) + p.add_argument("--token", default=os.environ.get("PRAGENT_BOT_TOKEN", "")) + p.add_argument("--db", default=os.environ.get( + "PRAGENT_FEEDBACK_DB", "/data/feedback.db", + )) + p.add_argument("--repo", default=os.environ.get( + "PRAGENT_FEEDBACK_ISSUE_REPO", REPO_DEFAULT, + )) + p.add_argument("--title", default=os.environ.get( + "PRAGENT_FEEDBACK_ISSUE_TITLE", TITLE_DEFAULT, + )) + p.add_argument("--since", type=int, default=None, + help="Unix timestamp; only include findings posted since") + args = p.parse_args() + + if not args.token: + print("PRAGENT_BOT_TOKEN required", flush=True) + return 2 + + logging.basicConfig(level=logging.INFO) + stats = deliver( + api=args.api, token=args.token, db_path=args.db, + repo=args.repo, title=args.title, since_ts=args.since, + ) + print(json.dumps(stats), flush=True) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/tests/pilot/test_feedback_post.py b/tests/pilot/test_feedback_post.py new file mode 100644 index 0000000..a835a35 --- /dev/null +++ b/tests/pilot/test_feedback_post.py @@ -0,0 +1,97 @@ +"""Tests for pilot/feedback_post.py — report delivery to Gitea. + +Mock `ai_review.gitea_get` + `gitea_post` so we exercise the find-or-create ++ comment-post flow without hitting the real API. +""" +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, os.path.join(HERE, "..", "..", "pilot")) + +import feedback # noqa: E402 +import feedback_analyze # noqa: E402 +import feedback_post # noqa: E402 + + +def _make_fake(method_routes: dict): + """`method_routes` maps HTTP path substring → (status, body, method). + + For our purposes both gitea_get and gitea_post share the same fake — + gitea_get is GET, gitea_post is POST, and the post helper also has a + body param. The fake returns whatever the route's body says. + """ + def fake_get(api, repo, path, token, accept="application/json"): + for needle in sorted(method_routes.keys(), key=len, reverse=True): + status, body, _m = method_routes[needle] + if needle in path: + return status, json.dumps(body).encode() + return 404, b'{"message":"not found"}' + + def fake_post(api, repo, path, token, body): + for needle in sorted(method_routes.keys(), key=len, reverse=True): + status, resp_body, _m = method_routes[needle] + if needle in path: + return status, json.dumps(resp_body).encode() + return 404, b'{"message":"not found"}' + + return fake_get, fake_post + + +class TestDeliver(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db = f"{self.tmp.name}/f.db" + conn = feedback.init(self.db) + rid = feedback.record_review(conn, repo="o/r", pr=1, head_sha="x") + feedback.record_inline_finding( + conn, review_id=rid, repo="o/r", pr=1, + path="a.ts", line=1, severity="HIGH", + problem="x", comment_id=99, + ) + conn.close() + + def tearDown(self): + self.tmp.cleanup() + + def test_creates_issue_then_posts_comment(self): + routes = { + "issues?state=open": (200, [], "GET"), # no existing issue + "issues": (201, {"id": 42, "number": 7, "title": "..."}, "POST"), + "issues/7/comments": (201, {"id": 777}, "POST"), + } + fake_get, fake_post = _make_fake(routes) + with patch("ai_review.gitea_get", side_effect=fake_get), \ + patch("ai_review.gitea_post", side_effect=fake_post): + stats = feedback_post.deliver( + api="http://x", token="t", db_path=self.db, + repo="gitea_admin/pragent", title="pragent feedback roll-up", + ) + self.assertEqual(stats["issue_id"], 7) + self.assertEqual(stats["comment_id"], 777) + + def test_reuses_existing_issue(self): + routes = { + "issues?state=open": (200, [ + {"id": 99, "number": 9, "title": "pragent feedback roll-up"}, + {"id": 100, "number": 10, "title": "something else"}, + ], "GET"), + "issues/9/comments": (201, {"id": 888}, "POST"), + } + fake_get, fake_post = _make_fake(routes) + with patch("ai_review.gitea_get", side_effect=fake_get), \ + patch("ai_review.gitea_post", side_effect=fake_post): + stats = feedback_post.deliver( + api="http://x", token="t", db_path=self.db, + repo="gitea_admin/pragent", title="pragent feedback roll-up", + ) + self.assertEqual(stats["issue_id"], 9) + self.assertEqual(stats["comment_id"], 888) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file