44class Session:
 45    """
 46    The main class for managing an execution session.
 47
 48    This class is responsible for setting up all the necessary components for running a Railtracks execution, including the coordinator, publisher, and state management.
 49
 50    For the configuration parameters of the setting. It will follow this precedence:
 51    1. The parameters in the `Session` constructor.
 52    2. The parameters in global context variables.
 53    3. The default values.
 54
 55    Default Values:
 56    - `name`: None
 57    - `timeout`: 150.0 seconds
 58    - `end_on_error`: False
 59    - `broadcast_callback`: None (no event listener)
 60    - `save_state`: True (the state of the execution will be saved to a file at the end of the run in the `.railtracks/data/sessions/` directory)
 61
 62
 63    Args:
 64        name (str | None, optional): Optional name for the session. This name will be included in the saved state file if `save_state` is True.
 65        context (Dict[str, Any], optional): A dictionary of global context variables to be used during the execution.
 66        flow_name (str | None, optional): The name of the flow this session is associated with.
 67        flow_id (str | None, optional): The unique identifier of the flow this session is associated with.
 68        timeout (float, optional): The maximum number of seconds to wait for a response to your top-level request.
 69        end_on_error (bool, optional): If True, the execution will stop when an exception is encountered.
 70        broadcast_callback (Callable[[str], None] | Callable[[str], Coroutine[None, None, None]] | None, optional): A passive listener for one-off events published with `rt.broadcast`.
 71        save_state (bool, optional): If True, the state of the execution will be saved to a file at the end of the run in the `.railtracks/data/sessions/` directory.
 72    """
 73
 74    def __init__(
 75        self,
 76        context: Dict[str, Any] | None = None,
 77        *,
 78        flow_name: str | None = None,
 79        flow_id: str | None = None,
 80        name: str | None = None,
 81        timeout: float | None = None,
 82        end_on_error: bool | None = None,
 83        broadcast_callback: (
 84            Callable[[str], None] | Callable[[str], Coroutine[None, None, None]] | None
 85        ) = None,
 86        save_state: bool | None = None,
 87        payload_callback: Callable[[dict[str, Any]], None] | None = None,
 88    ):
 89        # first lets read from defaults if nessecary for the provided input config
 90
 91        if flow_name is None:
 92            warnings.warn(
 93                "Sessions should be tied to a flow for better observability and state management. Please use the Flow object to create and manage your sessions (see __ for more details). This warning will become an error in future versions.",
 94                DeprecationWarning,
 95            )
 96
 97        self.executor_config = self.global_config_precedence(
 98            timeout=timeout,
 99            end_on_error=end_on_error,
100            broadcast_callback=broadcast_callback,
101            save_state=save_state,
102            payload_callback=payload_callback,
103        )
104
105        if context is None:
106            context = {}
107
108        self.name = name
109        self.flow_name = flow_name
110        self.flow_id = flow_id
111
112        self.publisher: RTPublisher = RTPublisher()
113
114        self._identifier = str(uuid.uuid4())
115
116        executor_info = ExecutionInfo.create_new()
117        self.coordinator = Coordinator(
118            execution_modes={
119                "async": AsyncioExecutionStrategy(
120                    scope_manager=ContextVarScopeManager()
121                )
122            }
123        )
124        self.rt_state = RTState(
125            executor_info, self.executor_config, self.coordinator, self.publisher
126        )
127
128        self.coordinator.start(self.publisher)
129        self._setup_subscriber()
130
131        # NOTE: `payload` still reports per-node details.internals
132        add_inline_listener(_node_internals.record)
133
134        # held so the context survives `delete_globals()` on close
135        self.context = register_globals(
136            session_id=self._identifier,
137            rt_publisher=self.publisher,
138            executor_config=self.executor_config,
139            global_context_vars=context,
140            flow_name=self.flow_name,
141            flow_id=self.flow_id,
142            session_name=self.name,
143        )
144
145        # set at exit, so payload() still works once the events have been released
146        self._internals: dict[str, Any] | None = None
147
148        self._start_time = time.time()
149
150        logger.debug("Session %s is initialized" % self._identifier)
151
152    @classmethod
153    def global_config_precedence(
154        cls,
155        timeout: float | None,
156        end_on_error: bool | None,
157        broadcast_callback: (
158            Callable[[str], None] | Callable[[str], Coroutine[None, None, None]] | None
159        ),
160        save_state: bool | None,
161        payload_callback: Callable[[dict[str, Any]], None] | None,
162    ) -> ExecutorConfig:
163        """
164        Uses the following precedence order to determine the configuration parameters:
165        1. The parameters in the method parameters.
166        2. The parameters in global context variables.
167        3. The default values.
168        """
169        global_executor_config = get_global_config()
170
171        return global_executor_config.precedence_overwritten(
172            timeout=timeout,
173            end_on_error=end_on_error,
174            subscriber=broadcast_callback,
175            save_state=save_state,
176            payload_callback=payload_callback,
177        )
178
179    def __enter__(self):
180        return self
181
182    def __exit__(self, exc_type, exc_val, exc_tb):
183        if self.executor_config.save_state:
184            try:
185                railtracks_dir = resolve_railtracks_home()
186                sessions_dir = railtracks_dir / "data" / "sessions"
187                sessions_dir.mkdir(
188                    parents=True, exist_ok=True
189                )  # Creates directory structure if doesn't exist, skips otherwise.
190
191                # Try to create file path with name, fallback to identifier only if there's an issue
192                if self.flow_name is not None:
193                    name = self.flow_name
194                elif self.name is not None:
195                    name = self.name
196                else:
197                    name = ""
198
199                candidate = sessions_dir / f"{name}_{self._identifier}.json"
200                try:
201                    candidate.touch()
202                    candidate.unlink()
203                    file_path = candidate
204                except OSError:
205                    logger.warning(
206                        get_message(
207                            ExceptionMessageKey.INVALID_SESSION_FILE_NAME_WARN
208                        ).format(name=name, identifier=self._identifier)
209                    )
210                    file_path = sessions_dir / f"{self._identifier}.json"
211
212                logger.info("Saving execution info to %s" % file_path)
213
214                content = json.dumps(self.payload())
215                file_path.write_text(content)
216
217            except OSError as exc:
218                logger.warning(
219                    "Could not persist session state to disk (%s: %s). "
220                    "Set RAILTRACKS_DISABLE_EVENTS=True to silence this warning.",
221                    type(exc).__name__,
222                    exc,
223                )
224            except Exception as e:
225                logger.error(
226                    "Error while saving execution info to file: %s",
227                    e,
228                    exc_info=True,
229                )
230        try:
231            if self.executor_config.payload_callback is not None:
232                self.executor_config.payload_callback(self.payload())
233        except Exception:
234            # TODO: add logging here.
235            pass
236
237        # Keep the folded result and release only the buffered events (the collector is shared)
238        self._internals = _node_internals.internals_for(self._identifier)
239        _node_internals.discard(self._identifier)
240
241        self._close()
242
243    def _setup_subscriber(self):
244        """
245        Prepares and attaches the saved `broadcast_callback` to the publisher: it listens on
246        the event lane for one-off `rt.broadcast` items.
247        """
248
249        if self.executor_config.subscriber is not None:
250            self.publisher.subscribe(
251                event_subscriber(self.executor_config.subscriber),
252                name="Broadcast Callback Subscriber",
253            )
254
255    def _close(self):
256        """
257        Closes the runner and cleans up all resources.
258
259        - Shuts down the state object
260        - Deletes all the global variables that were registered in the context
261        """
262        # FIX: Resource leak - publisher background task wasn't being shut down on Session exit
263        # VISION: Session owns publisher lifecycle and must clean up all resources when exiting
264        if self.publisher.is_running():
265            try:
266                # Signal shutdown by setting the flag - the loop will check this and exit
267                self.publisher._running = False
268
269                # Try to cancel the background task if it exists and isn't done
270                if (
271                    self.publisher.pub_loop is not None
272                    and not self.publisher.pub_loop.done()
273                ):
274                    try:
275                        # Cancel the task - it will check _running and exit naturally
276                        self.publisher.pub_loop.cancel()
277                    except Exception:
278                        # Task might be done or in a different loop, that's okay
279                        pass
280            except Exception:
281                # If shutdown fails for any reason, log it but don't crash
282                logger.warning(
283                    "Failed to shutdown publisher during Session cleanup. "
284                    "This may indicate a resource leak.",
285                    exc_info=True,
286                )
287
288        self.rt_state.shutdown()
289
290        delete_globals()
291        # by deleting all of the state variables we are ensuring that the next time we create a runner it is fresh
292
293    @property
294    def identifier(self) -> str:
295        """The unique identifier assigned to this session."""
296        return self._identifier
297
298    @property
299    def info(self) -> ExecutionInfo:
300        """
301        Returns the current state of the runner.
302
303        This is useful for debugging and viewing the current state of the run.
304        """
305        return self.rt_state.info
306
307    def payload(self) -> Dict[str, Any]:
308        """
309        Gets the complete json payload tied to this session.
310
311        The outputted json schema is maintained in (link here)
312        """
313        info = self.info
314
315        run_list = info.graph_serialization()
316        self._attach_node_internals(run_list)
317
318        full_dict = {
319            "flow_name": self.flow_name,
320            "flow_id": self.flow_id,
321            "session_id": self._identifier,
322            "session_name": self.name,
323            "start_time": self._start_time,
324            "end_time": time.time(),
325            "runs": run_list,
326        }
327
328        return json.loads(json.dumps(full_dict, cls=RTJSONEncoder))
329
330    def _attach_node_internals(self, runs: list[dict[str, Any]]) -> None:
331        """Refill each serialized node's ``details.internals`` from the event stream.
332
333        NOTE: Backward compatibility shim
334        """
335
336        if not isinstance(runs, list):
337            return
338
339        internals = (
340            self._internals
341            if self._internals is not None
342            else _node_internals.internals_for(self._identifier)
343        )
344
345        for run in runs:
346            if not isinstance(run, dict):
347                continue
348            for node in run.get("nodes", []):
349                block = internals.get(node.get("identifier")) or {}
350                node.setdefault("details", {})["internals"] = block
351
352                earlier = {k: v for k, v in block.items() if k != "latency"}
353                snapshot = node.get("parent")
354                while snapshot is not None:
355                    snapshot.setdefault("details", {})["internals"] = earlier
356                    snapshot = snapshot.get("parent")

