Stack Evaluator API Reference

The stack evaluator module (jsl.stack_evaluator) provides a stack-based virtual machine for executing JSL programs compiled to JPN (JSL Postfix Notation).

Overview

The stack evaluator offers an alternative execution model to the recursive evaluator: - Linear execution of postfix instructions - Natural resumption support for distributed computing - Efficient resource tracking with step counting - Dict-based closures for JSON serialization

Classes

StackEvaluator

The main stack-based evaluator class:

jsl.stack_evaluator.StackEvaluator(env=None, resource_budget=None, host_dispatcher=None)

Evaluator for postfix expressions using a value stack.

Initialize evaluator with optional environment and resource budget.

Parameters:

Name	Type	Description	Default
env	Optional[Env]	Environment for variable lookups (Env object)	None
resource_budget	Optional[ResourceBudget]	Optional resource budget for tracking gas/memory	None
host_dispatcher		Optional host dispatcher for side effects	None

Source code in jsl/stack_evaluator.py

def __init__(self, env: Optional[Env] = None, resource_budget: Optional[ResourceBudget] = None, host_dispatcher=None):
    """
    Initialize evaluator with optional environment and resource budget.

    Args:
        env: Environment for variable lookups (Env object)
        resource_budget: Optional resource budget for tracking gas/memory
        host_dispatcher: Optional host dispatcher for side effects
    """
    self.env = env or Env()
    self.resource_budget = resource_budget
    self.host_dispatcher = host_dispatcher
    self.builtins = self._setup_builtins()
    self.special_forms = SpecialFormEvaluator(self)

eval(instructions, state=None, env=None)

Evaluate postfix instructions.

Parameters:

Name	Type	Description	Default
instructions	List[Any]	List of postfix instructions	required
state	Optional[StackState]	Optional saved state for resumption	None
env	Optional[Env]	Optional environment override (Env object)	None

Returns:

Type	Description
Any	Result of evaluation

Raises:

Type	Description
ValueError	On invalid instructions or stack underflow
ResourceExhausted	When resource limits are exceeded

Source code in jsl/stack_evaluator.py

