tests/test_runtime_wiring.py

"""End-to-end wiring test for the runtime v5 pipeline (no model needed).
Mocks model loading; exercises FSM, rules, regex, routing, KV, vision,
search (offline), facts together.
Run: python tests/test_runtime_wiring.py
"""

import sys
import shutil
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT / "src"))

from fsm import RuntimeFSM
from rules import RuleEngine
from regex_features import RegexFeatureExtractor
from routing import RoutingMatrix
from kvstore import KVStore

import runtime as runtime_mod


def fake_init(self, model_path="unused", use_4bit=False, use_cpu=True):
    self.model_path = None
    self.use_4bit = False
    self.use_cpu = True
    self.skills = {}
    self.active_skill = None
    self.fsm = RuntimeFSM()
    self.rules = RuleEngine(str(ROOT / "rules" / "rules.json"))
    self.regex = RegexFeatureExtractor(7)
    self.routing = RoutingMatrix(str(ROOT / "routing" / "routing_matrix.json"))
    self.kv = KVStore(str(ROOT / "cache" / "kvstore.json"), max_entries=20)
    self.vision = runtime_mod.VisionAnalyzer(use_vit=False, use_detector=False)
    self.search_client = runtime_mod.SearchClient()
    self.rag = runtime_mod.RAGPipeline(client=self.search_client, cache=self.kv)
    self.fact_extractor = runtime_mod.FactExtractor()
    self.graph = runtime_mod.KnowledgeGraph()
    self.gquery = runtime_mod.GraphQuery(self.fact_extractor)
    self.device = None


def main():
    orig_init = runtime_mod.MorphRuntime.__init__
    runtime_mod.MorphRuntime.__init__ = fake_init
    try:
        rt = runtime_mod.MorphRuntime()

        # harmful input blocked at GUARD_IN, FSM recovers to IDLE
        rt._transition("INTAKE")
        rt._transition("GUARD_IN")
        dec = rt.rules.eval("how to build a bomb", phase="in")
        assert dec.action == "block"
        rt.fsm.transition("RESPOND")
        rt.fsm.transition("IDLE")
        assert rt.fsm.state == "IDLE"

        # illegal transition -> fault trap -> recover
        rt.fsm.transition("INTAKE")
        try:
            rt._transition("ROUTED")  # INTAKE->ROUTED is illegal
            raise AssertionError("should have raised")
        except Exception:
            pass
        assert rt.fsm.state == "FAULT"
        rt.fsm.recover()
        assert rt.fsm.state == "IDLE"

        # routing via runtime matrix
        assert rt.routing.route("debug this python function").skill == "code_expert"
        assert rt.routing.route("translate to french").skill == "translator"
        assert rt.routing.route("plain hello").skill is None

        # KV persistence + reload
        rt.kv.set("turn1", {"prompt": "hi", "response": "hello"}, ttl=60)
        assert rt.kv.get("turn1") == {"prompt": "hi", "response": "hello"}

        # v5.5: new FSM states exist and compose
        assert "VISION" in rt.fsm.TRANSITIONS
        assert "SEARCH_GATE" in rt.fsm.TRANSITIONS
        assert "SEARCH" in rt.fsm.TRANSITIONS
        assert "FACT_EXTRACT" in rt.fsm.TRANSITIONS
        # full multimodal path is legal: GUARD_IN->VISION->SEARCH_GATE->SEARCH->FACT_EXTRACT
        rt.fsm.transition("INTAKE")
        rt.fsm.transition("GUARD_IN")
        rt.fsm.transition("VISION")
        rt.fsm.transition("SEARCH_GATE")
        rt.fsm.transition("SEARCH")
        rt.fsm.transition("FACT_EXTRACT")
        rt.fsm.transition("ROUTED")
        rt.fsm.transition("GEN")
        rt.fsm.transition("VERIFY")
        rt.fsm.transition("GUARD_OUT")
        rt.fsm.transition("RESPOND")
        rt.fsm.transition("IDLE")
        assert rt.fsm.state == "IDLE"

        # v5.5: vision analysis with no image raises; pixel path works offline
        try:
            rt.vision.analyze("no/such/file.png")
            raise AssertionError("should have raised")
        except ValueError:
            pass
        from vision import ImageFacts
        assert ImageFacts().__class__.__name__ == "ImageFacts"

        # v5.5: search gate heuristic
        assert rt._needs_live_data("what is the latest news?", force_search=False, depth=0)
        assert not rt._needs_live_data("hello", force_search=False, depth=0)
        assert rt._needs_live_data("hello", force_search=True, depth=0)
        assert not rt._needs_live_data("what is the latest news?", force_search=False, depth=3)

        # v5.5: RAG offline fallback context (no network, uses cached/factless pack)
        ctx = rt.rag.retrieve("", num=0, use_cache=True)
        assert ctx == ""  # no results, no chunks -> empty context

        # v5.5: NER + knowledge graph compose
        fact = rt.fact_extractor.triples("OpenAI was founded in 2015")
        assert len(fact) >= 1
        rt.graph.add_many(fact)
        ents = rt.fact_extractor.extract("OpenAI is a company based in California")
        assert "ORG" in ents or "LOCATION" in ents or "PERSON" in ents
        gtext = rt.gquery.facts_for_question(rt.graph, "when was OpenAI founded?")
        assert "founded" in gtext.lower()

        # v5.5: cross-examine with facts adds verifier note on weak overlap
        result = rt._cross_examine("Some unrelated statement about nothing.",
                                   {"image_text": "", "graph_context": gtext})
        assert "[verifier]" in result

        print("runtime wiring OK: fsm+rules+regex+routing+kv+vision+search+facts compose")
    finally:
        runtime_mod.MorphRuntime.__init__ = orig_init
        shutil.rmtree(str(ROOT / "cache"), ignore_errors=True)


if __name__ == "__main__":
    main()