.cj{xUdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZejeZeZdad ed <ejZdad ed <ejZ iZ!d ed<dZ"dZ#dZ$dZ%dZ&ejZ'dZ(d_dZ)d`dZ*dadZ+dbdZ,dcd"Z-dad#Z.ddd$Z/ddd%Z0ded'Z1dfd)Z2dgd-Z3dfd.Z4dfd/Z5dfd0Z6dhd1Z7dhd2Z8did4Z9djd7Z:ddd8Z;dkd9Zddz_connect..zsVVV#s1vVVVrz$PRAGMA table_info(async_delegations))) owner_pidINTEGER)owner_started_atr)) task_jsonTEXT)delivery_claimr,)delivery_claimed_atREAL)origin_session_idr,z)ALTER TABLE async_delegations ADD COLUMN  ) hermes_staterrparentmkdirsqlite3connectexecute)rpathconncolumnsnamesql_types r_connectr=\s444444 ::DKdT222 ?4 , , ,DD+HIIIILL .WV.T!U!UVVVG  X Xh w   LLVTVVHVV W W W KrrecordDict[str, Any]Nonectj} ddlm}|td}n#t $rd}YnwxYwfddD}t 5t5}|dd dd  d d  d d |td|tj | d d f dddn #1swxYwYdddn #1swxYwYtdS)Nr)get_process_start_timeoscDi|]}|v||Srget)r%keyr>s r z%_persist_dispatch..s6  &== VZZ__ ==r)goalgoalscontexttoolsetsrolemodelis_batcha~INSERT OR REPLACE INTO async_delegations (delegation_id, origin_session, origin_ui_session_id, parent_session_id, state, dispatched_at, updated_at, delivery_state, delivery_attempts, owner_pid, owner_started_at, task_json, origin_session_id) VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?, ?) delegation_id session_keyorigin_ui_session_idparent_session_id dispatched_atr0) timegateway.statusrB __import__getpid Exception_DB_LOCKr=r7rFjsondumps_prune_durable_records)r>nowrBr* task_payloadr9s` r_persist_dispatchras )++C 99999911*T2B2B2I2I2K2KLL     XL    8::    Q O $fjj&C&C ZZ. 3 3VZZ@S5T5T O $c:d+;+;+B+B+D+D tz,77 ZZ+R 0 0  2                                sH0A AA.E =B-D6* E 6D: :E =D: >E  EErPstrct5t5}|d|fdddn #1swxYwYddddS#1swxYwYdS)Nz3DELETE FROM async_delegations WHERE delegation_id=?)r[r=r7rPr9s r_delete_durable_delegationres ^^8::^ J]L\]]]^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^s.A; A? A? AAAcrtj}|tz }t5t5}|d|f|dd}t d|tz }|r|d|f|dd}t d|tz }|r|d|fdddn #1swxYwYddddS#1swxYwYdS)zBBound terminal history, preferring delivered records for deletion.zQDELETE FROM async_delegations WHERE delivery_state='delivered' AND updated_at < ?zRSELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing')raXDELETE FROM async_delegations WHERE delegation_id IN ( SELECT delegation_id FROM async_delegations WHERE state NOT IN ('running','finalizing') ORDER BY CASE delivery_state WHEN 'delivered' THEN 0 ELSE 1 END, updated_at ASC LIMIT ? )z~SELECT COUNT(*) FROM async_delegations WHERE state NOT IN ('running','finalizing') AND delivery_state='pending'aDELETE FROM async_delegations WHERE delegation_id IN ( SELECT delegation_id FROM async_delegations WHERE state NOT IN ('running','finalizing') AND delivery_state='pending' ORDER BY updated_at ASC LIMIT ? )N) rV_DURABLE_RETENTION_SECONDSr[r=r7fetchonemax_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)r_cutoffr9terminal_countexcess pending_countoverflows rr^r^s )++C - -F   8::  _ I    `  (**QQ)@@AA   LL       [   (**Q q-*>>??   LL     3                                 s5D,CD D,D D,D D,,D03D0eventresultc tj}t5t5}|d|dd|d||t j|t j||dfdddn #1swxYwYddddS#1swxYwYdS)NzUPDATE async_delegations SET state=?, completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?status completed completed_atrP)rVr[r=r7rFr\r])rqrrr_r9s r_persist_completionrws8 )++C   8::   (YYx - -uyy/M/Ms Z   6 2 2E/4J L                                     s5B=A/B% B=%B) )B=,B) -B==CCct5t5}|dtj|fdddn #1swxYwYddddS#1swxYwYdS)NzfUPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?)r[r=r7rVrds r_note_delivery_attemptrys   8::  t Y[[- (                                    s4A%*A  A% A A%A A%%A),A)c f ddlm}m}n#t$rYdSwxYwt j}d}t 5t 5}|d}|D]}|\ }}} } } } } }}d}| rE|t| }|r+| )|t| t| k}|r[tj |pd}iddd |d |d | d |pd d| d| dd d| dd| dd| dd| dd| ddt| dddddddd| d|i}dd|dd}|d||tj|tj||f|dz } dddn #1swxYwYdddn #1swxYwY|S) zEClassify records whose owning process disappeared as outcome unknown.r) _pid_existsrBaSELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, owner_started_at, task_json, origin_session_id FROM async_delegations WHERE state IN ('running','finalizing')FNz{}typeasync_delegationrPrQrSr0rRrTrIrJrKrLrMrNrOrtunknownsummaryerrorzLDelegation owner exited before recording a terminal result; outcome unknown.rUrv)rtrrzUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r$)rWr{rBrZrVr[r=r7fetchallrr\loadsrFboolr])r{rBr_ recoveredr9rowsr&rPrQ origin_ui parent_idrUpidstartedr+r0livetaskrqrrs rrecover_abandoned_delegationsrsFFFFFFFFF qq )++CI ''8::'|| Q  (**   C;> 9]KI} '9&7D L"{3s88,,LG/11#c((;;s7||KD :i/400D * ,;] { ,BI  $%6%<"  $Y  178L8L '** -6txx 7J7J DHHZ00 39$((6:J:J '** -7TXXj=Q=Q8R8R ) &/ g   1?  E!*dU7^TTF LL,c4:e,,dj.@.@-P     NIIA ''''''''''''''''''''''''''''''P sB H&F;H H&H H&H H&&H*-H*ctt5t5}|d}|D]H\}}t j|}t|trd|d<| |I dddn #1swxYwYdddn #1swxYwYt|S)uzEnqueue durable pending completions as fresh turns after process start. Every restored event is stamped ``restored=True`` (in-memory only — the stamp is added after the durable payload is deserialized and is never persisted). Restored events originate from a *previous* process, so no consumer in THIS process implicitly owns them: drain paths that run without an ownership filter (the legacy single-session behavior) must leave them queued for a consumer that can positively prove ownership, otherwise a brand-new session adopts a dead session's delegation results seconds after boot (#64484). zSELECT delegation_id, event_json FROM async_delegations WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL ORDER BY completed_at, delegation_idTrestoredN) rr[r=r7rr\r isinstancedictputlen) target_queuer9r_delegation_idpayloadevts rrestore_undelivered_completionsrsV"###  " "8:: "|| 7   (** (, " " #NG*W%%C#t$$ '"&J   S ! ! ! !  " " " " " " " " " " " " " " " " " " " " " " " " " " " " " " " t99s5B<A3B% B<%B) )B<,B) -B<<CCrctj}t5t5}|d|||f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)zDAtomically acknowledge successful injection of a durable completion.zUPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=? WHERE delegation_id=? AND delivery_state!='delivered'r$NrVr[r=r7rowcount)rPr_r9curs rmark_completion_deliveredr7s) )++C !!8::!ll H #} %   |q !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!4A?$A' A?'A+ +A?.A+ /A??BBclaim_idc tj}t5t5}|d|f}| dddddddS|d|||||dz f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)zBClaim one pending completion across competing consumers/processes.zBSELECT delivery_state FROM async_delegations WHERE delegation_id=?NTaUPDATE async_delegations SET delivery_claim=?, delivery_claimed_at=?, delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND (delivery_claim IS NULL OR delivery_claimed_at < ?)i,r$)rVr[r=r7rhr)rPrr_r9r&rs rclaim_completion_deliveryrCs )++C !!8::!ll P     (**  ; !!!!!!!!!!!!!!ll LsCc :   |q !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!s@C-B0 C/(B0 C0B4 4C7B4 8CC C rconsumer Optional[str]c.|ddkrdSt|dpd}|sdS|dtddt jj}t||r|ndS)zCClaim a durable delegation event; non-durable events need no token.r|r}rRrP:rCN)rFrbrXrYuuiduuid4hexr)rrrPrs rclaim_event_deliveryrWs wwv,,,r006B77M rKKZ--4466KK9IKKH0II S88tSrc tj}t5t5}|d|||tf}|jdkr:t d|t dddddddS|d|||f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)uRelease a failed delivery claim so another consumer may retry. Attempts are counted at claim time, so a row that keeps being claimed and released has burned real delivery attempts. Once the budget is exhausted the row converges to a terminal ``dropped`` state instead of returning to ``pending`` — otherwise an undeliverable completion replays on every gateway restart forever (restore_undelivered_completions only restores pending rows). a UPDATE async_delegations SET delivery_state='dropped', delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=? AND delivery_attempts>=?r$znAsync delegation %s exhausted its %d delivery attempts; marking terminally dropped (result remains queryable).NTzUPDATE async_delegations SET delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?)rVr[r=r7_MAX_DELIVERY_ATTEMPTSrloggerwarning)rPrr_r9cappedrs rrelease_completion_deliveryrbs )++C !!8::! B-+A B    ?a   NNI5    !!!!!!!!!!!!!!ll )- *   |q -!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!sAC#A C 7 C##C 2 C# C C#C C##C'*C'ctj}t5t5}|d|||f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)uTerminally drop a claimed completion that can never be delivered. Used when the delivery target is permanently gone — the spawning session ended at an explicit user boundary (/new, reset) rather than a compression rotation. Marking the row ``dropped`` (not ``delivered``) keeps the ack honest, and (not ``pending``) keeps restart recovery from replaying a completion that will be fail-closed dropped again every time. aUPDATE async_delegations SET delivery_state='dropped', updated_at=?, delivery_claim=NULL, delivery_claimed_at=NULL WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?r$NrrPrr_r9rs rdrop_completion_deliveryrs, )++C  ! !8:: !ll ) - *   |q  ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !rc tj}t5t5}|d||||f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)z;Acknowledge acceptance for the consumer holding this claim.aUPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=?, delivery_claim=NULL, delivery_claimed_at=NULL WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?r$Nrrs rcomplete_completion_deliveryrs, )++C  ! !8:: !ll ) #}h /   |q  ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s4B%A( B(A, ,B/A, 0BBBc|rM|ddkr6tt|dpd|dSdSdSNr|r}rPrR)rFrrbrrs rcomplete_event_deliveryrsaTCGGFOO'999$S)A)A)GR%H%H(SSSSSTT99rc|rM|ddkr6tt|dpd|dSdSdSr)rFrrbrs rrelease_event_deliveryrsaSCGGFOO'999#C(@(@(FB$G$GRRRRRSS99rOptional[Dict[str, Any]]c t5t5}|d|f}dddn #1swxYwYdddn #1swxYwY|dS||d|d|d|d|drt j|dnd|d|d|d pd d S) NzSELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts, origin_session_id FROM async_delegations WHERE delegation_id=?rr$rrR) rPorigin_sessionstaterUrvrrdelivery_statedelivery_attemptsr0)r[r=r7rhr\r)rPr9r&s rget_durable_delegationrsV 8::ll ?BO@P   (**  {t&#a&3q6QQ(+A8$*SV$$$Da&s1v V\r   s4A$*A  A$ A A$A A$$A(+A( max_workersrct5t |tkrt|da|atcdddS#1swxYwYdS)uLazily create (or grow) the shared daemon executor. We never shrink — ThreadPoolExecutor can't resize — but if the configured cap grows between calls we rebuild a larger pool. Existing in-flight futures keep running on the old pool until it's garbage collected. Nzasync-delegate)rthread_name_prefix)_executor_lockrr_DaemonThreadPoolExecutor)rs r _get_executorrs    .C C C1'#3I%0 !s,AAAct5tdtDcdddS#1swxYwYdS)z.Number of async delegations currently running.c3HK|]}|ddvdVdS)rt>running finalizingr$NrEr%rs r zactive_count..s7``155??F_3_3_13_3_3_3_``rN) _records_locksumrvaluesrrr active_countrs aa``hoo//`````aaaaaaaaaaaaaaaaaas0AA  A cHdtjjddS)Ndeleg_r)rrrrrr_new_delegation_idrs" *DJLL$RaR( * **rc.dtD}t|tkrdS|d|dt|tz D] \}}t|d!dS)ziDrop the oldest completed records beyond the retention cap. Caller must hold ``_records_lock``. cNg|]"\}}|ddk||f#SrtrrE)r%ridrs r z+_prune_completed_locked..s? C 55??i ' ' a ' ' 'rNcr|ddp|ddpdS)Nr$rvrUrrE)kvs rz)_prune_completed_locked..s1"Q%))N";";"^r!uyy?Y?Y"^]^r)rG)ritemsrrjsortpop)rur_s r_prune_completed_lockedrs nn&&I  9~~000 NN^^N___Fc)nn/FFFG  Q S$  rcr ddlm}|dddkrdS|ddpdS#t$rYdSwxYw)uRaw session id of the ORIGINATING api_server request, or ``""``. The obvious source — ``HERMES_SESSION_ID`` via ``get_session_env`` — is NOT safe to read at dispatch time: constructing a child agent (``agent/agent_init.py``) calls ``set_current_session_id(child.session_id)``, clobbering that ContextVar *and* ``os.environ`` with the subagent's internal ``{timestamp}_{uuid}`` id moments before the dispatch code reads it, so the completion wake would self-post into the subagent's own (unread) session instead of the spawner's. The request-scoped ``HERMES_SESSION_CHAT_ID`` binding survives child construction: ``_bind_api_server_session`` binds ``chat_id`` to the raw ``X-Hermes-Session-Id``, and its only writer is ``set_session_vars`` — ``set_current_session_id`` never touches it. Gate on the platform: on push platforms ``chat_id`` is a chat, not a session, so yield ``""`` there. rget_session_envHERMES_SESSION_PLATFORMrR api_serverHERMES_SESSION_CHAT_ID)gateway.session_contextrrZrs r_current_origin_session_idrsp$;;;;;; ?4b 9 9\ I I27<<BB rrs( ( 66rR)rTrSr0 interrupt_fnmax_async_childrenrIrKrLOptional[List[str]]rMrNrQrTrunnerCallable[[], Dict[str, Any]]rSr0rOptional[Callable[[], None]]rc 6ttj|||rt|nd||||| |dd| d} t5t dt D} | | krdd| ddcdddS| t <dddn #1swxYwYt| t| }dfd } | t|nh#t$r[}t5t ddddn #1swxYwYtdd |dcYd}~Sd}~wwxYwtd |pd|pdddddS)a Spawn ``runner`` on the daemon executor and return a handle immediately. Parameters ---------- goal, context, toolsets, role, model The dispatch-time task spec, captured verbatim for the rich completion block. session_key The gateway session_key (from ``tools.approval.get_current_session_key``) captured on the parent thread BEFORE dispatch, because the daemon worker thread won't carry the contextvar. Used to route the completion back to the originating session. parent_session_id The durable ``state.db`` session id of the parent agent that spawned the delegation. Carried on the completion event so the gateway can pin routing to the spawning session instead of recovering the latest ``ended_at IS NULL`` row for the peer tuple (#57498). runner Zero-arg callable that builds + runs the child and returns the same result dict ``_run_single_child`` produces. Runs on the worker thread. interrupt_fn Optional callable to signal the child to stop (used on shutdown / explicit cancel). max_async_children Concurrency cap. When at capacity the dispatch is REJECTED (the caller should fall back to sync or tell the user) rather than queued, so a runaway model can't pile up unbounded background work. Returns ------- dict ``{"status": "dispatched", "delegation_id": ...}`` on success, or ``{"status": "rejected", "error": ...}`` when at capacity. Nr)rPrIrKrLrMrNrQrSr0rTrtrUrvrc3LK|]}|ddkdV dSrtrr$NrErs rrz,dispatch_async_delegation..Z?  AEE(OOy,H,HA,H,H,H,H  rrejected#Async delegation capacity reached (z running). Wait for one to finish (its result will re-enter the chat), or run this task synchronously (background=false). Raise delegation.max_concurrent_children in config.yaml to allow more concurrent background subagents.rtrrr@c i}d} pi}|dpd}nu#t$rh}tdddt |jd|dt tjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) NrrtruzAsync delegation %s crashed: rr)rtrr api_callsduration_seconds) rFrZr exceptionr|__name__roundrV _finalize)rrrtexcrPrUrs r_workerz*dispatch_async_delegation.._workerms!# 5VXX^FZZ))8[FF      :M J J J! II.77#77$)$)++ *Eq$I$I FFFFFFF  mVV 4 4 4 4 4ImVV 4 4 4 4s,#+B3 BABB3BB33Cz%Failed to schedule async delegation: z3Dispatched async delegation %s (session_key=%s): %srRP dispatchedrtrPrr@)rrVlistrrrrrarsubmitr rZrrerinfo)rIrKrLrMrNrQrTrrSr0rrr>rexecutorrrrPrUs ` @@rdispatch_async_delegationrsb'((MIKKM&&.8DNNND" 4..&$F& ))  ((      ( ( ($R:LRRR   ))))))))#))))))))))))))))"f/00H55555555&   3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 BSBB          KK={-g CRC/@#] C CCsZ?B0 B00B47B4""D E*E%D>2 E%>E E%E E%E*%E*rtct5t|}| ddddSd|d<tj|d<d|d<t |}dddn #1swxYwYt |||t5t|}|||d<t ddddS#1swxYwYdS)zDMark a record complete and push the completion event onto the queue.Nrrtrvr)rrrFrVr_push_completion_eventr)rPrrrtr> event_records rrrsz  $ $m,, >  $ $ $ $ $ $ $ $(x!%~!%~F||  $ $ $ $ $ $ $ $ $ $ $ $ $ $ $<888 ""m,,  %F8 !!! """"""""""""""""""s(A./A..A25A20CCCc 6 ddlm}nG#t$r:}td|d|Yd}~dSd}~wwxYw|d}|d}|dpt j}|d pt j}id d d|dd |d d d|dd d|dd d|dd|dd d|dd|dd|dd|dp|dd|d|d|d|ddd|dt||z dd|||dd} t| | |j | dS#t$r:}td|d|Yd}~dSd}~wwxYw)zPush a type='async_delegation' event onto the shared completion queue. Best-effort: a failure here must not crash the worker, but it WOULD mean a silently-lost result, so we log loudly. rprocess_registryzPAsync delegation %s finished but process_registry import failed; result lost: %srPNrrrUrvr|r}rQrRrSr0rTrIrKrLrMrNrtrrr exit_reason)rvrzHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryrrZrrrFrVrrwcompletion_queuer) r>rrrtrrrrrUrvrs rrrs&;;;;;;;    JJ ' '    jj##G JJw  EJJ//>49;;M::n--<L " O44  vzz-44    +A2 F F   VZZ(;R@@  VZZ(;<<   62&&  6::i((  FJJz**   6""  G$$; 7(;(;  &  7  ! " VZZ Q//# $ FJJ l]&BA F F  % * + ,%zz-00/   C2V$$$ )--c22222       JJ ' '          s, A /AA 8I J/JJ)rTrSr0rrrPrJ List[str]c   p t tjt|} | dkr|dn$| ddd|Dz}id d|dt |d |d |rt |nd d |d |d|d|d| d|ddddd d| dd}t 5t dtD}|| krdd| ddcd d d S|t <d d d n #1swxYwYt|t| }d& fd } | t|nh#t$r[}t 5t d d d d n #1swxYwYt dd!|dcYd }~Sd }~wwxYwt d" | |pd#d$ d%S)'u+Dispatch a WHOLE fan-out batch as ONE background unit. Unlike ``dispatch_async_delegation`` (which backs a single subagent), ``runner`` here runs the entire batch — it builds and joins on every child in parallel and returns the combined ``{"results": [...], "total_duration_seconds": N}`` dict that the synchronous path would have returned. We occupy ONE async slot for the whole batch (the in-batch parallelism is bounded separately by ``max_concurrent_children``), so a single ``delegate_task`` fan-out never exhausts the async pool by itself. When the batch finishes, a SINGLE completion event is pushed onto the shared ``process_registry.completion_queue`` carrying the full per-task ``results`` list, so the consolidated summaries re-enter the conversation as one message once every child is done — the chat is never blocked while they run. Returns ``{"status": "dispatched", "delegation_id": ...}`` on success or ``{"status": "rejected", "error": ...}`` when the async pool is at capacity. r$rz parallel subagents: z; c3*K|]}|ddVdS)N(r)r%gs rrz2dispatch_async_delegation_batch..s+G^G^ST#2#G^G^G^G^G^G^rrPrIrJrKrLNrMrNrQrSr0rTrtrrUrvrrOTc3LK|]}|ddkdV dSrrErs rrz2dispatch_async_delegation_batch..#rrrrz running). Wait for one to finish (its result will re-enter the chat), or raise delegation.max_concurrent_children in config.yaml to allow more concurrent background units.rrr@ci}d} pi}|dpg}|rtd|Drd}nd}ns#t$rf}tdgt |jd|ttjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nrresultsc3DK|]}|ddvVdS)rt)rusuccessNrErs rrzCdispatch_async_delegation_batch.._worker..<sF%%x(@@%%%%%%rruz!Async delegation batch %s crashedrr)r rtotal_duration_seconds) rFallrZrrr|rrrV_finalize_batch)combinedrt child_resultsrrPrUrs rrz0dispatch_async_delegation_batch.._worker5s5#% =vxx~2H$LL339rM %%%&%%%"" %!$      @- P P P II.77#77*/ m0KQ*O*OH FFFFFF  M8V < < < < rrrrrUs ` ` @rdispatch_async_delegation_batchr+sH"9%7%9%9MIKKM E AFFa1 ; ; ;diiG^G^X]G^G^G^>^>^ ^  e 7  h8DNNND     {  4 . . )     D!F$ ))  ((      ( ( ($N:LNNN ))))))))#)))))))))))))))) f/00H========2  3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 H3HH          KKKq+0#] C CCsZ?D" D""D&)D&"E77 GGF0$ G0F4 4G7F4 8GGGr&ct5t|}| ddddSd|d<tj|d<d|d<t |}dddn #1swxYwY ddlm}n4#t$r'}t d||Yd}~dSd}~wwxYw|d ptj}|dptj}id d d |d |d dd|ddd|ddd|dd|ddd|dd|dd|dd|dd|dd|ddd|dpgd|dd|d|d||d} t| | |j | n3#t$r&}t d||Yd}~nd}~wwxYwt5t|}|||d<tddddS#1swxYwYdS#t5t|}|||d<tdddw#1swxYwYwxYw) zDMark a batch record complete and push ONE combined completion event.NrrtrvrrrzVAsync delegation batch %s finished but process_registry import failed; result lost: %srUr|r}rPrQrRrSr0rTrIrJrKrLrMrNrOTr live_transcriptsrr#)r#rUrvzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s)rrrFrVrrrrZrrrwrrr) rPr&rtr>rrrrUrvrs rr%r%as $$m,, > $$$$$$$$(x!%~!%~F|| $$$$$$$$$$$$$$$;;;;;;;   & 3    !$$_55DM##N33Bty{{L "   |'' r::   0 01G L L   \--.A2FF   \--.ABB      ,,  !!'**  <##I..  L$$Z00     ((  !!'**  &  D " 8<< **0b# * HLL);<<+ , g&&- .#+,,/G"H"H&$3   C6X&&& &)--c2222       3          & &\\-00F!#)x # % % %  & & & & & & & & & & & & & & & & & &] & &\\-00F!#)x # % % %  & & & & & & & & & & & & & & & &sA./A..A25A2:B B2 B--B2!I<;L< J,J'"L'J,,L60K33K7:K7M0M8 MM M M MList[Dict[str, Any]]ct5dtDcdddS#1swxYwYdS)zSnapshot of async delegations (running + recently completed). Safe to call from any thread. Excludes the non-serialisable interrupt_fn. cJg|] }d|D!S)c&i|]\}}|dk ||S)rr)r%kvs rrHz5list_async_delegations...s( ? ? ?da1+>+>Q+>+>+>r)rrs rrz*list_async_delegations..s?    @ ?aggii ? ? ?   rN)rrrrrrlist_async_delegationsr4s     __&&                     s #8<<shutdownreasoncd}t5dtD}dddn #1swxYwY|D]}}|d}t |rW ||dz }7#t $r9}t d|d|Yd}~ud}~wwxYw~|rt d|||S) a*Signal every running async delegation to stop. Returns how many. Used on ``/stop`` and gateway shutdown so a dangling background subagent can't keep burning tokens with no one listening. The child still emits a completion event (status='interrupted') via the normal finalize path. rcDg|]}|ddk|SrrErs rrz!interrupt_all..s3   AEE(OOy,H,HA,H,H,HrNrr$z&interrupt_all: %s interrupt failed: %srPz'Interrupted %d async delegation(s) (%s) rrrrFcallablerZrdebugr )r6counttargetsrfnrs r interrupt_allr?se E     ((                     UU> " " B<<       <EE/**C    N =ufMMM Ls%$:>>.A>> C/B<<C session_endcsssdSd}t5fdtD}dddn #1swxYwY|D]}}|d}t |rW ||dz }7#t $r9}t d|d|Yd}~ud}~wwxYw~|rt d|||S) u{Signal running async delegations owned by ONE session to stop. A delegation's lifecycle is bound to the session that spawned it: when that session ends, its in-flight background subagents must end with it — a completed orphan would otherwise sit on the shared completion queue with no live owner, either leaking into another chat or burning tokens with no one listening (#55578). Selectors (any matching field claims the record): - ``origin_ui_session_id``: the live TUI tab/window that commissioned it. - ``session_key``: the durable routing key captured at dispatch. - ``parent_session_id``: the spawning agent's durable session-db id — the right selector for gateway chats, whose ``session_key`` (the platform conversation key) SURVIVES a ``/new`` reset while the session id rotates. Returns how many were interrupted. rcBg|]}|ddkrr(t|dpdksTr(t|dpdks*r*t|dpdk|S)rtrrSrRrQrT)rFrb)r%rrSrTrQs rrz)interrupt_for_session..s   uuX)++%,*-aee4J.K.K.Qr*R*RVj*j*j+k$'m(<(<(B$C$C{$R$R%%S*-aee4G.H.H.NB*O*OSd*d*d  +e*d*drNrr$z.interrupt_for_session: %s interrupt failed: %srPz:Interrupted %d async delegation(s) for ending session (%s)r9) rQrSrTr6r<r=rr>rs ``` rinterrupt_for_sessionrCs0 3 " " B<<       DEE/**C      H 6    Ls)(A  A A =B  C/C  Cct5ttddadadddn #1swxYwYt5t ddddS#1swxYwYdS)z6Test-only: clear all state and tear down the executor.NF)waitr)rrr5rrrclearrrr_reset_for_testsrGs ""    E  * * * ! """"""""""""""" s';?? A44A8;A8)rr)r>r?rr@)rPrbrr@r )rqr?rrr?rr@)rr)rPrbrr)rPrbrrbrr)rr?rrbrr)rr?rrbrr@)rPrbrr)rrrr)rrb)rIrbrKrrLrrMrbrNrrQrbrTrrrrSrbr0rbrrrrrr?)rPrbrrr?rtrbrr@)r>r?rrr?rtrbrr@)rJrrKrrLrrMrbrNrrQrbrTrrrrSrbr0rbrrrrrPrrr?)rPrbr&r?rtrbrr@)rr.)r5)r6rbrr)rRrRrRr@) rQrbrSrbrTrbr6rbrr)H__doc__ __future__rr\loggingr5 threadingrVrconcurrent.futuresrtypingrrrrr hermes_constantsr tools.daemon_poolr tools.thread_contextr getLoggerrrrr__annotations__Lockrrrr_DEFAULT_MAX_ASYNC_CHILDRENrjrgrkrr[rr=rarer^rwryrrrrrrrrrrrrrrrrrrrr+r%r4r?rCrGrrrrUs!!!F#"""""  11111166666666666666,,,,,,666666<<<<<<  8 $ $ 5+/ ....!!   ')((((-  9>  ***----`:^^^^ $$$$N         0000f6 ! ! ! !!!!!(TTTT!!!!!!!!H!!!!, ! ! ! !TTTT SSSS &&aaaa ++++    $H(, "159~D~D~D~D~D~DB"""",8 8 8 8 F(, "159#'yDyDyDyDyDyDxB&B&B&B&J     : " 55555p      r