rune/tests/test_runtime_architecture_characterization.py

"""RED characterization tests for the four unpinned runtime/TUI gaps.

These tests deliberately describe the approved architecture rather than the
candidate behavior.  Task 1 leaves them RED so later tasks can close each gap.
"""

from __future__ import annotations

import ast
import sys
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from rune import rune_operator  # noqa: E402
from rune.tool_protocol import OperatorResponse  # noqa: E402


def _function_node(path: Path, class_name: str, function_name: str) -> ast.FunctionDef:
    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
    for node in tree.body:
        if isinstance(node, ast.ClassDef) and node.name == class_name:
            for member in node.body:
                if isinstance(member, ast.FunctionDef) and member.name == function_name:
                    return member
    raise AssertionError(f"{class_name}.{function_name} was not found in {path}")


class RuntimeArchitectureCharacterizationTests(unittest.TestCase):
    def test_second_turn_includes_prior_user_and_assistant_context(self):
        with tempfile.TemporaryDirectory() as tempdir:
            operator = rune_operator.RuneOperator(str(Path(tempdir) / "operator.json"))
            runtime = Mock()
            runtime.run.side_effect = [
                OperatorResponse(
                    success=True, user_text="First answer", model="qwen3:8b"
                ),
                OperatorResponse(
                    success=True, user_text="Second answer", model="qwen3:8b"
                ),
            ]
            profile = SimpleNamespace(safety_tier=2)

            with (
                patch.object(rune_operator, "LocalToolRuntime", return_value=runtime),
                patch.object(rune_operator, "OllamaChatTransport", return_value=Mock()),
                patch.object(
                    rune_operator.model_profiles,
                    "get_or_create_profile",
                    return_value=profile,
                ),
                patch.object(
                    rune_operator.model_profiles,
                    "get_prompt_context",
                    return_value="",
                ),
                patch.object(rune_operator.model_profiles, "record_interaction"),
                patch(
                    "rune.attention_capsule.create_capsule",
                    return_value=SimpleNamespace(token_count=1),
                ),
                patch("rune.memory_store.connect", return_value=Mock()),
                patch("rune.memory_store.cached_recall", return_value=[]),
                patch.object(operator, "_log_token_usage"),
            ):
                operator.send_task("First question", "qwen3:8b")
                operator.send_task("Second question", "qwen3:8b")

            second_messages = runtime.run.call_args_list[1].args[0]
            role_content = [
                (message["role"], message["content"])
                for message in second_messages
                if message["role"] != "system"
            ]
            self.assertIn(
                ("user", "First question"),
                role_content,
                "the second provider turn omits the prior user message",
            )
            self.assertIn(
                ("assistant", "First answer"),
                role_content,
                "the second provider turn omits the prior assistant message",
            )

    def test_tui_send_path_projects_typed_tool_events(self):
        send = _function_node(
            ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model"
        )
        string_literals = {
            node.value
            for node in ast.walk(send)
            if isinstance(node, ast.Constant) and isinstance(node.value, str)
        }
        constructed_types = {
            node.func.id
            for node in ast.walk(send)
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Name)
        }

        self.assertIn(
            "tool_events",
            string_literals,
            "RuneApp drops the typed tool event stream instead of projecting it",
        )
        self.assertIn(
            "ToolCallWidget",
            constructed_types,
            "typed tool events are not rendered as distinct TUI objects",
        )

    def test_tui_updates_stable_tool_widgets_instead_of_appending_duplicates(self):
        source = (ROOT / "src" / "rune" / "rune_tui.py").read_text()
        self.assertIn("self._tool_widgets", source)
        self.assertIn("tool_widget.apply(tool_event)", source)

    def test_tui_send_path_does_not_auto_confirm_pending_mutation(self):
        send = _function_node(
            ROOT / "src" / "rune" / "rune_tui.py", "RuneApp", "_send_to_model"
        )
        calls = [
            node.func.attr
            for node in ast.walk(send)
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
        ]

        self.assertNotIn(
            "confirm_write",
            calls,
            "rendering a pending approval currently invokes mutation authority",
        )

    def test_generic_bubble_does_not_interpolate_protocol_markup(self):
        render = _function_node(
            ROOT / "src" / "rune" / "chat_bubble.py", "ChatBubble", "render"
        )
        direct_content_interpolations = [
            node
            for node in ast.walk(render)
            if isinstance(node, ast.FormattedValue)
            and isinstance(node.value, ast.Attribute)
            and isinstance(node.value.value, ast.Name)
            and node.value.value.id == "self"
            and node.value.attr == "content"
        ]

        self.assertEqual(
            direct_content_interpolations,
            [],
            "generic Rich rendering directly interpolates protocol-like/untrusted content",
        )


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