6jL UdZddlZddlZddlZddlZddlZddlZddlZddlm Z ddl m Z ddl m Z ejeZdedefdZde fd Zdefd ZGd d Zdaedzed <ejZdZdZdedzfdZdS)ah Hermes Web UI -- Gateway session watcher. Background daemon thread that polls state.db every 5 seconds for changes to gateway sessions (telegram, discord, slack, etc.). When changes are detected, it pushes notifications to all subscribed SSE clients. This enables real-time session list updates in the sidebar without requiring any changes to hermes-agent. N)Path)HOME)"read_importable_agent_session_rowssessionsreturncddt|dD}tj|dS)zMCreate a lightweight hash of session IDs and timestamps for change detection.|c 3K|]<}|dd|ddd|ddV=dS) session_id: updated_atr message_countN)get).0ss )/root/hermes-webui/api/gateway_watcher.py z!_snapshot_hash..sl  \?QQQUU<33QQaeeOQ6O6OQQc|dS)Nr )xs rz _snapshot_hash.. s ,r)keyF)usedforsecurity)joinsortedhashlibmd5encode hexdigest)rrs r_snapshot_hashr!sm ((&?&?@@@  C ;szz||U ; ; ; E E G GGrc h ddlm}t|}nk#t $r^tt jdttdz }YnwxYw|dz S)z-Resolve state.db path for the active profile.r)get_active_hermes_home HERMES_HOMEz.hermeszstate.db) api.profilesr#r expanduserresolve Exceptionosgetenvstrr)r# hermes_homes r_get_state_db_pathr-'sc777777113344??AAIIKK ccc29]Cy8H4I4IJJKKVVXX``bb c  ##sAAA%B,+B,ct}|sgS g}t|dtD]}||d|dpd|dpd|dp |d pd |d |d p|d |d pd|d|d|dd |S#t $rgcYSwxYw)znRead all non-webui sessions from state.db. Returns list of session dicts, or empty list on any error. )limitlogidtitlez Agent SessionmodelNractual_message_countr started_at last_activitysourcecli raw_sourcesession_source source_label) r r3r4r created_atr r8r:r;r<)r-existsrloggerappendrr()db_pathrrows r_get_agent_sessions_from_dbrC1s*!""G >>   5gSfUUU  C OO!$iW8W-!$_!5!Y=S9T!YXY!,/!/2Gc,6Gh-05!ggl33"%''*:";"; # 7 7        sB8C C.-C.cveZdZdZdZdZdZdZdefdZ dZ de j fd Z d e j fd Zd efd ZdZdS)GatewayWatcheraBackground thread that polls state.db for agent session changes. Usage: watcher = GatewayWatcher() watcher.start() q = watcher.subscribe() # ... receive change events via q.get() ... watcher.unsubscribe(q) watcher.stop() cg|_tj|_tj|_d|_d|_g|_dS)N) _subscribers threadingLock _sub_lockEvent _stop_event_thread _last_hash_last_sessionsselfs r__init__zGatewayWatcher.__init__^sE/1"))$?,,04 !$&rc|jr|jrdS|jt j|jdd|_|jdS)z Start the watcher daemon thread.NTzgateway-watcher)targetdaemonname)rPis_aliverOclearrKThread _poll_loopstartrSs rr^zGatewayWatcher.startfsq < DL1133  F     'ttRcddd  rrc@|j}|duo|S)aReturn True when the poll thread is running. Public accessor used by ``/api/sessions/gateway/stream`` probe mode and the live SSE handler to detect a watcher instance whose poll thread died silently (e.g. uncaught exception in ``_poll_loop``). Callers use this to decide whether to return 503 and trigger the client-side polling fallback, instead of handing out an SSE connection that would never emit events. N)rPrZ)rTts rrZzGatewayWatcher.is_alivens" L}--rcb|j|j5|jD]B} |d#t $rt dY?wxYw dddn #1swxYwY|jr$|j dd|_dSdS)zStop the watcher thread.Nz%Failed to send sentinel to subscriber)timeout) rOsetrMrJputr(r?debugrPrrTqs rstopzGatewayWatcher.stop{s(  ^ J J& J JJEE$KKKK JJJLL!HIIIIIJ J J J J J J J J J J J J J J J J < L  a  ( ( (DLLL  s4 A9AA9$A)&A9(A))A99A=A=ctjd}|j5|j|dddn #1swxYwY|S)zSubscribe to change events. Returns a queue.Queue. Events are dicts: {'type': 'sessions_changed', 'sessions': [...]} A None sentinel means the watcher is stopping. )maxsizeN)queueQueuerMrJr@rgs r subscribezGatewayWatcher.subscribes K # # # ^ ( (   $ $Q ' ' ' ( ( ( ( ( ( ( ( ( ( ( ( ( ( (sAA Arhc|j5 |j|n#t$rYnwxYwddddS#1swxYwYdS)zRemove a subscriber queue.N)rMrJremove ValueErrorrgs r unsubscribezGatewayWatcher.unsubscribes ^   !((++++                      s,A%A 2A2AA Arc d|d}|j5g}|jD]b} ||#tj$r||Y?t $r||Y_wxYw|D]n} |j|n#t$rYnwxYw |dE#t $rt dYkwxYw ddddS#1swxYwYdS)z%Push change event to all subscribers.sessions_changed)typerNz*Failed to send sentinel to dead subscriber) rMrJ put_nowaitrmFullr@r(rqrrr?rf)rTreventdeadrhs r_notify_subscribersz"GatewayWatcher._notify_subscriberss'   ^ O OD& # ##LL''''z###KKNNNNN ###KKNNNNN# O O%,,Q////!DOLL&&&& OOOLL!MNNNNNO O O O O O O O O O O O O O O O O O O Os C90C9$A7C9A74C96A77C9?BC9 B'$C9&B''C9+CC9$C(%C9'C((C99C=C=c|js t}t|}||jkr#||_||_||n,#t$rt ddYnwxYwt|j dzD]2}|jrdStj d3|jdSdS)z+Main polling loop. Runs in a daemon thread.z"Error in gateway watcher poll loopT)exc_inforkNg?)rOis_setrCr!rQrRr{r(r?rfrange POLL_INTERVALtimesleep)rTr current_hash_s rr]zGatewayWatcher._poll_loops"))++  R688-h77 4?22&2DO*2D',,X666 R R R AD QQQQQ R4-233  #**,,FF 3!"))++     sA A''&BBN)__name__ __module__ __qualname____doc__rSUBSCRIBER_TIMEOUTrUr^boolrZrirmrnrorslistr{r]rrrrErEOs  M''' .$ . . . .    5;U[ODOOOO6     rrE_watcherct5t'tatddddS#1swxYwYdS)z.Start the global gateway watcher (idempotent).N) _watcher_lockrrEr^rrr start_watcherrs   %''H NN   s/AA Act5ttdaddddS#1swxYwYdS)z Stop the global gateway watcher.N)rrrirrr stop_watcherrs    MMOOOHs #8<<cRt5tcdddS#1swxYwYdS)z9Get the global watcher instance (or None if not started).N)rrrrr get_watcherrso s   )rrjsonloggingr)rmrKrpathlibr api.configrapi.agent_sessionsr getLoggerrr?rr+r!r-rCrEr__annotations__rLrrrrrrrrs      AAAAAA  8 $ $ HTHcHHHH$D$$$$T