(self, level: LogLevel, message: str = None, node_id: str = None,
event_type: EventType = None, details: Dict[str, Any] = None,
duration: float = None)
| 15 | self.session_id = session_id |
| 16 | |
| 17 | def add_log(self, level: LogLevel, message: str = None, node_id: str = None, |
| 18 | event_type: EventType = None, details: Dict[str, Any] = None, |
| 19 | duration: float = None) -> LogEntry | None: |
| 20 | log_entry = super().add_log(level, message, node_id, event_type, details, duration) |
| 21 | if not log_entry: |
| 22 | return None |
| 23 | |
| 24 | # Send the message using the sync method which handles event loop properly |
| 25 | self.websocket_manager.send_message_sync(self.session_id, { |
| 26 | "type": "log", |
| 27 | "data": log_entry.to_dict() |
| 28 | }) |
| 29 | |
| 30 | return log_entry |
nothing calls this directly
no test coverage detected