:j3.dZddlZddlmZmZmZddlmZddlm Z m Z ddl m Z m Z ddlmZddlmZdd lmZmZdd lmZdd lmZmZmZmZdd lmZdd lmZddl m!Z!ddl"m#Z#m$Z$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-e'dZ.Gddeee+Z/y)z Dynamic pipeline system for asynchronous multi-provider task processing. Implements producer-consumer pattern with configurable worker threads and dynamic stage management. N)DictListOptional)Config)configure_authget_auth_provider) PipelineStage SystemState)PipelineStatus) ProviderTask)IPipelineStats IProvider)client)BasePipelineStage StageOutputStageResources StageUtils)StageRegistryMixin)DependencyResolver)MultiResultManager) get_session get_tokenget_user_agent) get_logger) RateLimiter)LifecycleManager) QueueManagermanagerceZdZdZdedeeeffdZde efdZ ddZ dd Z dd Z defd Zdefd Zdefd ZdefdZde eddfdZdedeefdZddZde efdZdeddfdZdedefdZy)PipelineaDynamic pipeline coordinator with registry-based stage management Inherits from PipelineBase to provide type-safe statistics interface, from StageRegistryMixin for stage management capabilities, and from LifecycleManager for lifecycle management. config providersc tj|d||_||_t t t tt|jj||jj|jj|jjt|jj |_t%|j&|_t+j,|jj.t+j0|j&t3|jj|jj4t|jj |_|jjs1 |j"j9|jj:nt>jCd|jDDcic]}|jFs|jH|c}|_%i|_&tO|jP|_)|jUd|_+|jYt[jZ|_.d|_/t>jadtc|jLd te|jLjgy#t<$r#}t>jAd|Yd}~d}~wwxYwcc}w) Nr!)session_providertoken_provideruser_agent_provider) workspacer# batch_size save_intervalsimpleshutdown_timeout)r(r*r,z$Failed to start periodic snapshots: z*Skipping periodic snapshots in simple moderz"Initialized dynamic pipeline with z stages: )4r__init__r"r#rrrrr global_configr( persistencer)r*r+floatr,result_managerr ratelimits rate_limiterr set_proxyproxyinit_github_clientrqueue_interval queue_managerstart_periodic_snapshotssnapshot_interval Exceptionloggererrordebugtasksenabledname task_configsstagesrregistryresolver_create_stages _order_cache_init_order_cachetime start_timeinitial_tasks_countinfolenlistkeys)selfr"r#etasks 7/root/.openclaw/workspace/harvester/manager/pipeline.pyr-zPipeline.__init__(sD!!$ 3 /8  Icqr1**44))44 ,,::%%,,"6#5#5#F#FG  ((9(9: --334 !!&"3"34)**44 ,,;;"6#5#5#F#FG !!(( I##<rAjoinrNrL)rP requestedrRr@results rS_aggregate_stageszPipeline._aggregate_stagesgsE KK%% GD||$006  ) r$))C '0B/C1EF  Gi 3F8<= Nc |j}|stjdy |jj |}t|j|j|j|jt}|jjj}|jjj}|D]}|j!|}|stj d|/ |j#||j$t'|j)|ddt'|j)|dd|jj*j,} | |j.|<y#t $r}tj d|d}~wwxYw#t $r!}tj d |d |d}~wwxYw) z;Create pipeline stages dynamically with hybrid architecturez+No stages requested, pipeline will be emptyNz&Failed to resolve stage dependencies: )limiterr#r"rBauthzStage definition not found: ri) resourceshandler thread_count queue_size max_retrieszFailed to create stage z: )r]r<warningrE resolve_orderr;r=rr3r#r"rBrpipelinethreads queue_sizes get_stage_def stage_class_handle_stage_outputmaxgetr.max_retries_requeuedrC) rPrequested_stagesordered_stagesrQrb thread_config queue_configrA definitionstages rSrFzPipeline._create_stagesws 113 NNH I  !]]889IJN #%%nn;;**"$   ,,44 {{++77 " D++D1J ;D6BC "..' 55!$]%6%6tQ%?!C"<#3#3D$#?C $ 9 9 N N /%* D!! #  LLA!E F  F  6tfBqcBC s0E;7BF%; F"FF"% G.G  Gc*|jstjdy|j}|D]0}|jj |}|s!|j 2tj dt|jdy)zStart all pipeline stageszNo stages to startNzStarted z pipeline stages)rCr<rg get_orderrpstartrLrM)rPrs stage_namerws rS _on_startzPipeline._on_startsv{{ NN/ 0 )( JKKOOJ/E    hs4;;/00@ABr^c|jsy|j}|jrdt|jz nd}t|D]1}|jj |}|s!|j |3|j j |jjtjdy)zStop all pipeline stagesNg>@zStopped all pipeline stages) rCryrMreversedrpstopr8r1stop_allr<rL)rPrs stage_timeoutr{rws rS_on_stopzPipeline._on_stops{{ )37;;s4;;//D ">2 *JKKOOJ/E =) * ! $$& 12r^c@|jsy|j}d}|D]y}|jj|}|s!|jr:|j |r)|j t jd|d|jrxd}{|S)z5Check if pipeline is finished and manage stage statesT[z] stopped accepting new tasksF) rCryrp accepting_can_stage_stop_acceptingstop_acceptingr<rL is_finished)rPrs all_finishedr{rws rSrzPipeline.is_finisheds{{) ) %JKKOOJ/E4#A#A*#M$$& a |+HIJ$$&$  %r^c"|jS)zGet statistics for all stages_get_pipeline_statusrPs rS get_all_statszPipeline.get_all_stats((**r^c"|jS)z%Get dynamic statistics for all stagesrrs rSget_dynamic_statszPipeline.get_dynamic_statsrr^c i}|jjD]\}}|j||<t|jrtj ntj t|jjDcgc]}|js|c}t|j|tj|jz }|Scc}w)z,Get pipeline status as PipelineStatus object)stateactivetotalrCruntime) rCitems get_statsr r RUNNINGSTOPPEDrMvaluesrunningrIrJ)rP stage_statusr{rwspipeline_statuss rSrzPipeline._get_pipeline_statuss "&!2!2!4 9 J','8L $ 9))-+%%+:M:M4;;#5#5#7Ea199EFdkk"IIK$//1   Fs C"C" initial_tasksc.t||_|jjtj j }|r|D]}|j|ntjdtjdt|dy)z$Add initial search tasks to pipelinez2Search stage not created, cannot add initial taskszAdded z initial tasks to pipelineN) rMrKrCrpr SEARCHvalueput_taskr<rgrL)rPr search_stagerRs rSadd_initial_taskszPipeline.add_initial_taskssx#&}#5 {{}';';'A'AB % ,%%d+ , NNO P fS/00JKLr^rAc8|jj|S)zGet stage by name)rCrp)rPrAs rS get_stagezPipeline.get_stages{{t$$r^c|jrd|jjt|jj |_t jd|j yy)z Initialize and cache stage orderzCached stage order: N)rCrErhrNrOrGr<r>rs rSrHzPipeline._init_order_cachesR ;; $ ; ;DAQAQASHandle pure functional stage output - core orchestration logicz Target stage z not found for task zStage z disabled for z, skipping taskN)resultsr1 add_resultlinks add_linksmodels add_models new_tasksrBrpproviderrcheckrCrr<rgr>) rPrr result_typedatarkeyrrR target_stager"rws rSrnzPipeline._handle_stage_outputsJ,2>> H 'Hk4    * *8[$ G H &|| ;OHe    ) )(E : ;&,]] B !Hc6    * *8S& A B#)"2"2 b D,&&**4==9F***6<@  5NN4(NN]<.@TUYUbUbTc#de vl^>$--P_`a br^r{c|jj|}|sy|jjsy|jdkDryg}|jj D]6}|j |}|s||jvs&|j|8|sy|D][}|jj|}|s!|jjsy|jdkDry|js[yy)zICheck if a stage can stop accepting new tasks based on precise conditionsFrT) rCrpqueueemptyactive_workersrOrl produces_forappendr)rPr{rwupstream_stagesother_stage_namerv upstream_nameupstream_stages rSrz"Pipeline._can_stage_stop_accepting:s  +{{  "   ! # $ 0 0 2 9 ++,<=JjJ,C,CC&&'78 9 - !M![[__];N%++113 !0014 !++  !r^)rTN)__name__ __module__ __qualname____doc__rrstrrr-rr]rFr|rboolrr rrrr rrrrrHryrrnrr^rSr!r! s=pv=p$sI~2F=p~49 2hC 3(T2+~++>+n$ MtL/A Md M%c%h/@&A%E E49E b;b4b6,C,D,r^r!)0rrItypingrrrconfig.schemasr core.authrr core.enumsr r core.metricsr core.modelsr core.typesr rsearchr stage.baserrrrstage.registryrstage.resolverrstorage.persistencertools.coordinatorrrr tools.loggerrtools.ratelimitrbaserrrr<r!rr^rSrsi  ''!71'$0QQ--2DD#'" I F~13CFr^