From 3d9ac4ca9ac4093e3cb072b4ea8ec4112b5762d9 Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Wed, 7 Oct 2026 08:47:21 -0500 Subject: [PATCH] next_functions: skip functions already in an open review item (by scope, or the addresses its title names), so agents are not offered work they or others have already done --- harbor/queue.py | 521 +++++++++++++++++++++++++----------------------- 1 file changed, 269 insertions(+), 252 deletions(-) diff --git a/harbor/queue.py b/harbor/queue.py index 1949c22..e491938 100644 --- a/harbor/queue.py +++ b/harbor/queue.py @@ -1,252 +1,269 @@ -"""Which functions to decompile next, which to set aside, and how far the project is, per function. - -- next_functions(agent): the functions in the agent's bands that are not decompiled, not deferred and not claimed by - someone else, easiest first. Ease is scored from the bytes and the team's records: small functions, callees that - already have names, stack-only conventions, game code, few earlier rejections and a readable near-duplicate rank - higher. A function rejected `attempt_cap` times or more is left out (defer it or ask the coordinator). -- defer(agent, address, reason, evidence): set a function aside with a reason slug; it becomes a note on fn:0x... - (`DEFERRED : `) so everyone sees it, and the queue skips it until the note is replaced. -- progress(): every function in a tier, best first: exact (byte match), verified (behaves the same in the emulator), - readable, deferred, raw, plus library and engine functions counted apart. - -Configured from the project's [reverse] table (facts, annotations, src, conventions) plus optional result files: -match_results (match71.py --json), diffrun_results (diffrun.py --json), library_results (libsweep csv).""" -import csv -import json -import re -from pathlib import Path - -from .reverse import Reverse -from .util import HarborError - -DEFER_REASONS = ("register-args", "needs-type", "callee-unnamed", "inline", "x87", "layout", "too-large", - "engine-template", "data-not-code", "other") -READABLE = " * status: readable" -TIER_ORDER = ("exact", "verified", "readable", "deferred", "raw") - - -class FunctionQueue: - def __init__(self, harbor): - self.h = harbor - self.config = harbor.project.reverse or {} - self.r = Reverse(self.config) - self.cap = int(self.config.get("attempt_cap", 2)) - self.min_size = int(self.config.get("queue_min_size", 8)) - # [[lo, hi, "label"], ...] ranked low, e.g. the compiler-generated static initializers at the end of .text - self.low_priority = [(int(str(lo), 0), int(str(hi), 0), label) - for lo, hi, label in self.config.get("low_priority_ranges", [])] - - # ------------------------------------------------------------------ records - def statuses(self): - """address -> (status, file name) from the headers of src/**/_.c.""" - out = {} - src = self.r.src - if not src or not src.exists(): - return out - for path in src.rglob("*.c"): - m = re.match(r"([0-9A-F]{6,8})_", path.name) - if not m: - continue - try: - with path.open(encoding="utf-8", errors="replace") as f: - head = [next(f, "") for _ in range(8)] - except (FileNotFoundError, PermissionError): - continue # a landing on main moved or rewrote it while we scanned; its new path is found - - status = "readable" if any(line.startswith(READABLE) for line in head) else \ - next((line.split("status:", 1)[1].strip() for line in head if "status:" in line), "raw") - out[int(m.group(1), 16)] = (status, path.name) - return out - - def rejections(self): - """address -> how many review items on fn:0x... were rejected or sent back.""" - out = {} - rows = self.h.conn.execute( - "SELECT i.scope, COUNT(*) AS n FROM item_events e JOIN items i ON i.id = e.item_id " - "WHERE e.to_state IN ('rejected', 'changes_requested') AND i.scope LIKE 'fn:0x%' GROUP BY i.scope") - for row in rows: - try: - out[int(row["scope"][3:], 16)] = row["n"] - except ValueError: - pass - return out - - def deferrals(self): - """address -> (reason, evidence, author) for live DEFERRED notes.""" - out = {} - for row in self.h.conn.execute("SELECT scope, body, author FROM notes WHERE replaced_by IS NULL " - "AND body LIKE 'DEFERRED %' AND scope LIKE 'fn:0x%'"): - m = re.match(r"DEFERRED ([\w-]+): (.*)", row["body"], re.S) - if m: - out[int(row["scope"][3:], 16)] = (m.group(1), m.group(2).strip(), row["author"]) - return out - - def bands(self, agent): - return [(row["lo"], row["hi"]) for row in self.h.conn.execute( - "SELECT lo, hi FROM claims WHERE kind='band' AND agent=? AND released_at IS NULL", (agent,)) - if row["lo"] is not None] - - def claimed_by_others(self, agent): - return [(row["lo"], row["hi"]) for row in self.h.conn.execute( - "SELECT lo, hi FROM claims WHERE kind IN ('draft','edit','naming') AND agent<>? AND released_at IS NULL", - (agent,)) if row["lo"] is not None] - - def _named(self): - names = self.r._csv("names.csv") - return {a for a, row in names.items() if len(row) > 1 and row[1] and not re.match(r"(FUN|sub|thunk)_", row[1])} - - def _category(self, addr): - row = self.r._csv("names.csv").get(addr) - if row and len(row) > 2 and row[2]: - return row[2] - return (self.r._facts().get(addr) or {}).get("category", "unsorted") - - def _results(self, key): - path = self.config.get(key) - if not path or not Path(path).exists(): - return {} - if str(path).endswith(".csv"): - with open(path, encoding="utf-8") as f: - return {int(row["address"], 16): row for row in csv.DictReader(f)} - return {int(row["address"], 16): row for row in json.loads(Path(path).read_text(encoding="utf-8"))} - - def similar_readable(self): - """Raw functions with a readable near-duplicate (from the [reverse] similar_clusters file).""" - path = self.config.get("similar_clusters") - if not path or not Path(path).exists(): - return frozenset() - data = json.loads(Path(path).read_text(encoding="utf-8")) - return frozenset(int(m["address"], 16) for c in data.get("clusters", []) for m in c.get("members", []) - if m.get("best_readable") and m.get("status") != "readable") - - def library_ranges(self): - """[(start, end)] of statically linked library blocks (the [reverse] library_ranges csv).""" - path = self.config.get("library_ranges") - if not path or not Path(path).exists(): - return [] - with open(path, encoding="utf-8") as f: - return [(int(r["start"], 16), int(r["end"], 16)) for r in csv.DictReader(f) - if r.get("verdict", "").startswith("D3DX") or "stubs" in r.get("verdict", "")] - # the D3DX9 block, and the compiler's import thunks and exception-unwind funclets: nothing to decompile - - def is_library(self, addr, library, ranges): - return addr in library or any(lo <= addr < hi for lo, hi in ranges) - - # ------------------------------------------------------------------ queue - def score(self, addr, facts, named, rejected, similar_readable=frozenset(), with_convention=True): - d = facts[addr] - size = int(d.get("size") or 0) - calls = [int(c, 16) if isinstance(c, str) else c for c in d.get("calls", [])] - named_share = (sum(1 for c in calls if c in named) / len(calls)) if calls else 1.0 - score = 1.0 / (1.0 + size / 250.0) # small first - score *= 0.5 + 0.5 * named_share # callees with names make readable code possible - conv = self.r.convention(addr) if with_convention and size <= 4000 else None - if conv and conv["summary"].startswith("custom"): - score *= 0.75 # register arguments: prototypes.h work too - if self._category(addr) not in ("game", "unsorted"): - score *= 0.6 - score *= 0.5 ** rejected.get(addr, 0) - low = next((label for lo, hi, label in self.low_priority if lo <= addr < hi), None) - if low: - score *= 0.2 # e.g. the compiler's static initializers - reasons_low = [low] - else: - reasons_low = [] - if addr in similar_readable: - score *= 1.5 # a readable near-duplicate to start from - reasons = [f"{size} bytes", f"{round(named_share * 100)}% of {len(calls)} callees named"] - if conv: - reasons.append(conv["summary"]) - if rejected.get(addr): - reasons.append(f"rejected {rejected[addr]}x") - if addr in similar_readable: - reasons.append("has a readable near-duplicate") - reasons += reasons_low - return score, reasons - - def next_functions(self, agent, n=10, similar_readable=frozenset()): - bands = self.bands(agent) - if not bands: - raise HarborError(f"{agent} has no band; ask the coordinator for one") - taken = self.claimed_by_others(agent) - statuses, deferred, rejected = self.statuses(), self.deferrals(), self.rejections() - facts, named = self.r._facts(), self._named() - library = self._results("library_results") - ranges = self.library_ranges() - ranked = [] - for addr in facts: - if not any(lo <= addr <= hi for lo, hi in bands) or any(lo <= addr <= hi for lo, hi in taken): - continue - if statuses.get(addr, ("raw",))[0] == "readable" or addr in deferred or self.is_library(addr, library, ranges): - continue - if rejected.get(addr, 0) >= self.cap: - continue - if self._category(addr) in ("library", "platform"): - continue - if int(facts[addr].get("size") or 0) < self.min_size or \ - re.match(r"(nullsub|thunk)_", self.r.name_of(addr) or "") or \ - re.match(r"thunk_", (statuses.get(addr) or (None, ""))[1].split("_", 1)[-1]): - continue # stubs and thunks are named in bulk, not decompiled - head = self.r.read(addr, 10) if int(facts[addr].get("size") or 0) == 10 else b"" - if head[:1] and 0xB8 <= head[0] <= 0xBF and head[5:6] == b"\xe9": - continue # mov reg, imm32; jmp: an at-exit destructor thunk or an SEH handler stub (compiler's) - score, reasons = self.score(addr, facts, named, rejected, similar_readable, with_convention=False) - ranked.append((score, addr, reasons)) - ranked.sort(reverse=True) - # the convention read from the bytes only lowers a score, and it is the slow part: read it for the head only - head = [(*self.score(a, facts, named, rejected, similar_readable), a) for _, a, _ in ranked[:max(4 * n, 40)]] - ranked = sorted(((s, a, why) for s, why, a in head), reverse=True) - return [{"address": f"0x{a:X}", "name": self.r.name_of(a), "score": round(s, 4), "why": ", ".join(why), - "file": statuses.get(a, (None, None))[1]} for s, a, why in ranked[:n]] - - def defer(self, agent, address, reason, evidence): - if reason not in DEFER_REASONS: - raise HarborError("reason must be one of " + ", ".join(DEFER_REASONS)) - if not (evidence or "").strip(): - raise HarborError("say what you found (evidence), so the next person does not start from zero") - addr = self.r.resolve(address) - func = self.r.function_at(addr) - start = func[0] if func else addr - return self.h.note(agent, f"fn:0x{start:X}", f"DEFERRED {reason}: {evidence.strip()}") - - # ------------------------------------------------------------------ progress - def progress(self): - facts, statuses, deferred = self.r._facts(), self.statuses(), self.deferrals() - exact = {a for a, row in self._results("match_results").items() - if str(row.get("status")) in ("match", "exact", "exact-relaxed")} - # behaves the same in the emulator, with at least one trial that ran to the end in both (not 'weak') - verified = {a for a, row in self._results("diffrun_results").items() - if str(row.get("verdict")) == "same" and not row.get("weak")} - library = set(self._results("library_results")) - ranges = self.library_ranges() - tiers = {t: {"functions": 0, "bytes": 0} for t in TIER_ORDER + ("library", "engine")} - for addr, d in facts.items(): - size = int(d.get("size") or 0) - category = self._category(addr) - readable = statuses.get(addr, ("raw",))[0] == "readable" - if self.is_library(addr, library, ranges) or category in ("library", "platform", "crt"): - tier = "library" - elif category == "engine" and not readable: - tier = "engine" - elif addr in exact: - tier = "exact" - elif addr in verified: - tier = "verified" - elif readable: - tier = "readable" - elif addr in deferred: - tier = "deferred" - else: - tier = "raw" - tiers[tier]["functions"] += 1 - tiers[tier]["bytes"] += size - sources = {k: bool(self.config.get(k) and Path(self.config[k]).exists()) - for k in ("match_results", "diffrun_results", "library_results")} - return {"tiers": tiers, "deferred_reasons": _count(r for r, _, _ in deferred.values()), "sources": sources} - - -def _count(values): - out = {} - for v in values: - out[v] = out.get(v, 0) + 1 - return out +"""Which functions to decompile next, which to set aside, and how far the project is, per function. + +- next_functions(agent): the functions in the agent's bands that are not decompiled, not deferred and not claimed by + someone else, easiest first. Ease is scored from the bytes and the team's records: small functions, callees that + already have names, stack-only conventions, game code, few earlier rejections and a readable near-duplicate rank + higher. A function rejected `attempt_cap` times or more is left out (defer it or ask the coordinator). +- defer(agent, address, reason, evidence): set a function aside with a reason slug; it becomes a note on fn:0x... + (`DEFERRED : `) so everyone sees it, and the queue skips it until the note is replaced. +- progress(): every function in a tier, best first: exact (byte match), verified (behaves the same in the emulator), + readable, deferred, raw, plus library and engine functions counted apart. + +Configured from the project's [reverse] table (facts, annotations, src, conventions) plus optional result files: +match_results (match71.py --json), diffrun_results (diffrun.py --json), library_results (libsweep csv).""" +import csv +import json +import re +from pathlib import Path + +from .reverse import Reverse +from .util import HarborError, parse_scope + +DEFER_REASONS = ("register-args", "needs-type", "callee-unnamed", "inline", "x87", "layout", "too-large", + "engine-template", "data-not-code", "other") +READABLE = " * status: readable" +TIER_ORDER = ("exact", "verified", "readable", "deferred", "raw") + + +class FunctionQueue: + def __init__(self, harbor): + self.h = harbor + self.config = harbor.project.reverse or {} + self.r = Reverse(self.config) + self.cap = int(self.config.get("attempt_cap", 2)) + self.min_size = int(self.config.get("queue_min_size", 8)) + # [[lo, hi, "label"], ...] ranked low, e.g. the compiler-generated static initializers at the end of .text + self.low_priority = [(int(str(lo), 0), int(str(hi), 0), label) + for lo, hi, label in self.config.get("low_priority_ranges", [])] + + # ------------------------------------------------------------------ records + def statuses(self): + """address -> (status, file name) from the headers of src/**/_.c.""" + out = {} + src = self.r.src + if not src or not src.exists(): + return out + for path in src.rglob("*.c"): + m = re.match(r"([0-9A-F]{6,8})_", path.name) + if not m: + continue + try: + with path.open(encoding="utf-8", errors="replace") as f: + head = [next(f, "") for _ in range(8)] + except (FileNotFoundError, PermissionError): + continue # a landing on main moved or rewrote it while we scanned; its new path is found + + status = "readable" if any(line.startswith(READABLE) for line in head) else \ + next((line.split("status:", 1)[1].strip() for line in head if "status:" in line), "raw") + out[int(m.group(1), 16)] = (status, path.name) + return out + + def rejections(self): + """address -> how many review items on fn:0x... were rejected or sent back.""" + out = {} + rows = self.h.conn.execute( + "SELECT i.scope, COUNT(*) AS n FROM item_events e JOIN items i ON i.id = e.item_id " + "WHERE e.to_state IN ('rejected', 'changes_requested') AND i.scope LIKE 'fn:0x%' GROUP BY i.scope") + for row in rows: + try: + out[int(row["scope"][3:], 16)] = row["n"] + except ValueError: + pass + return out + + def deferrals(self): + """address -> (reason, evidence, author) for live DEFERRED notes.""" + out = {} + for row in self.h.conn.execute("SELECT scope, body, author FROM notes WHERE replaced_by IS NULL " + "AND body LIKE 'DEFERRED %' AND scope LIKE 'fn:0x%'"): + m = re.match(r"DEFERRED ([\w-]+): (.*)", row["body"], re.S) + if m: + out[int(row["scope"][3:], 16)] = (m.group(1), m.group(2).strip(), row["author"]) + return out + + def bands(self, agent): + return [(row["lo"], row["hi"]) for row in self.h.conn.execute( + "SELECT lo, hi FROM claims WHERE kind='band' AND agent=? AND released_at IS NULL", (agent,)) + if row["lo"] is not None] + + def claimed_by_others(self, agent): + return [(row["lo"], row["hi"]) for row in self.h.conn.execute( + "SELECT lo, hi FROM claims WHERE kind IN ('draft','edit','naming') AND agent<>? AND released_at IS NULL", + (agent,)) if row["lo"] is not None] + + def in_review(self): + """(lo, hi) of every item not yet landed or closed, from any author: work already done on a branch""" + out = [] + for row in self.h.conn.execute("SELECT scope, title FROM items WHERE state IN " + "('submitted','in_review','approved','changes_requested','held')"): + lo = None + if row["scope"]: + try: + _, lo, hi = parse_scope(row["scope"]) + except HarborError: + lo = None + if lo is not None: + out.append((lo, hi)) + else: # no scope given: the addresses its title names + out += [(int(a, 16), int(a, 16)) for a in re.findall(r"\b0x([0-9A-Fa-f]{6})\b", row["title"] or "")] + return out + + def _named(self): + names = self.r._csv("names.csv") + return {a for a, row in names.items() if len(row) > 1 and row[1] and not re.match(r"(FUN|sub|thunk)_", row[1])} + + def _category(self, addr): + row = self.r._csv("names.csv").get(addr) + if row and len(row) > 2 and row[2]: + return row[2] + return (self.r._facts().get(addr) or {}).get("category", "unsorted") + + def _results(self, key): + path = self.config.get(key) + if not path or not Path(path).exists(): + return {} + if str(path).endswith(".csv"): + with open(path, encoding="utf-8") as f: + return {int(row["address"], 16): row for row in csv.DictReader(f)} + return {int(row["address"], 16): row for row in json.loads(Path(path).read_text(encoding="utf-8"))} + + def similar_readable(self): + """Raw functions with a readable near-duplicate (from the [reverse] similar_clusters file).""" + path = self.config.get("similar_clusters") + if not path or not Path(path).exists(): + return frozenset() + data = json.loads(Path(path).read_text(encoding="utf-8")) + return frozenset(int(m["address"], 16) for c in data.get("clusters", []) for m in c.get("members", []) + if m.get("best_readable") and m.get("status") != "readable") + + def library_ranges(self): + """[(start, end)] of statically linked library blocks (the [reverse] library_ranges csv).""" + path = self.config.get("library_ranges") + if not path or not Path(path).exists(): + return [] + with open(path, encoding="utf-8") as f: + return [(int(r["start"], 16), int(r["end"], 16)) for r in csv.DictReader(f) + if r.get("verdict", "").startswith("D3DX") or "stubs" in r.get("verdict", "")] + # the D3DX9 block, and the compiler's import thunks and exception-unwind funclets: nothing to decompile + + def is_library(self, addr, library, ranges): + return addr in library or any(lo <= addr < hi for lo, hi in ranges) + + # ------------------------------------------------------------------ queue + def score(self, addr, facts, named, rejected, similar_readable=frozenset(), with_convention=True): + d = facts[addr] + size = int(d.get("size") or 0) + calls = [int(c, 16) if isinstance(c, str) else c for c in d.get("calls", [])] + named_share = (sum(1 for c in calls if c in named) / len(calls)) if calls else 1.0 + score = 1.0 / (1.0 + size / 250.0) # small first + score *= 0.5 + 0.5 * named_share # callees with names make readable code possible + conv = self.r.convention(addr) if with_convention and size <= 4000 else None + if conv and conv["summary"].startswith("custom"): + score *= 0.75 # register arguments: prototypes.h work too + if self._category(addr) not in ("game", "unsorted"): + score *= 0.6 + score *= 0.5 ** rejected.get(addr, 0) + low = next((label for lo, hi, label in self.low_priority if lo <= addr < hi), None) + if low: + score *= 0.2 # e.g. the compiler's static initializers + reasons_low = [low] + else: + reasons_low = [] + if addr in similar_readable: + score *= 1.5 # a readable near-duplicate to start from + reasons = [f"{size} bytes", f"{round(named_share * 100)}% of {len(calls)} callees named"] + if conv: + reasons.append(conv["summary"]) + if rejected.get(addr): + reasons.append(f"rejected {rejected[addr]}x") + if addr in similar_readable: + reasons.append("has a readable near-duplicate") + reasons += reasons_low + return score, reasons + + def next_functions(self, agent, n=10, similar_readable=frozenset()): + bands = self.bands(agent) + if not bands: + raise HarborError(f"{agent} has no band; ask the coordinator for one") + taken = self.claimed_by_others(agent) + self.in_review() # claimed elsewhere, or already in an item + statuses, deferred, rejected = self.statuses(), self.deferrals(), self.rejections() + facts, named = self.r._facts(), self._named() + library = self._results("library_results") + ranges = self.library_ranges() + ranked = [] + for addr in facts: + if not any(lo <= addr <= hi for lo, hi in bands) or any(lo <= addr <= hi for lo, hi in taken): + continue + if statuses.get(addr, ("raw",))[0] == "readable" or addr in deferred or self.is_library(addr, library, ranges): + continue + if rejected.get(addr, 0) >= self.cap: + continue + if self._category(addr) in ("library", "platform"): + continue + if int(facts[addr].get("size") or 0) < self.min_size or \ + re.match(r"(nullsub|thunk)_", self.r.name_of(addr) or "") or \ + re.match(r"thunk_", (statuses.get(addr) or (None, ""))[1].split("_", 1)[-1]): + continue # stubs and thunks are named in bulk, not decompiled + head = self.r.read(addr, 10) if int(facts[addr].get("size") or 0) == 10 else b"" + if head[:1] and 0xB8 <= head[0] <= 0xBF and head[5:6] == b"\xe9": + continue # mov reg, imm32; jmp: an at-exit destructor thunk or an SEH handler stub (compiler's) + score, reasons = self.score(addr, facts, named, rejected, similar_readable, with_convention=False) + ranked.append((score, addr, reasons)) + ranked.sort(reverse=True) + # the convention read from the bytes only lowers a score, and it is the slow part: read it for the head only + head = [(*self.score(a, facts, named, rejected, similar_readable), a) for _, a, _ in ranked[:max(4 * n, 40)]] + ranked = sorted(((s, a, why) for s, why, a in head), reverse=True) + return [{"address": f"0x{a:X}", "name": self.r.name_of(a), "score": round(s, 4), "why": ", ".join(why), + "file": statuses.get(a, (None, None))[1]} for s, a, why in ranked[:n]] + + def defer(self, agent, address, reason, evidence): + if reason not in DEFER_REASONS: + raise HarborError("reason must be one of " + ", ".join(DEFER_REASONS)) + if not (evidence or "").strip(): + raise HarborError("say what you found (evidence), so the next person does not start from zero") + addr = self.r.resolve(address) + func = self.r.function_at(addr) + start = func[0] if func else addr + return self.h.note(agent, f"fn:0x{start:X}", f"DEFERRED {reason}: {evidence.strip()}") + + # ------------------------------------------------------------------ progress + def progress(self): + facts, statuses, deferred = self.r._facts(), self.statuses(), self.deferrals() + exact = {a for a, row in self._results("match_results").items() + if str(row.get("status")) in ("match", "exact", "exact-relaxed")} + # behaves the same in the emulator, with at least one trial that ran to the end in both (not 'weak') + verified = {a for a, row in self._results("diffrun_results").items() + if str(row.get("verdict")) == "same" and not row.get("weak")} + library = set(self._results("library_results")) + ranges = self.library_ranges() + tiers = {t: {"functions": 0, "bytes": 0} for t in TIER_ORDER + ("library", "engine")} + for addr, d in facts.items(): + size = int(d.get("size") or 0) + category = self._category(addr) + readable = statuses.get(addr, ("raw",))[0] == "readable" + if self.is_library(addr, library, ranges) or category in ("library", "platform", "crt"): + tier = "library" + elif category == "engine" and not readable: + tier = "engine" + elif addr in exact: + tier = "exact" + elif addr in verified: + tier = "verified" + elif readable: + tier = "readable" + elif addr in deferred: + tier = "deferred" + else: + tier = "raw" + tiers[tier]["functions"] += 1 + tiers[tier]["bytes"] += size + sources = {k: bool(self.config.get(k) and Path(self.config[k]).exists()) + for k in ("match_results", "diffrun_results", "library_results")} + return {"tiers": tiers, "deferred_reasons": _count(r for r, _, _ in deferred.values()), "sources": sources} + + +def _count(values): + out = {} + for v in values: + out[v] = out.get(v, 0) + 1 + return out