(self, session_id: str)
| 109 | return self._sessions.pop(session_id, None) |
| 110 | |
| 111 | def get_session_info(self, session_id: str) -> Optional[Dict[str, Any]]: |
| 112 | session = self._sessions.get(session_id) |
| 113 | if not session: |
| 114 | return None |
| 115 | return { |
| 116 | "session_id": session.session_id, |
| 117 | "yaml_file": session.yaml_file, |
| 118 | "status": session.status.value, |
| 119 | "created_at": session.created_at, |
| 120 | "updated_at": session.updated_at, |
| 121 | "current_node_id": session.current_node_id, |
| 122 | "waiting_for_input": session.waiting_for_input, |
| 123 | "error_message": session.error_message, |
| 124 | } |
| 125 | |
| 126 | def list_sessions(self) -> Dict[str, Dict[str, Any]]: |
| 127 | return {session_id: self.get_session_info(session_id) for session_id in self._sessions.keys()} |
no test coverage detected