411def session(
412    func: Callable[_P, Coroutine[Any, Any, _TOutput]] | None = None,
413    *,
414    name: str | None = None,
415    context: Dict[str, Any] | None = None,
416    timeout: float | None = None,
417    end_on_error: bool | None = None,
418    broadcast_callback: (
419        Callable[[str], None] | Callable[[str], Coroutine[None, None, None]] | None
420    ) = None,
421    save_state: bool | None = None,
422) -> (
423    Callable[_P, Coroutine[Any, Any, Tuple[_TOutput, Session]]]
424    | Callable[
425        [Callable[_P, Coroutine[Any, Any, _TOutput]]],
426        Callable[_P, Coroutine[Any, Any, Tuple[_TOutput, Session]]],
427    ]
428):
429    """
430    This decorator automatically creates and manages a Session context for the decorated function,
431    allowing async functions to use Railtracks operations without manually managing the session lifecycle.
432
433    Can be used as:
434    - @session (without parentheses) - uses default settings
435    - @session() (with empty parentheses) - uses default settings
436    - @session(name="my_task", timeout=30) (with configuration parameters)
437
438    When using this decorator, the function returns a tuple containing:
439    1. The original function's return value
440    2. The Session object used during execution
441
442    This allows access to session information (like execution state, logs, etc.) after the function completes,
443    while maintaining the simplicity of decorator usage.
444
445    Args:
446        name (str | None, optional): Optional name for the session. This name will be included in the saved state file if `save_state` is True.
447        context (Dict[str, Any], optional): A dictionary of global context variables to be used during the execution.
448        timeout (float, optional): The maximum number of seconds to wait for a response to your top-level request.
449        end_on_error (bool, optional): If True, the execution will stop when an exception is encountered.
450        broadcast_callback (Callable[[str], None] | Callable[[str], Coroutine[None, None, None]] | None, optional): A callback function that will be called with the broadcast messages.
451        save_state (bool, optional): If True, the state of the execution will be saved to a file at the end of the run in the `.railtracks/data/sessions/` directory.
452
453    Returns:
454        When used as @session (without parentheses): Returns the decorated function that returns (result, session).
455        When used as @session(...) (with parameters): Returns a decorator function that takes an async function
456        and returns a new async function that returns (result, session).
457    """
458
459    def decorator(
460        target_func: Callable[_P, Coroutine[Any, Any, _TOutput]],
461    ) -> Callable[_P, Coroutine[Any, Any, Tuple[_TOutput, Session]]]:
462        # Validate that the decorated function is async
463        if not inspect.iscoroutinefunction(target_func):
464            raise TypeError(
465                f"@session decorator can only be applied to async functions. "
466                f"Function '{target_func.__name__}' is not async. "
467                f"Add 'async' keyword to your function definition."
468            )
469
470        @wraps(target_func)
471        async def wrapper(
472            *args: _P.args, **kwargs: _P.kwargs
473        ) -> Tuple[_TOutput, Session]:
474            session_obj = Session(
475                context=context,
476                timeout=timeout,
477                end_on_error=end_on_error,
478                broadcast_callback=broadcast_callback,
479                name=name,
480                save_state=save_state,
481            )
482
483            with session_obj:
484                result = await target_func(*args, **kwargs)
485                return result, session_obj
486
487        return wrapper
488
489    # If used as @session without parentheses
490    if func is not None:
491        return decorator(func)
492
493    # If used as @session(...)
494    return decorator

 64async def call(
 65    node_: type[Node[_P, _TOutput]] | RTFunction[_P, _TOutput],
 66    *args: _P.args,
 67    **kwargs: _P.kwargs,
 68) -> _TOutput:
 69    """
 70    Call a node from within a node inside the framework. This will return a coroutine that you can interact with
 71    in whatever way using async/await logic.
 72
 73    Usage:
 74    ```python
 75    # for sequential operation
 76    result = await call(NodeA, "hello world", 42)
 77
 78    # for parallel operation
 79    tasks = [call(NodeA, "hello world", i) for i in range(10)]
 80    results = await asyncio.gather(*tasks)
 81    ```
 82
 83    Args:
 84        node: The node type you would like to create. This could be a function decorated with `@function_node`, a function, or a Node instance.
 85        *args: The arguments to pass to the node
 86        **kwargs: The keyword arguments to pass to the node
 87    """
 88    node: type[Node[_P, _TOutput]]
 89
 90    if hasattr(node_, "node_type"):
 91        # local import to prevent circular import issues (note it is a purely type checking import)
 92        from railtracks.built_nodes.function.base import RTFunction
 93
 94        assert isinstance(node_, RTFunction)
 95        node = extract_node_from_function(node_)
 96    else:
 97        node = node_
 98
 99    # if the context is none then we will need to create a wrapper for the state object to work with.
