next_functions, defer and decomp_progress: an easy-first queue per band, deferrals with reasons, tiered progress

harbor/cli.py

@@ -262,6 +262,13 @@ def build_parser():
     p.add_argument("--max-bytes", type=int, dest="max_bytes")
     p.add_argument("--count", type=int, default=16)
     p.add_argument("--as", dest="as_type", default="hex", choices=["hex", "int32", "uint32", "float", "double", "string"])
+    p = sub.add_parser("next-functions", help="functions in your bands to decompile next, easiest first")
+    p.add_argument("-n", type=int, default=10)
+    p = sub.add_parser("defer", help="set a function aside with a reason and evidence")
+    p.add_argument("address")
+    p.add_argument("reason")
+    p.add_argument("evidence")
+    sub.add_parser("progress", help="functions and bytes per tier: exact, verified, readable, deferred, raw")
     p = sub.add_parser("feed", help="the latest tool calls of the team (or one agent), from the hooks")
     p.add_argument("--agent", dest="filter_agent")
     p.add_argument("--limit", type=int, default=30)
@@ -706,6 +713,20 @@ def main(argv=None):
         else:
             res = r.lookup(args.target)
             out(res, json.dumps(res, indent=1))
+    elif cmd in ("next-functions", "defer", "progress"):
+        from .queue import FunctionQueue
+        q = FunctionQueue(harbor)
+        if cmd == "next-functions":
+            rows = q.next_functions(who(), args.n)
+            out(rows, "\n".join(f"{r['address']:<10} {r['score']:<7} {(r['name'] or r['file'] or ''):<34} {r['why']}"
+                                 for r in rows))
+        elif cmd == "defer":
+            res = q.defer(who(), args.address, args.reason, args.evidence)
+            out(res, f"deferred {args.address}: {args.reason}")
+        else:
+            res = q.progress()
+            out(res, "\n".join(f"{t:<9} {v['functions']:>6} functions {v['bytes']:>9} bytes"
+                                for t, v in res["tiers"].items()))
     elif cmd == "who":
         rows = harbor.experts(args.scope)
         out(rows, "\n".join(f"{e['agent']:<9} " + ", ".join(f"{v} {k}" for k, v in e["detail"].items()) for e in rows)

harbor/mcp_server.py

@@ -518,6 +518,44 @@ def re_match(targets: str, decomp: str | None = None) -> dict:
         raise ToolError(str(error)) from None
 
 
+def _queue():
+    from .queue import FunctionQueue
+    return FunctionQueue(harbor())
+
+
+@mcp.tool()
+def next_functions(n: int = 10) -> list:
+    """The functions in your bands to decompile next, easiest first: not readable yet, not deferred, not claimed
+    by someone else, rejected fewer than the attempt cap. Each row says why it ranks where it does (size, how
+    many callees already have names, the convention from the bytes, earlier rejections)."""
+    try:
+        return _queue().next_functions(me(), max(1, min(int(n), 50)))
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
+@mcp.tool()
+def defer(address: str, reason: str, evidence: str) -> dict:
+    """Set a function aside instead of trying it again: reason is one of register-args, needs-type,
+    callee-unnamed, inline, x87, layout, too-large, engine-template, data-not-code, other; evidence says what you
+    found so the next person starts from it. It becomes a DEFERRED note on the function; next_functions skips it
+    until the note is replaced. Defer after two failed attempts rather than looping."""
+    try:
+        return _queue().defer(me(), address, reason, evidence)
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
+@mcp.tool()
+def decomp_progress() -> dict:
+    """Every function in a tier: exact (byte match with the game's compiler), verified (same behaviour in the
+    emulator), readable, deferred, raw; library and engine counted apart. Functions and bytes per tier."""
+    try:
+        return _queue().progress()
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
 @mcp.tool()
 def ghidra_tools() -> list:
     """The tools of the team's shared Ghidra session (GhidrAssistMCP, read-only, the curated project with every

harbor/queue.py

@@ -0,0 +1,205 @@
+"""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 <reason>: <evidence>`) 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))
+
+    # ------------------------------------------------------------------ records
+    def statuses(self):
+        """address -> (status, file name) from the headers of src/**/<ADDR>_<Name>.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
+            with path.open(encoding="utf-8", errors="replace") as f:
+                head = [next(f, "") for _ in range(8)]
+            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"))}
+
+    # ------------------------------------------------------------------ queue
+    def score(self, addr, facts, named, rejected, similar_readable=frozenset()):
+        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 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)
+        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")
+        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")
+        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 addr in library:
+                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
+            score, reasons = self.score(addr, facts, named, rejected, similar_readable)
+            ranked.append((score, addr, reasons))
+        ranked.sort(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")) == "match"}
+        verified = {a for a, row in self._results("diffrun_results").items() if str(row.get("verdict")) == "same"}
+        library = set(self._results("library_results"))
+        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 addr in library or category in ("library", "platform"):
+                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

harbor/reverse.py

@@ -61,7 +61,7 @@ class Reverse:
         def load(p):
             out = {}
             if p:
-                for line in p.open(encoding="utf-8"):
+                for line in p.read_text(encoding="utf-8").splitlines():
                     if line.strip():
                         d = json.loads(line)
                         out[_int(d["address"])] = d
@@ -81,7 +81,7 @@ class Reverse:
 
     def _imports(self):
         return self._load("imports", self.imports_path, lambda p: {
-            int(d["slot"]): f"{d['module']}!{d['name']}" for d in json.load(p.open(encoding="utf-8"))} if p else {})
+            int(d["slot"]): f"{d['module']}!{d['name']}" for d in json.loads(p.read_text(encoding="utf-8"))} if p else {})
 
     def _csv(self, name):
         path = self.annotations / name if self.annotations else None
@@ -89,7 +89,7 @@ class Reverse:
         def load(p):
             rows = {}
             if p:
-                for row in csv.reader(p.open(encoding="utf-8")):
+                for row in csv.reader(p.read_text(encoding="utf-8").splitlines()):
                     if row and row[0].lower().startswith("0x"):
                         rows[_int(row[0])] = row
             return rows

tests/test_queue.py

@@ -0,0 +1,58 @@
+"""next_functions, defer and progress against a tiny synthetic project."""
+import json
+import unittest
+
+from helpers import HarborCase
+from harbor.queue import FunctionQueue
+from harbor.util import HarborError
+
+
+class FunctionQueueTests(HarborCase):
+    def setUp(self):
+        super().setUp()
+        d = self.dir
+        image = bytearray(0x200)
+        image[0x00:0x10] = bytes.fromhex("8b44240485c07e03b801000000c3cccc")   # 0x401000: stack argument
+        image[0x40:0x48] = bytes.fromhex("85c07e03b8010000")                   # 0x401040: reads EAX
+        image[0x48] = 0xC3
+        image[0x80:0x8C] = bytes.fromhex("8b442404c3cccccccccccccc")           # 0x401080: readable already
+        (d / "image.bin").write_bytes(bytes(image))
+        facts = [{"address": "0x401000", "size": 14, "calls": []},
+                 {"address": "0x401040", "size": 9, "calls": []},
+                 {"address": "0x401080", "size": 5, "calls": []}]
+        (d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8")
+        ann = d / "annotations"
+        ann.mkdir()
+        (ann / "names.csv").write_text("address,name,category,note\n0x00401000,First,game,x\n"
+                                       "0x00401040,Second,game,x\n0x00401080,Third,game,x\n", encoding="utf-8")
+        src = d / "src" / "game"
+        src.mkdir(parents=True)
+        (src / "401080_Third.c").write_text("/*\n * status: readable\n */\nint Third(int a) { return a; }\n",
+                                            encoding="utf-8")
+        (src / "401000_First.c").write_text("/*\n * status: raw\n */\n", encoding="utf-8")
+        self.h.project.reverse = {"image": str(d / "image.bin"), "base": "0x401000", "facts": str(d / "facts.jsonl"),
+                                  "annotations": str(ann), "src": str(d / "src")}
+        self.h.claim("Claude", "range:0x401000-0x401FFF", kind="band", on_behalf_of="DeepSeek")
+
+    def test_queue_skips_readable_and_deferred(self):
+        q = FunctionQueue(self.h)
+        q.min_size = 1
+        rows = q.next_functions("DeepSeek", 10)
+        self.assertEqual({r["address"] for r in rows}, {"0x401000", "0x401040"})
+        q.defer("DeepSeek", "0x401040", "register-args", "takes the value in EAX; prototypes.h row needed first")
+        self.assertEqual([r["address"] for r in q.next_functions("DeepSeek", 10)], ["0x401000"])
+        with self.assertRaises(HarborError):
+            q.defer("DeepSeek", "0x401000", "bored", "x")
+        with self.assertRaises(HarborError):
+            q.next_functions("Codex", 5)                       # no band
+
+    def test_progress_tiers(self):
+        q = FunctionQueue(self.h)
+        q.defer("DeepSeek", "0x401040", "x87", "precision differs")
+        tiers = q.progress()["tiers"]
+        self.assertEqual((tiers["readable"]["functions"], tiers["deferred"]["functions"], tiers["raw"]["functions"]),
+                         (1, 1, 1))
+
+
+if __name__ == "__main__":
+    unittest.main()