mnemosyne/tests/test_phantom_memory_query.py

"""Tests for the memory_query phantom tool addition to phantom.py."""

from __future__ import annotations

import copy

from mnemosyne.phantom import (
    PHANTOM_TOOL_DEFINITIONS,
    PHANTOM_TOOL_NAMES,
    PhantomCall,
    _handle_phantom_call,
    inject_tools,
)


# ── memory_query in PHANTOM_TOOL_NAMES ───────────────────────


def test_memory_query_in_phantom_tool_names():
    """memory_query must be a recognized phantom tool name."""
    assert "memory_query" in PHANTOM_TOOL_NAMES


def test_phantom_tool_names_is_frozenset():
    """PHANTOM_TOOL_NAMES must remain a frozenset (immutable)."""
    assert isinstance(PHANTOM_TOOL_NAMES, frozenset)


# ── memory_query tool definition schema ──────────────────────


def _get_memory_query_def() -> dict:
    """Helper: find the memory_query definition from PHANTOM_TOOL_DEFINITIONS."""
    for defn in PHANTOM_TOOL_DEFINITIONS:
        if defn["name"] == "memory_query":
            return defn
    raise AssertionError("memory_query not found in PHANTOM_TOOL_DEFINITIONS")


def test_memory_query_definition_exists():
    """memory_query must have a definition in PHANTOM_TOOL_DEFINITIONS."""
    defn = _get_memory_query_def()
    assert defn["name"] == "memory_query"


def test_memory_query_has_description():
    """memory_query definition must have a non-empty description."""
    defn = _get_memory_query_def()
    assert isinstance(defn["description"], str)
    assert len(defn["description"]) > 20


def test_memory_query_schema_properties():
    """memory_query input_schema must have question, scope, max_tokens."""
    defn = _get_memory_query_def()
    schema = defn["input_schema"]
    assert schema["type"] == "object"
    props = schema["properties"]
    assert "question" in props
    assert props["question"]["type"] == "string"
    assert "scope" in props
    assert props["scope"]["type"] == "string"
    assert "max_tokens" in props
    assert props["max_tokens"]["type"] == "integer"


def test_memory_query_required_fields():
    """Only 'question' should be required for memory_query."""
    defn = _get_memory_query_def()
    required = defn["input_schema"]["required"]
    assert required == ["question"]


# ── inject_tools adds memory_query ───────────────────────────


def test_inject_tools_adds_memory_query():
    """inject_tools should add memory_query to the tools array."""
    body = {"tools": [], "messages": []}
    inject_tools(body)
    tool_names = {t["name"] for t in body["tools"]}
    assert "memory_query" in tool_names


def test_inject_tools_does_not_duplicate_memory_query():
    """inject_tools should not duplicate memory_query if already present."""
    existing_def = {
        "name": "memory_query",
        "description": "already here",
        "input_schema": {"type": "object", "properties": {}},
    }
    body = {"tools": [existing_def], "messages": []}
    inject_tools(body)
    mq_count = sum(1 for t in body["tools"] if t["name"] == "memory_query")
    assert mq_count == 1


def test_inject_tools_returns_memory_query_as_observe_only_when_framework_provides():
    """If framework already has memory_query, it should be in observe_only set."""
    existing_def = {
        "name": "memory_query",
        "description": "framework provided",
        "input_schema": {"type": "object", "properties": {}},
    }
    body = {"tools": [existing_def], "messages": []}
    observe_only = inject_tools(body)
    assert "memory_query" in observe_only


# ── _handle_phantom_call returns pending placeholder ─────────


def test_handle_phantom_call_memory_query_returns_pending():
    """memory_query handler should return a pending placeholder string."""
    call = PhantomCall(
        name="memory_query",
        tool_use_id="toolu_test123",
        input={"question": "What auth library is used?"},
    )
    result = _handle_phantom_call(call, page_store=None)
    assert "[memory_query:pending]" in result
    assert "What auth library is used?" in result


def test_handle_phantom_call_memory_query_includes_scope():
    """memory_query handler should include scope in the placeholder."""
    call = PhantomCall(
        name="memory_query",
        tool_use_id="toolu_test456",
        input={
            "question": "What port does the server run on?",
            "scope": "config files",
        },
    )
    result = _handle_phantom_call(call, page_store=None)
    assert "[memory_query:pending]" in result
    assert "config files" in result


def test_handle_phantom_call_memory_query_includes_max_tokens():
    """memory_query handler should include max_tokens in the placeholder."""
    call = PhantomCall(
        name="memory_query",
        tool_use_id="toolu_test789",
        input={
            "question": "What is the DB schema?",
            "max_tokens": 100,
        },
    )
    result = _handle_phantom_call(call, page_store=None)
    assert "[memory_query:pending]" in result
    assert "100" in result


def test_handle_phantom_call_memory_query_defaults():
    """memory_query handler should use defaults for optional fields."""
    call = PhantomCall(
        name="memory_query",
        tool_use_id="toolu_defaults",
        input={"question": "test question"},
    )
    result = _handle_phantom_call(call, page_store=None)
    assert "[memory_query:pending]" in result
    # Default scope is None, default max_tokens is 200
    assert "None" in result
    assert "200" in result


# ── Existing tools still work ────────────────────────────────


def test_existing_phantom_tools_still_present():
    """All original phantom tool names must still be in PHANTOM_TOOL_NAMES."""
    for name in ("yuyay", "recall", "memory_fault", "qunqay", "tiqsiy"):
        assert name in PHANTOM_TOOL_NAMES, f"{name} missing from PHANTOM_TOOL_NAMES"


def test_existing_tool_definitions_count():
    """PHANTOM_TOOL_DEFINITIONS should now have 4 tools (3 original + memory_query)."""
    assert len(PHANTOM_TOOL_DEFINITIONS) == 4