100    if not is_context_present():
101        # we have to use lazy import here to prevent a circular import issue. This is a must have unfortunately.
102        from railtracks import Session
103
104        with Session():
105            result = await _start(node, args=args, kwargs=kwargs)
106            return result
107
108    # if the context is not active then we know this is the top level request
109    if not is_context_active():
110        result = await _start(node, args=args, kwargs=kwargs)
111        return result
112
113    # if the context is active then we can just run the node
114    result = await _run(node, args=args, kwargs=kwargs)
115    return result

265def astream(
266    node_: type[Node[_P, _TOutput]],
267    *args: _P.args,
268    **kwargs: _P.kwargs,
269) -> Stream[_TOutput]:
270    """
271    Invoke an agent node with streaming enabled and return a `Stream` over its emitted chunks.
272
273    Async-iterate the returned `Stream` for token chunks and read `.result` for the final
274    return value, or `await` it directly when you only want the result. Streaming is
275    frame-local: only the node invoked here streams its LLM responses; nested `rt.call`
276    children run buffered.
277
278    Args:
279        node_: The agent node class to invoke (built with `rt.agent_node(...)`). Only agent
280            nodes are accepted; a `@function_node` / tool node has no token stream to
281            surface, so use `rt.call` for those.
282        *args: The positional arguments to pass to the node.
283        **kwargs: The keyword arguments to pass to the node.
284
285    Returns:
286        Stream[_TOutput]: An async iterator over the chunks, with the final result available
287            via `.result` (or by awaiting the stream).
288
289    Raises:
290        NodeCreationError: If `node_` is not an agent node.
291    """
292    # local import to prevent circular import issues (mirrors rt.call)
293    from railtracks.nodes.nodes import Node
294
295    # rt.astream streams an agent's LLM tokens, so it only accepts agent nodes built with
296    # rt.agent_node(...). Validate the input up front: anything that is not an agent `Node`
297    # subclass (a @function_node / tool node, or a raw value) has no token stream to surface,
298    # so it is rejected here rather than unpacked use rt.call instead.
299    is_agent_node = (
300        isinstance(node_, type) and issubclass(node_, Node) and node_.type() == "Agent"
301    )
302    if not is_agent_node:
303        name = getattr(node_, "__name__", None) or repr(node_)
304        raise NodeCreationError(
305            message=f"rt.astream only supports agent nodes, but {name!r} is not one.",
306            notes=[
307                "Pass an agent built with rt.agent_node(...) to rt.astream(...).",
308                "To run a function/tool node, use rt.call(...) instead.",
309            ],
310        )
311    node = node_
312
313    # Session lifecycle is owned here, not by the Stream handle. When called outside any
314    # session we open one and hold it in this closure; the Stream calls `_close` back once
315    # it finishes (see Stream.on_start / on_close), so the session's open/close logic lives
316    # with astream rather than inside the returned object.
317    owned: list[Session] = []
318
319    def _open() -> None:
320        if not is_context_present():
321            from railtracks import Session  # lazy import to avoid a circular import
322
323            owned.append(Session())
324
325    def _close() -> None:
326        while owned:
327            owned.pop().__exit__(None, None, None)
328
329    return Stream(node, args, kwargs, on_start=_open, on_close=_close)

 8async def broadcast(item: str):
 9    """
10    Broadcasts a one-off **event** to the session bus.
11
12    This triggers the `broadcast_callback` you have provided to the `Session` (or via
13    `rt.set_config`). Each broadcast is a discrete event, independent of any LLM token
14    output an agent produces.
15
16    Args:
17        item (str): The item you want to broadcast.
18    """
19    publisher = get_publisher()
20
21    await publisher.publish(BroadcastEvent(node_id=get_parent_id(), item=item))

