yjQ8ddlZddlmZmZddlmZmZmZddlm Z ddl m Z m Z m Z ddlmZddlmZddlmZdd lmZdd lmZdd lmZmZdd lmZdd lmZmZm Z ddl!m"Z"ej#e$Z%de&e'e(efddfdZ)de de&e'e(efde(de(de&e'e(eff dZ*de de(de(de'e(e&e'e(efffdZ+ d1de'e(efdede(dzde&e(de(de'e(eff dZ,de'e(efdede(de-de'e(eff dZ.de(d e'e(e&e'e(effde/fd!Z0de de'e(efd e'e(e&e'e(effde(dede&e'e(eff d"Z1dddddd#de(d$e(de(d%e j2de(dzd&e(dzd'e(dzd(e-dzd)e-dzde'e(effd*Z3 d2de(d$e(de(d%e j2de(dzd&e(dzd'e(dzd(e-dzd)e-dzddfd+Z4de(d,ed-d.e(de'e(effd/Z5 d1de(d,ed-d.e(de dzddf d0Z6dS)3N)AnyLiteral)existsinsertselect) AsyncSession)crudmodelsschemas)settings) tracked_db)get_dream_scheduler)ValidationException) QueueItem)MessageConfigurationResolvedConfiguration)get_configuration)create_deletion_payloadcreate_dream_payloadcreate_payload)construct_work_unit_keypayloadreturncKt}|r|rt}|D]a}|d}|d}|r3|r1|||d{V}||b|r+t dt|dtd4d{V} |s dddd{VdS|dd}|dd}||td t||||d{V} | rbtt t} || | d{V|d{Vn[#t $rN} t d t$jjrddl} | | Yd} ~ nd} ~ wwxYwdddd{VdS#1d{VswxYwYdS) z~ Add message(s) to the deriver queue for processing. Args: payload: List of message payload dictionaries workspace_name peer_nameNz Cancelled z# pending dreams due to new activitymessage_enqueuer session_namez"Session and workspace are requiredzFailed to enqueue message(s)!)rsetgetcancel_dreams_for_observedupdateloggerinfolenr rhandle_sessionrr returningexecutecommit Exception exceptionr SENTRYENABLED sentry_sdkcapture_exception) rdream_schedulercancelled_dreamsmessagerr cancelled db_sessionr queue_recordsstmter.s ;/DATA/AppData/hermes/projects/honcho/src/deriver/enqueue.pyenqueuer9sZ*++O7%(UU 3 3G$[[)9::N K00I 3) 3"1"L"L"I## !'' 222   KKWS!122WWW   +,,0000000  0  00000000000000 %QZ(89N"1:n5L#~'=)*NOOO"0G^\##M *i((229== ((}========= ''))))))))) 0 0 0   < = = =& 0!!!!,,Q///  0)000000000000000000000000000000sD G; F !B+F  G; G%AG G; G%%G;; HHr4rrc Ktj|tj||d{Vj}tj||d{V}t d||}t|||d{V}g}|D]^} | d} | t | ||} n|} | t|| ||j | d{V_|S)a[ Handle enqueueing for normal session cases, creating appropriate queue items based on configurations. Args: db_session: The database session payload: List of message payloads workspace_name: Name of the workspace session_name: Name of the session Returns: List of queue records to insert )name)sessionrN)r configuration) r get_or_create_sessionr SessionCreateresource get_workspacerget_peers_with_configurationr extendgenerate_queue_recordsid) r4rrrr< workspacesession_level_configurationpeers_with_configurationr5r2message_configmessage_level_configurations r8r&r&Qs&( )|<<<)           (NSSSSSSSSSI#4D'9"M"M%ANL&&      +-M  6=kk/6R6R  %*;++ ' '+F '(( +            cKtj||d{V}||d{V}|}d|DS)a Retrieve peers with their configurations for a given session. Args: db_session: The database session workspace_name: Name of the workspace session_name: Name of the session Returns: Dictionary mapping peer names to their configurations )rrNcBi|]}|j|j|j|jgS)rpeer_configurationsession_peer_configuration is_active).0rows r8 z0get_peers_with_configuration..sB      "  * M    rK)r get_session_peer_configurationr(all)r4rrconfiguration_querypeers_with_configuration_resultpeers_with_configuration_lists r8rBrBs!% C%L!!!-7,>,>?R,S,S&S&S&S&S&S&S#$C$G$G$I$I!   1    rKr2conf session_id observersobservedc<|d}|d}t|tstdt|tstdt ||d||}t ||||d||dS)a Create a queue record for representation task. Args: message: The message payload conf: Resolved configuration for this particular message session_id: Optional session ID observers: List of observer peer names observed: Name of the sender Returns: Queue record dictionary with workspace_name and message_id as separate fields r message_id/workspace_name is required and must be a string-message_id is required and must be an integerrepresentation)r2r= task_typer\r] work_unit_keyrr[rcrr_)r isinstancestr TypeErrorintrr)r2rZr[r\r]rr_processed_payloads r8create_representation_recordrks*[[!122N\**J nc * *KIJJJ j# & &IGHHH(6" )))1ARSS$ %(   rKr=message_seq_in_sessionc:|d}|d}t|tstdt|tstdt ||d|}t ||||d||dS)a6 Create a queue record for summary task. Args: message: The message payload session_id: Session ID message_seq_in_session: The sequence number of the message in the session Returns: Queue record dictionary with workspace_name and message_id as separate fields rr_r`rasummary)r2r=rcrlrd)r rfrg ValueErrorrirr)r2r=r[rlrr_rjs r8create_summary_recordrps"[[!122N\**J nc * *LJKKK j# & &JHIII&#5 1ARSS$ (   rKrHc||iig}|drtjdi|dnd}|drtjdi|dntj}|r|j|jS|j|jndS)a( Determine the effective observe_me setting for a sender, considering session and peer configurations. Args: observed: Name of the sender peers_with_configuration: Dictionary of peer configurations Returns: True if observe_me is enabled, False otherwise NrTrN)r r SessionPeerConfig PeerConfig observe_me)r]rHr=sender_session_peer_configsender_peer_configs r8get_effective_observe_merxs 8;;Hr2hOOM9Fq9IS!55M!$4555t   "..]1-...   ! !"5&@&K&W)44  ( 4 %% rKc K|d}|d}t|dpd}|dkr*tj||d|d|d{V}g}|jjrL||jjzdks||jjzdkr&|t|||| t||} |j js|Sg} | r| || D]U\} } | |kr | d s| d rtjdi| d nd} | | js@| | V| r'|t!|||| | t"d ||t'|t'| |S)a Process a single message and generate queue records based on configurations. Args: db_session: The database session message: The message payload peers_with_configuration: Dictionary of peer configurations session_id: Session ID configuration: Resolved configuration for this particular message Returns: List of queue records for this message rr_seq_in_sessionrrr)rrr_N)r=r[rlrr)r]r\r[z;message %s from %s created %s queue items with %s observersrN)rir r get_message_seq_in_sessionrnenabledmessages_per_short_summarymessages_per_long_summaryappendrprx reasoningitemsr rsobserve_othersrkr#debugr%)r4r2rHr[rZr]r_rlrecordsshould_observer\r peer_confsession_peer_configs r8rDrD%sd({#Hl+J!-=!>!>!C!DD""'+'F "#34 0! ( ( ( " " " " " " %'G |  !HHAMM !DL$J Ja O O !"%'=       .h8PQQN > !I("""%=$B$B$D$D ( ( IyH$$Q< >Gq\S)99IaL999t #*2E2T*   Y ' ' ' '   (!#%        LLE G  I  NrK)rtrigger_reason delay_reason&documents_since_last_dream_at_scheduledocument_thresholdobserver dream_typerrrrc ^t||||||||} t|| | dd|ddS)a Create a queue record for a dream task. Args: workspace_name: Name of the workspace observer: Name of the observer peer observed: Name of the observed peer dream_type: Type of dream to execute session_name: Name of the session to scope the dream to if specified trigger_reason: what tripped the schedule delay_reason: what governed when it fires documents_since_last_dream_at_schedule: count snapshot at schedule time document_threshold: DOCUMENT_THRESHOLD snapshot at schedule time Returns: Queue record dictionary with workspace_name and other fields )rr]rrrrrNdreamrd)rr) rrr]rrrrrr dream_payloads r8create_dream_recordrs]:)!%!/U-   M1OO (   rKc PKtd4d{V} t||||||||| } | d} ttttjjtjj| k} | | d{V} | r7t d||||j  dddd{VdSttttjtj| ktjdk}| |d{V}|r7t d||||j  dddd{VdSttt}| || gd{V| d{Vt d||||j nW#t&$rJ}td t*jjrd dl}||d}~wwxYw dddd{VdS#1d{VswxYwYdS) u Enqueue a dream task for immediate processing by the deriver. Does not touch collection.internal_metadata["dream"] — both guard fields are written atomically in process_dream on successful completion. Deduplication: If a dream with the same work_unit_key is already in-progress (has an ActiveQueueSession) or pending in the queue, the enqueue is skipped. Args: workspace_name: Name of the workspace observer: Name of the observer peer observed: Name of the observed peer dream_type: Type of dream to execute session_name: Name of the session to scope the dream to if specified dream_enqueueN)rr]rrrrrrrezASkipping dream enqueue - already in progress: %s/%s/%s (type: %s)Fz3Dream already pending in queue: %s/%s/%s (type: %s)z+Enqueued dream task for %s/%s/%s (type: %s)zFailed to enqueue dream task!r)r rrrr ActiveQueueSessionrEwhererescalarr#r$valuer processedrr'r(r)r*r+r r,r-r.r/)rrr]rrrrrrr4 dream_recordrein_progress_checkis_in_progress pending_check is_pendingr6r7r.s r8 enqueue_dreamrs=6/**IIIIIIIjH .!!%)-)7]#5   L)9M &64788>>1?=P!! $.#4#45F#G#GGGGGGGN  W"$ CIIIIIIIIIIIIIIF#9<((..!/=@!+u4M *00????????J  I"$ kIIIIIIIIIIIIIIn)$$..y99D$$TL>:: : : : : : : :##%% % % % % % % % KK=           < = = =& 0!!!!,,Q///    wIIIIIIIIIIIIIIIIIIIIIIIIIIIIIIsDJB>H*,B#H*#BH*)J* I>4AI99I>>J J"J deletion_type)r< observationrF resource_idcRt||}t|||dd|ddS)a9 Create a queue record for a deletion task. Args: workspace_name: Name of the workspace deletion_type: Type of resource to delete ("session" or "observation") resource_id: ID of the resource to delete Returns: Queue record dictionary for insertion into the queue )rrNdeletionrd)rr)rrrdeletion_payloads r8create_deletion_recordr$sM /# 1AQRR#(   rKcKdtdtddffd } |||dd{VdStd4d{V}||d d{Vdddd{VdS#1d{VswxYwYdS#t$rJ}td t jjrd dl }| |d}~wwxYw) a: Enqueue a deletion task for processing by the deriver. This function adds a deletion task to the queue for asynchronous processing. The deletion will be handled by the queue consumer with retry support. Args: workspace_name: Name of the workspace deletion_type: Type of resource to delete ("session" or "observation") resource_id: ID of the resource to delete db_session: Optional database session. If provided, uses this session instead of creating a new one. The caller is responsible for committing. r< should_commitrNc2Kt}ttt}|||gd{V|r|d{Vt ddS)Nz=Enqueued deletion task: type=%s, resource_id=%s, workspace=%s)rrrr'r(r)r#r$)r<rdeletion_recordr6rrrs r8 _do_enqueuez%enqueue_deletion.._do_enqueueWs0      i  **955ood_$5666666666  #.."" " " " " " " " K         rKF)rdeletion_enqueueTz Failed to enqueue deletion task!r) rboolr r*r#r+r r,r-r.r/)rrrr4r new_sessionr7r.s``` r8enqueue_deletionrCs( <          (  !+j>>> > > > > > > > > >""455 C C C C C C C!k+TBBBBBBBBBB C C C C C C C C C C C C C C C C C C C C C C C C C C C C C C ;<<< ? " ,      ( ( + + + sABB A2B2 A<<B?A<B CACC)N)NNNNN)7loggingtypingrr sqlalchemyrrrsqlalchemy.ext.asynciorsrcr r r src.configr src.dependenciesr src.dreamer.dream_schedulerrsrc.exceptionsr src.modelsr src.schemasrrsrc.utils.config_helpersrsrc.utils.queue_payloadrrrsrc.utils.work_unitr getLogger__name__r#listdictrgr9r&rBrkrirprrxrD DreamTyperrrrrNrKr8rs----------//////%%%%%%%%%%'''''';;;;;;...... CCCCCCCC666666 877777  8 $ $504S#X/50D50505050p77 $sCx. !77 7  $sCx. 7777t.1AD #tDcN# #$@"++ #s(^+ +d + Cy +  + #s(^++++\& #s(^&(&& &  #s(^ &&&&R##-1#tDcN7K2K-L# ####Lcc #s(^c#3T#s(^(<#<=c c c  $sCx. ccccX $!%#9=%)///// / ! / * /$J/*/-0$J/d / #s(^////n $!%#9=%)ddddd! d * d $J d*d-0$Jdd d ddddN@A #s(^ F'+ 777@A77t# 7  777777rK