NexusOS/tests/test_slash_commands.py

"""synapse/slash_commands.py (the /tool_name(arg=val) parser) and its wiring
into chat_stream_endpoint (direct dispatch, no model call, no approval
round-trip) plus the ten curry_* tools it can now reach.
"""
import json

import pytest
from fastapi.testclient import TestClient

from synapse.slash_commands import SlashCommand, SlashCommandError, parse_slash_command
from synapse.main import app
from synapse import tools


# ---------------------------------------------------------------------------
# Parser
# ---------------------------------------------------------------------------

def test_parses_keyword_arguments_as_python_literals():
    result = parse_slash_command('/curry_call_function(name="x", version=1, args={"a": 1})')
    assert result == SlashCommand(
        tool="curry_call_function",
        args={"name": "x", "version": 1, "args": {"a": 1}},
    )


def test_parses_no_arguments():
    assert parse_slash_command("/curry_list_functions()") == SlashCommand(tool="curry_list_functions", args={})


def test_non_slash_message_returns_none():
    assert parse_slash_command("just chatting, not a command") is None


def test_slash_without_parens_returns_none():
    # The TUI's own local commands (/model foo, /new) use this shape — must
    # never be mistaken for a tool call.
    assert parse_slash_command("/model gpt") is None


def test_slash_embedded_in_prose_returns_none():
    assert parse_slash_command('hey /curry_call_function(name="x", version=1) run this') is None


def test_name_or_call_as_argument_value_is_rejected():
    # ast.literal_eval only accepts literals — a bare name or a call is a
    # parse failure, not a value, so nothing here is ever evaluated.
    result = parse_slash_command("/curry_call_function(x=some_name)")
    assert isinstance(result, SlashCommandError)
    result2 = parse_slash_command('/curry_call_function(x=__import__("os"))')
    assert isinstance(result2, SlashCommandError)


def test_positional_arguments_are_rejected():
    result = parse_slash_command("/curry_call_function(1, 2)")
    assert isinstance(result, SlashCommandError)


def test_double_star_unpacking_is_rejected():
    result = parse_slash_command('/curry_call_function(**{"a": 1})')
    assert isinstance(result, SlashCommandError)


def test_malformed_syntax_is_rejected():
    result = parse_slash_command("/curry_call_function(name=)")
    assert isinstance(result, SlashCommandError)


# ---------------------------------------------------------------------------
# Curry tool registration
# ---------------------------------------------------------------------------

_CURRY_ACTION_TOOLS = {
    "curry_declare_constant", "curry_retire_constant",
    "curry_declare_function", "curry_retire_function", "curry_call_function",
}
_CURRY_READ_TOOLS = {
    "curry_get_constant", "curry_get_constant_latest", "curry_list_constants",
    "curry_get_function", "curry_list_functions",
}


def test_all_curry_tools_registered():
    for name in _CURRY_ACTION_TOOLS | _CURRY_READ_TOOLS:
        assert name in tools.REGISTRY


def test_curry_write_and_execute_tools_are_gated_actions():
    for name in _CURRY_ACTION_TOOLS:
        assert tools.is_action(name), name
        assert name in tools.ALWAYS_ASK_ACTION_TOOLS, name


def test_curry_read_tools_are_not_actions():
    for name in _CURRY_READ_TOOLS:
        assert not tools.is_action(name), name


# ---------------------------------------------------------------------------
# End-to-end HTTP: direct dispatch, no model call, no approval round-trip
# ---------------------------------------------------------------------------

@pytest.fixture
def client():
    return TestClient(app)


def _sse_events(body: str) -> list[tuple[str, str]]:
    events = []
    event_type = "message"
    for block in body.split("\n\n"):
        for line in block.splitlines():
            if line.startswith("event: "):
                event_type = line[len("event: "):].strip()
            elif line.startswith("data: "):
                events.append((event_type, line[len("data: "):]))
                event_type = "message"
    return events


def test_slash_command_dispatches_without_model_call(client, monkeypatch):
    from synapse import chat as chatmod

    async def _boom(*a, **k):
        raise AssertionError("the model must not be called for a slash-command")
    monkeypatch.setattr(chatmod, "stream_chat_response", _boom)

    resp = client.post("/chat/stream", json={
        "message": '/curry_list_functions()',
        "conversation_id": "test-slash-http-1",
    })
    events = _sse_events(resp.text)
    assert ("status", json.dumps({"tool": "curry_list_functions"})) in events
    assert any(t == "done" for t, _ in events)


def test_slash_command_skips_approval_round_trip(client, monkeypatch):
    async def _fake_dispatch(name, args):
        return json.dumps({"ok": True, "result": "did it"})
    monkeypatch.setattr(tools, "dispatch", _fake_dispatch)

    resp = client.post("/chat/stream", json={
        "message": '/curry_call_function(name="x", version=1, args={})',
        "conversation_id": "test-slash-http-2",
    })
    events = _sse_events(resp.text)
    assert not any(t == "tool_request" for t, _ in events)
    assert any(t == "done" for t, _ in events)


def test_slash_command_uses_fence_from_result_when_present(client, monkeypatch):
    async def _fake_dispatch(name, args):
        return json.dumps({"ok": True, "fence": "```nexus-curry\n{\"kind\": \"x\"}\n```"})
    monkeypatch.setattr(tools, "dispatch", _fake_dispatch)

    resp = client.post("/chat/stream", json={
        "message": '/curry_call_function(name="x", version=1, args={})',
        "conversation_id": "test-slash-http-3",
    })
    events = _sse_events(resp.text)
    content = [d for t, d in events if t == "message"]
    assert content and "nexus-curry" in content[0]


def test_slash_command_unknown_tool_yields_error_not_a_chat_reply(client):
    resp = client.post("/chat/stream", json={
        "message": "/not_a_real_tool(a=1)",
        "conversation_id": "test-slash-http-4",
    })
    events = _sse_events(resp.text)
    assert any(t == "error" for t, _ in events)
    assert not any(t == "status" for t, _ in events)


def test_slash_command_malformed_yields_error(client):
    resp = client.post("/chat/stream", json={
        "message": "/curry_call_function(x=some_name)",
        "conversation_id": "test-slash-http-5",
    })
    events = _sse_events(resp.text)
    assert any(t == "error" for t, _ in events)


def test_message_with_leading_slash_but_not_command_shaped_goes_to_chat(client, monkeypatch):
    # e.g. "/model gpt" or plain prose starting with "/" - must still reach
    # the normal model path, not be swallowed as a broken slash-command.
    called = {}

    async def _fake_stream(*a, **k):
        called["hit"] = True
        return
        yield  # pragma: no cover - make this an async generator

    # main.py did `from .chat import stream_chat_response`, a separate name
    # binding from chat.stream_chat_response - patch the one main.py actually
    # calls.
    from synapse import main as mainmod
    monkeypatch.setattr(mainmod, "stream_chat_response", _fake_stream)

    client.post("/chat/stream", json={
        "message": "/model gpt",
        "conversation_id": "test-slash-http-6",
    })
    assert called.get("hit") is True