Prumo: zero-parameter rule-based decision engine (Decision Index 0.3, public 20.14)
b445c46 verified | """Quick checks for the release (no network, no GPU, ~20 s): | |
| cd pareto/zero/release && python -m unittest discover -s tests -v | |
| - sandbox: code from a request cannot open sockets, import network modules, spawn processes, write outside its | |
| temporary directory or read user files; infinite loops stop on the deterministic line budget; | |
| - determinism: identical answers (choices and probabilities, bit for bit) across two runs in separate processes | |
| with different PYTHONHASHSEED values; | |
| - latency: per-request wall time on a fixed mixed sample stays far below the board's 1,000 ms limit. | |
| """ | |
| import json | |
| import os | |
| import random | |
| import statistics | |
| import subprocess | |
| import sys | |
| import tempfile | |
| import time | |
| import unittest | |
| from pathlib import Path | |
| HERE = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(HERE)) | |
| from prumo_rules import sandbox # noqa: E402 | |
| def sample_requests(): | |
| """A fixed, synthetic mix that exercises every capability (no benchmark items).""" | |
| reqs = [] | |
| reqs.append(({"code": "def f(xs):\n return sorted(xs)[::-1][:2]", "input": "[3, 9, 1, 4]"}, | |
| {"a": {"type": "choice", "instructions": "Value returned by f?", "criteria": {"x": "[9, 4]", "y": "[1, 3]", "z": "[9, 3]"}}})) | |
| reqs.append(({"code": "def add(a, b):\n return a + b", "tests": "def check(c):\n assert c(2, 3) == 5"}, | |
| {"p": {"type": "noul", "instructions": "The code passes the tests."}})) | |
| reqs.append(({"fen": "r1bqkbnr/pppp1ppp/2n5/4p3/4P3/5N2/PPPP1PPP/RNBQKB1R w KQkq - 2 3"}, | |
| {"m": {"type": "choice", "instructions": "Best move?", "criteria": {"f1b5": "Bb5", "d2d4": "d4", "a2a3": "a3", "h2h4": "h4"}}})) | |
| tools = [{"name": "get_weather", "description": "Current weather for a city", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}}, | |
| {"name": "convert_currency", "description": "Convert an amount between currencies", "parameters": {"type": "object", "properties": {"amount": {"type": "number"}, "to": {"type": "string"}}, "required": ["amount", "to"]}}] | |
| reqs.append(({"tools": tools, "question": "What's the weather like in Lisbon right now?"}, | |
| {"r": {"type": "choice", "instructions": "Which reply?", "criteria": { | |
| "A": '{"name": "get_weather", "arguments": {"city": "Lisbon"}}', "B": "Which city do you mean?", | |
| "C": "Sorry, I cannot do that with these tools.", "D": "It is sunny."}}})) | |
| reqs.append(({"functions": tools, "conversation": [{"role": "user", "content": "Convert 20 dollars to euros"}]}, | |
| {"t1": {"type": "choice", "instructions": "Call this function?\nFunction: get_weather", "criteria": {"y": "Yes, call it.", "n": "No, do not call it."}}, | |
| "t2": {"type": "choice", "instructions": "Call this function?\nFunction: convert_currency", "criteria": {"y": "Yes, call it.", "n": "No, do not call it."}}})) | |
| reqs.append(("not ( True and False ) or False is", {"b": {"type": "choice", "instructions": "Evaluate.", "criteria": {"t": "True", "f": "False"}}})) | |
| reqs.append(({"question": "Today is 03/01/2024. What is the date one week ago?"}, | |
| {"d": {"type": "choice", "instructions": "Pick the date.", "criteria": {"a": "02/23/2024", "b": "02/22/2024", "c": "03/08/2024"}}})) | |
| reqs.append(({"premises": "If the lamp is on, then the room is bright. The lamp is on.", "claim": "The room is bright."}, | |
| {"v": {"type": "choice", "instructions": "Status of the claim?", "criteria": {"follows": "Guaranteed.", "contradicted": "Ruled out.", "undetermined": "Either way."}}})) | |
| reqs.append(("There are three books on a shelf: a red book, a blue book, and a green book. The red book is the leftmost. The green book is to the right of the blue book.", | |
| {"o": {"type": "choice", "instructions": "Which is true?", "criteria": {"a": "The blue book is the second from the left.", "b": "The green book is the second from the left.", "c": "The red book is the second from the left."}}})) | |
| reqs.append(({"question": "Sam has 3 boxes with 12 pencils each. How many pencils does he have?"}, | |
| {"n": {"type": "choice", "instructions": "Answer.", "criteria": {"a": "36", "b": "40", "c": "50", "d": "7"}}})) | |
| reqs.append(("Some unrelated free text about the weather in spring.", | |
| {"u": {"type": "choice", "instructions": "Which is the best answer?", "criteria": {"a": "alpha", "b": "beta", "c": "gamma"}}})) | |
| return reqs | |
| class Sandbox(unittest.TestCase): | |
| def run_code(self, code, call="f()", cands=("1",)): | |
| return sandbox.run({"mode": "output", "code": code, "call": call, "candidates": list(cands)}) | |
| def test_value(self): | |
| r = self.run_code("def f(x):\n return x * 2", "f(21)", ["42", "21"]) | |
| self.assertTrue(r["ok"]) | |
| self.assertEqual(r["matches"], [True, False]) | |
| def test_no_network(self): | |
| for code in ("import socket\ndef f():\n return 1", | |
| "def f():\n import urllib.request\n return urllib.request.urlopen('http://example.com').status", | |
| "def f():\n __import__('_socket')\n return 1"): | |
| self.assertFalse(self.run_code(code).get("ok"), code) | |
| def test_no_write_outside_tmp(self): | |
| target = Path.home() / f".zr_sandbox_probe_{os.getpid()}" | |
| r = self.run_code(f"def f():\n open({str(target)!r}, 'w').write('x')\n return 1") | |
| self.assertFalse(r.get("ok")) | |
| self.assertFalse(target.exists()) | |
| r = self.run_code("def f():\n open('scratch.txt', 'w').write('ok')\n return open('scratch.txt').read()", "f()", ["'ok'"]) | |
| self.assertTrue(r["ok"]) | |
| self.assertEqual(r["matches"], [True]) | |
| def test_no_read_user_files(self): | |
| self.assertFalse(self.run_code("def f():\n return open('/etc/passwd').read()[:1]").get("ok")) | |
| def test_no_processes(self): | |
| for code in ("def f():\n import os\n return os.system('true')", "def f():\n import os\n return os.fork()", | |
| "def f():\n import subprocess\n return 1"): | |
| self.assertFalse(self.run_code(code).get("ok"), code) | |
| def test_line_budget(self): | |
| t = time.perf_counter() | |
| r = self.run_code("def f():\n while True:\n pass") | |
| self.assertFalse(r.get("ok")) | |
| self.assertEqual(r.get("error"), "budget") | |
| self.assertLess(time.perf_counter() - t, sandbox.WALL + 0.5) | |
| def test_wall_timeout(self): | |
| t = time.perf_counter() | |
| r = self.run_code("def f():\n return sum(range(10**11))") | |
| self.assertFalse(r.get("ok")) | |
| self.assertLess(time.perf_counter() - t, sandbox.WALL + 0.5) | |
| DUMP = r""" | |
| import json, sys | |
| sys.path.insert(0, sys.argv[1]) | |
| sys.path.insert(0, sys.argv[1] + "/tests") | |
| from decision_index_engine import DecisionEngine | |
| from test_engine import sample_requests | |
| eng = DecisionEngine() | |
| out = [eng(s, q)[0] for s, q in sample_requests()] | |
| print(json.dumps(out, sort_keys=True)) | |
| """ | |
| class Determinism(unittest.TestCase): | |
| def test_same_answers_across_processes(self): | |
| outs = [] | |
| for seed in ("1", "12345"): | |
| env = dict(os.environ, PYTHONHASHSEED=seed) | |
| p = subprocess.run([sys.executable, "-c", DUMP, str(HERE)], capture_output=True, text=True, env=env, timeout=120) | |
| self.assertEqual(p.returncode, 0, p.stderr[-2000:]) | |
| outs.append(p.stdout) | |
| self.assertEqual(outs[0], outs[1]) | |
| def test_expected_answers(self): | |
| from decision_index_engine import DecisionEngine | |
| eng = DecisionEngine() | |
| got = [eng(s, q)[0]["answers"] for s, q in sample_requests()] | |
| self.assertEqual(got[0]["a"]["choice"], "x") | |
| self.assertGreater(got[1]["p"]["noul"], 0.5) | |
| self.assertEqual(got[3]["r"]["choice"], "A") | |
| self.assertEqual((got[4]["t1"]["choice"], got[4]["t2"]["choice"]), ("n", "y")) | |
| self.assertEqual(got[5]["b"]["choice"], "t") | |
| self.assertEqual(got[6]["d"]["choice"], "a") | |
| self.assertEqual(got[7]["v"]["choice"], "follows") | |
| self.assertEqual(got[8]["o"]["choice"], "a") | |
| self.assertEqual(got[9]["n"]["choice"], "a") | |
| self.assertAlmostEqual(got[10]["u"]["probabilities"]["a"], 1 / 3) | |
| class Latency(unittest.TestCase): | |
| def test_latency(self): | |
| from decision_index_engine import DecisionEngine | |
| eng = DecisionEngine() | |
| reqs = sample_requests() * 3 | |
| ms = [] | |
| for s, q in reqs: | |
| t = time.perf_counter() | |
| eng(s, q) | |
| ms.append(1000 * (time.perf_counter() - t)) | |
| ms.sort() | |
| med, mean, p80 = statistics.median(ms), statistics.mean(ms), ms[int(0.8 * len(ms))] | |
| print(f"\nlatency ms: median {med:.1f} mean {mean:.1f} p80 {p80:.1f} max {ms[-1]:.1f}") | |
| self.assertLess(med, 200) | |
| self.assertLess(mean, 200) | |
| self.assertLess(p80, 300) | |
| self.assertLess(ms[-1], 1000) | |
| if __name__ == "__main__": | |
| unittest.main() | |