Internal node that forces loop termination (break-like behavior).
| 100 | self._attributes_store["current_iteration"] = 0 |
| 101 | self._attributes_store["max_iterations"] = max_iterations |
| 102 | class TerminateNode(InternalGraphNode): |
| 103 | """Internal node that forces loop termination (break-like behavior).""" |
| 104 | def __init__(self, name, gate_close_callback:Callable|None=None, pull_keys:dict[str,dict|str]|None=None, push_keys:dict[str,dict|str]|None=None): |
| 105 | super().__init__(name, gate_close_callback, pull_keys, push_keys) |
| 106 | @property |
| 107 | def is_ready(self) -> bool: |
| 108 | for in_edge in self.in_edges: |
| 109 | if in_edge.is_congested: |
| 110 | return True |
| 111 | return False |
| 112 | @masf_hook(Node.Hook.FORWARD) |
| 113 | def _forward(self,input:dict[str,object]) -> dict[str,object]: |
| 114 | return input.copy() |
| 115 | # def _message_aggregate_in(self) -> dict[str,object]: |
| 116 | # input_msg:dict[str,object] = dict() |
| 117 | # for in_edge in self.in_edges: |
| 118 | # if in_edge.is_congested: |
| 119 | # message:dict[str,object] = in_edge.receive_message() |
| 120 | # input_msg = {**input_msg,**message} |
| 121 | # return input_msg |
| 122 | def _message_dispatch_out(self,message:dict[str,object]): |
| 123 | self._output = message |
| 124 | self._gate = Gate.OPEN |
| 125 | class Controller(InternalGraphNode): |
| 126 | """Internal loop controller that decides continue vs terminate.""" |
| 127 | def __init__(self, |