flux-policy-tester 0.1.2

Testing framework for FLUX bytecode agent policies

🧪 FLUX Policy Tester

Testing framework for FLUX bytecode agent policies — verify behavior, fuzz edge cases, enforce conservation bounds.

FLUX policies are bytecode programs that govern AI agent behavior — conservation laws enforced by a register-based VM. But bytecode is notoriously hard to test. A policy that correctly blocks 99% of violations but has an off-by-one in its budget calculation could let through a critical overflow. The FLUX Policy Tester brings disciplined testing to bytecode policies: unit tests with expected outcomes, adversarial fuzzing, property-based testing, and conservation bound verification. It uses the same VM as production — not a mock — so you're testing the real execution path.

What It Does

The Policy Tester provides a PolicyTester class that wraps a FLUX bytecode policy and runs it against test inputs. Three testing modes cover the full quality surface:

Unit testing — define specific inputs with expected outputs. "Given temperature=72, the deadband controller should return action=idle." The tester executes the bytecode, reads the output registers, and compares to expected values. Fast, deterministic, and CI-friendly.

Adversarial fuzzing — throw extreme, edge-case, and malformed inputs at the policy to find cracks. What happens with temperature=-999? temperature=NaN? An empty string? A megabyte of input? The fuzz tester generates adversarial inputs automatically based on the policy's declared input types and runs them, flagging any that cause crashes, hangs, or unexpected outputs.

Conservation bound verification — given a policy and a corpus of inputs, verify that the policy never exceeds its declared conservation limits (max steps, memory budget, execution time). This catches policies that are correct in behavior but have performance pathologies on certain inputs — a policy that loops 10,000 times on one specific input is a denial-of-service risk.

The tester includes an inline FLUX VM (the same one used in conservation-enforcer) with zero external dependencies. This means tests run in CI without installing additional packages, and test results are guaranteed to match production execution — because it's the same VM.

Install

pip install flux-policy-tester

For development:

git clone https://github.com/SuperInstance/flux-policy-tester.git
cd flux-policy-tester
pip install -e ".[dev]"

Quick Start

from flux_policy_tester import PolicyTester

# Load a policy from the registry or from raw bytecode
tester = PolicyTester.from_registry("deadband-controller")

# ── Unit Tests ──────────────────────────────────────────

# Test specific inputs with expected outputs
tester.test_input(
    inputs={"temperature": 72},
    expected={"action": 0},      # idle
    description="comfortable temperature → idle",
)

tester.test_input(
    inputs={"temperature": 80},
    expected={"action": 1},      # cool
    description="hot → cool",
)

tester.test_input(
    inputs={"temperature": 60},
    expected={"action": 2},      # heat
    description="cold → heat",
)

# Run all unit tests
results = tester.run_unit_tests()
print(f"{results.passed}/{results.total} tests passed")
for failure in results.failures:
    print(f"  ✗ {failure.description}: {failure.error}")

Adversarial Testing

# Fuzz with extreme inputs
tester.test_adversarial(
    inputs={"temperature": 99999},
    description="extreme high temperature",
)

tester.test_adversarial(
    inputs={"temperature": -99999},
    description="extreme low temperature",
)

tester.test_adversarial(
    inputs={"temperature": 0},
    description="zero boundary",
)

# Auto-generate adversarial inputs based on declared types
fuzz_results = tester.fuzz_type(
    input_name="temperature",
    input_type="float",
    strategies=["extremes", "boundaries", "nan", "negative"],
    iterations=100,
)
print(f"Fuzz: {fuzz_results.crashed} crashes, {fuzz_results.unexpected} unexpected")

Conservation Bounds

# Verify the policy never exceeds conservation limits
tester.test_conservation_bounds(
    corpus=all_test_inputs,
    max_budget=100,
    max_steps=1000,
    max_memory=256,
)

# Test for performance pathologies
perf = tester.profile(
    inputs=stress_test_inputs,
    iterations=10000,
)
print(f"Avg cycles: {perf.avg_cycles}")
print(f"Max cycles: {perf.max_cycles}")
print(f"P99 cycles: {perf.p99_cycles}")

YAML Test Suites

# suites/deadband.yaml
policy: deadband-controller
unit_tests:
  - inputs: {temperature: 72}
    expected: {action: 0}
    description: "comfortable → idle"
  - inputs: {temperature: 80}
    expected: {action: 1}
    description: "hot → cool"
  - inputs: {temperature: 60}
    expected: {action: 2}
    description: "cold → heat"

adversarial:
  - inputs: {temperature: 99999}
    description: "extreme high"
  - inputs: {temperature: -99999}
    description: "extreme low"

conservation:
  max_steps: 100
  max_budget: 256