def eval(self, instructions: List[Any], state: Optional[StackState] = None, env: Optional[Env] = None) -> Any:
    """
    Evaluate postfix instructions.

    Args:
        instructions: List of postfix instructions
        state: Optional saved state for resumption
        env: Optional environment override (Env object)

    Returns:
        Result of evaluation

    Raises:
        ValueError: On invalid instructions or stack underflow
        ResourceExhausted: When resource limits are exceeded
    """
    if state:
        # Resume from saved state
        stack = state.stack.copy()
        pc = state.pc
        instructions = state.instructions
        # Restore resource state if available
        if state.resource_checkpoint and self.resource_budget:
            self.resource_budget.restore(state.resource_checkpoint)
        # Restore environment if available
        if state.env:
            self.env = state.env
    else:
        # Start fresh
        stack = []
        pc = 0
        self._last_tracked_memory = 0  # Reset memory tracking

    # Use provided env or default
    old_env = None
    if env is not None:
        old_env = self.env
        self.env = env

    while pc < len(instructions):
        # Check resources before each operation
        self._check_resources()
        self._track_memory_for_stack(len(stack))

        instr = instructions[pc]

        # Check for special form marker
        if instr == Opcode.SPECIAL_FORM:
            # Handle special form
            pc += 1
            if pc >= len(instructions):
                raise ValueError("Special form marker without form")

            special_expr = instructions[pc]
            pc += 1

            if isinstance(special_expr, list) and len(special_expr) > 0:
                form = special_expr[0]
                args = special_expr[1:]

                # Consume gas for special form
                self._consume_gas(GasCost.FUNCTION_CALL, f"special form: {form}")

                # Evaluate special form
                result = self.special_forms.eval_special_form(form, args, self.env)

                # Push result to stack
                stack.append(result)
            else:
                raise ValueError(f"Invalid special form: {special_expr}")

        # Check if this is an arity (number followed by operator)
        elif (isinstance(instr, int) and 
            pc + 1 < len(instructions) and 
            isinstance(instructions[pc + 1], str) and
            (instructions[pc + 1] in self.builtins or 
             instructions[pc + 1] in self.env or
             instructions[pc + 1] == '__apply__' or
             instructions[pc + 1] == '__dict__' or
             instructions[pc + 1] == '__empty_list__')):
            # This is an arity-operator pair
            arity = instr
            operator = instructions[pc + 1]
            pc += 2  # Skip both arity and operator

            # Check if it's a special __apply__ operator
            if operator == '__apply__':
                # Pop arguments from stack
                if len(stack) < arity + 1:  # +1 for the function itself
                    raise ValueError(f"Stack underflow: apply needs function + {arity} args, have {len(stack)}")

                # Pop arguments
                args = []
                for _ in range(arity):
                    args.insert(0, stack.pop())

                # Pop function/closure
                func = stack.pop()

                # Apply the function
                if isinstance(func, Closure):
                    # It's a Closure - apply it
                    self._consume_gas(GasCost.FUNCTION_CALL, "closure application")

                    # Check arity
                    if len(args) != len(func.params):
                        raise ValueError(f"Arity mismatch: closure expects {len(func.params)} args, got {len(args)}")

                    # Create new environment extending the closure's captured environment
                    call_env = func.env.extend(dict(zip(func.params, args)))

                    # Import compiler here to avoid circular dependency
                    from .compiler import compile_to_postfix

                    # Compile and evaluate body in new environment
                    body_jpn = compile_to_postfix(func.body)
                    result = self.eval(body_jpn, env=call_env)

                    # Check result constraints if we have a resource budget
                    if self.resource_budget:
                        self.resource_budget.check_result(result)

                    stack.append(result)
                else:
                    raise ValueError(f"Cannot apply non-closure: {type(func).__name__}")

            elif operator in self.builtins:
                # Regular builtin operator
                # Consume gas based on operation type and arity
                if operator in ['+', '-', '*', '/', '%']:
                    base_cost = GasCost.ARITHMETIC
                elif operator in ['=', '!=', '<', '>', '<=', '>=']:
                    base_cost = GasCost.COMPARISON
                elif operator in ['not', 'and', 'or']:
                    base_cost = GasCost.LOGICAL
                else:
                    # Built-in function call (list operations, etc.)
                    base_cost = GasCost.FUNCTION_CALL

                if arity == 2:
                    self._consume_gas(base_cost, f"binary {operator}")
                else:
                    self._consume_gas(base_cost + arity, f"n-ary {operator}")

                # Pop arguments from stack
                if len(stack) < arity:
                    raise ValueError(f"Stack underflow: {operator} needs {arity} args, have {len(stack)}")

                args = []
                for _ in range(arity):
                    args.insert(0, stack.pop())

                # Apply operator
                result = self.builtins[operator](args)

                # Check result constraints if we have a resource budget
                if self.resource_budget:
                    self.resource_budget.check_result(result)

                stack.append(result)

            else:
                # Not a builtin - could be a variable that's a function
                # Pop arguments from stack
                if len(stack) < arity:
                    raise ValueError(f"Stack underflow: {operator} needs {arity} args, have {len(stack)}")

                args = []
                for _ in range(arity):
                    args.insert(0, stack.pop())

                # Look up the function
                if operator in self.env:
                    func = self.env.get(operator)
                    if isinstance(func, Closure):
                        # It's a closure - apply it
                        self._consume_gas(GasCost.FUNCTION_CALL, f"closure call: {operator}")

                        # Check arity
                        if len(args) != len(func.params):
                            raise ValueError(f"Arity mismatch: {operator} expects {len(func.params)} args, got {len(args)}")

                        # Create new environment extending the closure's captured environment
                        call_env = func.env.extend(dict(zip(func.params, args)))

                        # Import compiler here to avoid circular dependency
                        from .compiler import compile_to_postfix

                        # Compile and evaluate body in new environment
                        body_jpn = compile_to_postfix(func.body)
                        result = self.eval(body_jpn, env=call_env)

                        # Check result constraints if we have a resource budget
                        if self.resource_budget:
                            self.resource_budget.check_result(result)

                        stack.append(result)
                    elif callable(func):
                        # Built-in function stored in env
                        self._consume_gas(GasCost.FUNCTION_CALL, f"builtin call: {operator}")
                        result = func(*args)
                        if self.resource_budget:
                            self.resource_budget.check_result(result)
                        stack.append(result)
                    else:
                        raise ValueError(f"'{operator}' is not a function")
                else:
                    raise ValueError(f"Undefined function: {operator}")

        elif isinstance(instr, (int, float, bool, type(None))):
            # Push literal number/bool/null
            self._consume_gas(GasCost.LITERAL, "literal")
            stack.append(instr)
            pc += 1

        elif isinstance(instr, str):
            if instr.startswith('@'):
                # Literal string (@ prefix)
                self._consume_gas(GasCost.LITERAL, "string literal")
                result = instr[1:]
                if self.resource_budget:
                    self.resource_budget.check_string_length(len(result))
                stack.append(result)
            else:
                # Variable lookup
                self._consume_gas(GasCost.VARIABLE, f"variable {instr}")
                try:
                    value = self.env.get(instr)
                    stack.append(value)
                except:
                    raise ValueError(f"Undefined variable: {instr}")
            pc += 1

        elif isinstance(instr, list):
            # Push list literal
            self._consume_gas(GasCost.LIST_CREATE + len(instr) * GasCost.LIST_PER_ITEM, "list literal")
            if self.resource_budget:
                self.resource_budget.check_collection_size(len(instr))
            stack.append(instr)
            pc += 1

        elif isinstance(instr, dict):
            # Push dict literal
            self._consume_gas(GasCost.DICT_CREATE + len(instr) * GasCost.DICT_PER_ITEM, "dict literal")
            if self.resource_budget:
                self.resource_budget.check_collection_size(len(instr))
            stack.append(instr)
            pc += 1

        else:
            # Other types - push as-is
            self._consume_gas(GasCost.LITERAL, "literal")
            stack.append(instr)
            pc += 1

    if len(stack) != 1:
        raise ValueError(f"Invalid expression: stack has {len(stack)} items at end")

    # Restore original env if we overrode it
    if old_env is not None:
        self.env = old_env

    return stack[0]

