gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(feedback): move feedback analyzer from WIP into pilot/

Browse Source
This commit is contained in:
Marcos
2026-08-22 14:46:07 +00:00
parent 69e1fc06a2
commit 8472f35a58
2 changed files with 650 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/feedback_analyze.py
+419
View File
@@ -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())
tests/pilot/test_feedback_analyze.py
+231
View File
@@ -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()