"""Offline, reduced reproductions of three defects in a research recorder/evaluator. Run: python reproduce.py Export the shared browser fixtures: python reproduce.py --fixtures Python 3.9+, standard library only. No network, wallets, account access or orders. PUMP/OTHER are symbolic program labels, not deployable program identifiers. The source decoder validates a consumed prefix; this model reproduces selected boundary checks, not the binary decoder or a full current IDL. """ import argparse import json import math import re import unittest def boolean_result(value): if type(value) is not int or not 0 <= value <= 255: raise ValueError("Expected one byte") return {"legacy": "BUY" if bool(value) else "SELL", "fixed": "BUY" if value == 1 else "SELL" if value == 0 else "REJECT", "accepted": value in (0, 1)} def attributed_events(logs, transaction_error=None): """Reduced frame-buffering model; TRADE:1 substitutes for valid binary data. A log prefix alone identifies neither the emitter nor a successful outcome. Child buffers merge only on success. A failed ancestor discards descendants. Any unbalanced/malformed stack discards the whole notification. """ if transaction_error is not None or not isinstance(logs, list): return 0 stack, accepted = [], 0 for line in logs: if not isinstance(line, str): return 0 start = re.fullmatch(r"Program (PUMP|OTHER) invoke \[([1-9][0-9]*)\]", line) end = re.fullmatch(r"Program (PUMP|OTHER) (success|failed: .+)", line) if start: if int(start[2]) != len(stack) + 1: return 0 stack.append([start[1], 0]) elif end: if not stack or stack[-1][0] != end[1]: return 0 program, pending = stack.pop() if end[2] == "success": if stack: stack[-1][1] += pending else: accepted += pending elif not stack: return 0 # Top-level failure contradicts the reported success. elif line.startswith("Program data: "): if not stack: return 0 if stack[-1][0] == "PUMP" and line == "Program data: TRADE:1": stack[-1][1] += 1 elif line.startswith("Program log: "): continue # Text from a program cannot become a runtime stack marker. else: return 0 # This deliberately small model supports only these lines. return 0 if stack else accepted def gap_result(opening): if type(opening) not in (int, float) or not math.isfinite(opening): raise ValueError("Expected a finite opening price") # Zero-spread long: entry100, initial stop95, risk5. If opening >=95, # this fixture assumes the bar later reaches the standing stop. return {"legacy": -1.0, "fixed": (min(opening, 95) - 100) / 5} def fixtures(): event = "Program data: TRADE:1" traces = [ ("direct", "Direct program event", ["Program PUMP invoke [1]", event, "Program PUMP success"], None, 1), ("foreign", "Foreign event in the same transaction", ["Program PUMP invoke [1]", "Program PUMP success", "Program OTHER invoke [1]", event, "Program OTHER success"], None, 0), ("foreign-cpi", "Foreign CPI inside the program", ["Program PUMP invoke [1]", "Program OTHER invoke [2]", event, "Program OTHER success", "Program PUMP success"], None, 0), ("pump-cpi", "Program invoked through CPI", ["Program OTHER invoke [1]", "Program PUMP invoke [2]", event, "Program PUMP success", "Program OTHER success"], None, 1), ("truncated", "Missing runtime completion", ["Program PUMP invoke [1]", event], None, 0), ("failed-ancestor", "Failed ancestor contradicts reported success", ["Program OTHER invoke [1]", "Program PUMP invoke [2]", event, "Program PUMP success", "Program OTHER failed: synthetic rollback"], None, 0), ("later-failure", "Later top-level failure rejects earlier events", ["Program PUMP invoke [1]", event, "Program PUMP success", "Program OTHER invoke [1]", "Program OTHER failed: synthetic rollback"], None, 0), ("caught-child", "Caught failed CPI preserves parent event", ["Program PUMP invoke [1]", event, "Program OTHER invoke [2]", "Program PUMP invoke [3]", event, "Program PUMP success", "Program OTHER failed: synthetic rollback", "Program PUMP success"], None, 1), ("mismatched", "Mismatched stack completion", ["Program PUMP invoke [1]", event, "Program OTHER success"], None, 0), ("error", "Transaction error overrides events", ["Program PUMP invoke [1]", event, "Program PUMP success"], {"synthetic": "failure"}, 0), ("depth", "Noncontiguous invocation depth", ["Program PUMP invoke [2]", event, "Program PUMP success"], None, 0), ("orphan", "Event without an active frame", [event], None, 0), ] return {"schema_version": 1, "synthetic": True, "scope": "Reduced offline models; no market, RPC or executable transaction.", "boolean": [{"value": x, **boolean_result(x)} for x in (0, 1, 2, 127, 255)], "attribution": [{"id": key, "label": label, "logs": lines, "transaction_error": err, "legacy": 0 if err is not None else sum(line == event for line in lines), "fixed": count} for key, label, lines, err, count in traces], "gap": [{"opening": value, **gap_result(value)} for value in (85, 90, 93, 95, 99)]} class Reproductions(unittest.TestCase): def test_boolean_boundary(self): for item in fixtures()["boolean"]: with self.subTest(byte=item["value"]): self.assertEqual(boolean_result(item["value"])["accepted"], item["value"] in (0, 1)) self.assertEqual(boolean_result(2), {"legacy": "BUY", "fixed": "REJECT", "accepted": False}) def test_attribution_boundaries(self): for item in fixtures()["attribution"]: with self.subTest(trace=item["id"]): self.assertEqual(attributed_events(item["logs"], item["transaction_error"]), item["fixed"]) # The earlier stream already rejected a reported transaction error. self.assertEqual(next(x for x in fixtures()["attribution"] if x["id"] == "error")["legacy"], 0) def test_gap_accounting(self): self.assertEqual(gap_result(85), {"legacy": -1.0, "fixed": -3.0}) self.assertEqual(gap_result(93)["fixed"], -1.4) self.assertEqual(gap_result(99)["fixed"], -1.0) def test_program_text_cannot_forge_a_runtime_marker(self): logs = ["Program OTHER invoke [1]", "Program log: Program PUMP invoke [2]", "Program data: TRADE:1", "Program OTHER success"] self.assertEqual(attributed_events(logs), 0) def test_balanced_earlier_events_discarded_on_later_truncation(self): logs = fixtures()["attribution"][0]["logs"] + ["Program OTHER invoke [1]"] self.assertEqual(attributed_events(logs), 0) def test_invalid_inputs_reject(self): for value in (-1, 256, True, 1.5, "1"): with self.subTest(value=value), self.assertRaises(ValueError): boolean_result(value) for value in (float("nan"), float("inf"), True): with self.subTest(value=value), self.assertRaises(ValueError): gap_result(value) if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--fixtures", action="store_true", help="Print the deterministic browser fixtures as JSON") args = parser.parse_args() if args.fixtures: print(json.dumps(fixtures(), indent=2)) else: unittest.main(argv=["reproduce.py"], verbosity=2)