i[dZddlZddlZddlmZmZmZmZmZm Z ddl m Z m Z ddl mZddlmZddlmZddlmZGd d Zy) z:Cycle executor that runs workflow graphs containing loops.N)DictListCallableAnySetOptional)NodeEdgeLink) LogManager) CycleManager)ParallelExecutor)GraphTopologyBuildercLeZdZdZdedeeefdeeee fde de egdff dZ d)d Z d eeee fd dfd Zd eeee fd dfd Zdeee fd dfdZded dfdZdeee fd dfdZdedeed edzfdZdedeededed eef dZdeeded eeefdZ d*deedeed eeeffdZ d+deededeeeded eeee ff d Z d,d!eeeee fd"ed#eedeeded eef d$Z d+ded%eed&ed eefd'Zdedeed efd(Zy)- CycleExecutorzExecute workflow graphs that contain cycles. Features: - Scheduling is based on "super nodes" - Parallel execution inside cycles - Automatic detection of exit conditions log_managernodescycle_execution_order cycle_managerexecute_node_funcNcl||_||_||_||_||_t |||_y)afInitialize the cycle executor. Args: log_manager: Logger instance nodes: Mapping of node ids to nodes cycle_execution_order: Super-node execution order with cycles cycle_manager: Cycle manager coordinating iterations execute_node_func: Callable that executes a single node N)rrrrrr parallel_executor)selfrrrrrs M/Users/bowang/.openclaw/workspace/ChatDev/workflow/executor/cycle_executor.py__init__zCycleExecutor.__init__s:"' %:"*!2!1+u!Ereturnc |jjdt|jD]A\}}|jjd|dt |d|j |Cy)z&Run the workflow that contains cycles.z6Executing graph with cycles using super-node schedulerzExecuting super-node layer z with z itemsN)rdebug enumeraterlen_execute_super_layer)r layer_idx layer_itemss rexecutezCycleExecutor.execute/sm WX&/0J0J&K "I{    " "%@ 6RUVaRbQcci#j k  % %k 2'Lrr#c&|j|y)z"Execute a single super-node layer.N)_execute_super_layer_parallel)rr#s rr!z"CycleExecutor._execute_super_layer7s **;7rcdtttfdtfd}|jj ||j |y)z'Execute a super-node layer in parallel.itemrcX|ddk(rd|dS|ddk(rd|dSd|dd S) Ntypecyclecycle cycle_idnodenode node_idrrr(s ritem_desc_funczCCycleExecutor._execute_super_layer_parallel..item_desc_func=sXF|w&Z 0122f'tI/00tG}Q/011rN)rstrrrexecute_items_parallel_execute_super_item)rr#r3s rr&z+CycleExecutor._execute_super_layer_parallel;sB 2c3h 2C 2 55   $ $  rr(c|ddk(r|j|ddy|ddk(r|j|dy|ddk(r|j|yy) z1Execute a single super-node item (node or cycle).r*layerrrr.r0r+N)_execute_single_node_execute_cycle)rr(s rr6z!CycleExecutor._execute_super_itemMsb <7 "  % %d7mA&6 7 &\V #  % %d9o 6 &\W $    %%rr0c|jjd||j|}|jr|j |y|jj d|dy)zExecute a non-cycle node.zExecuting non-cycle node: Node z% is not triggered, skipping executionN)rrr is_triggeredrwarning)rr0r.s rr9z"CycleExecutor._execute_single_nodeXsd !;G9EFzz'"      " "4 (    $ $uWI5Z%[ \r cycle_infoc|d}|d}|jjd|d| |j||}| |jjd|dy||j j|_|jjd|d||j j||j||||j j|j |j j||jjd|d y#t$r*}|jj t |d}~wwxYw) z0Execute a cycle using the multi-iteration logic.r-rzExecuting cycle z with nodes: NCycle z= has no triggered entry node in this pass; skipping executionz initial node: max_iterationsz completed) rr_validate_cycle_entry ValueErrorerrorr4rcycles initial_nodeactivate_cycle_execute_cycle_with_iterationsget_max_iterationsdeactivate_cycle)rr?r-rinitial_node_ides rr:zCycleExecutor._execute_cyclebslj)7# !1(=PQ "885IO  "    " " "_`  QQS ,  ++H5 z<=7     " "3q6 *  sD00 E#9%EE#r-c g}|D]q}|j|}|jD]Q}|j|vs|j|}|s&|js3|j s@|j |qs|jjj|}|r |jnd} t|dk(r| r| Syt|dkDrtd|d|d|d} | r| | k7rtd|d| d| d | S) aT Validate that exactly one node in the cycle is triggered by external edges. Args: cycle_id: The cycle ID nodes: List of node IDs in the cycle Returns: The ID of the unique initial node Raises: ValueError: If no node or multiple nodes are triggered NrrAz% has multiple triggered entry nodes: z>. Only one entry node must be triggered when entering a cycle.z entry mismatch: configured 'z' but triggered '') r predecessorsidfind_outgoing_edgetrigger triggeredappendrrGgetconfigured_entry_noder rE) rr-rtriggered_nodesr0r. predecessoredger?configured_entry entry_nodes rrDz#CycleExecutor._validate_cycle_entrys8&(G::g&D#00 >>.&99'BD '..w7 1''..228< ?I:;;t  1 $''  !A % "GGXYOO %Q'  .> > "?@P?QR"",Q0  r cycle_nodesrMrCcd}||kr|jjd|d|dzd||j||}|j||||dk(}|j |||||dk(}|r,|jjd|dt ||S|j ||s |jjd|d n |dz }||kr||k\r"|jjd|d |d tS) a Execute a cycle with multiple iterations. Args: cycle_id: Cycle ID cycle_nodes: List of all nodes in the cycle initial_node_id: Initial node ID max_iterations: Maximum number of iterations Returns: A tuple of two sets: - exit_nodes: nodes triggered outside the *current* cycle scope - external_nodes: subset of exit_nodes that are also outside the provided parent_cycle_nodes scope rrAz iteration rP/)is_first_iteration)rMrbz$ exited - external nodes triggered: z) completed - initial node not retriggeredz reached max iterations ()) rr_detect_cycles_in_scope"_build_topological_layers_in_scope_execute_scope_layerssorted_is_initial_node_retriggeredr>set) rr-r_rMrC iteration inner_cyclesexecution_layersexternal_nodess rrJz,CycleExecutor._execute_cycle_with_iterationssf, .(    " " +i!m_An=MN   77 _UL $FF_l$-N G   "77  /$-N 8N  &&XJ&J6R`KaJbc&%44_kR  &&XJ&OP NII.(L  &    $ $ ";N;K1M u r scope_nodesc|j||}tj|}|Dcgc]}t|dkDr|}}|Scc}w)a# Detect nested cycles within the scoped subgraph. Constructs a subgraph containing only: 1. Nodes in scope_nodes 2. Edges where both source and target are in scope_nodes 3. Initial node's incoming edges are REMOVED (to break the outer cycle) Args: scope_nodes: List of node IDs in the current scope initial_node_id: Initial node ID (whose incoming edges are removed) Returns: List of detected nested cycles (excluding the current cycle itself) clear_entry_noderP)_build_scoped_nodesr detect_cyclesr )rrnrM scoped_nodes all_cyclesr+ nested_cycless rrdz%CycleExecutor._detect_cycles_in_scopesd*// o/^ *77 E  * )e5zA~ z    sArqc  i}t|}|D]U}|j|}tj|}|jDcgc]A}|jj |vr'|j r|jj |k7r|C} }| |_||k(rg|_n[g} |jD]C} | j |vs| j|} | s&| j s3| j| E| |_|jD cgc]C j |vr3 j |k7r$t fd|jDr E}} ||_ |||<X|Scc}wcc} w)a Build a scoped subgraph containing only nodes and edges within the scope. Args: scope_nodes: List of node IDs in the scope clear_entry_node: If specified, this node's incoming edges will be removed (used to break the outer cycle when detecting nested cycles) Returns: Dictionary of scoped nodes c3K|]5}|jjjk(xr |j7ywN)targetrSrU).0 edge_linksuccs r z4CycleExecutor._build_scoped_nodes..Us<%H $$''4772Hy7H7HH%Hs;>) rircopyiter_outgoing_edgesrzrSrU_outgoing_edgesrRrTrW successorsany)rrnrqrtscope_nodes_setr0 original_node scoped_noder| scoped_edgesscoped_predecessorspredr\r}scoped_successorss ` rrrz!CycleExecutor._build_scoped_nodess  k*"G JJw/M))M2K ,9+L+L+N+Ni##&&/9%%$$''+;;+N  +7K '**+- (&(#)66Dww/1#66w?DLL/66t< 7 ,? ( "/!9!9!!977o-GG//%2%F%F%H!9 !&7K "$/L !U#XK0!s AE,AE1rkrbc |j|d}|r||vr:g||_n/|D]*}|j|js!g||_,g}t }|r|j |n6|D]1}|j|js!|j |3|D]}|j |} | s| jD]Y} | jj|vr|j|| jj| j| jd[|stj|} | Stj|||} tj | |} | S)a Build topological execution order for the scoped subgraph. Args: scope_nodes: List of node IDs in the scope initial_node_id: Initial node ID inner_cycles: List of nested cycles detected in the scope is_first_iteration: Whether this is the first iteration (affects initial node handling) Returns: List of execution layers, each containing execution items Nrp)fromtorU condition)rrrRrr=riaddrXrrzrSrWrUrrbuild_dag_layerscreate_super_node_graphtopological_sort_super_nodes) rrnrMrkrbrtr0rexclude_targetsrr|layers super_graphs rrez0CycleExecutor._build_topological_layers_in_scope`s*// d/S ,.=? _-:'::g&3359;L)6'  %     0'::g&335#''0'#G&**73K!,!@!@!BI ''**o=  '' ''..11#,#4#4%.%8%8 ) "C # )::<HFM/FFlLK*FF\FMrrlparent_cycle_idparent_cycle_nodesc  t| t tjtj dtt ddf fd dt t tfdt fd}|D]d}jrnRdt t tfddf fd }jj|||jsdn r$ D]} j| j! S)a Execute scoped layers with parallelism, supporting nested cycles. Args: execution_layers: List of execution layers parent_cycle_id: Parent cycle ID parent_cycle_nodes: List of nodes in the parent cycle initial_node_id: Initial node ID (for first iteration special handling) is_first_iteration: Whether this is the first iteration Returns: external_nodes: subset of exit_nodes outside parent_cycle_nodes_set rrNc|sy5|r!j|jdddy#1swYyxYwry)updateri)rrm result_lock stop_events rrecord_externalz.record_externals1"))%0NN$s$4=r(cD|ddk(rd|dS|ddk(rd|dSy) Nr*r.r/r0r+r,r- layer_itemr1r2s r item_descz6CycleExecutor._execute_scope_layers..item_descsBF|v%tI/00F|w&Z 0122rcHjry|ddk(r.|d} xr| k(}j||}|r |yy|ddk(r|d}|d}jjd|d  j ||}| jjd |d yj|||d }|r|D chc]} | vr|  } } | r | yyyy#t $r*}jj t|d}~wwxYwcc} w)Nr*r.r0) force_executer+rr-zExecuting nested cycle z within cycle z Nested cycle z! has no triggered entry; skippingdrB) is_set#_execute_single_cycle_node_in_scoperrrDrErFr4rJ)r(_node_idrtargetsinner_cycle_nodesinner_cycle_idinner_initial_noderNinner_external_nodesr.filteredrMrbrrscope_node_setrrs r executor_funcz:CycleExecutor._execute_scope_layers..executor_funcs$$&<6)#IH$6$XH9!(<!$ $+H5$ ,;-&((..s1v6$$s=C) D) D2%DD) ri threadingEventLockrr4rrrrr5rreset_triggers)rrlrrrMrbrr8rr0rmrrrrs` ` `` @@@@@rrfz#CycleExecutor._execute_scope_layerss*/0#&5__& nn&  %3s8 % % DcN s &E  "3 6DcN3 6t3 63 6j  " " 9 9    "A&D ) 7#224*rrrc|j|}|s|js tS|jD] }d|_ |j |t}|jD]}|j j|vs|js)|jjd|d|j j|j|j j|S)a Execute a single node within a cycle scope. Args: node_id: Node ID to execute scope_node_set: Nodes that belong to the current scoped cycle force_execute: If True, execute even if not triggered (for initial node in first iteration) Returns: Set of node IDs triggered outside the current scoped cycle Fr<z triggered external node ) rr=rirrVrrzrSrrr)rr0rrr.r|external_targetss rrz1CycleExecutor._execute_single_cycle_node_in_scope$s"zz'"$$&u 113I"'I 4 t$&)U113I"".8Y=P=P  &&G9$=i>N>N>Q>Q=RS!$$Y%5%5%8%89 4 rc|j|}|jD]@}|j|vs|j|}|s&|js3|j s@yy)a2 Check if the initial node is retriggered by any internal edge (from within the cycle). Args: initial_node_id: Initial node ID cycle_nodes: List of nodes in the cycle Returns: True if the initial node is retriggered by an internal edge TF)rrRrSrTrUrV)rrMr_rHr[r\s rrhz*CycleExecutor._is_initial_node_retriggeredNsUzz/2 '44K~~,"55oFDLLT^^ 5r)rNry)F)NF)__name__ __module__ __qualname____doc__r rr4r rrr rrr$r!r&r6r9r:rDintrrJrdrrrboolrerfrrhr1rrrrsFFCIF $DcN3 F $ F $TFDL1 F038T#s(^0D88 d38n9M RV $ &S#X &4 &]C]D]%>c3h%>D%>R-c-$s)-d -^BB#YB B  B S BF #Y   c#h J+/?#Y?#3-? c4i ?L$) P#YPP3s8n P ! P d38n  Pn*.#( ptDcN34pp!I p "# p ! p Spl$ ( ( C(  ( S ( T#Y  rr)rrrtypingrrrrrrentity.configsr r utils.log_managerr workflow.cycle_managerr #workflow.executor.parallel_executorr workflow.topology_builderrrr1rrrs/@ ;;)(/@:X X r