"""Tests for the idle / learning / readahead mode. No kernel, no encoder, no network: a fake assistant supplies exactly the surface ``AutonomousMode`` touches (``d``, ``threshold``, ``intent_vectors``, ``classify``, ``chat``, ``kernel``), so what is under test is the autonomy logic and not the machine underneath it. """ import math import unittest from hdc.autonomy import ACT_FLOOR, SAFE_COMMANDS, AutonomousMode from hdc.hypervec import hvc #: One lap of the behaviour the mode has to learn: a four-command cycle. CYCLE = ("status", "registers", "disassemble", "help") #: Utterance -> intent, deterministic on purpose. UTTERANCES = { "show the statistics": "status", "show the registers": "registers", "show the instruction set": "disassemble", "what can you do": "help", "run the program": "run", "boot the machine": "boot", } class FakeKernel: """Records what was executed, so "did it act?" is a list, not a flag.""" def __init__(self): self.calls = [] def execute(self, command): self.calls.append(command) return "output of " + command class FakeAssistant: """The slice of ``Assistant`` that ``AutonomousMode`` depends on.""" def __init__(self, d=2000): self.d = d self.threshold = 0.42 self.kernel = FakeKernel() # every intent the mode may predict, including the unsafe ones it must # refuse to execute self.intent_vectors = { name: hvc("intent/" + name, d) for name in ("status", "registers", "disassemble", "help", "run", "boot") } self.history = [] def vector(self, intent): return self.intent_vectors[intent] def classify(self, text): intent = UTTERANCES.get(text, "") return intent, (0.9 if intent else 0.0) def chat(self, text, execute=None): intent, confidence = self.classify(text) command = {"registers": "regs", "disassemble": "disasm", "status": "stat", "help": "help", "run": "run", "boot": "boot"}.get(intent, "") output = "" if command: output = (execute or self.kernel.execute)(command) record = {"input": text, "intent": intent, "confidence": confidence, "executed": command, "output": output} self.history.append(record) return record def mode_with(**kwargs): return AutonomousMode(FakeAssistant(d=kwargs.pop("d", 2000)), **kwargs) def lap(mode, learn=True, cycles=1): """Observe whole laps of CYCLE, closing the loop back to 'status'.""" for _ in range(cycles): for name in CYCLE + ("status",): mode.observe(name, learn=learn) return mode def prediction(intent="registers", command="regs", confidence=0.5, depth=1, cached=False): return {"intent": intent, "command": command, "confidence": confidence, "path": confidence, "depth": depth, "cached": cached} class TestObserveAndLearn(unittest.TestCase): def test_observe_records_the_intent_and_returns_a_verdict(self): mode = mode_with() verdict = mode.observe("status") self.assertEqual(mode.sequence, ["status"]) self.assertEqual(verdict["actual"], "status") self.assertIsNone(verdict["predicted"]) # nothing to grade yet self.assertFalse(verdict["hit"]) self.assertEqual(mode.graded, 0) # an ungraded first turn is not a miss def test_cycle_is_predicted_after_one_lap(self): mode = lap(mode_with()) self.assertEqual(mode.readahead[0]["intent"], "registers") def test_learning_is_what_makes_the_prediction_exist(self): learned = lap(mode_with()) unlearned = lap(mode_with(), learn=False) self.assertGreater(learned.model.transitions, 0) self.assertEqual(unlearned.model.transitions, 0) self.assertEqual(learned.readahead[0]["intent"], "registers") self.assertEqual(unlearned.readahead, []) def test_hits_and_graded_track_the_predictions(self): mode = lap(mode_with()) # the exact key 'status' now has a learned successor, so this turn is # predicted before the answer is known verdict = mode.observe("registers") self.assertTrue(verdict["hit"]) self.assertEqual(verdict["predicted"], "registers") self.assertGreaterEqual(mode.graded, 1) self.assertGreaterEqual(mode.hits, 1) self.assertLessEqual(mode.hits, mode.graded) self.assertEqual(mode.stats()["accuracy"], round(mode.hits / mode.graded, 3)) class TestReadahead(unittest.TestCase): def test_chain_follows_the_learned_cycle(self): mode = lap(mode_with(), cycles=2) lookahead = mode.predict(horizon=3) self.assertEqual([item["intent"] for item in lookahead], ["registers", "disassemble", "help"]) self.assertEqual([item["depth"] for item in lookahead], [1, 2, 3]) self.assertEqual([item["command"] for item in lookahead], ["regs", "disasm", "help"]) for item in lookahead: for key in ("intent", "command", "confidence", "path", "depth", "cached"): self.assertIn(key, item) def test_path_is_the_running_confidence_product(self): mode = lap(mode_with(), cycles=2) lookahead = mode.predict(horizon=3) expected = 1.0 for item in lookahead: expected *= item["confidence"] # `path` multiplies the unrounded confidences, so compare at a # precision that cannot hinge on the last printed digit self.assertAlmostEqual(item["path"], expected, places=2) paths = [item["path"] for item in lookahead] self.assertEqual(paths, sorted(paths, reverse=True)) def test_empty_lookahead_before_anything_is_observed(self): self.assertEqual(mode_with().predict(), []) class TestAutonomy(unittest.TestCase): def test_safe_prediction_is_executed_once_and_cached(self): mode = mode_with(enabled=True) mode.readahead = [prediction()] self.assertEqual(len(mode._act()), 1) self.assertEqual(mode.assistant.kernel.calls, ["regs"]) self.assertEqual(mode.stats()["cached_commands"], ["regs"]) # a second idle cycle must not run the same command again mode.readahead = [prediction()] self.assertEqual(mode._act(), []) self.assertEqual(mode.assistant.kernel.calls, ["regs"]) self.assertEqual(mode.stats()["autonomous_actions"], 1) def test_unsafe_commands_are_never_executed(self): mode = mode_with(enabled=True) mode.readahead = [prediction("run", "run", 0.9), prediction("boot", "boot", 0.9)] self.assertEqual(mode._act(), []) self.assertEqual(mode.assistant.kernel.calls, []) for command in ("run", "boot", "asm", "clear", "step"): self.assertNotIn(command, SAFE_COMMANDS) def test_budget_caps_the_actions(self): mode = mode_with(enabled=True, budget=2) mode.readahead = [ prediction("help", "help", 0.5), prediction("status", "stat", 0.5), prediction("registers", "regs", 0.5), prediction("disassemble", "disasm", 0.5), ] acted = mode._act() self.assertEqual(len(acted), 2) self.assertEqual(len(mode.assistant.kernel.calls), 2) def test_prediction_below_the_floor_is_skipped(self): mode = mode_with(enabled=True) mode.readahead = [prediction(confidence=ACT_FLOOR - 0.05)] self.assertEqual(mode._act(), []) self.assertEqual(mode.assistant.kernel.calls, []) self.assertEqual(mode.at_floor_skips, 1) def test_idle_does_nothing_when_disabled(self): mode = lap(mode_with(enabled=False), cycles=2) entry = mode.tick() self.assertEqual(entry["acted"], []) self.assertEqual(mode.assistant.kernel.calls, []) self.assertEqual(mode.ticks, 1) class TestChatWithReadahead(unittest.TestCase): def test_warm_cache_serves_the_answer_and_records_the_saving(self): mode = mode_with() mode.cache["regs"] = {"output": "cached regs", "ms": 12.5, "confidence": 0.6} record = mode.chat("show the registers") self.assertTrue(record["cached"]) self.assertEqual(record["output"], "cached regs") self.assertEqual(record["saved_ms"], 12.5) self.assertEqual(mode.served, 1) self.assertEqual(mode.ms_saved, 12.5) self.assertEqual(mode.stats()["served_from_readahead"], 1) # the turn is still observed, so it is still graded and learned from self.assertEqual(mode.sequence, ["registers"]) self.assertIsNotNone(record["graded"]) class TestReport(unittest.TestCase): def test_report_quotes_accuracy_and_chance(self): mode = lap(mode_with(), cycles=2) text = mode.report() self.assertIn("next-intent accuracy", text) self.assertIn("chance", text) self.assertIn("autonomous actions", text) def test_stats_expose_the_floor_and_the_chance_level(self): mode = mode_with() stats = mode.stats() expected = round(1.0 / math.sqrt(len(mode.intents)), 3) self.assertEqual(stats["chance"], expected) self.assertEqual(stats["act_floor"], ACT_FLOOR) self.assertEqual(stats["budget"], mode.budget) self.assertIn("readahead", stats) if __name__ == "__main__": unittest.main()