rune/tests/test_builder_bridge.py

#!/usr/bin/env python3
"""tests/test_builder_bridge.py — MiMo builder bridge canaries (Option C).

The bridge is a thin adapter over Rune Core: zero model calls, bounded
capsule, same authority the operator consumes. These tests prove the
contract, not the cognition (the cognition is the core's, tested there).
"""

import json
import os
import sys
import unittest

PKG_ROOT = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, os.path.join(PKG_ROOT, "src"))

from rune import builder_bridge  # noqa: E402
from rune.builder_bridge import (  # noqa: E402
    BuilderCapsule,
    MAX_LESSON_LINES,
    MAX_MEMORY_LINES,
    preflight,
)


class BridgeContract(unittest.TestCase):
    def test_preflight_local_task_is_cheap_and_bounded(self):
        cap = preflight("Fix the telemetry summary log level constant in stage_telemetry.py")
        self.assertIn(cap.task_route, ("LOCAL_FIX", "STANDARD_CHANGE"))
        self.assertFalse(cap.floor)
        self.assertLessEqual(len(cap.memory_lines), MAX_MEMORY_LINES)
        self.assertLessEqual(len(cap.lesson_lines), MAX_LESSON_LINES)
        d = cap.to_dict()
        self.assertIn("generated_at", d)
        self.assertEqual(d["core_version"], "rune-core/operator-path")

    def test_preflight_research_task_carries_floor(self):
        cap = preflight("New subsystem: evaluate and adopt an external vector database dependency")
        self.assertEqual(cap.task_route, "RESEARCHED_CHANGE")
        self.assertTrue(cap.floor)
        self.assertTrue(any("floor" in c.lower() for c in cap.constraints))

    def test_lesson_lines_empty_until_pl1(self):
        # Handoff §9: raw SEED records are outcome telemetry, not lessons.
        # The bridge must not inject unvalidated process pressure.
        cap = preflight("Fix typo in CLI help text")
        self.assertEqual(cap.lesson_lines, [])

    def test_markdown_renders_without_empty_sections(self):
        cap = BuilderCapsule(task_route="LOCAL_FIX", unknowns=["verifiability"])
        md = cap.to_markdown()
        self.assertIn("LOCAL_FIX", md)
        self.assertIn("verifiability", md)
        self.assertNotIn("Relevant Rune memory", md)  # empty section omitted
        self.assertNotIn("Validated process lessons", md)

    def test_memory_selector_plumbs_through(self):
        class FakeSelector:
            def select_for_task(self, text):
                class Block:
                    content = "memory line one\nmemory line two"

                class Result:
                    blocks = [Block()]

                return Result()

        import rune.builder_bridge as bb

        saved = bb._SELECTOR
        bb._SELECTOR = FakeSelector()
        try:
            lines = bb._retrieve_memory("Fix typo")
            self.assertTrue(any("memory line" in m for m in lines))
        finally:
            bb._SELECTOR = saved

    def test_selector_failure_degrades_not_blocks(self):
        class BrokenSelector:
            def select_for_task(self, text):
                raise RuntimeError("boom")

        cap = preflight("Fix typo", memory_selector=BrokenSelector())
        self.assertEqual(cap.memory_lines, [])
        self.assertIn(cap.task_route, ("LOCAL_FIX", "STANDARD_CHANGE", "RESEARCHED_CHANGE"))


class ResearchRequiredLifecycle(unittest.TestCase):
    """research_required is a nonterminal disposition, NOT failure (§2)."""

    def test_state_machine_allows_lifecycle(self):
        import sys as _sys

        _sys.path.insert(0, os.path.join(PKG_ROOT, "src"))
        from rune.orchestration.conversation_state import TASK_STATUSES, TASK_TRANSITIONS, OPEN_TASK_STATUSES

        self.assertIn("research_required", TASK_STATUSES)
        self.assertIn("research_required", OPEN_TASK_STATUSES)
        # running -> research_required -> running (resume) and -> completed
        self.assertIn("research_required", TASK_TRANSITIONS["running"])
        self.assertIn("running", TASK_TRANSITIONS["research_required"])
        self.assertIn("completed", TASK_TRANSITIONS["research_required"])
        self.assertIn("failed", TASK_TRANSITIONS["research_required"])

    def test_experience_consumer_skips_research_dispatch(self):
        import tempfile
        import json as _json

        sys.path.insert(0, os.path.join(PKG_ROOT, "scripts"))
        from process_experience import record_operator_outcomes

        with tempfile.TemporaryDirectory() as tmp:
            state = {
                "tasks": [
                    {
                        "task_id": "t1",
                        "title": "research task",
                        "status": "research_required",
                        "outcome_ref": "research_dispatch",
                        "assigned_model": "m",
                        "result": "RESEARCH_REQUIRED",
                    },
                    {
                        "task_id": "t2",
                        "title": "real failure",
                        "status": "failed",
                        "outcome_ref": "",
                        "assigned_model": "m",
                    },
                ]
            }
            sp = os.path.join(tmp, "task_state.json")
            with open(sp, "w") as f:
                _json.dump(state, f)
            lp = os.path.join(tmp, "led.jsonl")
            recs = record_operator_outcomes(sp, lp)
            self.assertEqual(len(recs), 1)  # only the real failure
            self.assertEqual(recs[0]["result_type"], "FAILURE")
            self.assertNotIn("research", recs[0]["symptom"].lower())


class RouteRiskFlagsContract(unittest.TestCase):
    def test_detect_flags_vocabulary(self):
        from rune.route_risk_flags import RISK_FLAGS, UNCERTAINTY_FLAGS, detect_flags

        risk, unc = detect_flags("this adds a new dependency and changes memory architecture")
        self.assertIn("new_dependency", risk)
        self.assertIn("memory_architecture", risk)
        risk2, unc2 = detect_flags("competing designs with expensive reversal")
        self.assertIn("competing_designs", unc2)
        self.assertIn("expensive_reversal", unc2)
        risk3, unc3 = detect_flags("fix typo in help text")
        self.assertEqual(risk3, [])
        self.assertEqual(unc3, [])


if __name__ == "__main__":
    unittest.main()