domain/paging layer: FunctionIndex, block-cached DisasmModel, decompile, resolve

- Program: model registry + small prefetch pool; cache invalidation hooks
- FunctionIndex: lazy pagination (advance by len, clamp page=500), filter globs,
  viewport windows; full 10,092-func enumerate matches survey in ~3.1s
- DisasmModel: block-cached windowed disasm (256/block), forward+back prefetch,
  cached instruction totals; deep window revisit 196ms -> 0ms (cache proven)
- Decompilation: truncation-aware, hard-failure surfaced as data (monster funcs)
- resolve(): int/hex/symbol via lookup_funcs (string-array query shape)
- tests/test_domain.py: 17/17 on both ls (~290 funcs) and libcrypto (10k+52k-insn)
- smoke_client no longer assumes 'main' exists (works on libraries)
- doc: idle_ttl worker self-exit finding ("not reachable" needs re-open)

2 files changed, 168 insertions, 3 deletions

diff --git a/tests/smoke_client.py b/tests/smoke_client.py
index 647b589..5466284 100644
--- a/tests/smoke_client.py
+++ b/tests/smoke_client.py
@@ -26,6 +26,13 @@ from idatui.client import (  # noqa: E402
 PASS, FAIL = 0, 0
 
 
+def _query_data(payload):
+    res = payload.get("result", payload) if isinstance(payload, dict) else payload
+    if isinstance(res, list):
+        res = res[0] if res else {}
+    return res.get("data", []) if isinstance(res, dict) else []
+
+
 def check(name, cond, detail=""):
     global PASS, FAIL
     if cond:
@@ -82,8 +89,11 @@ def main(argv):
     check("health is dict", isinstance(h, dict), type(h).__name__)
     funcs = ida.call("list_funcs", queries=[{"count": 5}])
     check("list_funcs shape", isinstance(funcs, (list, dict)), type(funcs).__name__)
-    dis = ida.call("disasm", addr="main", max_instructions=10)
-    lines = dis.get("asm", {}).get("lines") if isinstance(dis, dict) else None
+    # Pick a real function (don't assume 'main' exists — libraries have none).
+    some = ida.call("list_funcs", queries=[{"filter": "sub_*", "count": 1}])
+    target = _query_data(some)[0]["addr"]
+    dis = ida.call("disasm", addr=target, max_instructions=10)
+    lines = (dis.get("asm") or {}).get("lines") if isinstance(dis, dict) else None
     check("disasm main has lines", bool(lines), str(dis)[:120])
     check("disasm line carries addr", bool(lines and "addr" in lines[0]),
           str(lines[0]) if lines else "no lines")
@@ -136,7 +146,7 @@ def main(argv):
 
     print("[concurrency]")
     def worker(i):
-        return ida.call("disasm", addr="main", max_instructions=5)
+        return ida.call("disasm", addr=target, max_instructions=5)
 
     with concurrent.futures.ThreadPoolExecutor(max_workers=8) as ex:
         results = list(ex.map(worker, range(40)))
diff --git a/tests/test_domain.py b/tests/test_domain.py
new file mode 100644
index 0000000..6875b66
--- /dev/null
+++ b/tests/test_domain.py
@@ -0,0 +1,155 @@
+#!/usr/bin/env python3
+"""Validate the domain/paging layer against a real, large binary.
+
+Run with a session open on a big binary (e.g. libcrypto.so.3, 10k funcs):
+
+    python3 tests/test_domain.py --db <session_id>
+
+Checks: correct full pagination (advance by len, not next_offset), viewport
+slicing across block boundaries, window caching (revisit is instant), prefetch
+warming, cached instruction totals, decompile success + hard-failure handling,
+and address resolution.
+"""
+import os
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from idatui.client import IDAClient  # noqa: E402
+from idatui.domain import DISASM_BLOCK, LIST_PAGE, Program  # noqa: E402
+
+URL = "http://127.0.0.1:8745/mcp"
+PASS = FAIL = 0
+
+
+def check(name, cond, detail=""):
+    global PASS, FAIL
+    if cond:
+        PASS += 1
+        print(f"  ok   {name}")
+    else:
+        FAIL += 1
+        print(f"  FAIL {name}   {detail}")
+
+
+def ms(fn):
+    t = time.time()
+    r = fn()
+    return (time.time() - t) * 1e3, r
+
+
+def main(argv):
+    db = None
+    it = iter(argv)
+    for a in it:
+        if a == "--db":
+            db = next(it)
+    c = IDAClient(URL, db=db)
+    c.connect()
+    if db is None:
+        c.set_db(c.resolve_db())
+    prog = Program(c)
+    module = c.health().get("module")
+    total_funcs = c.call("survey_binary").get("statistics", {}).get("total_functions")
+    print(f"db={c.db} module={module} survey_total_functions={total_funcs}")
+
+    # ---- function index: full enumeration correctness ------------------ #
+    print("\n[function index]")
+    idx = prog.functions()
+    dt, _ = ms(lambda: idx.load_all())
+    n = len(idx)
+    check("enumerated all functions (matches survey)", n == total_funcs,
+          f"got {n} vs survey {total_funcs}")
+    print(f"    loaded {n} funcs in {dt:.0f}ms ({dt / max(n,1):.3f}ms/func), "
+          f"page={LIST_PAGE}")
+    # no duplicates, monotonic-ish uniqueness by addr
+    addrs = [idx.get(i).addr for i in range(min(n, 3000))]
+    check("no duplicate addrs in first 3000", len(addrs) == len(set(addrs)))
+    # viewport slice (adapt to binary size)
+    wstart = min(1000, max(n - 50, 0))
+    wlen = min(50, n - wstart)
+    w = idx.window(wstart, wlen)
+    check(f"window({wstart},{wlen}) returns {wlen}", len(w) == wlen, str(len(w)))
+
+    # ---- filtered index uses server-side glob -------------------------- #
+    print("\n[filtered index]")
+    sub = prog.functions(filter="sub_*")
+    sub.ensure(10)
+    check("filter sub_* yields sub_ names", all(f.name.startswith("sub_") for f in sub.window(0, 10)),
+          str([f.name for f in sub.window(0, 5)]))
+
+    # ---- pick the fattest function for disasm stress ------------------- #
+    print("\n[disasm windowing]")
+    fattest = max((idx.get(i) for i in range(n)), key=lambda f: f.size)
+    dm = prog.disasm(fattest.addr, fattest.name)
+    dt, total = ms(dm.total)
+    check("total() returns positive count", total > 0, str(total))
+    dt2, total2 = ms(dm.total)
+    check("total() cached (2nd call ~instant)", dt2 < dt / 2 + 1, f"{dt:.0f}ms -> {dt2:.1f}ms")
+    print(f"    fattest {fattest.name}: {total} insns, total() {dt:.0f}ms then {dt2:.1f}ms")
+
+    # viewport across a block boundary
+    start = DISASM_BLOCK - 5
+    win = dm.lines(start, 60, prefetch=False)
+    check("viewport spans block boundary, right length",
+          len(win) == min(60, max(total - start, 0)), f"got {len(win)}")
+    # addresses strictly increasing and contiguous slice
+    eas = [ln.ea for ln in win]
+    check("viewport addrs strictly increasing", all(b > a for a, b in zip(eas, eas[1:])),
+          str(eas[:4]))
+
+    # deep window: first slow (O(offset)), revisit instant (cached)
+    if total > DISASM_BLOCK * 4:
+        deep = (total // DISASM_BLOCK - 1) * DISASM_BLOCK
+        dt_cold, a = ms(lambda: dm.lines(deep, 60, prefetch=False))
+        dt_warm, b = ms(lambda: dm.lines(deep, 60, prefetch=False))
+        check("deep window revisit is cached/instant", dt_warm < dt_cold / 2 + 1,
+              f"cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms")
+        check("cached window identical", [l.ea for l in a] == [l.ea for l in b])
+        print(f"    deep@{deep}: cold={dt_cold:.0f}ms warm={dt_warm:.1f}ms")
+
+    # prefetch warms the next block
+    print("\n[prefetch]")
+    dm2 = prog.disasm(fattest.addr + 0)  # same model (cached by ea)
+    fresh = prog.disasm(idx.get(0).addr, idx.get(0).name)
+    fresh.lines(0, 60, prefetch=True)  # should prefetch block 1
+    time.sleep(0.3)
+    check("prefetch warmed a neighbor block", fresh.cached_blocks() >= 2,
+          f"cached_blocks={fresh.cached_blocks()}")
+
+    # ---- decompile: success + hard-failure ----------------------------- #
+    print("\n[decompile]")
+    # a small function likely decompiles
+    small = min((idx.get(i) for i in range(n)), key=lambda f: f.size if f.size > 4 else 1 << 30)
+    d_small = prog.decompile(small.addr)
+    check("small func decompiles or fails cleanly", isinstance(d_small.failed, bool))
+    dt_c, _ = ms(lambda: prog.decompile(small.addr))
+    check("decompile cached (2nd ~instant)", dt_c < 5, f"{dt_c:.1f}ms")
+    # the monster should hard-fail as a soft error (not raise)
+    d_big = prog.decompile(fattest.addr)
+    check("monster decompile handled (failed flag, no raise)",
+          d_big.failed or d_big.code is not None,
+          f"failed={d_big.failed} err={d_big.error}")
+    print(f"    small {small.name}: failed={d_small.failed} "
+          f"trunc={d_small.truncated} chars={d_small.total_chars}")
+    print(f"    monster {fattest.name}: failed={d_big.failed} err={d_big.error}")
+
+    # ---- resolve ------------------------------------------------------- #
+    print("\n[resolve]")
+    check("resolve hex", prog.resolve(hex(fattest.addr)) == fattest.addr)
+    check("resolve int passthrough", prog.resolve(fattest.addr) == fattest.addr)
+    named = next((idx.get(i) for i in range(n) if not idx.get(i).name.startswith("sub_")), None)
+    if named:
+        try:
+            r = prog.resolve(named.name)
+            check("resolve symbol name", r == named.addr, f"{hex(r)} vs {hex(named.addr)} ({named.name})")
+        except KeyError as e:
+            check("resolve symbol name", False, str(e))
+
+    prog.close()
+    print(f"\n{PASS} passed, {FAIL} failed")
+    return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))