"""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" " " ) 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())