class InteractiveDebugHook[T: State, S: Shared](GraphHook[T, S]):
"""
A hook that prints the state and shared data at each step of the graph
execution and pauses for user input. Useful for interactive debugging.
"""
def __init__(self) -> None:
self._renderer = GraphRenderer[T, S]()
def _pause(self) -> None:
self._renderer.render_rule(style="dim")
input("Press Enter to continue...")
async def on_graph_start(self, state: T, shared: S) -> None:
self._renderer.render_graph_start(state, shared)
self._pause()
async def on_step_start(self, state: T, shared: S, nodes: list[NextNode[T, S]]) -> None:
self._renderer.render_step_start(nodes)
self._pause()
async def on_merge_conflict(
self,
state: T,
changes: list[dict[tuple[Hashable, ...], Change]],
conflicts: dict[tuple[Hashable, ...], list[Change]],
) -> None:
self._renderer.render_merge_conflict(conflicts)
self._pause()
async def on_merge_end(
self,
state: T,
result_states: list[T],
changes: list[dict[tuple[Hashable, ...], Change]],
merged_state: T,
) -> None:
self._renderer.render_merge_end(changes)
self._pause()
async def on_step_end(self, state: T, shared: S, nodes: list[NextNode[T, S]]) -> None:
self._renderer.render_step_end(state, shared, nodes)
self._pause()
async def on_graph_end(self, state: T, shared: S) -> None:
self._renderer.render_graph_end(state, shared)
self._pause()