cleveragents-core/robot/helper_session_model.py

"""Robot Framework helper for Session domain model smoke tests.

Provides a CLI-style interface for Robot to invoke session model
operations and verify outputs.  Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_session_model.py create
    python robot/helper_session_model.py append-messages
    python robot/helper_session_model.py export
    python robot/helper_session_model.py cli-dict
    python robot/helper_session_model.py export-key-order
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from ulid import ULID  # noqa: E402

from cleveragents.domain.models.core.session import (  # noqa: E402
    EXPORT_SCHEMA_VERSION,
    MessageRole,
    Session,
)


def _make_session() -> Session:
    """Create a minimal valid Session."""
    return Session(session_id=str(ULID()))


def _cmd_create() -> int:
    """Create a session and verify fields."""
    session = _make_session()
    if not session.session_id:
        print("session-create-fail: missing session_id")
        return 1
    if session.namespace != "local":
        print(f"session-create-fail: namespace={session.namespace}")
        return 1
    if not session.is_empty:
        print("session-create-fail: session should be empty")
        return 1
    print(f"session-create-ok: id={session.session_id}")
    return 0


def _cmd_append_messages() -> int:
    """Create a session, append messages, verify ordering."""
    session = _make_session()
    session.append_message(MessageRole.USER, "Hello")
    session.append_message(MessageRole.ASSISTANT, "Hi there")
    session.append_message(MessageRole.USER, "How are you?")

    if session.message_count != 3:
        print(f"session-append-fail: count={session.message_count}")
        return 1

    # Verify sequences are 0, 1, 2
    sequences = [m.sequence for m in session.messages]
    if sequences != [0, 1, 2]:
        print(f"session-append-fail: sequences={sequences}")
        return 1

    # Verify last message
    last = session.last_message
    if last is None or last.content != "How are you?":
        print("session-append-fail: wrong last message")
        return 1

    print("session-append-ok: 3 messages in order")
    return 0


def _cmd_export() -> int:
    """Create a session with messages, export, verify keys."""
    session = _make_session()
    session.append_message(MessageRole.USER, "Hello")
    session.append_message(MessageRole.ASSISTANT, "Hi there")
    export = session.as_export_dict()

    required_keys = [
        "schema_version",
        "session_id",
        "actor_name",
        "namespace",
        "messages",
        "linked_plan_ids",
        "token_usage",
        "metadata",
        "created_at",
        "updated_at",
        "checksum",
    ]
    for key in required_keys:
        if key not in export:
            print(f"session-export-fail: missing key '{key}'")
            return 1

    if export["schema_version"] != EXPORT_SCHEMA_VERSION:
        print(f"session-export-fail: schema_version={export['schema_version']}")
        return 1

    if len(export["messages"]) != 2:
        print(f"session-export-fail: messages count={len(export['messages'])}")
        return 1

    print("session-export-ok: all keys present")
    return 0


def _cmd_cli_dict() -> int:
    """Create a session with messages, get CLI dict, verify keys."""
    session = _make_session()
    session.append_message(MessageRole.USER, "Hello")
    cli = session.as_cli_dict()

    required_keys = [
        "session_id",
        "namespace",
        "message_count",
        "created_at",
        "updated_at",
        "token_usage",
    ]
    for key in required_keys:
        if key not in cli:
            print(f"session-cli-dict-fail: missing key '{key}'")
            return 1

    if cli["message_count"] != 1:
        print(f"session-cli-dict-fail: message_count={cli['message_count']}")
        return 1

    print("session-cli-dict-ok: all keys present")
    return 0


def _cmd_export_key_order() -> int:
    """Verify export dict keys appear in expected serialization order."""
    session = _make_session()
    session.append_message(MessageRole.USER, "Hello")
    export = session.as_export_dict()

    expected_order = [
        "schema_version",
        "session_id",
        "actor_name",
        "namespace",
        "messages",
        "linked_plan_ids",
        "token_usage",
        "metadata",
        "created_at",
        "updated_at",
        "checksum",
    ]

    actual_keys = list(export.keys())
    if actual_keys != expected_order:
        print(f"session-export-order-fail: {actual_keys}")
        return 1

    # Verify messages within export are ordered by sequence
    seqs = [m["sequence"] for m in export["messages"]]
    if seqs != sorted(seqs):
        print(f"session-export-order-fail: message sequences={seqs}")
        return 1

    print("session-export-order-ok")
    return 0


_COMMANDS: dict[str, object] = {
    "create": _cmd_create,
    "append-messages": _cmd_append_messages,
    "export": _cmd_export,
    "cli-dict": _cmd_cli_dict,
    "export-key-order": _cmd_export_key_order,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            "Usage: helper_session_model.py"
            " <create|append-messages|export|cli-dict|export-key-order>"
        )
        return 1

    command = sys.argv[1]
    handler = _COMMANDS.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1

    return handler()  # type: ignore[operator]


if __name__ == "__main__":
    sys.exit(main())