Download source/icl/vm.py from invariant-computing-lab/icl-native-r5-opcode: direct link, hf CLI and curl.
- Browser
- Download file 25.6 kB
-
https://huggingface.co/invariant-computing-lab/icl-native-r5-opcode/resolve/main/source/icl/vm.py
- Command line
-
hf download hf://invariant-computing-lab/icl-native-r5-opcode/source/icl/vm.py
-
curl -L -o vm.py https://huggingface.co/invariant-computing-lab/icl-native-r5-opcode/resolve/main/source/icl/vm.py
25.6 kB
| """ICL bytecode compiler and independent stack interpreter. | |
| This backend shares scalar arithmetic, builtins, and resource accounting with | |
| the tree interpreter; it does not call the tree interpreter or execute Python | |
| source. Checkpoint hashes detect accidental modification, not adversarial | |
| forgery: they are unkeyed checksums and confer no trust or authority. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import math | |
| from copy import deepcopy | |
| from .runtime import ICLError, RuntimeContext, builtin, checked_binary, checked_unary, DEFAULT_LIMITS | |
| from .checker import valid_type, validate_value | |
| BYTECODE_SCHEMA = "icl-bytecode/1" | |
| CHECKPOINT_SCHEMA = "icl-vm-checkpoint/1" | |
| OPCODES = {"PUSH", "LOAD", "LET", "STORE", "POP", "BINARY", "UNARY", "CALL", | |
| "LIST", "TUPLE", "RECORD", "INDEX", "FIELD", "JUMP", "JUMP_FALSE", | |
| "LOOP_RESET", "LOOP_GUARD", "ENTER_SCOPE", "EXIT_SCOPE", "RETURN"} | |
| _DEFAULT_LIMITS = dict(DEFAULT_LIMITS) | |
| def _fail(code, message, span=None): | |
| code = {"E_BYTECODE": "BYTECODE_CORRUPT", "E_CHECKPOINT": "CHECKPOINT_CORRUPT", | |
| "E_LIMIT": "MEMORY_LIMIT", "E_NUMERIC": "NONFINITE", "E_VALUE": "TYPE", | |
| "E_TYPE": "TYPE", "E_ENTRY": "NAME", "E_NAME": "NAME", "E_ARITY": "TYPE", | |
| "E_IMMUTABLE": "IMMUTABLE", "E_INDEX": "BOUNDS", "E_FIELD": "TYPE"}.get(code, code) | |
| raise ICLError(code, message, span) | |
| def _canonical(value): | |
| try: | |
| return json.dumps(value, sort_keys=True, separators=(",", ":"), | |
| ensure_ascii=False, allow_nan=False) | |
| except (TypeError, ValueError, RecursionError): | |
| _fail("E_BYTECODE", "Value is not finite canonical JSON") | |
| def _hash(value): | |
| return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest() | |
| def compile_bytecode(program): | |
| """Lower a checked ICL AST to a JSON-serializable instruction stream.""" | |
| functions = {} | |
| for function in program.get("functions", []): | |
| name = function["name"] | |
| if name in functions: | |
| _fail("E_BYTECODE", "Duplicate function: " + name) | |
| code = [] | |
| loop_sequence = [0] | |
| def emit(op, span=None, **kwargs): | |
| index = len(code) | |
| code.append({"op": op, **kwargs, "span": span or function.get("span")}) | |
| return index | |
| def expr(node): | |
| kind = node["kind"] | |
| span = node.get("span") | |
| if kind == "literal": | |
| emit("PUSH", span, value=node["value"]) | |
| elif kind == "var": | |
| emit("LOAD", span, name=node["name"]) | |
| elif kind == "binary": | |
| # Preserve short-circuit evaluation without calling AST eval. | |
| if node["op"] in ("&&", "and"): | |
| expr(node["left"]) | |
| branch = emit("JUMP_FALSE", span, target=0) | |
| expr(node["right"]) | |
| finish = emit("JUMP", span, target=0) | |
| code[branch]["target"] = len(code) | |
| emit("PUSH", span, value=False) | |
| code[finish]["target"] = len(code) | |
| elif node["op"] in ("||", "or"): | |
| expr(node["left"]) | |
| branch = emit("JUMP_FALSE", span, target=0) | |
| emit("PUSH", span, value=True) | |
| finish = emit("JUMP", span, target=0) | |
| code[branch]["target"] = len(code) | |
| expr(node["right"]) | |
| code[finish]["target"] = len(code) | |
| else: | |
| expr(node["left"]) | |
| expr(node["right"]) | |
| emit("BINARY", span, operator=node["op"]) | |
| elif kind == "unary": | |
| operand = node["expr"] | |
| if node["op"] == "-" and operand["kind"] == "literal" and type(operand["value"]) is int and operand["value"] == 1 << 63: | |
| emit("PUSH", span, value=-(1 << 63)) | |
| else: | |
| expr(operand) | |
| emit("UNARY", span, operator=node["op"]) | |
| elif kind == "call": | |
| for argument in node["args"]: | |
| expr(argument) | |
| emit("CALL", span, name=node["name"], argc=len(node["args"])) | |
| elif kind in ("list", "tuple"): | |
| for item in node["items"]: | |
| expr(item) | |
| emit(kind.upper(), span, count=len(node["items"])) | |
| elif kind == "record": | |
| ordered_fields = sorted(node["fields"]) | |
| for name in ordered_fields: | |
| expr(node["fields"][name]) | |
| emit("RECORD", span, fields=ordered_fields) | |
| elif kind == "index": | |
| expr(node["target"]) | |
| expr(node["index"]) | |
| emit("INDEX", span) | |
| elif kind == "field": | |
| expr(node["target"]) | |
| emit("FIELD", span, name=node["name"]) | |
| else: | |
| _fail("E_BYTECODE", "Unsupported expression: " + str(kind), span) | |
| def statements(nodes, scoped=False): | |
| if scoped: | |
| emit("ENTER_SCOPE") | |
| for node in nodes: | |
| kind, span = node["kind"], node.get("span") | |
| if kind == "let": | |
| expr(node["expr"]) | |
| emit("LET", span, name=node["name"], mutable=node.get("mutable", False)) | |
| elif kind == "assign": | |
| expr(node["expr"]) | |
| emit("STORE", span, name=node["name"]) | |
| elif kind == "return": | |
| expr(node["expr"]) if node.get("expr") is not None else emit("PUSH", span, value=None) | |
| emit("RETURN", span) | |
| elif kind == "expr": | |
| expr(node["expr"]) | |
| emit("POP", span) | |
| elif kind == "if": | |
| expr(node["cond"]) | |
| branch = emit("JUMP_FALSE", span, target=0) | |
| statements(node["then"], scoped=True) | |
| finish = emit("JUMP", span, target=0) | |
| code[branch]["target"] = len(code) | |
| statements(node.get("else", []), scoped=True) | |
| code[finish]["target"] = len(code) | |
| elif kind == "while": | |
| loop_sequence[0] += 1 | |
| key = str(loop_sequence[0]) | |
| emit("LOOP_RESET", span, key=key) | |
| start = len(code) | |
| expr(node["cond"]) | |
| branch = emit("JUMP_FALSE", span, target=0) | |
| emit("LOOP_GUARD", span, key=key, limit=node["limit"]) | |
| statements(node["body"], scoped=True) | |
| emit("JUMP", span, target=start) | |
| code[branch]["target"] = len(code) | |
| else: | |
| _fail("E_BYTECODE", "Unsupported statement: " + str(kind), span) | |
| if scoped: | |
| emit("EXIT_SCOPE") | |
| # Parameters occupy the call frame; the function body is a child lexical | |
| # scope, matching the source rule that a body local may shadow a parameter. | |
| statements(function["body"], scoped=True) | |
| emit("PUSH", value=None) | |
| emit("RETURN") | |
| functions[name] = {"params": function["params"], "return_type": function["return_type"], | |
| "code": code} | |
| result = {"schema": BYTECODE_SCHEMA, "module": program.get("module", "main"), "functions": functions} | |
| validate_bytecode(result) | |
| return result | |
| def _integer(value, lower=0, upper=10000000): | |
| return type(value) is int and lower <= value <= upper | |
| def _validate_bytecode_impl(bytecode): | |
| """Fail closed on malformed code before executing any instruction.""" | |
| if not isinstance(bytecode, dict) or bytecode.get("schema") != BYTECODE_SCHEMA: | |
| _fail("E_BYTECODE", "Unsupported bytecode schema") | |
| if len(_canonical(bytecode)) > 20000000: | |
| _fail("E_BYTECODE", "Bytecode exceeds serialized size limit") | |
| functions = bytecode.get("functions") | |
| if not isinstance(functions, dict) or not functions or len(functions) > 10000: | |
| _fail("E_BYTECODE", "Invalid function table") | |
| for name, function in functions.items(): | |
| if not isinstance(name, str) or not name or not isinstance(function, dict): | |
| _fail("E_BYTECODE", "Invalid function descriptor") | |
| params, code = function.get("params"), function.get("code") | |
| if not isinstance(params, list) or any(not isinstance(p, (list, tuple)) or len(p) != 2 | |
| or not isinstance(p[0], str) or not isinstance(p[1], str) for p in params): | |
| _fail("E_BYTECODE", "Invalid parameter table") | |
| if len({p[0] for p in params}) != len(params): | |
| _fail("E_BYTECODE", "Duplicate parameter") | |
| if any(not valid_type(p[1]) for p in params) or not isinstance(function.get("return_type"), str) or not valid_type(function["return_type"]): | |
| _fail("E_BYTECODE", "Invalid function type signature") | |
| if not isinstance(code, list) or not code or len(code) > 1000000: | |
| _fail("E_BYTECODE", "Invalid instruction stream") | |
| for instruction in code: | |
| if not isinstance(instruction, dict) or not isinstance(instruction.get("op"), str) or instruction["op"] not in OPCODES: | |
| _fail("E_BYTECODE", "Unknown or malformed opcode") | |
| op = instruction["op"] | |
| if op in {"LOAD", "LET", "STORE", "CALL", "FIELD"} and not isinstance(instruction.get("name"), str): | |
| _fail("E_BYTECODE", "Invalid identifier operand") | |
| if op == "PUSH" and "value" not in instruction: | |
| _fail("E_BYTECODE", "Missing literal operand") | |
| if op == "LET" and type(instruction.get("mutable")) is not bool: | |
| _fail("E_BYTECODE", "Invalid mutability operand") | |
| if op in {"LIST", "TUPLE"} and not _integer(instruction.get("count")): | |
| _fail("E_BYTECODE", "Invalid collection size") | |
| if op == "CALL" and not _integer(instruction.get("argc")): | |
| _fail("E_BYTECODE", "Invalid argument count") | |
| if op == "RECORD" and (not isinstance(instruction.get("fields"), list) | |
| or any(not isinstance(k, str) for k in instruction["fields"]) | |
| or len(instruction["fields"]) != len(set(instruction["fields"]))): | |
| _fail("E_BYTECODE", "Invalid record fields") | |
| if op in {"JUMP", "JUMP_FALSE"} and not _integer(instruction.get("target"), 0, len(code) - 1): | |
| _fail("E_BYTECODE", "Jump target outside instruction stream") | |
| if op == "BINARY" and (not isinstance(instruction.get("operator"), str) or instruction["operator"] not in {"+", "-", "*", "/", "%", "==", "!=", "<", "<=", ">", ">=", "&&", "||"}): | |
| _fail("E_BYTECODE", "Invalid binary operator") | |
| if op == "UNARY" and (not isinstance(instruction.get("operator"), str) or instruction["operator"] not in {"-", "!"}): | |
| _fail("E_BYTECODE", "Invalid unary operator") | |
| if op in {"LOOP_RESET", "LOOP_GUARD"} and not isinstance(instruction.get("key"), str): | |
| _fail("E_BYTECODE", "Invalid loop key") | |
| if op == "LOOP_GUARD" and not _integer(instruction.get("limit"), 0): | |
| _fail("E_BYTECODE", "Invalid loop limit") | |
| return True | |
| def validate_bytecode(bytecode): | |
| """Validate untrusted JSON data and expose only deterministic ICL errors.""" | |
| try: | |
| return _validate_bytecode_impl(bytecode) | |
| except ICLError: | |
| raise | |
| except (TypeError, KeyError, IndexError, ValueError, AttributeError, RecursionError): | |
| raise ICLError("BYTECODE_CORRUPT", "Malformed bytecode descriptor") from None | |
| def _normalized_limits(limits): | |
| result = dict(_DEFAULT_LIMITS) | |
| if limits: | |
| for key, value in limits.items(): | |
| if key not in result or not _integer(value, 1, result[key]): | |
| _fail("RESOURCE_POLICY", "Invalid resource limit: " + str(key)) | |
| result[key] = value | |
| return result | |
| def _validate_value(value, limits, depth=0, budget=None): | |
| if budget is None: | |
| budget = [limits["max_items"]] | |
| budget[0] -= 1 | |
| if budget[0] < 0 or depth > 128: | |
| _fail("E_LIMIT", "Value exceeds collection or nesting limit") | |
| if type(value) is int and not -(1 << 63) <= value < (1 << 63): | |
| _fail("OVERFLOW", "Int64 overflow") | |
| if value is None or type(value) in (bool, int): | |
| return | |
| if type(value) is float: | |
| if not math.isfinite(value): | |
| _fail("E_NUMERIC", "Non-finite floating-point value") | |
| return | |
| if isinstance(value, str): | |
| if len(value) > limits["max_output"]: | |
| _fail("E_LIMIT", "String exceeds output limit") | |
| return | |
| if isinstance(value, (list, tuple)): | |
| for item in value: | |
| _validate_value(item, limits, depth + 1, budget) | |
| return | |
| if isinstance(value, dict) and all(isinstance(k, str) for k in value): | |
| for item in value.values(): | |
| _validate_value(item, limits, depth + 1, budget) | |
| return | |
| _fail("E_VALUE", "Unsupported runtime value") | |
| def _new_frame(name, args, functions, ctx): | |
| params = functions[name]["params"] | |
| if len(args) != len(params): | |
| _fail("E_ARITY", "Wrong argument count for " + name) | |
| for (_, typ), value in zip(params, args): | |
| validate_value(value, typ, ctx) | |
| scope = {"locals": {p[0]: deepcopy(v) for p, v in zip(params, args)}, | |
| "mutable": {p[0]: False for p in params}} | |
| return {"function": name, "pc": 0, "scopes": [scope], "stack": [], "loops": {}} | |
| def _execute_bytecode(bytecode, entry="main", args=None, limits=None, policy=None, | |
| checkpoint=None, stop_after=None): | |
| """Execute, pause, or resume verified bytecode. No ambient I/O is available.""" | |
| validate_bytecode(bytecode) | |
| functions = bytecode["functions"] | |
| limits = _normalized_limits(limits) | |
| # Authority always comes from this trusted host argument, including replay. | |
| # A checkpoint is data with an unkeyed checksum, never an authorization token. | |
| policy = {} if policy is None else policy | |
| if not isinstance(policy, dict): | |
| _fail("RESOURCE_POLICY", "Host policy must be an object") | |
| if stop_after is not None and not _integer(stop_after, 0): | |
| _fail("E_LIMIT", "stop_after must be a nonnegative integer") | |
| if checkpoint is not None: | |
| if not isinstance(checkpoint, dict) or checkpoint.get("schema") != CHECKPOINT_SCHEMA: | |
| _fail("E_CHECKPOINT", "Unsupported checkpoint schema") | |
| payload = {k: v for k, v in checkpoint.items() if k != "integrity"} | |
| if checkpoint.get("integrity") != _hash(payload) or checkpoint.get("bytecode_hash") != _hash(bytecode): | |
| _fail("E_CHECKPOINT", "Checkpoint integrity or bytecode identity mismatch") | |
| if len(_canonical(checkpoint)) > 20000000: | |
| _fail("E_CHECKPOINT", "Checkpoint too large") | |
| previous_limits = _normalized_limits(checkpoint.get("limits")) | |
| limits = {k: min(v, previous_limits[k]) for k, v in limits.items()} | |
| prior_policy = checkpoint.get("policy") | |
| if policy != prior_policy: | |
| _fail("E_CHECKPOINT", "Replay requires an independently supplied identical host policy") | |
| ctx = RuntimeContext(limits=limits, policy=policy) | |
| if checkpoint is None: | |
| if entry not in functions: | |
| _fail("E_ENTRY", "Unknown entry function: " + entry) | |
| args = [] if args is None else args | |
| if not isinstance(args, list): | |
| _fail("E_ARITY", "Entry arguments must be a list") | |
| _validate_value(args, limits) | |
| frames = [_new_frame(entry, args, functions, ctx)] | |
| else: | |
| frames = deepcopy(checkpoint.get("frames")) | |
| steps, trace, effects = checkpoint.get("steps"), checkpoint.get("trace"), checkpoint.get("effects") | |
| if not _integer(steps, 0, limits["max_steps"]) or not isinstance(trace, list) or not isinstance(effects, list): | |
| _fail("E_CHECKPOINT", "Malformed checkpoint accounting") | |
| if not isinstance(frames, list) or not frames or len(frames) > limits["max_stack"]: | |
| _fail("E_CHECKPOINT", "Malformed checkpoint frame stack") | |
| for frame in frames: | |
| if not isinstance(frame, dict) or frame.get("function") not in functions: | |
| _fail("E_CHECKPOINT", "Unknown checkpoint function") | |
| if not _integer(frame.get("pc"), 0, len(functions[frame["function"]]["code"]) - 1): | |
| _fail("E_CHECKPOINT", "Invalid checkpoint instruction pointer") | |
| if not isinstance(frame.get("scopes"), list) or not frame["scopes"] or not isinstance(frame.get("stack"), list): | |
| _fail("E_CHECKPOINT", "Invalid checkpoint values") | |
| if len(frame["scopes"]) > limits["max_steps"] + 1: | |
| _fail("E_CHECKPOINT", "Too many lexical scopes") | |
| for scope in frame["scopes"]: | |
| if not isinstance(scope, dict) or not isinstance(scope.get("locals"), dict) or not isinstance(scope.get("mutable"), dict) or set(scope["mutable"]) != set(scope["locals"]) or any(type(v) is not bool for v in scope["mutable"].values()): | |
| _fail("E_CHECKPOINT", "Invalid checkpoint mutability table") | |
| _validate_value(scope["locals"], limits) | |
| if not isinstance(frame.get("loops"), dict) or any(not _integer(v) for v in frame["loops"].values()): | |
| _fail("E_CHECKPOINT", "Invalid checkpoint loop counters") | |
| _validate_value(frame["stack"], limits) | |
| ctx.steps, ctx.trace, ctx.effects = steps, deepcopy(trace), deepcopy(effects) | |
| if not _integer(checkpoint.get("prng"), 0, (1 << 31) - 1): | |
| _fail("E_CHECKPOINT", "Invalid checkpoint PRNG state") | |
| ctx.prng = checkpoint["prng"] | |
| start_steps = ctx.steps | |
| def result(value, paused=False): | |
| output = {"value": value, "trace": ctx.trace, "effects": ctx.effects, | |
| "steps": ctx.steps, "halted": not paused} | |
| if paused: | |
| payload = {"schema": CHECKPOINT_SCHEMA, "bytecode_hash": _hash(bytecode), | |
| "frames": deepcopy(frames), "steps": ctx.steps, "trace": deepcopy(ctx.trace), | |
| "effects": deepcopy(ctx.effects), "limits": limits, "policy": ctx.policy, | |
| "prng": ctx.prng} | |
| payload["integrity"] = _hash(payload) | |
| output["checkpoint"] = payload | |
| return output | |
| while frames: | |
| if stop_after is not None and ctx.steps - start_steps >= stop_after: | |
| return result(None, True) | |
| if len(frames) > limits["max_stack"]: | |
| _fail("STACK_LIMIT", "VM call stack limit exceeded") | |
| frame = frames[-1] | |
| code = functions[frame["function"]]["code"] | |
| if not 0 <= frame["pc"] < len(code): | |
| _fail("E_BYTECODE", "Instruction pointer outside function") | |
| instruction = code[frame["pc"]] | |
| input_state = {"frames": frames, "prng": ctx.prng} | |
| trace_index = len(ctx.trace) | |
| effects_before = len(ctx.effects) | |
| ctx.step("BC:" + instruction["op"], inputs=input_state) | |
| frame["pc"] += 1 | |
| op, span, stack = instruction["op"], instruction.get("span"), frame["stack"] | |
| def finish_trace(return_value=None, returning=False): | |
| state = {"frames": frames, "prng": ctx.prng} | |
| if returning: | |
| state["return_value"] = return_value | |
| event = ctx.trace[trace_index] | |
| event["output_hash"] = _hash(state) | |
| event["state_ref"] = event["output_hash"] | |
| if len(ctx.effects) > effects_before: | |
| effect = ctx.effects[-1] | |
| event["effect_request"] = {"action": effect["action"], "payload_hash": effect["payload_hash"]} | |
| event["effect_decision"] = effect["decision"] | |
| def pop(count=1): | |
| if len(stack) < count: | |
| _fail("E_BYTECODE", "Operand stack underflow", span) | |
| if count == 0: | |
| return [] | |
| values = stack[-count:] | |
| del stack[-count:] | |
| return values | |
| if op == "PUSH": | |
| stack.append(deepcopy(instruction["value"])) | |
| elif op == "LOAD": | |
| name = instruction["name"] | |
| scope = next((s for s in reversed(frame["scopes"]) if name in s["locals"]), None) | |
| if scope is None: | |
| _fail("E_NAME", "Undefined variable: " + name, span) | |
| stack.append(deepcopy(scope["locals"][name])) | |
| elif op == "LET": | |
| name = instruction["name"] | |
| scope = frame["scopes"][-1] | |
| if name in scope["locals"]: | |
| _fail("E_NAME", "Duplicate local declaration: " + name, span) | |
| scope["locals"][name] = pop()[0] | |
| scope["mutable"][name] = instruction["mutable"] | |
| elif op == "STORE": | |
| name = instruction["name"] | |
| scope = next((s for s in reversed(frame["scopes"]) if name in s["locals"]), None) | |
| if scope is None: | |
| _fail("E_NAME", "Undefined variable: " + name, span) | |
| if not scope["mutable"][name]: | |
| _fail("E_IMMUTABLE", "Cannot assign immutable variable: " + name, span) | |
| scope["locals"][name] = pop()[0] | |
| elif op == "ENTER_SCOPE": | |
| frame["scopes"].append({"locals": {}, "mutable": {}}) | |
| elif op == "EXIT_SCOPE": | |
| if len(frame["scopes"]) <= 1: | |
| _fail("E_BYTECODE", "Cannot exit function root scope", span) | |
| frame["scopes"].pop() | |
| elif op == "POP": | |
| pop() | |
| elif op == "BINARY": | |
| a, b = pop(2) | |
| stack.append(checked_binary(instruction["operator"], a, b)) | |
| elif op == "UNARY": | |
| stack.append(checked_unary(instruction["operator"], pop()[0])) | |
| elif op == "CALL": | |
| arguments = pop(instruction["argc"]) | |
| name = instruction["name"] | |
| if name in functions: | |
| if len(frames) >= limits["max_stack"]: | |
| _fail("STACK_LIMIT", "VM call stack limit exceeded", span) | |
| frames.append(_new_frame(name, arguments, functions, ctx)) | |
| else: | |
| stack.append(builtin(name, arguments, ctx)) | |
| elif op in {"LIST", "TUPLE"}: | |
| values = pop(instruction["count"]) | |
| stack.append(values) | |
| elif op == "RECORD": | |
| stack.append(dict(zip(instruction["fields"], pop(len(instruction["fields"]))))) | |
| elif op == "INDEX": | |
| target, index = pop(2) | |
| if not isinstance(target, (list, tuple, str)) or type(index) is not int or not 0 <= index < len(target): | |
| _fail("E_INDEX", "Invalid or out-of-range index", span) | |
| stack.append(target[index]) | |
| elif op == "FIELD": | |
| target = pop()[0] | |
| if not isinstance(target, dict) or instruction["name"] not in target: | |
| _fail("E_FIELD", "Missing record field: " + instruction["name"], span) | |
| stack.append(target[instruction["name"]]) | |
| elif op == "JUMP": | |
| frame["pc"] = instruction["target"] | |
| elif op == "JUMP_FALSE": | |
| condition = pop()[0] | |
| if type(condition) is not bool: | |
| _fail("E_TYPE", "Branch condition must be Bool", span) | |
| if not condition: | |
| frame["pc"] = instruction["target"] | |
| elif op == "LOOP_RESET": | |
| frame["loops"][instruction["key"]] = 0 | |
| elif op == "LOOP_GUARD": | |
| key = instruction["key"] | |
| if key not in frame["loops"]: | |
| _fail("E_BYTECODE", "Loop guard without reset", span) | |
| frame["loops"][key] += 1 | |
| if frame["loops"][key] > instruction["limit"]: | |
| _fail("LOOP_LIMIT", "Bounded loop limit exceeded", span) | |
| elif op == "RETURN": | |
| value = pop()[0] | |
| validate_value(value, functions[frame["function"]]["return_type"], ctx) | |
| if stack: | |
| _fail("E_BYTECODE", "Unbalanced operand stack at return", span) | |
| frames.pop() | |
| if not frames: | |
| _validate_value(value, limits) | |
| finish_trace(value, returning=True) | |
| return result(value) | |
| frames[-1]["stack"].append(value) | |
| if len(stack) > limits["max_items"]: | |
| _fail("E_LIMIT", "VM operand stack limit exceeded", span) | |
| if stack: | |
| _validate_value(stack[-1], limits) | |
| ctx.check_value(stack[-1]) | |
| finish_trace() | |
| _fail("E_BYTECODE", "VM terminated without return") | |
| def execute_bytecode(bytecode, entry="main", args=None, limits=None, policy=None, | |
| checkpoint=None, stop_after=None): | |
| """Public fail-closed boundary; malformed data cannot escape as host errors.""" | |
| try: | |
| return _execute_bytecode(bytecode, entry, args, limits, policy, checkpoint, stop_after) | |
| except ICLError: | |
| raise | |
| except (TypeError, KeyError, IndexError, ValueError, OverflowError, ZeroDivisionError, AttributeError) as error: | |
| raise ICLError("TYPE", "Malformed bytecode operand or primitive argument: " + type(error).__name__) from None | |
| except RecursionError: | |
| raise ICLError("STACK_LIMIT", "Bytecode structure exceeds host recursion limit") from None | |