238 lines
9.2 KiB
Python
238 lines
9.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Unit tests for zubble doom-loop escape hatch detectors.
|
||
|
|
|
||
|
|
Pure-logic tests — no I/O, no mocking. The detectors are deterministic
|
||
|
|
state machines; we feed them synthetic token streams and action
|
||
|
|
sequences and verify threshold crossings, resets, and the accepted
|
||
|
|
period-37 blind spot.
|
||
|
|
|
||
|
|
python3 -m unittest discover -s tests -v
|
||
|
|
"""
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
|
||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
|
||
|
|
from zubble.doom import ( # noqa: E402
|
||
|
|
ActionRepetitionDetector,
|
||
|
|
ActionSignal,
|
||
|
|
StreamRepetitionDetector,
|
||
|
|
Level,
|
||
|
|
InsuranceCounter,
|
||
|
|
ResetCap,
|
||
|
|
canonical_form,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# ── Helpers ────────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
def _feed_blocks(detector, text, n_blocks, block_size=50):
|
||
|
|
"""Feed n_blocks identical 50-delta blocks to the detector."""
|
||
|
|
for _ in range(n_blocks):
|
||
|
|
for _ in range(block_size):
|
||
|
|
detector.feed("content", text)
|
||
|
|
|
||
|
|
|
||
|
|
def _feed_cycle(detector, cycle_len, n_cycles, block_size=50):
|
||
|
|
"""Feed a repeating pattern with period cycle_len. Each position in
|
||
|
|
the cycle gets a unique token so blocks only match if the period
|
||
|
|
divides the block size."""
|
||
|
|
for _ in range(n_cycles):
|
||
|
|
for i in range(cycle_len):
|
||
|
|
detector.feed("content", f"t{i} ")
|
||
|
|
|
||
|
|
|
||
|
|
# ── ActionRepetitionDetector ───────────────────────────────────────
|
||
|
|
|
||
|
|
class ActionRepetitionTest(unittest.TestCase):
|
||
|
|
|
||
|
|
def test_two_identical_silent(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
|
||
|
|
self.assertEqual(d.feed(a), ActionSignal.NONE)
|
||
|
|
self.assertEqual(d.feed(a), ActionSignal.NONE)
|
||
|
|
|
||
|
|
def test_third_identical_notes(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a)
|
||
|
|
self.assertEqual(d.feed(a), ActionSignal.NOTE)
|
||
|
|
|
||
|
|
def test_fourth_identical_kills(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a)
|
||
|
|
self.assertEqual(d.feed(a), ActionSignal.KILL)
|
||
|
|
|
||
|
|
def test_different_action_resets(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a = {"action": "run", "command": "kubectl get pods", "reasoning": "r", "plan_step_id": 1}
|
||
|
|
b = {"action": "run", "command": "kubectl get nodes", "reasoning": "r2", "plan_step_id": 2}
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a) # count=3 → NOTE
|
||
|
|
self.assertEqual(d.feed(b), ActionSignal.NONE) # reset
|
||
|
|
self.assertEqual(d.feed(b), ActionSignal.NONE) # count=2
|
||
|
|
self.assertEqual(d.feed(b), ActionSignal.NOTE) # count=3
|
||
|
|
|
||
|
|
def test_reasoning_excluded_from_canonical(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a1 = {"action": "run", "command": "ls", "reasoning": "thinking A", "plan_step_id": 1}
|
||
|
|
a2 = {"action": "run", "command": "ls", "reasoning": "thinking B", "plan_step_id": 1}
|
||
|
|
self.assertEqual(d.feed(a1), ActionSignal.NONE)
|
||
|
|
self.assertEqual(d.feed(a2), ActionSignal.NONE) # still count=2, different reasoning
|
||
|
|
self.assertEqual(d.feed(a1), ActionSignal.NOTE) # count=3
|
||
|
|
|
||
|
|
def test_reset_clears_state(self):
|
||
|
|
d = ActionRepetitionDetector()
|
||
|
|
a = {"action": "run", "command": "ls", "reasoning": "r", "plan_step_id": 1}
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a)
|
||
|
|
d.feed(a) # count=3 → NOTE
|
||
|
|
d.reset()
|
||
|
|
self.assertEqual(d.feed(a), ActionSignal.NONE) # count=1 after reset
|
||
|
|
|
||
|
|
def test_canonical_form_excludes_reasoning(self):
|
||
|
|
a1 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 1}
|
||
|
|
a2 = {"action": "run", "command": "ls", "reasoning": "B", "plan_step_id": 1}
|
||
|
|
self.assertEqual(canonical_form(a1), canonical_form(a2))
|
||
|
|
|
||
|
|
def test_canonical_form_includes_other_fields(self):
|
||
|
|
a1 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 1}
|
||
|
|
a2 = {"action": "run", "command": "ls", "reasoning": "A", "plan_step_id": 2}
|
||
|
|
self.assertNotEqual(canonical_form(a1), canonical_form(a2))
|
||
|
|
|
||
|
|
|
||
|
|
# ── StreamRepetitionDetector ───────────────────────────────────────
|
||
|
|
|
||
|
|
class StreamRepetitionTest(unittest.TestCase):
|
||
|
|
|
||
|
|
def test_nonrepeating_stream_never_fires(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
for i in range(200):
|
||
|
|
s.feed("content", f"unique{i} ")
|
||
|
|
self.assertEqual(s.level, Level.NONE)
|
||
|
|
self.assertEqual(s.streak, 0)
|
||
|
|
|
||
|
|
def test_l1_fires_at_streak_50(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_blocks(s, "hello ", n_blocks=51) # 51 blocks → streak=50
|
||
|
|
self.assertEqual(s.level, Level.L1)
|
||
|
|
|
||
|
|
def test_l2_fires_at_streak_500(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_blocks(s, "x", n_blocks=501) # 501 blocks → streak=500
|
||
|
|
self.assertEqual(s.level, Level.L2)
|
||
|
|
|
||
|
|
def test_l3_fires_at_streak_1500(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_blocks(s, "x", n_blocks=1501) # 1501 blocks → streak=1500
|
||
|
|
self.assertEqual(s.level, Level.L3)
|
||
|
|
|
||
|
|
def test_level_escalates_l1_to_l2_to_l3(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_blocks(s, "x", n_blocks=51)
|
||
|
|
self.assertEqual(s.level, Level.L1)
|
||
|
|
_feed_blocks(s, "x", n_blocks=450) # total streak=500
|
||
|
|
self.assertEqual(s.level, Level.L2)
|
||
|
|
_feed_blocks(s, "x", n_blocks=1000) # total streak=1500
|
||
|
|
self.assertEqual(s.level, Level.L3)
|
||
|
|
|
||
|
|
def test_different_block_resets_streak(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_blocks(s, "A", n_blocks=10) # streak=9
|
||
|
|
_feed_blocks(s, "B", n_blocks=1) # different block → streak=0
|
||
|
|
self.assertEqual(s.streak, 0)
|
||
|
|
self.assertEqual(s.level, Level.NONE)
|
||
|
|
|
||
|
|
def test_reasoning_and_content_combined(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
seg = "tok "
|
||
|
|
for _ in range(51):
|
||
|
|
for _ in range(50):
|
||
|
|
s.feed("reasoning", seg) # pure reasoning stream
|
||
|
|
self.assertEqual(s.level, Level.L1)
|
||
|
|
|
||
|
|
def test_mixed_kinds_combined_in_order(self):
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
seg = "z "
|
||
|
|
for _ in range(51):
|
||
|
|
for i in range(50):
|
||
|
|
kind = "reasoning" if i % 2 == 0 else "content"
|
||
|
|
s.feed(kind, seg)
|
||
|
|
self.assertEqual(s.level, Level.L1)
|
||
|
|
|
||
|
|
def test_period_37_blind_spot(self):
|
||
|
|
"""A 37-delta cycle evades 50-delta block detection because
|
||
|
|
gcd(37, 50)=1 — each block starts at a different cycle position.
|
||
|
|
This is an accepted blind spot, pinned here as a regression."""
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_cycle(s, cycle_len=37, n_cycles=200) # 7400 deltas, 148 blocks
|
||
|
|
self.assertEqual(s.streak, 0)
|
||
|
|
self.assertEqual(s.level, Level.NONE)
|
||
|
|
|
||
|
|
def test_period_25_detected(self):
|
||
|
|
"""A 25-delta cycle IS detected because 25 divides 50 — every
|
||
|
|
block is identical."""
|
||
|
|
s = StreamRepetitionDetector()
|
||
|
|
_feed_cycle(s, cycle_len=25, n_cycles=200) # 5000 deltas, 100 blocks
|
||
|
|
self.assertEqual(s.level, Level.L1) # streak=99 ≥ 50
|
||
|
|
|
||
|
|
|
||
|
|
# ── InsuranceCounter ───────────────────────────────────────────────
|
||
|
|
|
||
|
|
class InsuranceCounterTest(unittest.TestCase):
|
||
|
|
|
||
|
|
def test_nine_aborts_no_kill(self):
|
||
|
|
ic = InsuranceCounter()
|
||
|
|
for _ in range(9):
|
||
|
|
self.assertFalse(ic.register_abort())
|
||
|
|
self.assertEqual(ic.count, 9)
|
||
|
|
|
||
|
|
def test_tenth_abort_kills(self):
|
||
|
|
ic = InsuranceCounter()
|
||
|
|
for _ in range(9):
|
||
|
|
ic.register_abort()
|
||
|
|
self.assertTrue(ic.register_abort()) # 10th
|
||
|
|
|
||
|
|
def test_eleventh_still_kills(self):
|
||
|
|
ic = InsuranceCounter()
|
||
|
|
for _ in range(10):
|
||
|
|
ic.register_abort()
|
||
|
|
self.assertTrue(ic.register_abort()) # 11th, still >= 10
|
||
|
|
|
||
|
|
|
||
|
|
# ── ResetCap ───────────────────────────────────────────────────────
|
||
|
|
|
||
|
|
class ResetCapTest(unittest.TestCase):
|
||
|
|
|
||
|
|
def test_three_resets_ok(self):
|
||
|
|
rc = ResetCap()
|
||
|
|
for _ in range(3):
|
||
|
|
self.assertFalse(rc.register_reset())
|
||
|
|
self.assertEqual(rc.count, 3)
|
||
|
|
self.assertEqual(rc.remaining, 0)
|
||
|
|
|
||
|
|
def test_fourth_reset_exceeds_cap(self):
|
||
|
|
rc = ResetCap()
|
||
|
|
for _ in range(3):
|
||
|
|
rc.register_reset()
|
||
|
|
self.assertTrue(rc.register_reset()) # 4th → doom_loop
|
||
|
|
|
||
|
|
def test_remaining_decrements(self):
|
||
|
|
rc = ResetCap()
|
||
|
|
self.assertEqual(rc.remaining, 3)
|
||
|
|
rc.register_reset()
|
||
|
|
self.assertEqual(rc.remaining, 2)
|
||
|
|
rc.register_reset()
|
||
|
|
self.assertEqual(rc.remaining, 1)
|
||
|
|
rc.register_reset()
|
||
|
|
self.assertEqual(rc.remaining, 0)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|