re_similar, re_struct_audit; next_functions favours raw functions with a readable near-duplicate

harbor/cli.py

@@ -255,7 +255,7 @@ def build_parser():
     p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)")
     p.add_argument("scope")
     p = sub.add_parser("re", help="reverse-engineering lookups from the project's image and annotations")
-    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools", "match"])
+    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools", "match", "similar", "struct-audit"])
     p.add_argument("target", nargs="?", default="", help="an address (0x4AD850) or a name; for ghidra, the tool")
     p.add_argument("arguments", nargs="?", default="{}", help="ghidra: the tool's arguments as JSON")
     p.add_argument("--decomp", help="match: the checkout or worktree to compile from")
@@ -684,7 +684,12 @@ def main(argv=None):
     elif cmd == "re":
         from .reverse import Ghidra, Reverse
         r = None if args.action.startswith(("ghidra", "debugger", "match")) else Reverse(harbor.project.reverse)
-        if args.action == "match":
+        if args.action in ("similar", "struct-audit"):
+            from .reverse import run_toolkit
+            res = (run_toolkit(harbor.project.reverse, "similar", [args.target]) if args.action == "similar" else
+                   run_toolkit(harbor.project.reverse, "struct_audit", ["--struct", args.target], json_to_file=True))
+            out(res, json.dumps(res, indent=1))
+        elif args.action == "match":
             from .reverse import match_functions
             res = match_functions(harbor.project.reverse, args.target, args.decomp)
             out(res, res["summary"] + "\n" + "\n".join(
@@ -717,7 +722,7 @@ def main(argv=None):
         from .queue import FunctionQueue
         q = FunctionQueue(harbor)
         if cmd == "next-functions":
-            rows = q.next_functions(who(), args.n)
+            rows = q.next_functions(who(), args.n, similar_readable=q.similar_readable())
             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":

harbor/mcp_server.py

@@ -518,6 +518,30 @@ def re_match(targets: str, decomp: str | None = None) -> dict:
         raise ToolError(str(error)) from None
 
 
+@mcp.tool()
+def re_similar(target: str, k: int = 10, min_score: float = 0.7) -> dict:
+    """Functions whose instructions look like this one (normalized: addresses blanked, registers and small
+    constants kept), best first, with each one's status. A readable near-duplicate is a head start: read it
+    first. 0.85+ is usually the same code (template copies); 0.75-0.85 the same idiom; check before reusing."""
+    try:
+        from .reverse import run_toolkit
+        return run_toolkit(harbor().project.reverse, "similar",
+                           [str(target), "--k", str(max(1, min(int(k), 50))), "--min", str(float(min_score))])
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
+@mcp.tool()
+def re_struct_audit(struct: str) -> list:
+    """Check a struct's field types against how the game's code accesses it through its globals: width,
+    signedness (movsx/movzx), float vs int, and unnamed bytes the code uses. Use it before changing types.h."""
+    try:
+        from .reverse import run_toolkit
+        return run_toolkit(harbor().project.reverse, "struct_audit", ["--struct", str(struct)], json_to_file=True)
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
 def _queue():
     from .queue import FunctionQueue
     return FunctionQueue(harbor())
@@ -529,7 +553,8 @@ def next_functions(n: int = 10) -> list:
     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)))
+        q = _queue()
+        return q.next_functions(me(), max(1, min(int(n), 50)), similar_readable=q.similar_readable())
     except (HarborError, projects.ProjectError) as error:
         raise ToolError(str(error)) from None
 

harbor/queue.py

@@ -103,6 +103,15 @@ class FunctionQueue:
                 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")
+
     # ------------------------------------------------------------------ queue
     def score(self, addr, facts, named, rejected, similar_readable=frozenset()):
         d = facts[addr]

harbor/reverse.py

@@ -509,3 +509,25 @@ def match_functions(config, targets, decomp=None, flags=None):
     errors = [line for line in result.stdout.splitlines() if "compile errors" in line]
     return {"results": rows, "summary": result.stdout.strip().splitlines()[0] if result.stdout.strip() else "",
             **({"compile": errors[0]} if errors else {})}
+
+
+def run_toolkit(config, key, argv, json_to_file=False, timeout=900):
+    """Run the toolkit script configured as [reverse] <key> with argv and return its JSON (stdout, or a temp file
+    passed as `--json <file>` when json_to_file)."""
+    import subprocess
+    import sys
+    import tempfile
+    script = (config or {}).get(key)
+    if not script or not Path(script).exists():
+        raise HarborError(f"this project has no {key} script in its [reverse] table")
+    with tempfile.TemporaryDirectory(dir=(config or {}).get("debugger_temp")) as tmp:
+        out = Path(tmp) / "out.json"
+        command = [sys.executable, "-B", str(script), *argv] + (["--json", str(out)] if json_to_file else ["--json"])
+        result = subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="replace",
+                                timeout=timeout, cwd=str(Path(script).parent),
+                                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
+        text = out.read_text(encoding="utf-8") if json_to_file and out.exists() else result.stdout
+        try:
+            return json.loads(text)
+        except ValueError:
+            raise HarborError(f"{key} failed: " + (result.stdout + result.stderr).strip()[-800:]) from None