Load a saved session into the current context. Args: session_id: Session ID to load Returns: True if loaded successfully
(self, session_id: str)
| 885 | return None |
| 886 | |
| 887 | def load_session(self, session_id: str) -> bool: |
| 888 | """Load a saved session into the current context. |
| 889 | |
| 890 | Args: |
| 891 | session_id: Session ID to load |
| 892 | |
| 893 | Returns: |
| 894 | True if loaded successfully |
| 895 | """ |
| 896 | data = self._session_store.load(session_id) |
| 897 | if data is None: |
| 898 | return False |
| 899 | |
| 900 | try: |
| 901 | # Inject messages into the session manager |
| 902 | for msg_dict in data.messages: |
| 903 | role = msg_dict.get("role", "") |
| 904 | content = msg_dict.get("content", "") |
| 905 | |
| 906 | if role == MessageRole.SYSTEM: |
| 907 | continue # System prompt is regenerated |
| 908 | elif role == MessageRole.USER: |
| 909 | event = SessionEvent( |
| 910 | type=EventType.MESSAGE, |
| 911 | source=EventSource.USER, |
| 912 | message=content, |
| 913 | ) |
| 914 | elif role == MessageRole.ASSISTANT: |
| 915 | # Assistant messages with tool_calls need full dict |
| 916 | if msg_dict.get("tool_calls"): |
| 917 | event = SessionEvent( |
| 918 | type=EventType.TOOL_CALL, |
| 919 | source=EventSource.SYSTEM, |
| 920 | message=msg_dict, |
| 921 | ) |
| 922 | else: |
| 923 | event = SessionEvent( |
| 924 | type=EventType.MESSAGE, |
| 925 | source=EventSource.LLM, |
| 926 | message=content, |
| 927 | ) |
| 928 | elif role == MessageRole.TOOL: |
| 929 | # Tool result messages stored as full dict for reconstruction |
| 930 | event = SessionEvent( |
| 931 | type=EventType.TOOL_CALL, |
| 932 | source=EventSource.SYSTEM, |
| 933 | message=msg_dict, |
| 934 | ) |
| 935 | else: |
| 936 | continue |
| 937 | |
| 938 | self.session._session.events.append(event) |
| 939 | |
| 940 | logger.info( |
| 941 | f"Loaded session {session_id} with {len(data.messages)} messages" |
| 942 | ) |
| 943 | return True |
| 944 | except Exception as e: |