19async def call_batch(
20    node: type[Node[..., _TOutput]] | RTFunction[..., _TOutput],
21    *iterables: Iterable[Any],
22    return_exceptions: bool = True,
23):
24    """
25    Complete a node over multiple iterables, allowing for parallel execution.
26
27    Note the results will be returned in the order of the iterables, not the order of completion.
28
29    If one of the nodes returns an exception, the thrown exception will be included as a response.
30
31    Args:
32        node: The node type to create.
33        *iterables: The iterables to map the node over.
34        return_exceptions: If True, exceptions will be returned as part of the results.
35            If False, exceptions will be raised immediately, and you will lose access to the results.
36            Defaults to true.
37
38    Returns:
39        An iterable of results from the node.
40
41    Usage:
42        ```python
43        results = await batch(NodeA, ["hello world"] * 10)
44        for result in results:
45            handle(result)
46        ```
47    """
48    # this is big typing disaster but there is no way around it. Try if if you want to.
49    contracts = [call(node, *args) for args in zip(*iterables)]
50
51    results = await asyncio.gather(*contracts, return_exceptions=return_exceptions)
52    return results

 19class ExecutionInfo:
 20    """
 21    A class that contains the full details of the state of a run at any given point in time.
 22
 23    The class is designed to be used as a snapshot of state that can be used to display the state of the run, or to
 24    create a graphical representation of the system.
 25    """
 26
 27    def __init__(
 28        self,
 29        request_forest: RequestForest,
 30        node_forest: NodeForest,
 31        stamper: StampManager,
 32    ):
 33        self.request_forest = request_forest
 34        self.node_forest = node_forest
 35        self.stamper = stamper
 36
 37    @classmethod
 38    def default(cls) -> ExecutionInfo:
 39        """Creates a new "empty" instance of the ExecutionInfo class with the default values."""
 40        return cls.create_new()
 41
 42    @classmethod
 43    def create_new(
 44        cls,
 45    ) -> ExecutionInfo:
 46        """
 47        Creates a new empty instance of state variables with the provided executor configuration.
 48
 49        """
 50        request_heap = RequestForest()
 51        node_heap = NodeForest()
 52        stamper = StampManager()
 53
 54        return ExecutionInfo(
 55            request_forest=request_heap,
 56            node_forest=node_heap,
 57            stamper=stamper,
 58        )
 59
 60    @property
 61    def answer(self):
 62        """Convenience method to access the answer of the run."""
 63        return self.request_forest.answer
 64
 65    @property
 66    def all_stamps(self) -> List[Stamp]:
 67        """Convenience method to access all the stamps of the run."""
 68        return self.stamper.all_stamps
 69
 70    @property
 71    def name(self):
 72        """
 73        Gets the name of the graph by pulling the name of the insertion request. It will raise a ValueError if the insertion
 74        request is not present or there are multiple insertion requests.
 75        """
 76        insertion_requests = self.insertion_requests
 77
 78        # The name is only defined for the length of 1.
 79        # NOTE: Maybe we should send a warning once to user in other cases.
 80        if len(insertion_requests) != 1:
 81            return None
 82
 83        i_r = insertion_requests[0]
 84
 85        return self.node_forest.get_node_type(i_r.sink_id).name()
 86
 87    @property
 88    def insertion_requests(self):
 89        """A convenience method to access all the insertion requests of the run."""
 90        return self.request_forest.insertion_request
 91
 92    def _get_info(self, ids: List[str] | str | None = None) -> ExecutionInfo:
 93        """
 94        Gets a subset of the current state based on the provided node ids. It will contain all the children of the provided node ids
 95
 96        Note: If no ids are provided, the full state is returned.
 97
 98        Args:
 99            ids (List[str] | str | None): A list of node ids to filter the state by. If None, the full state is returned.
100
101        Returns:
102            ExecutionInfo: A new instance of ExecutionInfo containing only the children of the provided ids.
103
104        """
105        if ids is None:
106            return self
107        else:
108            # firstly lets
109            if isinstance(ids, str):
110                ids = [ids]
111
112            # we need to quickly check to make sure these ids are valid
113            for identifier in ids:
114                if identifier not in self.request_forest:
115                    raise ValueError(
116                        f"Identifier '{identifier}' not found in the current state."
117                    )
118
119            new_node_forest, new_request_forest = create_sub_state_info(
120                self.node_forest.heap(),
121                self.request_forest.heap(),
122                ids,
123            )
124            return ExecutionInfo(
125                node_forest=new_node_forest,
126                request_forest=new_request_forest,
127                stamper=self.stamper,
128            )
129
130    def _to_graph(self) -> Tuple[List[Vertex], List[Edge]]:
131        """
132        Converts the current state into its graph representation.
133
134        Returns:
135            List[Node]: An iterable of nodes in the graph.
136            List[Edge]: An iterable of edges in the graph.
137        """
138        return self.node_forest.to_vertices(), self.request_forest.to_edges()
139
140    def graph_serialization(self) -> dict[str, Any]:
141        """
142                Creates a string (JSON) representation of this info object designed to be used to construct a graph for this
143                info object.
144
145                Some important notes about its structure are outlined below:
146                - The `nodes` key contains a list of all the nodes in the graph, represented as `Vertex` objects.
147                - The `edges` key contains a list of all the edges in the graph, represented as `Edge` objects.
148                - The `stamps` key contains an ease of use list of all the stamps associated with the run, represented as `Stamp` objects.
149
150                - The "nodes" and "requests" key will be outlined with normal graph details like connections and identifiers in addition to a loose details object.
151                - However, both will carry an addition param called "stamp" which is a timestamp style object.
152                - They also will carry a "parent" param which is a recursive structure that allows you to traverse the graph in time.
153
154
155        ```
156        """
157        parent_nodes = [x.identifier for x in self.insertion_requests]
158
159        infos = [self._get_info(parent_node) for parent_node in parent_nodes]
160
161        runs = []
162
163        for info, parent_node_id in zip(infos, parent_nodes):
164            insertion_requests = info.request_forest.insertion_request
165
166            assert len(insertion_requests) == 1
167            parent_request = insertion_requests[0]
168
169            all_parents = parent_request.get_all_parents()
170
171            start_time = all_parents[-1].stamp.time
172
173            assert len([x for x in all_parents if x.status == "Completed"]) <= 1
174            end_time = None
175            for req in all_parents:
176                if req.status in ["Completed", "Failed"]:
177                    end_time = req.stamp.time
178                    break
179
180            entry = {
181                "name": info.name,
182                "run_id": parent_node_id,
183                "nodes": info.node_forest.to_vertices(),
184                "status": parent_request.status,
185                "edges": info.request_forest.to_edges(),
186                "steps": _get_stamps_from_forests(
187                    info.node_forest, info.request_forest
188                ),
189                "start_time": start_time,
190                "end_time": end_time,
191            }
192            runs.append(entry)
193
194        return json.loads(
195            json.dumps(
196                runs,
197                cls=RTJSONEncoder,
198            )
199        )