"""The web board's Decomp view: progress tiers, a function inspector, each agent's next functions, deferrals and the state of the reverse-engineering tools. Everything here reads; nothing writes. It runs on its own Harbor connection with its own lock and short caches, because the answers take seconds (they scan src/ and score whole bands) and must not hold up the chat.""" import json import socket import subprocess import threading import time from pathlib import Path from .core import Harbor from .queue import FunctionQueue from .reverse import run_toolkit from .util import HarborError PROGRESS_TTL = 120 QUEUE_TTL = 600 class DecompAPI: def __init__(self, project): self.project = project self.lock = threading.Lock() self.cache = {} self._local = threading.local() @property def h(self): # SQLite connections belong to the thread that opened them, and the server answers on many threads if getattr(self._local, "h", None) is None: self._local.h = Harbor(self.project) return self._local.h def _cached(self, key, ttl, fn): hit = self.cache.get(key) if hit and time.monotonic() - hit[0] < ttl: return hit[1] value = fn() self.cache[key] = (time.monotonic(), value) return value def _queue(self): return FunctionQueue(self.h) # ------------------------------------------------------------------ answers def progress(self): with self.lock: return self._cached("progress", PROGRESS_TTL, lambda: self._queue().progress()) def queues(self, n=6): def build(): q = self._queue() similar = q.similar_readable() out = [] for agent in self.project.ai_names: if not q.bands(agent): continue try: rows = q.next_functions(agent, n, similar_readable=similar) except HarborError as error: rows, note = [], str(error) else: note = None out.append({"agent": agent, "bands": [[f"0x{lo:X}", f"0x{hi:X}"] for lo, hi in q.bands(agent)], "next": rows, **({"note": note} if note else {})}) return out with self.lock: return self._cached("queues", QUEUE_TTL, build) def deferred(self): with self.lock: q = self._queue() return [{"address": f"0x{a:X}", "name": q.r.name_of(a), "reason": reason, "evidence": evidence, "author": author} for a, (reason, evidence, author) in sorted(q.deferrals().items())] def function(self, target): """Everything about one function, for the inspector.""" with self.lock: q = self._queue() r = q.r info = r.lookup(target) addr = int(info.get("function", {}).get("start", info["address"]), 16) statuses = self._cached("statuses", PROGRESS_TTL, q.statuses) status, filename = statuses.get(addr, ("none", None)) info["status"] = status if filename: info["src_file"] = filename for key, label in (("match_results", "match"), ("diffrun_results", "behaviour")): row = q._results(key).get(addr) if row: info[label] = {k: row[k] for k in ("status", "verdict", "similarity", "first_difference", "registers_in", "ours_registers_in", "reason", "weak") if k in row} deferral = q.deferrals().get(addr) if deferral: info["deferred"] = {"reason": deferral[0], "evidence": deferral[1], "by": deferral[2]} info["category"] = q._category(addr) library = q._results("library_results") if q.is_library(addr, library, q.library_ranges()): info["library"] = True try: info["disassembly"] = r.disassemble(f"0x{addr:X}") if int(info.get("function", {}).get("size", 0)) <= 0x1800 \ else r.disassemble(f"0x{addr:X}", 0x600) + "\n; ... (first 0x600 bytes of a large function)" except HarborError as error: info["disassembly"] = f"; {error}" try: similar = run_toolkit(self.project.reverse, "similar", [f"0x{addr:X}", "--k", "6"], timeout=60) info["similar"] = similar.get("results", []) except (HarborError, subprocess.TimeoutExpired, OSError): info["similar"] = [] return info def tools(self): """Is each reverse-engineering tool in place, and how fresh are its results?""" config = self.project.reverse or {} out = [] def fresh(path): p = Path(path) if not p.exists(): return None return round((time.time() - p.stat().st_mtime) / 60) url = config.get("ghidra_url", "") port = int(url.rsplit(":", 1)[1].split("/")[0]) if url.count(":") >= 2 else 0 up = False if port: try: with socket.create_connection(("127.0.0.1", port), timeout=1): up = True except OSError: up = False decomp = Path(config.get("annotations", "")).parent if config.get("annotations") else None head = applied = None if decomp and (decomp / ".git").exists(): try: head = subprocess.run(["git", "-C", str(decomp), "rev-parse", "HEAD"], capture_output=True, text=True, timeout=10).stdout.strip() or None except (OSError, subprocess.TimeoutExpired): head = None applied_file = Path(config.get("debugger_temp", "")).parent / "ghidra-mcp" / "applied-commit.txt" \ if config.get("debugger_temp") else None if applied_file and applied_file.exists(): applied = applied_file.read_text(encoding="utf-8").strip() out.append({"tool": "Shared Ghidra session", "ok": up and (not head or head == applied), "detail": ("up" if up else "down") + (f"; names from main {applied[:8]}" if applied else "") + ("" if not head or head == applied else f"; main is at {head[:8]}: run ghidra-serve.ps1 -IfStale")}) compiler = Path(r"C:\Program Files (x86)\Microsoft Visual C++ Toolkit 2003\bin\cl.exe") out.append({"tool": "Game compiler (Visual C++ 2003 Toolkit)", "ok": compiler.exists(), "detail": "cl 13.10.3052 installed" if compiler.exists() else "not found"}) if config.get("debugger_path"): out.append({"tool": "Debugger (x32dbg + automation plugin)", "ok": Path(config["debugger_path"]).exists(), "detail": config["debugger_path"]}) for key, label in (("match_results", "Whole-program match results"), ("diffrun_results", "Emulator results"), ("library_results", "Library sweep"), ("similar_clusters", "Near-duplicate clusters"), ("source_files", "Source files from the Xbox build"), ("mac_names", "Names from the Mac build")): if config.get(key): age = fresh(config[key]) out.append({"tool": label, "ok": age is not None, "detail": "missing" if age is None else f"updated {age} min ago"}) return out _APIS = {} _LOCK = threading.Lock() def api_for(project): with _LOCK: if project.name not in _APIS: _APIS[project.name] = DecompAPI(project) return _APIS[project.name] def handle(project, path, query): """GET /api/decomp/: returns (status, body).""" api = api_for(project) what = path.rsplit("/", 1)[-1] if not (project.reverse or {}).get("facts"): return 404, {"error": "this project has no [reverse] table"} if what == "progress": return 200, api.progress() if what == "queues": return 200, api.queues() if what == "deferred": return 200, api.deferred() if what == "function": target = (query.get("target") or "").strip() if not target: return 400, {"error": "give an address or a name"} return 200, api.function(target) if what == "tools": return 200, api.tools() return 404, {"error": "not found"} def _json_safe(value): return json.loads(json.dumps(value, default=str))