cleveragents-core/robot/helper_tui_llm_dispatch.py

"""Helper script for tui_llm_dispatch.robot integration tests.

Each subcommand is a self-contained check that prints a sentinel on success.

Uses a stub LLM and real SQLite DB — no API keys required.
Follows the same fixture pattern as helper_session_tell_llm.py.
"""

# ruff: noqa: E402
from __future__ import annotations

import sys
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock, patch

_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from cleveragents.application.services.session_service import PersistentSessionService
from cleveragents.application.services.session_workflow import SessionWorkflow
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
    SessionMessageRepository,
    SessionRepository,
)

# ---------------------------------------------------------------------------
# Stub LLM — deterministic responses without real API calls
# ---------------------------------------------------------------------------

_STUB_TEXT = "Hello from the TUI integration stub LLM."


class _StubLLM:
    """Minimal LLM stub compatible with LangChain chat model interface."""

    def invoke(self, messages: Any, **kwargs: Any) -> Any:
        class _Resp:
            content = _STUB_TEXT

            def __init__(self) -> None:
                self.tool_calls: list[Any] = []
                self.response_metadata: dict[str, Any] = {
                    "usage": {"input_tokens": 5, "output_tokens": 10}
                }

        return _Resp()


# ---------------------------------------------------------------------------
# Fixture helpers — mirrors helper_session_tell_llm.py pattern
# ---------------------------------------------------------------------------


def _make_db() -> tuple[Any, Any]:
    """Create in-memory SQLite DB; return (session_factory, get_db callable)."""
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)
    factory = sessionmaker(bind=engine, expire_on_commit=False)
    db_session = factory()

    def get_db() -> Any:
        return db_session

    return factory, get_db


def _make_service(get_db: Any) -> PersistentSessionService:
    """Build a PersistentSessionService from repository instances."""
    return PersistentSessionService(
        session_repo=SessionRepository(get_db),
        message_repo=SessionMessageRepository(get_db),
    )


def _make_workflow(
    service: PersistentSessionService,
) -> SessionWorkflow:
    """Build a SessionWorkflow using the stub LLM via llm_factory injection."""
    stub = _StubLLM()
    return SessionWorkflow(
        session_service=service,
        llm_factory=lambda _actor_name: stub,
    )


# ---------------------------------------------------------------------------
# Subcommands
# ---------------------------------------------------------------------------


def cmd_create_session() -> None:
    """Verify _create_tui_session() creates a real DB-backed session."""
    from cleveragents.tui import commands as cmd_mod

    _, get_db = _make_db()
    service = _make_service(get_db)
    container = MagicMock()
    container.session_service.return_value = service

    with patch.object(cmd_mod, "get_container", return_value=container):
        sid = cmd_mod._create_tui_session()

    assert sid != "default", "Expected a real session ID, got 'default'"
    assert len(sid) > 0, "session_id must not be empty"

    session_obj = service.get(sid)
    assert session_obj is not None, "Session not found in database"
    assert session_obj.session_id == sid

    print("tui-create-session-ok")


def cmd_build_facade() -> None:
    """Verify _build_tui_facade() returns a non-None A2aLocalFacade."""
    from cleveragents.a2a.facade import A2aLocalFacade
    from cleveragents.tui import commands as cmd_mod

    _, get_db = _make_db()
    service = _make_service(get_db)
    container = MagicMock()
    container.session_service.return_value = service
    container.plan_lifecycle_service.side_effect = RuntimeError("not needed")
    container.resource_registry_service.side_effect = RuntimeError("not needed")
    container.tool_registry.side_effect = RuntimeError("not needed")

    with (
        patch.object(cmd_mod, "get_container", return_value=container),
        patch("cleveragents.a2a.cli_bootstrap._facade_instance", None),
    ):
        facade = cmd_mod._build_tui_facade()

    assert facade is not None, "_build_tui_facade() must return a facade"
    assert isinstance(facade, A2aLocalFacade), (
        f"Expected A2aLocalFacade, got {type(facade)}"
    )
    print("tui-build-facade-ok")


