yj1<UddlZddlZddlmZmZddlmZddlZddlmZm Z m Z ddl m Z ddl mZddlmZddlmZdd lmZdd lmZmZeeZdad ed <ddZddZGddZde dejde fdZ!dS)N)datetimetimezone) getLogger)existsfuncselect) AsyncSession)models)settings) tracked_db) DreamType)construct_work_unit_keyparse_work_unit_keyDreamScheduler | None_dream_schedulerdream_schedulerDreamSchedulerreturnc |adS)z)Set the global dream scheduler reference.Nr)rs C/DATA/AppData/hermes/projects/honcho/src/dreamer/dream_scheduler.pyset_dream_schedulerrs'ctS)z)Get the global dream scheduler reference.rrrget_dream_schedulerrs rceZdZUdZded<dZeed<fdZdZe dd Z ddddd d e d e d e de de de de dzde dzde dzde dzddfdZd e defdZd e de dee fdZddddd d e d e d e de de de de dzde dzde dzde dzddfdZddddd d e de de de de dzde dzde dzde dzddfdZddZxZS)rNr _instanceF _initializedcl|j&t||_|jSN)rsuper__new__)cls __class__s rr#zDreamScheduler.__new__%s* = !GGOOC00CM}rcHtjsi|_dt_dSdS)NT)rrpending_dreams)selfs r__init__zDreamScheduler.__init__*s-* /ACD *.N ' ' ' / /rrc"d|_d|_dS)z5Reset the singleton instance. Only use this in tests.NF)rr)r$s rreset_singletonzDreamScheduler.reset_singleton0s  r)trigger_reason delay_reason&documents_since_last_dream_at_scheduledocument_threshold work_unit_keyworkspace_name delay_minutes dream_typeobserverobservedr,r-r.r/cKtjjsdSd{Vt j|||||||| |  } | j<| fddS)zSchedule a dream for a collection after a delay. telemetry kwargs are captured at schedule time and threaded through the queue payload so DreamRunEvent can attribute the dream back to its scheduling context. Nr4r5r,r-r.r/c:jdSr!)r'pop)tr(r0s rz/DreamScheduler.schedule_dream.._s)<)@)@PT)U)Ur) r DREAMENABLED cancel_dreamasyncio create_task_delayed_dreamr'add_done_callback) r(r0r1r2r3r4r5r,r-r.r/tasks `` rschedule_dreamzDreamScheduler.schedule_dream6s(~%  F ........."   !!-)7]#5     .2M* UUUUUVVVVVrcK||jvrn|j|}|tjt j5|d{Vdddn #1swxYwYdSdS)z>Cancel a pending dream. Returns True if a dream was cancelled.NTF)r'r9cancel contextlibsuppressr?CancelledError)r(r0rCs rr>zDreamScheduler.cancel_dreamas D/ / /&**=99D KKMMM$W%;<<                  4us A--A14A1cKt}g}|jD]<}t|}|j|kr |j|kr||=|D]2}||d{Vr||3|S)a Cancel all pending dreams where the observed peer matches. This handles both self-observation (observer=observed) and peer-to-peer observation (observer!=observed) dreams. Args: workspace_name: The workspace to match observed: The observed peer name to match Returns: Set of work_unit_keys that were cancelled N)setr'rr1r5appendr>add)r(r1r5 cancelledkeys_to_cancelr0parsedkeys rcancel_dreams_for_observedz)DreamScheduler.cancel_dreams_for_observedls "ee %'!0 5 5M(77F$666?h;V;V%%m444" # #C&&s++++++++ # c"""rc K tj|dzd{V|||||||| | d{Vtd|dS#tj$rtd|YdSt$rR} td|| tj j rtj | Yd} ~ dSYd} ~ dSd} ~ wwxYw)N<r7zExecuted dream for %szDream task cancelled for %sz!Error in delayed dream for %s: %s)r?sleep execute_dreamloggerinforIdebug Exceptionerrorr SENTRYr= sentry_sdkcapture_exception) r(r0r1r2r3r4r5r,r-r.r/es rrAzDreamScheduler._delayed_dreams\ 0-  233 3 3 3 3 3 3 3$$!!-)7]#5%          KK/ ? ? ? ? ?% G G G LL6 F F F F F F 0 0 0 LL>>>>>455       v344O2nDO,8O,8O)Z7  &/499;;<<q "$4000000L dnddxdd(ddd%              (!,,> -G#00N0SSSSSSSSI--dGYGGM &.  Z>ZZLZZZ?              6 7                           Bm !%)%3Y1    sD G#A6G## G-0G-cjK|jrtdt|jd|jD]}|t j|jddid{V|jdSdS)z*Cancel all pending dreams during shutdown.z Cancelling z pending dreams...return_exceptionsTN) r'rWrXlenvaluesrFr?gatherclear)r(rCs rshutdownzDreamScheduler.shutdowns   ( KKRc$*=&>&>RRR S S S+2244   .$"5"<"<">">WRVWW W W W W W W W   % % ' ' ' ' '  ( (r)rN)__name__ __module__ __qualname__r__annotations__rboolr#r) classmethodr+strintr rDr>rKrRrArVr __classcell__)r%s@rrr!s)-I&---L$ /// !!![!&*#'=A)-)W)W)W)W)W )W  )W)W)Wd )WDj)W14d )W $J)W )W)W)W)WV      !-0 SR&*#'=A)-"0"0"0"0"0 "0  "0"0"0d "0Dj"014d "0 $J"0 "0"0"0"0V&*#'=A)-< < < < <  <  < d < Dj< 14d <  $J<  < < < < |((((((((rry collectionc KtjjsdSjdi}|dd}|d}t t jtj j  tj j j ktj j j ktj jjktj jdk}t!||d{Vpd}||z }t$dj j j|||tjjd  |tjjkrd }tjjdkrd nd } |r t-j|} t-jt2j| z dz } | tjjkrGt$dj djd| dddtjjdzdSn>#t<t>f$r*} t$ d|d| Yd} ~ nd} ~ wwxYwtjj!} fd| D}|t tEt tj#j  tj#j$dktj#j%dktj#j&'|d{V}|r(t$dj jdStQ}|r| D]}tSj dj j|d}|*|j tjjtW|j j|| |tjj d{Vt$dj j j|tjj|d dSdS)u` From the moment a dream is scheduled until it completes or fails, no second dream may be enqueued for the same (workspace, observer, observed) — and the baseline count advances only when consolidation actually happened. Check if a collection has reached the explicit-observation threshold and schedule a timer-based dream. This function only schedules a timer-based dream if: 1. Dreams are enabled 2. Explicit-observation threshold is reached (dreamer output does not count) 3. Minimum hours between dreams have passed 4. No dream is already pending in the queue for this collection (in-flight check) 5. No dream is already scheduled for this collection Args: db: Database session collection: Collection model to check Returns: True if a dream timer was scheduled, False otherwise Frwlast_dream_document_countr last_dream_atreNz Dream check)r1r4r5current_explicit_countrdocuments_since_last_dreamr/)extrar/ idle_timeout immediateizSkipping dream for rgz: only z.1fz hours zsince last dream (minimum: )z!Invalid last_dream_at timestamp: z , error: c Xg|]&}tjdjj|d'S)rw task_typer4r5r3)rr1r4r5).0r3rs r z,check_and_schedule_dream..RsW     $)!( * 3 * 3",       rzASkipping dream schedule for %s/%s: pending dream already in queuer)r3r4r5r,r-r.r/zScheduled dream)r1r4r5rr/r3T),r r<r=internal_metadatagetrrcountr rlidrmr1r4r5rnrrsrWrYDOCUMENT_THRESHOLDIDLE_TIMEOUT_MINUTESr fromisoformatnowrutc total_secondsMIN_HOURS_BETWEEN_DREAMSrX ValueError TypeErrorrt ENABLED_TYPESr QueueItemr processedr0in_rrrDr )ryrdream_metadatarr count_stmtrrr,r-last_dream_timehours_since_last_dreamr_enabled_dream_types pending_keyspending_existsrr3dream_work_unit_keys ` rcheck_and_schedule_dreamrs2 > !u155grBBN . 2 23NPQ R R"&&77M 6?#56677==&**CC J$77 J$77+ J !ryy'<'rs''''''''++++++++++//////''''''!!!!!!LLLLLLLL 8  -1)000''''  T(T(T(T(T(T(T(T(n^^!^ ^^^^^^r