eval_partial(instructions, max_steps, state=None)

Evaluate with step limit for resumption.

Parameters:

Name	Type	Description	Default
instructions	List[Any]	Postfix instructions	required
max_steps	int	Maximum steps to execute	required
state	Optional[StackState]	Optional saved state	None

Returns:

Type	Description
tuple[Optional[Any], Optional[StackState]]	Tuple of (result, state) - result is None if not complete

Raises:

Type	Description
ResourceExhausted	When resource limits are exceeded

Source code in jsl/stack_evaluator.py

def eval_partial(self, instructions: List[Any], max_steps: int, 
                 state: Optional[StackState] = None) -> tuple[Optional[Any], Optional[StackState]]:
    """
    Evaluate with step limit for resumption.

    Args:
        instructions: Postfix instructions
        max_steps: Maximum steps to execute
        state: Optional saved state

    Returns:
        Tuple of (result, state) - result is None if not complete

    Raises:
        ResourceExhausted: When resource limits are exceeded
    """
    if state:
        stack = state.stack.copy()
        pc = state.pc
        instructions = state.instructions
        # Restore resource state if available
        if state.resource_checkpoint and self.resource_budget:
            self.resource_budget.restore(state.resource_checkpoint)
        # Restore user environment if available
        if state.env:
            # Use the restored environment
            self.env = state.env
    else:
        stack = []
        pc = 0
        self._last_tracked_memory = 0  # Reset memory tracking

    steps = 0

    while pc < len(instructions) and steps < max_steps:
        # Check resources before each operation
        self._check_resources()
        self._track_memory_for_stack(len(stack))

        instr = instructions[pc]

        # Check if this is an arity (number followed by operator)
        if (isinstance(instr, int) and 
            pc + 1 < len(instructions) and 
            isinstance(instructions[pc + 1], str) and
            (instructions[pc + 1] in self.builtins or
             instructions[pc + 1] in self.env)):
            # This is an arity-operator pair
            arity = instr
            operator = instructions[pc + 1]
            pc += 2  # Skip both arity and operator

            # Consume gas based on operation type and arity
            # Use different gas costs based on operator type
            if operator in ['+', '-', '*', '/', '%']:
                base_cost = GasCost.ARITHMETIC
            elif operator in ['=', '!=', '<', '>', '<=', '>=']:
                base_cost = GasCost.COMPARISON
            elif operator in ['not', 'and', 'or']:
                base_cost = GasCost.LOGICAL
            else:
                # Built-in function call (list operations, etc.)
                base_cost = GasCost.FUNCTION_CALL

            if arity == 2:
                self._consume_gas(base_cost, f"binary {operator}")
            else:
                self._consume_gas(base_cost + arity, f"n-ary {operator}")

            # Pop arguments from stack
            if len(stack) < arity:
                raise ValueError(f"Stack underflow: {operator} needs {arity} args, have {len(stack)}")

            args = []
            for _ in range(arity):
                args.insert(0, stack.pop())

            # Apply operator
            if operator in self.builtins:
                result = self.builtins[operator](args)

                # Check result constraints if we have a resource budget
                if self.resource_budget:
                    self.resource_budget.check_result(result)

                stack.append(result)
            else:
                # Not a builtin - could be a user-defined function in env
                if operator in self.env:
                    func = self.env.get(operator)
                    if isinstance(func, Closure):
                        # It's a closure - apply it
                        self._consume_gas(GasCost.FUNCTION_CALL, f"closure call: {operator}")

                        # Check arity
                        if len(args) != len(func.params):
                            raise ValueError(f"Arity mismatch: {operator} expects {len(func.params)} args, got {len(args)}")

                        # Create new environment extending the closure's captured environment
                        call_env = func.env.extend(dict(zip(func.params, args)))

                        # Import compiler here to avoid circular dependency
                        from .compiler import compile_to_postfix

                        # Compile and evaluate body in new environment
                        body_jpn = compile_to_postfix(func.body)
                        result = self.eval(body_jpn, env=call_env)

                        # Check result constraints if we have a resource budget
                        if self.resource_budget:
                            self.resource_budget.check_result(result)

                        stack.append(result)
                    elif callable(func):
                        # Built-in function stored in env
                        self._consume_gas(GasCost.FUNCTION_CALL, f"builtin call: {operator}")
                        result = func(*args)
                        if self.resource_budget:
                            self.resource_budget.check_result(result)
                        stack.append(result)
                    else:
                        raise ValueError(f"'{operator}' is not a function")
                else:
                    raise ValueError(f"Undefined function: {operator}")

        elif isinstance(instr, (int, float, bool, type(None))):
            # Push literal number/bool/null
            self._consume_gas(GasCost.LITERAL, "literal")
            stack.append(instr)
            pc += 1

        elif isinstance(instr, str):
            if instr.startswith('@'):
                # Literal string (@ prefix)
                self._consume_gas(GasCost.LITERAL, "string literal")
                result = instr[1:]
                if self.resource_budget:
                    self.resource_budget.check_string_length(len(result))
                stack.append(result)
            else:
                # Variable lookup
                self._consume_gas(GasCost.VARIABLE, f"variable {instr}")
                try:
                    value = self.env.get(instr)
                    stack.append(value)
                except:
                    raise ValueError(f"Undefined variable: {instr}")
            pc += 1

        elif isinstance(instr, list):
            # Push list literal
            self._consume_gas(GasCost.LIST_CREATE + len(instr) * GasCost.LIST_PER_ITEM, "list literal")
            if self.resource_budget:
                self.resource_budget.check_collection_size(len(instr))
            stack.append(instr)
            pc += 1

        elif isinstance(instr, dict):
            # Push dict literal
            self._consume_gas(GasCost.DICT_CREATE + len(instr) * GasCost.DICT_PER_ITEM, "dict literal")
            if self.resource_budget:
                self.resource_budget.check_collection_size(len(instr))
            stack.append(instr)
            pc += 1

        else:
            # Other types - push as-is
            self._consume_gas(GasCost.LITERAL, "literal")
            stack.append(instr)
            pc += 1

        steps += 1

    if pc >= len(instructions) and len(stack) == 1:
        # Complete
        return stack[0], None
    else:
        # Incomplete - save state with resource checkpoint
        resource_checkpoint = None
        if self.resource_budget:
            resource_checkpoint = self.resource_budget.checkpoint()

        state = StackState(
            stack=stack, 
            pc=pc, 
            instructions=instructions,
            resource_checkpoint=resource_checkpoint,
            env=self.env
        )
        return None, state

