SuperInstance/flux-policy-tester
GitHub: SuperInstance/flux-policy-tester
专为 FLUX 字节码 AI agent 策略设计的测试框架，通过单元测试、对抗性模糊测试和守恒边界验证确保策略行为正确且资源安全。
Stars: 0 | Forks: 0
# 🧪 FLUX Policy Tester
[](https://python.org)
[](LICENSE)
[](tests/)
FLUX 策略是控制 AI agent 行为的字节码程序——由基于寄存器的 VM 强制执行的守恒定律。但字节码出了名地难以测试。一个能正确阻止 99% 违规行为的策略，如果在其预算计算中存在 off-by-one 错误，就可能导致严重的溢出。FLUX Policy Tester 为字节码策略引入了规范化的测试：带有预期结果的单元测试、对抗性模糊测试、基于属性的测试以及守恒边界验证。它使用与生产环境相同的 VM（而非模拟），因此您测试的是真实的执行路径。
## 它的功能
Policy Tester 提供了一个 `PolicyTester` 类，它封装了 FLUX 字节码策略并针对测试输入运行它。三种测试模式覆盖了完整的质量面：
**单元测试** —— 定义具有预期输出的特定输入。“给定 temperature=72，死区控制器应返回 action=idle。”测试器执行字节码，读取输出寄存器，并与预期值进行比较。快速、确定性强且对 CI 友好。
**对抗性模糊测试** —— 向策略输入极端、边缘和格式错误的输入以寻找漏洞。如果 temperature=-999 会怎样？temperature=NaN 呢？空字符串呢？一兆字节的输入呢？模糊测试器会根据策略声明的输入类型自动生成对抗性输入并运行它们，标记出任何导致崩溃、挂起或意外输出的输入。
**守恒边界验证** —— 给定一个策略和输入语料库，验证策略永远不会超过其声明的守恒限制（最大步数、内存预算、执行时间）。这可以捕获行为正确但在某些输入上存在性能病理的策略——一个在某个特定输入上循环 10,000 次的策略就是一个拒绝服务风险。
该测试器包含一个内联的 FLUX VM（与 conservation-enforcer 中使用的相同），零外部依赖。这意味着测试可以在 CI 中运行而无需安装额外的包，并且测试结果保证与生产执行相匹配——因为它是同一个 VM。
## 安装
```
pip install flux-policy-tester
```
用于开发：
```
git clone https://github.com/SuperInstance/flux-policy-tester.git
cd flux-policy-tester
pip install -e ".[dev]"
```
## 快速开始
```
from flux_policy_tester import PolicyTester
# 从 registry 或原始 bytecode 加载 policy
tester = PolicyTester.from_registry("deadband-controller")
# ── Unit Tests ──────────────────────────────────────────
# 使用预期输出测试特定输入
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",
)
# 运行所有 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}")
```
### 对抗性测试
```
# 使用极端输入进行 Fuzz
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",
)
# 根据声明的类型自动生成对抗性输入
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")
```
### 守恒边界
```
# 验证 policy 永不超出 conservation 限制
tester.test_conservation_bounds(
corpus=all_test_inputs,
max_budget=100,
max_steps=1000,
max_memory=256,
)
# 测试性能病态
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 测试套件
```
# 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
```
```
# 运行 YAML suite
tester.run_suite("suites/deadband.yaml")
```
## 架构
```
┌──────────────────────────────────────────────────────┐
│ 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 参考
### `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
```
### 结果类型
```
@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
```
## 测试
```
pip install -e ".[dev]"
# 运行 tester 自身的测试
pytest tests/ -v
# 运行特定的 test suite
pytest tests/test_unit_testing.py -v
pytest tests/test_fuzzing.py -v
pytest tests/test_conservation.py -v
```
## CLI
```
# 从命令行运行 test suite
flux-test suites/deadband.yaml
# 以详细输出运行
flux-test suites/deadband.yaml --verbose
# 针对自动生成的 fuzz 输入运行单个 policy
flux-test --policy deadband-controller --fuzz --iterations 1000
```
## 跨实现
此组件存在于两种语言中：
- **Python** (`pip install flux-policy-tester`) —— 此仓库
- **Rust** (`cargo add flux-policy-tester`) —— [SuperInstance/flux-policy-tester-rs](https://github.com/SuperInstance/flux-policy-tester-rs)
两者都实现了相同的规范。请根据您的 runtime 进行选择。
## 理念
守恒执行的可靠性仅取决于它的测试。一个存在单个未测试代码路径的策略就是一个存在潜在漏洞的策略。FLUX Policy Tester 将您应用于任何安全关键代码的同等严谨性应用于字节码策略：用于预期行为的单元测试、用于意外输入的对抗性测试，以及用于性能安全的守恒边界验证。
这是 SuperInstance 守恒执行栈的测试层。策略被编写、编译为 FLUX 字节码、使用此框架进行测试、发布到 [flux-registry](https://github.com/SuperInstance/flux-registry)，并在运行时由 [conservation-enforcer](https://github.com/SuperInstance/conservation-enforcer) 执行。每一层都有其职责；这一层的职责是确保策略执行其所声明的内容。
有关理论基础，请参阅 [AI-Writings](https://github.com/SuperInstance/AI-Writings)。
## 生态系统
### FLUX Runtime
- [flux-vm](https://github.com/SuperInstance/flux-vm) —— Python VM (`pip install flux-vm`)
- [flux-core](https://github.com/SuperInstance/flux-core) —— Rust VM (`cargo add fluxvm`)
- [flux-js](https://github.com/SuperInstance/flux-js) —— JavaScript VM (`npm install flux-js`)
### 守恒
- [conservation-enforcer](https://github.com/SuperInstance/conservation-enforcer) —— 针对 LLM 输出的守恒定律执行
- [flux-registry](https://github.com/SuperInstance/flux-registry) —— 预编译策略注册表
- **[flux-policy-tester](https://github.com/SuperInstance/flux-policy-tester)** —— **此仓库** —— 测试框架
### 理念
- [AI-Writings](https://github.com/SuperInstance/AI-Writings) —— 论文、小说、诗歌
- [NEXT_HORIZONS](https://github.com/SuperInstance/SuperInstance/blob/main/NEXT_HORIZONS.md) —— 策略
## 许可证
MIT —— 请参阅 [LICENSE](LICENSE)。
标签：AI代理, Python, 单元测试, 字节码, 属性测试, 无后门, 测试框架, 逆向工具