yj%xUdZddlZddlZddlZddlmZmZmZddlZddlm Z ddl m Z m Z ddl mZddlmZddlmZdd lmZdd lmZejeZGd d e Zd ZeddejjeddedZee efe!d<da"de!d<ddZ#ddZ$GddZ%dS)a: Reconciler scheduler for self-healing background tasks. This module provides a scheduler for running reconciliation and cleanup tasks like vector store sync and soft-delete cleanup. It ensures only one task of each type runs at a time across multiple deriver instances by using the queue table for coordination. N)datetime timedeltatimezone) BaseModel)existsselect)IntegrityError)models)settings) tracked_db) QueueItemc2eZdZUdZeed<eed<eed<dS)ReconcilerTaskz Definition of a reconciler task.name work_unit_keyinterval_secondsN)__name__ __module__ __qualname____doc__str__annotations__int@/DATA/AppData/hermes/projects/honcho/src/reconciler/scheduler.pyrrs9** IIIrri sync_vectorszreconciler:sync_vectors)rrr cleanup_queuezreconciler:cleanup_queue)rrRECONCILER_TASKSReconcilerScheduler | None_reconciler_scheduler schedulerReconcilerSchedulerreturnc |adS)z.Set the global reconciler scheduler reference.Nr!)r"s rset_reconciler_schedulerr'9s&rctS)z.Get the global reconciler scheduler reference.r&rrrget_reconciler_schedulerr)?s  rceZdZUdZdZded<dZeed<fdZdZ e dd Z dd Z dd Z dd Zded efdZxZS)r#a Scheduler for self-healing reconciliation and cleanup tasks. Ensures only one task of each type runs at a time across multiple deriver instances by using the queue table for coordination. This provides: - Vector store synchronization (syncing pending documents/embeddings) - Soft-delete cleanup (removing soft-deleted documents from vector stores) - Self-healing behavior (retrying failed syncs) - Extensible task registry for adding new maintenance tasks Each task type has its own interval and work_unit_key, allowing multiple different tasks to be queued simultaneously while preventing duplicates of the same task type. Nr _instanceF _initializedcl|j&t||_|jS)N)r+super__new__)cls __class__s rr/zReconcilerScheduler.__new__Ws* = !GGOOC00CM}rctjs4d|_tj|_i|_dt_dSdS)NT)r#r,_scheduler_taskasyncioEvent_shutdown_event _next_runselfs r__init__zReconcilerScheduler.__init__\sC"/ 4>BD 29-//D 24DN/3  , , ,  4 4rr$c"d|_d|_dS)z5Reset the singleton instance. Only use this in tests.NF)r+r,)r0s rreset_singletonz#ReconcilerScheduler.reset_singletones  rc6K|jtddS|jt jtj}t D]%\}}|t|j z|j |<&tj||_tdt%tt'tdS)z$Start the reconciler scheduler loop.Nz#ReconcilerScheduler already runningsecondsz-ReconcilerScheduler started with %d tasks: %s)r3loggerwarningr6clearrnowrutcritemsrrr7r4 create_task_scheduler_loopinfolenlistkeys)r9rC task_nametasks rstartzReconcilerScheduler.startks   + NN@ A A A F ""$$$l8<((/5577 W WOIt(+i@U.V.V.V(VDN9 % %&243G3G3I3IJJ ;  ! ! !&&(( ) )     rcRK|jdStd|j t j|jdd{Vn#t j$rytd|j tj t j 5|jd{Vdddn #1swxYwYYnwxYwd|_|j tddS)zStop the reconciler scheduler.Nz$Shutting down ReconcilerScheduler...g@timeoutz7ReconcilerScheduler shutdown timed out, cancelling taskzReconcilerScheduler stopped)r3r@rHr6setr4wait_for TimeoutErrorrAcancel contextlibsuppressCancelledErrorr7rBr8s rshutdownzReconcilerScheduler.shutdown~sr   ' F :;;;   """ +"4#7EEE E E E E E E E E# + + + NNT U U U  ' ' ) ) )$W%;<< + +******** + + + + + + + + + + + + + + + + $  122222s7!A""A C*C C*C C*#C $C*)C*cjK |jstjtj}t D]\}}|j ||}||kr | |d{V}|rt d|nW#t$rJ}td|tjjrt%j|Yd}~nd}~wwxYw|t)|jz|j|<|jrht-|j}t1d|tjtjz }nd} t5j|j|d{VdS#t4j$rYnwxYw|jdSdS#t4j$rt dwxYw) z Main scheduler loop that enqueues tasks based on their intervals. Each task has its own interval and the loop checks all tasks on each iteration, enqueueing any that are due. NzEnqueued task: %szError enqueueing task %sr>g?gN@rPz"ReconcilerScheduler loop cancelled)r6is_setrrCrrDrrEr7get_try_enqueue_taskr@debug Exception exceptionr SENTRYENABLED sentry_sdkcapture_exceptionrrminvaluesmax total_secondsr4rSwaitrTrX) r9rCrLrMnext_runenqueuedenext_task_time sleep_secondss rrGz#ReconcilerScheduler._scheduler_loops- *1133( l8<00(8'='='?'?OIt#~11)SAAHh@-1-C-CD-I-I'I'I'I'I'I'IH'M & -@) L L L(@@@",,-GSSS'6@ * )%()>)>)@)@%A%AN$''(,x|*D*DDSSUU%%MM %)M!*,1133 -E+DM*1133( ( ( ( ( T%    LL= > > >  sVA7H<8B54H5 D ?AD?HD  BH3GG&#H%G&&H+H2rMc Ktd4d{V}ttttjjtjj|jk}||d{V}|r4t d|j  dddd{VdSttttjtj|jktj dk}||d{V}|r4t d|j  dddd{VdSt|jd|j idddd}|| |d{Vn\#t $rO|d{Vt d |j Ydddd{VdSwxYwtd |j  dddd{Vd S#1d{VswxYwYdS) aY Attempt to enqueue a reconciler task. This is idempotent - if the task is already in-progress (has an ActiveQueueSession) or pending in the queue, the enqueue is skipped. Args: task: The task definition to enqueue Returns: True if a task was enqueued, False if skipped reconciler_enqueueNz-Task %s already in progress, skipping enqueueFz2Task %s already pending in queue, skipping enqueuereconciler_type reconciler)rpayload session_id task_typeworkspace_name message_idz6Task %s already enqueued by another instance, skippingzEnqueued reconciler task: %sT)r rrr ActiveQueueSessionidwhererscalarr@r^rr processedaddcommitr rollbackrH)r9rMdbin_progress_checkis_in_progress pending_check is_pending queue_items rr]z%ReconcilerScheduler._try_enqueue_tasks2337 7 7 7 7 7 7 r &64788>>1?4CUU!! $&99->#?#???????N  LdiXXX7 7 7 7 7 7 7 7 7 7 7 7 7 7  #9<((..!/43EE!+u4M "yy77777777J  H$)=7 7 7 7 7 7 7 7 7 7 7 7 7 7 B#"0%ty &#   J FF:    iikk!!!!!!!!!   kkmm####### Ldii7 7 7 7 7 7 7 7 7 7 7 7 7 7 \  KK6 B B Bo7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 7 sDB"I( B&I(6I(=GI(AH1I(0H11$I(( I25I2)r$N)rrrrr+rr,boolr/r: classmethodr<rNrYrGrr] __classcell__)r1s@rr#r#Ds  /3I+222L$ 444!!![!     &3333(4444lDNDtDDDDDDDDr)r"r#r$N)r$r )&rr4rVloggingrrrrcpydanticr sqlalchemyrrsqlalchemy.excr srcr src.configr src.dependenciesr src.modelsr getLoggerrr@rQUEUE_CLEANUP_INTERVAL_SECONDS VECTOR_STORERECONCILIATION_INTERVAL_SECONDSrdictrrr!r'r)r#rrrrs 2222222222%%%%%%%%))))))''''''  8 $ $Y"+#N /!.N $^ 07 / /$sN*+   7;3:::&&&& !!!! HHHHHHHHHHr