# Run a YAML suite
tester.run_suite("suites/deadband.yaml")

Architecture

┌──────────────────────────────────────────────────────┐
│               FLUX Policy Tester                      │
│                                                       │
│  ┌──────────────┐                                    │
│  │ PolicyTester │                                    │
│  │              │     ┌──────────────────────┐       │
│  │ .test_input()│────▶│   Inline FLUX VM     │       │
│  │ .fuzz_type() │     │  (zero-dep, from     │       │
│  │ .profile()   │     │   conservation-      │       │
│  │ .run_suite() │     │   enforcer)          │       │
│  └──────┬───────┘     └──────────┬───────────┘       │
│         │                        │                    │
│         ▼                        ▼                    │
│  ┌──────────────┐     ┌──────────────────────┐       │
│  │ Test Results │     │  Execution Trace     │       │
│  │ .passed      │     │  .cycles             │       │
│  │ .failed      │     │  .register_states    │       │
│  │ .errors[]    │     │  .memory_accesses    │       │
│  └──────────────┘     └──────────────────────┘       │
└──────────────────────────────────────────────────────┘

API Reference

PolicyTester

class PolicyTester:
    @classmethod
    def from_registry(cls, policy_name: str) -> PolicyTester
    @classmethod
    def from_bytecode(cls, bytecode: bytes) -> PolicyTester
    @classmethod
    def from_file(cls, path: str | Path) -> PolicyTester

    # Unit testing
    def test_input(self, inputs: dict, expected: dict,
                   description: str = "") -> TestResult
    def run_unit_tests(self) -> UnitTestResults

    # Adversarial
    def test_adversarial(self, inputs: dict,
                         description: str = "") -> AdversarialResult
    def fuzz_type(self, input_name: str, input_type: str,
                  strategies: list[str] | None = None,
                  iterations: int = 100) -> FuzzResults

    # Conservation
    def test_conservation_bounds(self, corpus: list[dict],
                                 max_budget: int,
                                 max_steps: int = 1000,
                                 max_memory: int = 256) -> ConservationResult

    # Profiling
    def profile(self, inputs: list[dict],
                iterations: int = 1000) -> ProfileResult

    # Suites
    def run_suite(self, path: str | Path) -> SuiteResult

Result Types

@dataclass
class TestResult:
    passed: bool
    description: str
    expected: dict
    actual: dict
    cycles: int              # VM cycles consumed

@dataclass
class FuzzResults:
    total_runs: int
    crashed: int
    unexpected: int          # non-crash but wrong output
    crashes: list[FuzzCrash]
    coverage: float          # 0.0-1.0

@dataclass
class ProfileResult:
    avg_cycles: float
    max_cycles: int
    p99_cycles: int
    avg_memory: float
    max_memory: int

Testing

pip install -e ".[dev]"

# Run the tester's own tests
pytest tests/ -v

# Run a specific test suite
pytest tests/test_unit_testing.py -v
pytest tests/test_fuzzing.py -v
pytest tests/test_conservation.py -v

CLI

# Run a test suite from the command line
flux-test suites/deadband.yaml

# Run with verbose output
flux-test suites/deadband.yaml --verbose

# Run a single policy against auto-generated fuzz inputs
flux-test --policy deadband-controller --fuzz --iterations 1000

Cross-Implementation

This component exists in two languages:

• Python (pip install flux-policy-tester) — this repo
• Rust (cargo add flux-policy-tester) — SuperInstance/flux-policy-tester-rs

Both implement the same specification. Choose based on your runtime.

Philosophy

Conservation enforcement is only as trustworthy as its tests. A policy with a single untested code path is a policy with a potential exploit. The FLUX Policy Tester applies the same rigor to bytecode policies that you'd apply to any safety-critical code: unit tests for expected behavior, adversarial testing for unexpected inputs, and conservation bound verification for performance safety.

This is the testing layer of the SuperInstance conservation enforcement stack. Policies are written, compiled to FLUX bytecode, tested with this framework, published to the flux-registry, and enforced at runtime by the conservation-enforcer. Each layer has a job; this layer's job is making sure policies do what they claim.

For the theoretical foundation, see AI-Writings.

Ecosystem

FLUX Runtime

• flux-vm — Python VM (pip install flux-vm)
• flux-core — Rust VM (cargo add fluxvm)
• flux-js — JavaScript VM (npm install flux-js)

Conservation

• conservation-enforcer — Conservation-law enforcement for LLM outputs
• flux-registry — Pre-compiled policy registry
• flux-policy-tester — This repo — testing framework

Philosophy

• AI-Writings — Essays, fiction, poetry
• NEXT_HORIZONS — Strategy

License

MIT — see LICENSE.