HDCSTORAGE: hyperdimensional computer, webOS and agentic MCP layer (stdlib-only Python)

ad474c9 verified 5 days ago

9.53 kB

"""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()

	"""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()