beaver-agent/tests/test_vibegram.py

from urllib.parse import parse_qsl, urlsplit

import pytest
from beaver_gateway.mcp.types import McpServer
from beaver_gateway.mcp.wrap import build_python_tool_server
from fastmcp import Client
from fastmcp.exceptions import ToolError

from beaver_agent.hands.vibegram import PAGE, PAGES, Vibegram, _cards, _event, _work


def test_events_render_and_unknown_kinds_are_skipped():
    msg = _event(
        {
            "id": 61,
            "nick": "claude-neighbour",
            "kind": "message",
            "payload": {"body": "@beaver-test привет"},
            "createdAt": "2026-08-30T10:00:00.000Z",
        }
    )
    assert (
        msg is not None
        and msg.line() == "[08-30 10:00] claude-neighbour: @beaver-test привет"
    )
    assert _event({"id": 1, "kind": "tree", "payload": {}}) is None


def test_mentioned_and_cards():
    v = Vibegram(hub="http://127.0.0.1:9", token="t", nick="beaver-test")
    ev = _event(
        {
            "id": 2,
            "nick": "x",
            "kind": "message",
            "payload": {"body": "эй @Beaver-Test"},
        }
    )
    assert ev is not None and v.mentioned([ev])
    text = _cards(
        [
            {
                "nick": "beaver-test",
                "status": "online",
                "skills": [],
                "focus": {"holding": ["a.md"]},
            }
        ],
        "beaver-test",
    )
    assert "(это ты)" in text and "держит: a.md" in text
    assert "плана в комнате пока нет" in _work(
        {"free": [], "taken": [], "busy": [], "cards": [], "planRevision": 0}, "me"
    )


async def test_tool_schema_lists_actions_and_needs_text_for_send():
    tool = Vibegram(hub="http://127.0.0.1:9", token="t", nick="beaver-test")
    server = build_python_tool_server(
        McpServer.python_tool(name="vibegram", tools=[tool.vibegram])
    )
    async with Client(server) as client:
        (spec,) = await client.list_tools()
        assert {"read", "send", "who", "claim", "release"} <= set(
            spec.inputSchema["properties"]["action"]["enum"]
        )
        with pytest.raises(ToolError, match="text"):
            await client.call_tool("vibegram", {"action": "send"})


def _raw(eid, kind, body=None):
    return {
        "id": eid,
        "roomId": "r",
        "nick": "claude-kai",
        "kind": kind,
        "payload": {} if body is None else {"body": body},
        "createdAt": "2026-09-23T05:00:02.000Z",
    }


def _hub(monkeypatch, events):
    calls = []

    async def call(self, method, path, body):
        calls.append(path)
        if path == "/api/room":
            return {"room": {"id": "r", "name": "kai-rem-watch", "viewToken": "v 1"}}
        query = dict(parse_qsl(urlsplit(path).query))
        assert query["view"] == "v 1"  # ключ комнаты уезжает экранированным
        after = int(query.get("after", 0))
        return {"events": [e for e in events if e["id"] > after][:PAGE]}

    monkeypatch.setattr(Vibegram, "_call", call)
    return calls, Vibegram(hub="http://hub", token="t", nick="beaver-test")


async def test_window_pages_forward_to_the_end_and_drops_joins(monkeypatch):
    # 520 событий комнаты - три страницы по 200; join/leave в окно не идут
    events = [
        _raw(i, "message" if i % 4 else "agent_leave", f"строка {i}")
        for i in range(1, 521)
    ]
    calls, room = _hub(monkeypatch, events)

    window = await room.window(25)

    assert window.full
    assert len(window.events) == 25
    assert window.events[-1].text.endswith("строка 519")
    assert all("вошёл" not in e.text and "вышел" not in e.text for e in window.events)
    assert calls.count("/api/room") == 1
    assert sum(p.startswith("/api/state") for p in calls) == 3
    assert "view=v%201" in calls[1]
    assert not any("pending" in p for p in calls)  # курсор сторожу не окно


async def test_window_that_ran_out_of_pages_says_it_is_not_full(monkeypatch):
    events = [_raw(i, "message", f"строка {i}") for i in range(1, 521)]
    _, room = _hub(monkeypatch, events)

    assert not (await room.window(25, pages=2)).full
    assert (await room.window(25, pages=3)).full


async def test_window_of_an_empty_room_is_full_and_empty(monkeypatch):
    _, room = _hub(monkeypatch, [])
    window = await room.window(25)
    assert window.full and window.events == []


async def test_read_since_walks_history_without_touching_the_cursor(monkeypatch):
    events = [
        _raw(i, "message" if i % 4 else "agent_join", f"строка {i}")
        for i in range(1, 12)
    ]
    calls, room = _hub(monkeypatch, events)

    out = await room.vibegram("read", since=8)

    assert out.splitlines() == [
        "#9 [09-23 05:00] claude-kai: строка 9",
        "#10 [09-23 05:00] claude-kai: строка 10",
        "#11 [09-23 05:00] claude-kai: строка 11",
    ]
    assert not any("pending" in p for p in calls)


async def test_read_last_shows_only_the_tail(monkeypatch):
    events = [_raw(i, "message", f"строка {i}") for i in range(1, 12)]
    _, room = _hub(monkeypatch, events)

    out = await room.vibegram("read", last=2)

    assert out.splitlines() == [
        "#10 [09-23 05:00] claude-kai: строка 10",
        "#11 [09-23 05:00] claude-kai: строка 11",
    ]
    with pytest.raises(ValueError, match="last"):
        await room.vibegram("read", last=0)


async def test_read_since_on_a_silent_room_says_history_is_read_out(monkeypatch):
    _, room = _hub(monkeypatch, [_raw(3, "message", "строка 3")])

    assert await room.vibegram("read", since=3) == (
        "после #3 сообщений нет (история дочитана)"
    )


async def test_read_since_admits_an_unfinished_window(monkeypatch):
    events = [_raw(i, "message", f"строка {i}") for i in range(1, PAGES * PAGE + 2)]
    _, room = _hub(monkeypatch, events)

    out = await room.vibegram("read", since=0)

    assert out.splitlines()[0] == "окно неполное: до конца истории не дочитали"


async def test_plain_read_keeps_the_cursor_and_numbers_the_lines(monkeypatch):
    seen = []

    async def call(self, method, path, body):
        seen.append(path)
        return {"events": [_raw(7, "message", "привет")]}

    monkeypatch.setattr(Vibegram, "_call", call)
    room = Vibegram(hub="http://hub", token="t", nick="beaver-test")

    assert await room.vibegram("read") == "#7 [09-23 05:00] claude-kai: привет"
    assert seen == ["/api/pending?limit=50"]


async def test_tool_schema_exposes_since_and_last():
    tool = Vibegram(hub="http://127.0.0.1:9", token="t", nick="beaver-test")
    server = build_python_tool_server(
        McpServer.python_tool(name="vibegram", tools=[tool.vibegram])
    )
    async with Client(server) as client:
        (spec,) = await client.list_tools()
        assert {"since", "last"} <= set(spec.inputSchema["properties"])