def cmd_dispatch_message() -> None:
    """Verify facade.dispatch(message/send) with stub LLM persists and responds."""
    from cleveragents.a2a.facade import A2aLocalFacade
    from cleveragents.a2a.models import A2aRequest

    _, get_db = _make_db()
    service = _make_service(get_db)
    workflow = _make_workflow(service)

    facade = A2aLocalFacade(
        services={"session_service": service, "session_workflow": workflow}
    )

    session = service.create(actor_name="openai/gpt-4o")
    session_id = session.session_id

    request = A2aRequest(
        method="message/send",
        params={"session_id": session_id, "message": "hello integration"},
    )
    response = facade.dispatch(request)

    assert response.error is None, f"Unexpected error: {response.error}"
    assert response.result is not None, "Expected result dict"
    assistant_msg: str = response.result.get("assistant_message", "")
    assert len(assistant_msg) > 0, "Expected non-empty assistant message"
    assert _STUB_TEXT in assistant_msg, (
        f"Expected stub text {_STUB_TEXT!r} in response, got: {assistant_msg!r}"
    )

    messages = service.get_messages(session_id)
    # get_messages() returns SessionMessage objects — access .role attribute
    roles = [m.role.value for m in messages]
    assert "user" in roles, "User message not persisted"
    assert "assistant" in roles, "Assistant message not persisted"

    print("tui-dispatch-message-ok")


def cmd_no_actor_error() -> None:
    """Verify SessionActorNotConfiguredError is raised when no actor configured.

    The A2A facade re-raises domain exceptions (SessionActorNotConfiguredError,
    SessionNotFoundError, DatabaseError) so callers can handle them by type.
    This mirrors the CLI pattern in _facade_dispatch() in session.py.
    """
    from cleveragents.a2a.facade import A2aLocalFacade
    from cleveragents.a2a.models import A2aRequest
    from cleveragents.domain.models.core.session import (
        SessionActorNotConfiguredError,
    )

    _, get_db = _make_db()
    service = _make_service(get_db)
    workflow = SessionWorkflow(session_service=service)

    facade = A2aLocalFacade(
        services={"session_service": service, "session_workflow": workflow}
    )

    session = service.create(actor_name=None)
    session_id = session.session_id

    request = A2aRequest(
        method="message/send",
        params={"session_id": session_id, "message": "hello"},
    )

    try:
        facade.dispatch(request)
        raise AssertionError("Expected SessionActorNotConfiguredError to be raised")
    except SessionActorNotConfiguredError:
        pass  # expected — facade re-raises domain exceptions

    print("tui-no-actor-error-ok")


def cmd_database_error() -> None:
    """Verify _run_llm_dispatch returns a friendly message on DatabaseError.

    Simulates a DatabaseError raised by the facade and confirms the TUI
    dispatch function converts it to a user-readable error string rather
    than propagating the exception.
    """
    from unittest.mock import MagicMock

    from cleveragents.core.exceptions import DatabaseError
    from cleveragents.tui.app import _run_llm_dispatch

    facade = MagicMock()
    facade.dispatch.side_effect = DatabaseError("disk full")

    result = _run_llm_dispatch(facade, "test-session", "hello")

    assert "Database error" in result, (
        f"Expected 'Database error' in result, got: {result!r}"
    )
    print("tui-database-error-ok")


def cmd_session_not_found() -> None:
    """Verify _run_llm_dispatch returns a friendly message on SessionNotFoundError.

    Simulates the session being deleted between creation and dispatch (e.g.
    concurrent process or manual DB edit) and confirms the TUI surfaces a
    clear error message rather than crashing.
    """
    from unittest.mock import MagicMock

    from cleveragents.domain.models.core.session import SessionNotFoundError
    from cleveragents.tui.app import _run_llm_dispatch

    facade = MagicMock()
    facade.dispatch.side_effect = SessionNotFoundError("session gone")

    result = _run_llm_dispatch(facade, "test-session", "hello")

    assert "Session not found" in result, (
        f"Expected 'Session not found' in result, got: {result!r}"
    )
    print("tui-session-not-found-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

COMMANDS = {
    "create-session": cmd_create_session,
    "build-facade": cmd_build_facade,
    "dispatch-message": cmd_dispatch_message,
    "no-actor-error": cmd_no_actor_error,
    "database-error": cmd_database_error,
    "session-not-found": cmd_session_not_found,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
        print(f"Usage: {sys.argv[0]} [{' | '.join(COMMANDS)}]", file=sys.stderr)
        sys.exit(1)
    COMMANDS[sys.argv[1]]()