#!/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()