(method: str, params: dict)
| 3663 | # Set up notification handler for session events |
| 3664 | # Note: This handler is called from the event loop (thread-safe scheduling) |
| 3665 | def handle_notification(method: str, params: dict): |
| 3666 | if method == "session.event": |
| 3667 | session_id = params["sessionId"] |
| 3668 | event_dict = params["event"] |
| 3669 | # Convert dict to SessionEvent object |
| 3670 | event = session_event_from_dict(event_dict) |
| 3671 | with self._sessions_lock: |
| 3672 | session = self._sessions.get(session_id) |
| 3673 | if session: |
| 3674 | session._dispatch_event(event) |
| 3675 | elif method == "session.lifecycle": |
| 3676 | # Handle session lifecycle events |
| 3677 | lifecycle_event = _session_lifecycle_event_from_dict(params) |
| 3678 | self._dispatch_lifecycle_event(lifecycle_event) |
| 3679 | |
| 3680 | self._client.set_notification_handler(handle_notification) |
| 3681 | self._client.set_request_handler("userInput.request", self._handle_user_input_request) |
nothing calls this directly
no test coverage detected