Update current graph context and prepare graph payload.
(self, root_graph: object)
| 289 | t.join(timeout=1.0) |
| 290 | |
| 291 | def set_active_graph(self, root_graph: object) -> None: |
| 292 | """ |
| 293 | Update current graph context and prepare graph payload. |
| 294 | """ |
| 295 | try: |
| 296 | from .serialize import serialize_root_graph |
| 297 | |
| 298 | graph_name = getattr(root_graph, "name", None) |
| 299 | if not isinstance(graph_name, str) or not graph_name: |
| 300 | graph_name = "unknown" |
| 301 | |
| 302 | payload = serialize_root_graph(root_graph).graph |
| 303 | except Exception as e: |
| 304 | graph_name = getattr(root_graph, "name", "unknown") |
| 305 | payload = {"nodes": ["entry", "exit"], "nodeTypes": {"entry": "entry", "exit": "exit"}, "edges": [], "warnings": [str(e)]} |
| 306 | |
| 307 | with self._lock: |
| 308 | self._graph_name = graph_name |
| 309 | self._graph_payload = payload |
| 310 | self._graph_version += 1 |
| 311 | # Keep a handle to RootGraph entry/exit for stable node id mapping ("entry"/"exit"). |
| 312 | self._root_entry_obj = getattr(root_graph, "_entry", None) |
| 313 | self._root_exit_obj = getattr(root_graph, "_exit", None) |
| 314 | |
| 315 | self.start() |
| 316 | |
| 317 | def _resolve_node_id(self, node: object) -> str | None: |
| 318 | if node is None: |
nothing calls this directly
no test coverage detected