StackState

Represents the state of the stack machine for resumption:

jsl.stack_evaluator.StackState(stack, pc, instructions, resource_checkpoint=None, env=None) dataclass

State of the stack evaluator, can be serialized for resumption.

from_dict(data, prelude_env=None) classmethod

Restore from dictionary.

Source code in jsl/stack_evaluator.py

@classmethod
def from_dict(cls, data: Dict, prelude_env: Optional[Env] = None) -> 'StackState':
    """Restore from dictionary."""
    return cls(
        stack=from_json(data['stack'], prelude_env),
        pc=data['pc'],
        instructions=data['instructions'],
        resource_checkpoint=data.get('resource_checkpoint'),
        env=from_json(data['env'], prelude_env) if data.get('env') else prelude_env
    )

to_dict()

Convert to serializable dictionary.

Source code in jsl/stack_evaluator.py

def to_dict(self) -> Dict:
    """Convert to serializable dictionary."""
    return {
        'stack': to_json(self.stack),
        'pc': self.pc,
        'instructions': self.instructions,
        'resource_checkpoint': self.resource_checkpoint,
        'env': to_json(self.env) if self.env else None
    }

Usage Examples

Basic Evaluation

from jsl.stack_evaluator import StackEvaluator
from jsl.compiler import compile_to_postfix

