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

Merge agent-A: move feedback*.py from WIP into tree

Browse Source
This commit is contained in:
Marcos
2026-08-22 14:46:46 +00:00
parent a9e1b7ddfc 4e5f43ada7
commit f3a09b9397
8 changed files with 2177 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pilot/feedback.py
+356
View File
@@ -0,0 +1,356 @@
"""pragent pilot — feedback storage.
A thin SQLite layer that records every bot review comment + the reactions /
thread-state / replies it accumulates over time. Powers the daily analysis
that produces suggested addenda for `.pr-review.json:instructions` and
`PRAGENT_ADDITIONAL_CONTEXT_URL` (see `feedback_analyze.py`).
Why SQLite: stdlib, no extra deps in the container, single writer (the
webhook server is one process per pod). Mount at `/data/feedback.db`
via the `feedback-data` PVC.
Schema (idempotent — safe to call `init` at every boot):
review(repo, pr, head_sha, body_comment_id, posted_at, review_id_gitea)
inline_finding(review_id → review.id, repo, pr, path, line,
severity, problem, fix, suggestion,
comment_id, posthash UNIQUE, posted_at)
reaction(comment_id, user, content, created_at,
PRIMARY KEY (comment_id, user, content))
thread_state(finding_id → inline_finding.id, resolved, checked_at,
PRIMARY KEY (finding_id))
reply(finding_id → inline_finding.id, author, body, created_at,
PRIMARY KEY (finding_id, created_at))
`posthash` is a short hash of (path|line|severity|first 80 chars of problem).
It survives across reviews of the same finding on the same line — same
finding on PR #5 and PR #12 of the same file de-duplicate, so the daily
analyzer can count votes across reviews instead of one-at-a-time.
Everything is best-effort. The webhook server never aborts a review
because the feedback DB had a hiccup — `record_*` functions log and
swallow.
"""
from __future__ import annotations
import hashlib
import logging
import sqlite3
import time
from typing import Iterable, Optional
log = logging.getLogger("pragent.feedback")
# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
_SCHEMA = """
CREATE TABLE IF NOT EXISTS review (
id INTEGER PRIMARY KEY AUTOINCREMENT,
repo TEXT NOT NULL,
pr INTEGER NOT NULL,
head_sha TEXT NOT NULL,
review_id_gitea INTEGER,
body_comment_id INTEGER,
posted_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS review_repo_pr ON review(repo, pr);
CREATE TABLE IF NOT EXISTS inline_finding (
id INTEGER PRIMARY KEY AUTOINCREMENT,
review_id INTEGER REFERENCES review(id),
repo TEXT NOT NULL,
pr INTEGER NOT NULL,
path TEXT NOT NULL,
line INTEGER NOT NULL,
severity TEXT NOT NULL,
problem TEXT NOT NULL,
fix TEXT,
suggestion TEXT,
comment_id INTEGER,
posthash TEXT NOT NULL,
posted_at INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS inline_finding_posthash_idx ON inline_finding(posthash);
CREATE INDEX IF NOT EXISTS inline_finding_repo_pr ON inline_finding(repo, pr);
CREATE INDEX IF NOT EXISTS inline_finding_posthash ON inline_finding(posthash);
CREATE TABLE IF NOT EXISTS reaction (
comment_id INTEGER NOT NULL,
user TEXT NOT NULL,
content TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (comment_id, user, content)
);
CREATE INDEX IF NOT EXISTS reaction_comment ON reaction(comment_id);
CREATE TABLE IF NOT EXISTS thread_state (
finding_id INTEGER NOT NULL REFERENCES inline_finding(id),
resolved INTEGER NOT NULL,
checked_at INTEGER NOT NULL,
PRIMARY KEY (finding_id)
);
CREATE TABLE IF NOT EXISTS reply (
finding_id INTEGER NOT NULL REFERENCES inline_finding(id),
author TEXT NOT NULL,
body TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (finding_id, created_at)
);
"""
def init(db_path: str) -> sqlite3.Connection:
"""Open (or create) the DB, ensure schema. Returns a Connection."""
conn = sqlite3.connect(db_path)
conn.row_factory = sqlite3.Row # so callers can use row["name"]
conn.executescript(_SCHEMA)
conn.commit()
return conn
# ---------------------------------------------------------------------------
# Posthash — cross-review finding dedup
# ---------------------------------------------------------------------------
def posthash(path: str, line: int, severity: str, problem: str) -> str:
"""Short stable hash of the finding's identifying triple + a problem
fingerprint. Designed so two reviews of the SAME finding (same file,
same line, same severity, same core complaint) collapse to one row —
reactions across PRs aggregate.
`line` is the post-change (RIGHT-side) line — the agent anchors on it
and so does this hash. Different lines = different finding, by design.
`severity` participates because "this is a CRITICAL bug" and "this is a
LOW nitpick" at the same line on the same problem text are different
signals to learn from.
"""
h = hashlib.sha256()
h.update(f"{path}\n".encode())
h.update(f"{line}\n".encode())
h.update(f"{severity.upper()}\n".encode())
h.update(problem[:80].strip().lower().encode())
return h.hexdigest()[:16]
# ---------------------------------------------------------------------------
# Write helpers — all best-effort. Log + swallow.
# ---------------------------------------------------------------------------
def record_review(
conn: sqlite3.Connection,
*,
repo: str,
pr: int,
head_sha: str,
review_id_gitea: Optional[int] = None,
body_comment_id: Optional[int] = None,
posted_at: Optional[int] = None,
) -> Optional[int]:
"""Insert a review row. Returns the new row id, or None on failure."""
try:
cur = conn.execute(
"INSERT INTO review(repo, pr, head_sha, review_id_gitea, body_comment_id, posted_at) "
"VALUES(?,?,?,?,?,?)",
(repo, pr, head_sha, review_id_gitea, body_comment_id, posted_at or int(time.time())),
)
conn.commit()
return cur.lastrowid
except Exception as e:
log.warning("record_review failed: %s", e)
return None
def record_inline_finding(
conn: sqlite3.Connection,
*,
review_id: Optional[int],
repo: str,
pr: int,
path: str,
line: int,
severity: str,
problem: str,
fix: str = "",
suggestion: str = "",
comment_id: Optional[int] = None,
posted_at: Optional[int] = None,
) -> Optional[int]:
"""Insert an inline-finding row, deduped on posthash.
`comment_id` is filled in by the harvester when it discovers the
Gitea-assigned comment id for this finding. The post path returns the
`review_id` only; the inline ids come from a follow-up fetch.
"""
ph = posthash(path, line, severity, problem)
ts = posted_at or int(time.time())
# Every call inserts a fresh row. Aggregation by posthash is the
# caller's job — see `findings_with_votes` which GROUP BYs posthash.
# Letting each finding be its own row means reactions on different
# comment_ids across multiple PR reviews are not lost when one of
# those comment_ids becomes stale.
try:
cur = conn.execute(
"INSERT INTO inline_finding(review_id, repo, pr, path, line, severity, "
"problem, fix, suggestion, comment_id, posthash, posted_at) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",
(review_id, repo, pr, path, line, severity, problem, fix, suggestion,
comment_id, ph, ts),
)
conn.commit()
return cur.lastrowid
except Exception as e:
log.warning("record_inline_finding failed: %s", e)
return None
def record_reaction(
conn: sqlite3.Connection,
*,
comment_id: int,
user: str,
content: str,
created_at: Optional[int] = None,
) -> bool:
"""Upsert one reaction. PK = (comment_id, user, content)."""
try:
conn.execute(
"INSERT OR IGNORE INTO reaction(comment_id, user, content, created_at) "
"VALUES(?,?,?,?)",
(comment_id, user, content, created_at or int(time.time())),
)
conn.commit()
return True
except Exception as e:
log.warning("record_reaction failed: %s", e)
return False
def record_thread_state(
conn: sqlite3.Connection,
*,
finding_id: int,
resolved: bool,
checked_at: Optional[int] = None,
) -> bool:
"""Upsert the latest thread-state check."""
try:
conn.execute(
"INSERT INTO thread_state(finding_id, resolved, checked_at) "
"VALUES(?,?,?) "
"ON CONFLICT(finding_id) DO UPDATE SET "
" resolved = excluded.resolved, checked_at = excluded.checked_at",
(finding_id, 1 if resolved else 0, checked_at or int(time.time())),
)
conn.commit()
return True
except Exception as e:
log.warning("record_thread_state failed: %s", e)
return False
def record_reply(
conn: sqlite3.Connection,
*,
finding_id: int,
author: str,
body: str,
created_at: int,
) -> bool:
"""Insert one reply. PK includes created_at → re-imports are idempotent."""
try:
conn.execute(
"INSERT OR IGNORE INTO reply(finding_id, author, body, created_at) "
"VALUES(?,?,?,?)",
(finding_id, author, body, created_at),
)
conn.commit()
return True
except Exception as e:
log.warning("record_reply failed: %s", e)
return False
# ---------------------------------------------------------------------------
# Read helpers — for the analyzer
# ---------------------------------------------------------------------------
def findings_with_votes(
conn: sqlite3.Connection,
*,
repo: Optional[str] = None,
since_ts: Optional[int] = None,
) -> Iterable[sqlite3.Row]:
"""Stream every inline finding with rolled-up votes attached.
Joins:
inline_finding ◀ reaction (count by content)
inline_finding ◀ thread_state (latest resolved flag)
inline_finding ◀ reply (count + concatenation of bodies for negation
pattern matching)
Yielded rows expose:
id, repo, pr, path, line, severity, problem, fix, suggestion,
comment_id, posthash, posted_at,
upvotes INT, downvotes INT,
resolved INT (0/1/NULL),
reply_count INT,
reply_bodies TEXT ('\\n\\n'-joined for substring match),
review_posted_at INT
"""
where = []
params: list = []
if repo:
where.append("f.repo = ?")
params.append(repo)
if since_ts is not None:
where.append("COALESCE(r.posted_at, f.posted_at) >= ?")
params.append(since_ts)
where_sql = ("WHERE " + " AND ".join(where)) if where else ""
sql = f"""
SELECT
f.posthash AS id, -- alias for compat — every row IS an aggregated posthash
f.repo, MAX(f.pr) AS pr, f.path, f.line, MAX(f.severity) AS severity,
MAX(f.problem) AS problem, MAX(f.fix) AS fix, MAX(f.suggestion) AS suggestion,
MAX(f.comment_id) AS comment_id, f.posthash, MAX(f.posted_at) AS posted_at,
COUNT(*) AS occurrences,
r.posted_at AS review_posted_at,
COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes,
COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes,
MAX(ts.resolved) AS resolved,
COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id IN (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), 0) AS reply_count,
COALESCE((SELECT GROUP_CONCAT(body, char(10)||char(10)) FROM reply WHERE finding_id IN (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), '') AS reply_bodies
FROM inline_finding f
LEFT JOIN review r ON r.id = f.review_id
LEFT JOIN reaction rct ON rct.comment_id = f.comment_id
LEFT JOIN thread_state ts ON ts.finding_id = f.id
{where_sql}
GROUP BY f.posthash, f.repo, f.path, f.line
ORDER BY posted_at DESC
"""
return conn.execute(sql, params)
def known_posthashes_for_repo(conn: sqlite3.Connection, repo: str) -> set[str]:
"""For the harvester: which findings on this repo have already been
recorded? Used to skip re-fetching reactions we already harvested this
round."""
return {
row[0]
for row in conn.execute(
"SELECT DISTINCT posthash FROM inline_finding WHERE repo = ?", (repo,)
).fetchall()
}
def comment_ids_for_finding(conn: sqlite3.Connection, posthash: str) -> Optional[int]:
"""Return the current Gitea comment_id for an existing finding (used to
harvest votes for findings the harvester discovers on a brand-new PR that
ALSO has older bot comments on prior PRs)."""
row = conn.execute(
"SELECT comment_id FROM inline_finding WHERE posthash = ?", (posthash,)
).fetchone()
return row[0] if row else None
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())
pilot/feedback_harvest.py
+386
View File
@@ -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"<!--\s*pragent:sha=([0-9a-f]+)\s*-->", 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())
pilot/feedback_post.py
+128
View File
@@ -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())
tests/pilot/test_feedback.py
+317
View File
@@ -0,0 +1,317 @@
"""Tests for pilot/feedback.py — SQLite storage for review feedback signals.
Covers: schema bootstrap, posthash stability, dedup-on-insert, reaction /
thread-state / reply upserts, the analyzer-side `findings_with_votes` join,
and graceful failure on bad inputs.
"""
import sqlite3
import tempfile
import unittest
from pilot import feedback
class TestPosthash(unittest.TestCase):
def test_stable_across_calls(self):
a = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
b = feedback.posthash("src/api/foo.ts", 42, "HIGH", "Race condition in handler")
self.assertEqual(a, b)
def test_length_is_short(self):
h = feedback.posthash("a", 1, "low", "x")
self.assertEqual(len(h), 16)
def test_different_path_different_hash(self):
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("b", 1, "LOW", "x"),
)
def test_different_line_different_hash(self):
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("a", 2, "LOW", "x"),
)
def test_different_severity_different_hash(self):
# Same line, same problem, different severity → different signal.
self.assertNotEqual(
feedback.posthash("a", 1, "LOW", "x"),
feedback.posthash("a", 1, "CRITICAL", "x"),
)
def test_problem_prefix_used_only(self):
# First 80 chars participate; rest is ignored.
self.assertEqual(
feedback.posthash("a", 1, "LOW", "x" * 80 + "tail"),
feedback.posthash("a", 1, "LOW", "x" * 80),
)
def test_case_and_whitespace_normalized_in_problem(self):
# Lowercased + stripped → same hash.
self.assertEqual(
feedback.posthash("a", 1, "LOW", " Same Finding "),
feedback.posthash("a", 1, "LOW", "same finding"),
)
class TestInit(unittest.TestCase):
def test_init_creates_db(self):
with tempfile.TemporaryDirectory() as d:
db = f"{d}/f.db"
conn = feedback.init(db)
# Application tables exist (sqlite_sequence is a bookkeeping table
# created by AUTOINCREMENT — not part of the contract).
tables = {r[0] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()}
self.assertTrue(
{"review", "inline_finding", "reaction", "thread_state", "reply"}.issubset(tables),
f"missing tables: got {tables}",
)
conn.close()
def test_init_is_idempotent(self):
with tempfile.TemporaryDirectory() as d:
db = f"{d}/f.db"
feedback.init(db)
# Second call must not raise.
feedback.init(db)
class TestRecordReview(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_returns_id_and_row(self):
rid = feedback.record_review(
self.conn, repo="o/r", pr=1, head_sha="abc",
review_id_gitea=99, body_comment_id=42,
)
self.assertIsNotNone(rid)
row = self.conn.execute("SELECT * FROM review WHERE id = ?", (rid,)).fetchone()
self.assertEqual(row[1], "o/r")
self.assertEqual(row[4], 99)
self.assertEqual(row[5], 42)
class TestRecordInlineFinding(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def _new_review(self):
return feedback.record_review(
self.conn, repo="o/r", pr=1, head_sha="x",
)
def test_insert_returns_id(self):
rid = self._new_review()
fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH",
problem="bug", fix="patch", suggestion="code",
comment_id=555,
)
self.assertIsNotNone(fid)
def test_dedup_on_posthash(self):
# Two reviews of the SAME finding on different PRs insert two
# rows — deduplication by posthash is the *analyzer's* job
# (findings_with_votes GROUP BY posthash). Storing one row per
# review preserves per-comment reactions across PRs.
rid1 = self._new_review()
feedback.record_inline_finding(
self.conn, review_id=rid1, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH", problem="race",
comment_id=100,
)
rid2 = feedback.record_review(self.conn, repo="o/r", pr=2, head_sha="y")
feedback.record_inline_finding(
self.conn, review_id=rid2, repo="o/r", pr=2,
path="a/b.ts", line=10, severity="HIGH", problem="race",
comment_id=200,
)
rows = self.conn.execute(
"SELECT id, comment_id FROM inline_finding WHERE path='a/b.ts' AND line=10 ORDER BY id"
).fetchall()
self.assertEqual(len(rows), 2)
# Both comment_ids preserved (PK dedup is the *reaction* table's job).
self.assertEqual([r[1] for r in rows], [100, 200])
def test_posthash_set(self):
rid = self._new_review()
feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="nit",
)
ph = self.conn.execute(
"SELECT posthash FROM inline_finding LIMIT 1"
).fetchone()[0]
expected = feedback.posthash("a", 1, "LOW", "nit")
self.assertEqual(ph, expected)
class TestRecordReaction(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_insert_upsert(self):
ok = feedback.record_reaction(
self.conn, comment_id=10, user="alice", content="+1",
)
self.assertTrue(ok)
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 1)
# Re-insert same PK → no duplicate.
feedback.record_reaction(self.conn, comment_id=10, user="alice", content="+1")
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 1)
def test_distinct_users_can_react(self):
feedback.record_reaction(self.conn, comment_id=10, user="a", content="+1")
feedback.record_reaction(self.conn, comment_id=10, user="b", content="-1")
n = self.conn.execute("SELECT COUNT(*) FROM reaction").fetchone()[0]
self.assertEqual(n, 2)
class TestRecordThreadState(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
self.fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_upsert_overwrites(self):
feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=True)
row = self.conn.execute(
"SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
).fetchone()
self.assertEqual(row[0], 1)
feedback.record_thread_state(self.conn, finding_id=self.fid, resolved=False)
row = self.conn.execute(
"SELECT resolved FROM thread_state WHERE finding_id = ?", (self.fid,)
).fetchone()
self.assertEqual(row[0], 0)
class TestRecordReply(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
self.fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def test_insert_idempotent(self):
feedback.record_reply(
self.conn, finding_id=self.fid, author="a",
body="hi", created_at=1000,
)
feedback.record_reply(
self.conn, finding_id=self.fid, author="a",
body="hi", created_at=1000, # same PK
)
n = self.conn.execute("SELECT COUNT(*) FROM reply").fetchone()[0]
self.assertEqual(n, 1)
class TestFindingsWithVotes(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.conn = feedback.init(f"{self.tmp.name}/f.db")
def tearDown(self):
self.conn.close(); self.tmp.cleanup()
def _seed(self):
rid = feedback.record_review(self.conn, repo="o/r", pr=1, head_sha="x")
fid = feedback.record_inline_finding(
self.conn, review_id=rid, repo="o/r", pr=1,
path="a/b.ts", line=10, severity="HIGH",
problem="race", comment_id=500,
)
feedback.record_reaction(self.conn, comment_id=500, user="u1", content="+1")
feedback.record_reaction(self.conn, comment_id=500, user="u2", content="-1")
feedback.record_thread_state(self.conn, finding_id=fid, resolved=True)
feedback.record_reply(
self.conn, finding_id=fid, author="u3",
body="this is fine because of X", created_at=2000,
)
return fid
def test_join_rolls_up_votes(self):
self._seed()
rows = list(feedback.findings_with_votes(self.conn))
self.assertEqual(len(rows), 1)
r = rows[0]
self.assertEqual(r["upvotes"], 1)
self.assertEqual(r["downvotes"], 1)
self.assertEqual(r["resolved"], 1)
self.assertEqual(r["reply_count"], 1)
self.assertIn("this is fine", r["reply_bodies"])
def test_repo_filter(self):
self._seed()
# Add a finding under a different repo.
feedback.record_inline_finding(
self.conn, review_id=None, repo="other/r", pr=99,
path="x", line=1, severity="LOW", problem="y",
)
rows = list(feedback.findings_with_votes(self.conn, repo="o/r"))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["repo"], "o/r")
def test_findings_with_no_signals_return_zero_votes(self):
feedback.record_inline_finding(
self.conn, review_id=None, repo="x/y", pr=1,
path="p", line=1, severity="LOW", problem="z",
)
rows = list(feedback.findings_with_votes(self.conn))
self.assertEqual(len(rows), 1)
self.assertEqual(rows[0]["upvotes"], 0)
self.assertEqual(rows[0]["downvotes"], 0)
self.assertIsNone(rows[0]["resolved"])
class TestKnownPosthashes(unittest.TestCase):
def test_returns_distinct_set(self):
with tempfile.TemporaryDirectory() as d:
conn = feedback.init(f"{d}/f.db")
feedback.record_inline_finding(
conn, review_id=None, repo="o/r", pr=1,
path="a", line=1, severity="LOW", problem="x",
)
feedback.record_inline_finding(
conn, review_id=None, repo="o/r", pr=1,
path="a", line=2, severity="LOW", problem="y",
)
phs = feedback.known_posthashes_for_repo(conn, "o/r")
self.assertEqual(len(phs), 2)
conn.close()
if __name__ == "__main__":
unittest.main()
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()
tests/pilot/test_feedback_harvest.py
+243
View File
@@ -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()
tests/pilot/test_feedback_post.py
+97
View File
@@ -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()