💻 which code review skill work the best?
ranked by score ↓Source
Paste as source: in your trap.yaml
git+https://github.com/trapstreet/trapstreet-tasks@d82626d5683b718828ab613a90ba4d0e3fe33484#subdirectory=tasks/code_review_skill/python_bugfix_diffShare
python-bugfix-diff
A code-review Claude Skill (SKILL.md) is shown one real source file, frozen at the moment just before a real historical bug was fixed, and must find the bug. Ground truth is the actual fix commit — not a synthetic injected bug.
10 cases
Each case feeds files from inputs/<id>/ to the solution, expects files in expected/<id>/, and is scored by judge.py then aggregated by grader.py.
cases (10)
▸case_01Real bugfix-commit review: off-by-one / boundary condition
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: src/croniter/croniter.py
1267| def _get_low_from_current_date_number(cls, field_index, step, from_timestamp):
1268| dt = datetime.datetime.fromtimestamp(from_timestamp, tz=UTC_DT)
1269| if field_index == MINUTE_FIELD:
1270| return dt.minute % step
1271| if field_index == HOUR_FIELD:
1272| return dt.hour % step
1273| if field_index == DAY_FIELD:
1274| return ((dt.day - 1) % step) + 1
1275| if field_index == MONTH_FIELD:
1276| return dt.month % step
1277| if field_index == DOW_FIELD:
1278| return (dt.weekday() + 1) % step
1279|
1280| raise ValueError("Can't get current date number for index larger than 4")
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_01",
"bug_category": "off_by_one",
"file_path": "src/croniter/croniter.py",
"buggy_line": 1276,
"line_tolerance": 2,
"keywords": [
"off-by-one",
"off by one",
"1-based",
"1-indexed",
"boundary condition",
"wrong range",
"out of range",
"+1 correction",
"1-12 range",
"valid 1-12",
"falls outside",
"outside the valid range",
"no correction to keep"
],
"bug_description": "The month field is 1-based (1-12), but its low-bound is computed with `dt.month % step` instead of the 1-based formula `((dt.month - 1) % step) + 1` already used for the day field a few lines above. When step evenly divides the month, this yields 0, which is outside the valid 1-12 range.",
"source_commit_url": "https://github.com/pallets-eco/croniter/commit/3ddcd1385b53adabcce42bd14c6b8dfb83899fa6",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_02Real bugfix-commit review: null/None dereference
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: Lib/gftools/push/trafficjam.py
523| for item in board_items:
524| if item["type"] != "PULL_REQUEST":
525| raise ValueError(
526| "Traffic Jam contains issues! All items must be pull requests. "
527| "Please remove the issues and rerun the tool, "
528| "https://github.com/orgs/google/projects/74/views/1"
529| )
530| # sort items by pr number
531| board_items.sort(key=lambda k: k["content"]["url"])
532|
533| # get files for prs which have more than 100 changed files
534| for item in board_items:
535| changed_files = item["content"]["files"]["totalCount"]
536| if changed_files <= 100:
537| continue
538| pr_number = item["content"]["number"]
539| pr_url = item["content"]["url"]
540| log.warn(
541| f"{pr_url} has {changed_files} changed files. Attempting to fetch them."
542| )
543| files = g.pr_files(pr_number)
544| item["content"]["files"]["nodes"] = [
545| {"path": f["filename"]} for f in files
546| ]
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_02",
"bug_category": "null_deref",
"file_path": "Lib/gftools/push/trafficjam.py",
"buggy_line": 535,
"line_tolerance": 2,
"keywords": [
"missing null check",
"missing none check",
"null check",
"none check",
"not subscriptable"
],
"bug_description": "The loop indexes item[\"content\"][\"files\"][\"totalCount\"] for every board item without checking whether the PR is closed-but-not-merged; for such PRs the API returns files: null, so subscripting it raises \"TypeError: 'NoneType' object is not subscriptable\".",
"source_commit_url": "https://github.com/googlefonts/gftools/commit/f28317cab96bda1d8e07f5b044f645e6e4c07cab",
"license": "Apache-2.0"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_03Real bugfix-commit review: logic error (always-true condition)
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: rest_access_policy/access_policy.py
88| def _get_statements_matching_principal(
89| self, request, statements: List[dict]
90| ) -> List[dict]:
91| user = request.user
92| user_roles = None
93| matched = []
94|
95| for statement in statements:
96| principals = statement["principal"]
97| found = False
98|
99| if "*" in principals:
100| found = True
101| elif "admin" in principals:
102| found = user.is_superuser
103| elif "staff" in principals:
104| found = user.is_staff
105| elif "authenticated" in principals:
106| found = not user.is_anonymous
107| elif "anonymous" in principals:
108| found = user.is_anonymous
109| elif self.id_prefix + str(user.pk) in principals:
110| found = True
111| else:
112| if not user_roles:
113| user_roles = self.get_user_group_values(user)
114|
115| for user_role in user_roles:
116| if self.group_prefix + user_role in principals:
117| found = True
118| break
119|
120| if found:
121| matched.append(statement)
122|
123| return matched
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_03",
"bug_category": "elif_short_circuit_wrong_deny",
"file_path": "rest_access_policy/access_policy.py",
"buggy_line": 102,
"line_tolerance": 6,
"keywords": [
"elif chain stops",
"stops evaluating",
"doesn't fall through",
"should fall through",
"multiple principal types",
"list of principals",
"wrongly denies access",
"terminates the check",
"first matching keyword",
"should continue checking",
"prevents checking the remaining",
"elif short-circuits",
"stops checking"
],
"bug_description": "principals is a list that can declare multiple principal types at once (e.g. ['admin', 'id:5']), but the elif chain only tests which keyword is present, not whether that keyword's own check actually passes. If a user matches the 'admin' keyword but isn't a superuser, found=False and the elif chain stops there -- it never falls through to check the id_prefix (or any other) keyword also present in the same list, wrongly denying access the statement should have granted.",
"source_commit_url": "https://github.com/rsinger86/drf-access-policy/commit/fd628ad7cf0c30f039ca33f12881ac00d4c2d99d",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_04Real bugfix-commit review: race condition
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: backoff/_sync.py
23| def retry_predicate(target, wait_gen, predicate,
24| *,
25| max_tries, max_time, jitter,
26| on_success, on_backoff, on_giveup,
27| wait_gen_kwargs):
28|
29| @functools.wraps(target)
30| def retry(*args, **kwargs):
31|
32| # update variables from outer function args
33| nonlocal max_tries, max_time
34| max_tries = _maybe_call(max_tries)
35| max_time = _maybe_call(max_time)
36|
37| tries = 0
38| start = datetime.datetime.now()
39| wait = _init_wait_gen(wait_gen, wait_gen_kwargs)
40| while True:
41| tries += 1
42| elapsed = timedelta.total_seconds(datetime.datetime.now() - start)
43| details = {
44| "target": target,
45| "args": args,
46| "kwargs": kwargs,
47| "tries": tries,
48| "elapsed": elapsed,
49| }
50|
51| ret = target(*args, **kwargs)
52| if predicate(ret):
53| max_tries_exceeded = (tries == max_tries)
54| max_time_exceeded = (max_time is not None and
55| elapsed >= max_time)
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_04",
"bug_category": "stale_closure_state",
"file_path": "backoff/_sync.py",
"buggy_line": 34,
"line_tolerance": 2,
"keywords": [
"frozen after the first call",
"resolved only once",
"permanently overwrites",
"reassigning the nonlocal",
"mutates the enclosing scope",
"never re-evaluated",
"silently frozen",
"stale after the first call",
"only resolved on the first",
"overwrites the outer",
"no longer callable"
],
"bug_description": "retry() declares max_tries/max_time as nonlocal and reassigns them to the already-resolved value; because of nonlocal this mutates the enclosing retry_predicate scope permanently. max_tries/max_time can be passed as a callable specifically so it's re-evaluated on every call, but after the first invocation the reassignment replaces it with a plain value, so _maybe_call's callable(f) check is never true again -- every subsequent call to the decorated function silently reuses whatever the first call happened to resolve, forever.",
"source_commit_url": "https://github.com/litl/backoff/commit/732eaa34f782e31e891bbbd4aa524e42b27f3bc7",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_05Real bugfix-commit review: resource leak
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: src/pytest_socket/__init__.py
344| def guarded_connect(inst: socket.socket, *args: Any) -> None:
345| host = host_from_connect_args(args)
346| if host in allowed_ip_hosts_and_hostnames or (
347| _is_unix_socket(inst.family) and allow_unix_socket
348| ):
349| return _true_connect(inst, *args)
350|
351| if host and networks and is_ipaddress(host):
352| ip = ipaddress.ip_address(host)
353| if any(ip in net for net in networks):
354| return _true_connect(inst, *args)
355|
356| raise SocketConnectBlockedError(allowed_list, host)
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_05",
"bug_category": "resource_leak",
"file_path": "src/pytest_socket/__init__.py",
"buggy_line": 356,
"line_tolerance": 2,
"keywords": [
"resource leak",
"file descriptor leak",
"file descriptor leaks",
"socket not closed",
"not closed",
"leaked fd",
"fd leak"
],
"bug_description": "When a connection is disallowed, the function raises SocketConnectBlockedError without first closing the already-created real socket `inst`, leaking its file descriptor.",
"source_commit_url": "https://github.com/miketheman/pytest-socket/commit/2aaaee1bc226ddb996dd5498f3f64e9387f91db5",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_06Real bugfix-commit review: missing authentication check
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: src/ha_mcp/settings_ui.py
3033| def _mount(prefix: str) -> None:
3034| for path, methods, handler_key in routes:
3035| mcp.custom_route(f"{prefix}{path}", methods=methods)(handlers[handler_key])
3036|
3037| if is_addon:
3038| # Root mount lets HA ingress proxy localhost:9583/ → settings UI.
3039| # Direct port 9583 LAN access also reaches these routes; in this
3040| # respect they share the existing add-on networking model where
3041| # port 9583 is exposed via host_network and the secret path is
3042| # the auth for direct access. Document this in DOCS.md.
3043| mcp.custom_route("/", methods=["GET"])(handlers["root_page"])
3044| _mount("")
3045|
3046| if secret_prefix:
3047| # Mount under the MCP secret path so Docker / standalone clients
3048| # need the same secret to reach the UI as they do for the MCP
3049| # endpoint.
3050| _mount(secret_prefix)
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_06",
"bug_category": "security_missing_auth",
"file_path": "src/ha_mcp/settings_ui.py",
"buggy_line": 3044,
"line_tolerance": 2,
"keywords": [
"missing authentication",
"no authentication",
"without authentication",
"no auth check",
"no auth",
"unauthenticated access",
"authorization bypass",
"publicly reachable",
"without requiring the secret",
"bypassing the secret",
"secret-based auth",
"without the secret path",
"no secret required",
"without the secret"
],
"bug_description": "In add-on mode, the entire settings/backup/policy API is mounted at the bare root path via `_mount(\"\")` with zero authentication check, while the secret-path mount is the only place a secret is required; any client reaching the port directly can invoke these state-changing endpoints unauthenticated.",
"source_commit_url": "https://github.com/homeassistant-ai/ha-mcp/commit/9f5b085ad4a7b38b067c9da0dc5b45462c4d796e",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_07Real bugfix-commit review: mutable default argument
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: kolibri/content/api.py
238| def filter_in_lesson(self, queryset, name, value):
239| """
240| Show only content associated with this lesson
241|
242| :param queryset: all content nodes
243| :param value: id of target lesson
244| :return: content nodes for this lesson
245| """
246| try:
247| resources = Lesson.objects.get(id=value).resources
248| contentnode_id_list = [node['contentnode_id'] for node in resources]
249| # adapted from https://codybonney.com/creating-a-queryset-from-a-list-while-preserving-order-using-django/
250| clauses = ' '.join(["WHEN {}.id='{}' THEN {}".format(models.ContentNode._meta.db_table,
251| pk, i) for i, pk in enumerate(contentnode_id_list)])
252| ordering = 'CASE {} END'.format(clauses)
253| return queryset.filter(pk__in=contentnode_id_list) \
254| .extra(select={'ordering': ordering}, order_by=('ordering',))
255| except (Lesson.DoesNotExist, ValueError): # also handles invalid uuid
256| queryset.none()
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_07",
"bug_category": "sql_injection",
"file_path": "kolibri/content/api.py",
"buggy_line": 250,
"line_tolerance": 2,
"keywords": [
"sql injection",
"not parameterized",
"raw sql",
"unsanitized",
"string formatting into sql",
"break out of the string literal",
"user-controlled value",
"injectable",
"extra() with unsanitized",
"escape the quote",
"inject into the order by"
],
"bug_description": "clauses interpolates pk (a content-node id sourced from a lesson's resources field, editable by any coach/admin via the API) directly into a raw SQL fragment via .format(), which is passed to Django's .extra() -- a documented SQL-injection vector. A malicious id containing a single quote could break out of the string literal and inject arbitrary SQL into the resulting ORDER BY clause.",
"source_commit_url": "https://github.com/learningequality/kolibri/commit/520b0778568694b5cbb4961eaec5b1979b28dc5c",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_08Real bugfix-commit review: overly broad except clause
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: xrspatial/geotiff/_vrt.py
347| src_r0 = sr.y_off + int((clip_r0 - dst_r0) * scale_y)
348| src_c0 = sr.x_off + int((clip_c0 - dst_c0) * scale_x)
349| src_r1 = sr.y_off + int((clip_r1 - dst_r0) * scale_y)
350| src_c1 = sr.x_off + int((clip_c1 - dst_c0) * scale_x)
351|
352| # Read from source file using windowed read
353| try:
354| src_arr, _ = read_to_array(
355| src.filename,
356| window=(src_r0, src_c0, src_r1, src_c1),
357| band=src.band - 1, # convert 1-based to 0-based
358| )
359| except Exception as e:
360| # Under XRSPATIAL_GEOTIFF_STRICT=1, surface the read failure
361| # so partial mosaics are caught in CI. Default mode warns
362| # once per missing source then continues, preserving the
363| # historical behaviour. See issue #1662.
364| import warnings
365| from . import _geotiff_strict_mode, GeoTIFFFallbackWarning
366| if _geotiff_strict_mode():
367| raise
368| warnings.warn(
369| f"VRT source {src.filename!r} could not be read "
370| f"({type(e).__name__}: {e}); skipping. The output "
371| f"mosaic will have a hole at this tile.",
372| GeoTIFFFallbackWarning,
373| stacklevel=2,
374| )
375| continue # skip missing/unreadable sources
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_08",
"bug_category": "broad_except",
"file_path": "xrspatial/geotiff/_vrt.py",
"buggy_line": 359,
"line_tolerance": 2,
"keywords": [
"bare except",
"except exception",
"too broad",
"overly broad",
"broadest exception",
"broadest `exception`",
"swallowing exceptions",
"catches too much",
"catches everything",
"should be narrowed",
"narrowed to",
"can mask genuine",
"mask genuine errors",
"masks genuine",
"masking genuine",
"hiding real defects",
"hiding genuine defects",
"hides real defects",
"hides genuine defects"
],
"bug_description": "The windowed read of each VRT source tile is wrapped in `except Exception as e`, far broader than the I/O errors read_to_array actually raises, silently catching genuine unrelated bugs like RuntimeError or MemoryError.",
"source_commit_url": "https://github.com/xarray-contrib/xarray-spatial/commit/c6daae54b8776b7c6ac67c77fcb9f9531cb56069",
"license": "MIT"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_09Real bugfix-commit review: cache.invalidate() raises KeyError on already-missing key
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: funcy/calc.py
100| def cache(timeout, key_func=None):
101| """Caches a function results for timeout seconds."""
102| if isinstance(timeout, int):
103| timeout = timedelta(seconds=timeout)
104|
105| if key_func is None:
106| key_func = lambda *a, **kw: a + tuple(sorted(kw.items())) if kw else a
107|
108| def decorator(func):
109| cache = {}
110|
111| @wraps(func)
112| def wrapper(*args, **kwargs):
113| key = key_func(*args, **kwargs)
114| if key in cache:
115| result, timestamp = cache[key]
116| if datetime.now() - timestamp < timeout:
117| return result
118| else:
119| del cache[key]
120|
121| result = func(*args, **kwargs)
122| cache[key] = result, datetime.now()
123| return result
124|
125| def invalidate(*args, **kwargs):
126| cache.pop(key_func(*args, **kwargs))
127| wrapper.invalidate = invalidate
128|
129| def invalidate_all():
130| cache.clear()
131| wrapper.invalidate_all = invalidate_all
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_09",
"bug_category": "cache_invalidate_keyerror",
"file_path": "funcy/calc.py",
"buggy_line": 126,
"line_tolerance": 2,
"keywords": [
"dict.pop without default",
"missing default value",
"pop raises keyerror",
"race condition",
"concurrent invalidation",
"cache invalidate crash",
"works in tests fails in production",
"already removed",
"raises a keyerror",
"raises keyerror if",
"key is not present",
"was never cached",
"already expired"
],
"bug_description": "The invalidate() closure calls cache.pop(key_func(*args, **kwargs)) without a default value; if the key was already removed -- e.g. it expired via the timeout-based eviction on the next read, or a concurrent thread already invalidated it -- pop() raises KeyError. Because unit tests rarely hit this timing/race window, the code appears correct until it crashes in production.",
"source_commit_url": "https://github.com/Suor/funcy/commit/a96b449ee08d7355805e4c6037e9782e74241ff7",
"license": "BSD-3-Clause"
}Scored by judge.py — see Scoring logic below for the full rule.
▸case_10Real bugfix-commit review: float truncated before scaling, losing microsecond precision
input
question.txt
You are reviewing a code change. Below is one file from the change, shown with its real line numbers.
File: freezegun/api.py
176| def fake_time():
177| if _should_use_real_time():
178| return real_time()
179| current_time = get_current_time()
180| return calendar.timegm(current_time.timetuple()) + current_time.microsecond / 1000000.0
181|
182| if _TIME_NS_PRESENT:
183| def fake_time_ns():
184| if _should_use_real_time():
185| return real_time_ns()
186| return int(int(fake_time()) * 1e9)
Find the most significant bug in this code and report it. Respond with ONLY a JSON object in this exact shape, and nothing else:
{"findings": [{"file": "<file path>", "line": <line number>, "description": "<1-2 sentence description of the issue>"}, ...]}
List findings in order of confidence, most likely real bug first. You may report more than one finding, but only your first 5 will be scored.
expected output
answer.json
{
"id": "case_10",
"bug_category": "floating_point_truncation",
"file_path": "freezegun/api.py",
"buggy_line": 186,
"line_tolerance": 0,
"keywords": [
"truncation order",
"int() called too early",
"microsecond precision lost",
"precision loss",
"floating point truncation",
"rounds down to the second",
"sub-second precision discarded",
"double truncation",
"multiply before truncating",
"uuid collision",
"loses microseconds",
"discarding all sub-second precision",
"discarding sub-second precision",
"discards sub-second precision",
"rounded down to a whole second",
"before multiplying by 1e9",
"loses the microseconds"
],
"bug_description": "fake_time_ns() computes int(int(fake_time()) * 1e9): the inner int() truncates fake_time()'s fractional (microsecond) part to zero BEFORE multiplying by 1e9, discarding all sub-second precision. It should multiply first and truncate once: int(fake_time() * 1e9). The practical symptom is that uuid1() calls made within the same frozen second collide, since time_ns() loses the microseconds that would otherwise differentiate them.",
"source_commit_url": "https://github.com/spulec/freezegun/commit/9591645e0fd49b54bab688ac7eeeec14b87a9226",
"license": "Apache-2.0"
}Scored by judge.py — see Scoring logic below for the full rule.
scoring logic
judge.py runs once per case and prints a score per case. grader.py runs once at the end and folds case scores into a run-level summary. Without grader.py, the run's score is simply the average of case scores.
▸judge.py137 lines · view on GitHub
"""Per-case judge for the code_review_skill/python_bugfix_diff task.
The skill under test receives one real (pre-fix) source file, shown with its
real line numbers, and reports findings as JSON. A finding "hits" the gold
bug only if it names the right file, points within a small line-number
window of the real bug line, AND its description contains at least one of
the case's pre-declared keywords. Only the FIRST 5 findings are considered
(anti-shotgun) -- a skill can't win by flagging every line.
This is deterministic (no LLM judge): the ground truth is a real historical
bugfix commit, and "keywords" are phrases a competent human reviewer of that
exact bug would naturally use, curated per case in gold.cases.json.
I/O contract: reads TRAPTASK_MANIFEST (trap-cli).
"""
from __future__ import annotations
import json
import os
import re
from pathlib import Path
from typing import Any
MAX_FINDINGS_SCORED = 5
def _strip_fences(text: str) -> str:
text = text.strip()
if not text.startswith("```"):
return text
lines = text.split("\n")
if lines and lines[0].startswith("```"):
lines = lines[1:]
if lines and lines[-1].startswith("```"):
lines = lines[:-1]
return "\n".join(lines).strip()
def _finding_matches(finding: Any, expected: dict) -> dict[str, bool]:
"""Per-signal match flags for one finding against the gold bug."""
if not isinstance(finding, dict):
return {"file_match": False, "line_match": False, "keyword_match": False}
file_match = False
f = finding.get("file")
if isinstance(f, str) and f.strip():
file_match = Path(f.strip()).name == Path(expected["file_path"]).name
line_match = False
line = finding.get("line")
if isinstance(line, (int, float)) and not isinstance(line, bool):
try:
line_match = abs(int(line) - expected["buggy_line"]) <= expected.get("line_tolerance", 2)
except (OverflowError, ValueError):
# int(float("inf"))/-inf raises OverflowError, int(float("nan")) raises
# ValueError. json.loads accepts these non-standard literals by default,
# so a solution can emit e.g. {"line": Infinity} and must degrade to a
# clean miss here rather than crashing score_case().
line_match = False
keyword_match = False
desc = finding.get("description")
if isinstance(desc, str) and desc:
keyword_match = any(
re.search(rf"\b{re.escape(kw)}\b", desc, re.IGNORECASE)
for kw in expected["keywords"]
)
return {"file_match": file_match, "line_match": line_match, "keyword_match": keyword_match}
def score_case(stdout: str, expected: dict) -> dict[str, Any]:
s = _strip_fences(stdout)
try:
obj = json.loads(s)
except (json.JSONDecodeError, TypeError, RecursionError):
return {"score": 0.0, "hit_index": None, "n_findings_considered": 0,
"format_ok": False, "reason": "output is not valid JSON"}
if not isinstance(obj, dict) or not isinstance(obj.get("findings"), list):
return {"score": 0.0, "hit_index": None, "n_findings_considered": 0,
"format_ok": False, "reason": "missing 'findings' list"}
# Anti-shotgun: only the first MAX_FINDINGS_SCORED findings are considered.
considered = obj["findings"][:MAX_FINDINGS_SCORED]
hit_index = None
best_match_count = -1
best_signals = {"file_match": False, "line_match": False, "keyword_match": False}
for i, finding in enumerate(considered):
signals = _finding_matches(finding, expected)
match_count = sum(signals.values())
if match_count > best_match_count:
best_match_count = match_count
best_signals = signals
if all(signals.values()) and hit_index is None:
hit_index = i
score = 1.0 if hit_index is not None else 0.0
return {
"score": score,
"format_ok": True,
"hit_index": hit_index,
"n_findings_total": len(obj["findings"]),
"n_findings_considered": len(considered),
"best_match_signals": best_signals,
}
def main() -> None:
m = json.loads(os.environ["TRAPTASK_MANIFEST"])
stdout = Path(m["run"]["stdout"]).read_text()
exit_code = json.loads(Path(m["run"]["meta"]).read_text())["exit_code"]
expected = json.loads((Path(m["expected_dir"]) / "answer.json").read_text())
base = {"id": expected.get("id"), "bug_category": expected.get("bug_category")}
if exit_code != 0:
print(json.dumps({**base, "score": 0.0, "reason": f"solution exited {exit_code}",
"agent_output": stdout.strip()[:500]}))
return
if not stdout.strip():
print(json.dumps({**base, "score": 0.0, "reason": "agent produced no output",
"agent_output": ""}))
return
metrics = score_case(stdout, expected)
metrics.update(base)
metrics["agent_output"] = stdout.strip()[:500]
print(json.dumps(metrics))
if __name__ == "__main__":
main()
▸grader.py73 lines · view on GitHub
"""Overall grader for the code_review_skill/python_bugfix_diff task.
Aggregates per-case judge results (the trap-cli TRAPTASK_MANIFEST list) into
a run-level verdict. Same shape as wildlife_camera_trap/grader.py and
connections/word_groups/grader.py.
"""
from __future__ import annotations
import json
import os
from collections import Counter
PASS_THRESHOLD = 0.5
def main() -> None:
cases = json.loads(os.environ["TRAPTASK_MANIFEST"])
scored = [c for c in cases if c.get("metrics") and c["metrics"].get("score") is not None]
skipped = [c for c in cases if not c.get("metrics") or c["metrics"].get("score") is None]
accuracy = sum(c["metrics"]["score"] for c in scored) / len(scored) if scored else 0.0
by_category_score: Counter[str] = Counter()
by_category_total: Counter[str] = Counter()
for c in scored:
cat = c["metrics"].get("bug_category")
if cat:
by_category_total[cat] += 1
by_category_score[cat] += c["metrics"]["score"]
by_category_pct = {
k: round(by_category_score[k] / by_category_total[k], 3) for k in by_category_total
}
durations = [c.get("duration", 0.0) for c in cases if c.get("duration") is not None]
if durations:
ds = sorted(durations)
latency_ms_median = round(ds[len(ds) // 2] * 1000, 1)
latency_ms_p95 = round(ds[int(0.95 * len(ds))] * 1000, 1) if len(ds) > 1 else latency_ms_median
latency_ms_total = round(sum(ds) * 1000, 1)
else:
latency_ms_median = latency_ms_p95 = latency_ms_total = 0.0
case_costs = [
c["cost"]["cost_usd"]
for c in cases
if isinstance(c.get("cost"), dict) and c["cost"].get("cost_usd") is not None
]
cost_usd_total = round(sum(case_costs), 4) if case_costs else None
n_passed = sum(1 for c in scored if c["metrics"]["score"] == 1.0)
passed = bool(scored) and accuracy >= PASS_THRESHOLD
print(json.dumps({
"passed": passed,
"score": round(accuracy, 3),
"n_passed": n_passed,
"n_total": len(cases),
"n_scored": len(scored),
"n_skipped_no_gold": len(skipped),
"threshold": PASS_THRESHOLD,
"by_category": by_category_pct,
"latency_ms_median": latency_ms_median,
"latency_ms_p95": latency_ms_p95,
"latency_ms_total": latency_ms_total,
"cost_usd_total": cost_usd_total,
}))
if __name__ == "__main__":
main()