# Create evaluator
evaluator = StackEvaluator()

# Compile S-expression to JPN
expr = ["+", 1, 2, 3]
jpn = compile_to_postfix(expr)  # [1, 2, 3, 3, "+"]

# Evaluate
result = evaluator.eval(jpn)
print(result)  # Output: 6

Resumable Execution

from jsl.stack_evaluator import StackEvaluator
from jsl.compiler import compile_to_postfix

evaluator = StackEvaluator()

# Compile a complex expression
expr = ["*", ["+", 10, 20], ["-", 100, 50]]
jpn = compile_to_postfix(expr)

# Execute with step limit
result, state = evaluator.eval_partial(jpn, max_steps=3)

if state:  # Execution paused
    print(f"Paused at PC: {state.pc}")
    print(f"Stack: {state.stack}")

    # Resume execution
    final_result, _ = evaluator.eval_partial(jpn, state=state)
    print(f"Result: {final_result}")  # Output: 1500

Working with Closures

from jsl.stack_evaluator import StackEvaluator
from jsl.compiler import compile_to_postfix

evaluator = StackEvaluator()

# Define a function
program = [
    "do",
    ["def", "square", ["lambda", ["x"], ["*", "x", "x"]]],
    ["square", 5]
]

jpn = compile_to_postfix(program)
result = evaluator.eval(jpn)
print(result)  # Output: 25

# The closure is stored as a dict
closure = evaluator.env.get("square")
print(closure["type"])    # "closure"
print(closure["params"])  # ["x"]
print(closure["body"])    # ["*", "x", "x"]

JPN Instruction Set

The stack evaluator processes these instruction types:

Values

• Literals: Numbers, strings, booleans, null, lists, dicts push themselves
• Variables: String identifiers trigger environment lookup

Operators

• N-ary operators: Preceded by arity count (e.g., 3, "+" for 3-argument addition)
• Built-in functions: Called like operators with arity

Special Forms

• Opcode.SPECIAL_FORM: Marks special form instructions
• Opcode.JUMP: Conditional/unconditional jumps
• Opcode.JUMP_IF_FALSE: Jump if top of stack is false
• Opcode.LAMBDA: Create closure from body and params

Stack Machine Architecture

Execution Model

1. Program Counter (PC): Points to current instruction
2. Operand Stack: Holds intermediate values
3. Environment: Variable bindings (dict-based)
4. Call Stack: For function calls (managed internally)

Instruction Processing

# Simplified execution loop
while pc < len(program):
    instruction = program[pc]

    if isinstance(instruction, (int, float, str, bool, type(None))):
        stack.append(instruction)
    elif instruction in operators:
        arity = program[pc - 1]
        args = [stack.pop() for _ in range(arity)]
        result = operators[instruction](*reversed(args))
        stack.append(result)

    pc += 1

Differences from Recursive Evaluator

Feature	Recursive Evaluator	Stack Evaluator
Execution	Tree walking	Linear instruction stream
Closures	Closure objects	Dict representations
Resumption	Difficult	Natural with StackState
Performance	Good for small programs	Better for large programs
Debugging	Natural call stack	Requires PC tracking

Integration with JSLRunner

The JSLRunner can use either evaluator:

from jsl.runner import JSLRunner

# Use stack evaluator (default)
runner_stack = JSLRunner(use_recursive_evaluator=False)

# Use recursive evaluator
runner_recursive = JSLRunner(use_recursive_evaluator=True)

# Both produce identical results
result1 = runner_stack.execute(["+", 1, 2, 3])
result2 = runner_recursive.execute(["+", 1, 2, 3])
assert result1 == result2  # True