"Yjؖ^UdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZejeZeZdad ed <ejZdad ed <ejZ iZ!d ed<dZ"dZ#dZ$dZ%ejZ&dZ'dZdZ(d[dZ)d\dZ*d]dZ+d^d!Z,d\d"Z-d_d#Z.d_d$Z/d`d&Z0dad(Z1dbd,Z2dad-Z3dad.Z4dcd/Z5dcd0Z6ddd2Z7ded5Z8d_d6Z9dfd7Z:d]d8Z;dd9de"d:dgdIZdd9de"d:djdOZ?dkdQZ@dldSZAdmdndVZB dodpdXZCd]dYZDdS)qu Async (background) delegation registry. Backs ``delegate_task(background=true)``: the parent agent dispatches a subagent that runs on a module-level daemon executor and returns a handle immediately, so the user and the model can keep working while the child runs. When the child finishes, a completion event is pushed onto the SHARED ``process_registry.completion_queue`` with ``type="async_delegation"``. The CLI (``cli.py`` process_loop) and gateway (``_run_process_watcher`` / ``completion_queue`` drain) already poll that queue while the agent is idle and forge a fresh user/internal turn from each event. We deliberately reuse that rail rather than reaching into a running agent loop: - completions surface as a NEW turn when the agent is idle, never spliced between a tool result and an assistant message. That keeps strict message-role alternation legal and the prompt cache intact (hard invariant: never mutate past context). - we inherit the queue's de-dup, crash-recovery checkpoint, and the existing CLI + gateway drain wiring for free — no new drain loops in the two largest files in the repo. The completion payload carries a RICH, self-contained task-source block (the original goal, the context the parent supplied, toolsets, model, dispatch time, status, and the full result summary). When the result re-enters the conversation the parent may be deep in unrelated context and won't remember why the subagent existed; the block lets it either use the result or re-dispatch if the world has moved on. This module owns ONLY the async lifecycle. The actual child build + run is delegated back to ``delegate_tool._run_single_child`` via an injected runner, so all the credential leasing, heartbeat, timeout, and result-shaping logic stays in one place. ) annotationsN)ThreadPoolExecutor)AnyCallableDictListOptionalget_hermes_home)DaemonThreadPoolExecutor)propagate_context_to_threadzOptional[ThreadPoolExecutor] _executorint_executor_max_workerszDict[str, Dict[str, Any]]_records2i: ic$tdz S)Nzstate.dbr KC:\Users\server\AppData\Local\hermes\hermes-agent\tools\async_delegation.py_db_pathrSs   z ))rreturnsqlite3.Connectioncft}|jddtj|d}|d|dd|dD}d D]$\}}||vr|d |d |%|S) NT)parentsexist_ok )timeoutzPRAGMA journal_mode=WALaCREATE TABLE IF NOT EXISTS async_delegations ( delegation_id TEXT PRIMARY KEY, origin_session TEXT NOT NULL, origin_ui_session_id TEXT NOT NULL DEFAULT '', parent_session_id TEXT, state TEXT NOT NULL, dispatched_at REAL NOT NULL, completed_at REAL, updated_at REAL NOT NULL, event_json TEXT, result_json TEXT, delivery_state TEXT NOT NULL DEFAULT 'pending', delivery_attempts INTEGER NOT NULL DEFAULT 0, delivered_at REAL, owner_pid INTEGER, owner_started_at INTEGER, task_json TEXT, delivery_claim TEXT, delivery_claimed_at REAL )ch|] }|d S)r).0rows r z_connect..rsVVV#s1vVVVrz$PRAGMA table_info(async_delegations))) owner_pidINTEGER)owner_started_atr&) task_jsonTEXT)delivery_claimr))delivery_claimed_atREALz)ALTER TABLE async_delegations ADD COLUMN  )rparentmkdirsqlite3connectexecute)pathconncolumnsnamesql_types r_connectr8Ws ::DKdT222 ?4 , , ,DLL*+++LL ,WV.T!U!UVVVGXXh 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 |f dddn #1swxYwYdddn #1swxYwYtdS) Nr)get_process_start_timeoscDi|]}|v||Srget)r"keyr9s r z%_persist_dispatch..s6  &== VZZ__ ==r)goalgoalscontexttoolsetsrolemodelis_batchahINSERT 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) VALUES (?, ?, ?, ?, 'running', ?, ?, 'pending', 0, ?, ?, ?) delegation_id session_keyorigin_ui_session_idparent_session_id dispatched_at) timegateway.statusr= __import__getpid Exception_DB_LOCKr8r2rAjsondumps_prune_durable_records)r9nowr=r' task_payloadr4s` r_persist_dispatchr\s )++C 99999911*T2B2B2I2I2K2KLL     XL    8::    N O $fjj&C&C ZZ. 3 3VZZ@S5T5T O $c:d+;+;+B+B+D+D tz,77 9                                sH0A AA.D8=BD! D8!D% %D8(D% )D88D<?D<rKstrct5t5}|d|fdddn #1swxYwYddddS#1swxYwYdS)Nz3DELETE FROM async_delegations WHERE delegation_id=?)rVr8r2rKr4s r_delete_durable_delegationr`s ^^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) rQ_DURABLE_RETENTION_SECONDSrVr8r2fetchonemax_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)rZcutoffr4terminal_countexcess pending_countoverflows rrYrYs )++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_atrK)rQrVr8r2rArWrX)rlrmrZr4s r_persist_completionrrs8 )++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=?)rVr8r2rQr_s r_note_delivery_attemptrts   8::  t Y[[- (                                    s4A%*A  A% A A%A A%%A),A)c X ddlm}m}n#t$rYdSwxYwt j}d}t 5t 5}|d}|D]}|\}}} } } } } }d}| rE|t| }|r+| )|t| t| k}|rZtj |pd}iddd |d |d | d | d | d dd| dd| dd| dd| dd| ddt| dddddddd| d|}dd|dd}|d||tj|tj||f|dz } dddn #1swxYwYdddn #1swxYwY|S)zEClassify records whose owning process disappeared as outcome unknown.r) _pid_existsr=zSELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, owner_started_at, task_json FROM async_delegations WHERE state IN ('running','finalizing')FNz{}typeasync_delegationrKrLrNrOrDrMrErFrGrHrIrJrounknownsummaryerrorzLDelegation owner exited before recording a terminal result; outcome unknown.rPrq)rorzr{zUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r!)rRrvr=rUrQrVr8r2fetchallrrWloadsrAboolrX)rvr=rZ recoveredr4rowsr#rKrL origin_ui parent_idrPpidstartedr(livetaskrlrms rrecover_abandoned_delegationsrseFFFFFFFFF qq )++CI ##8::#|| Q  (**   Cgj dM; 9mSRY[dD L"{3s88,,LG/11#c((;;s7||KD :i/400D * ,;] { ,BI $Y 178L8L '** -6txx 7J7J  DHHZ00  39$((6:J:J  '**  -7TXXj=Q=Q8R8R ) &/ g   1? E!*dU7^TTF LL,c4:e,,dj.@.@-P     NII9 ##############################H sB HF4H< HH HH HH#&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) rrVr8r2r|rWr} isinstancedictputlen) target_queuer4r_delegation_idpayloadevts rrestore_undelivered_completionsr sV"###  " "8:: "|| 7   (** (, " " #NG*W%%C#t$$ '"&J   S ! ! ! !  " " " " " " " " " " " " " " " " " " " " " " " " " " " " " " " t99s5B<A3B% B<%B) )B<,B) -B<<CCr~ctj}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!NrQrVr8r2rowcount)rKrZr4curs rmark_completion_deliveredr%s) )++C !!8::!ll H #} %   |q !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!s4A?$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!)rQrVr8r2rcr)rKrrZr4r#rs rclaim_completion_deliveryr1s )++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.rwrxrMrK:r>N)rAr]rSrTuuiduuid4hexr)rrrKrs rclaim_event_deliveryrEs wwv,,,r006B77M rKKZ--4466KK9IKKH0II S88tSrct5t5}|dtj||f}|jdkcdddcdddS#1swxYwYddddS#1swxYwYdS)z>Release a failed delivery claim so another consumer may retry.zUPDATE async_delegations SET delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?r!N)rVr8r2rQr)rKrr4rs rrelease_completion_deliveryrPs% !!8::!ll )Y[[- 2   |q !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!s4A=5A% A=%A) )A=,A) -A==BBc 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!Nr)rKrrZr4rs rcomplete_completion_deliveryr]s, )++C  ! !8:: !ll ) #}h /   |q  ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s4B%A( B(A, ,B/A, 0BBBc|rM|ddkr6tt|dpd|dSdSdSNrwrxrKrM)rArr]rrs rcomplete_event_deliveryrlsaTCGGFOO'999$S)A)A)GR%H%H(SSSSSTT99rc|rM|ddkr6tt|dpd|dSdSdSr)rArr]rs rrelease_event_deliveryrqsaSCGGFOO'999#C(@(@(FB$G$GRRRRRSS99rOptional[Dict[str, Any]]c t5t5}|d|f}dddn #1swxYwYdddn #1swxYwY|dS||d|d|d|d|drt j|dnd|d|dd S) NzSELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts FROM async_delegations WHERE delegation_id=?rr!r)rKorigin_sessionstaterPrqrmdelivery_statedelivery_attempts)rVr8r2rcrWr})rKr4r#s rget_durable_delegationrvsG 8::ll ?AN@P   (**   {t&#a&3q6QQ(+A8$*SV$$$Da&s1v   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)ro>running finalizingr!Nr@r"rs r zactive_count..s7``155??F_3_3_13_3_3_3_``rN) _records_locksumrvaluesrrr active_countrs aa``hoo//`````aaaaaaaaaaaaaaaaaas0AA  A cHdtjjddS)Ndeleg_)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#Srorr@)r"ridrs r z+_prune_completed_locked..s? C 55??i ' ' a ' ' 'rNcr|ddp|ddpdS)Nr!rqrPrr@)kvs rz)_prune_completed_locked..s1"Q%))N";";"^r!uyy?Y?Y"^]^r)rB)ritemsrresortpop)rpr_s r_prune_completed_lockedrs nn&&I  9~~000 NN^^N___Fc)nn/FFFG  Q S$  rrM)rOrN interrupt_fnmax_async_childrenrDrFrGOptional[List[str]]rHrIrLrOrunnerCallable[[], Dict[str, Any]]rNrOptional[Callable[[], None]]rc 4ttj|||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) rKrDrFrGrHrIrLrNrOrorPrqrc3LK|]}|ddkdV dSrorr!Nr@rs rrz,dispatch_async_delegation..?  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.ror{rr;c i}d} pi}|dpd}nu#t$rh}tdddt |jd|dt tjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nr{rorpzAsync delegation %s crashed: rr)rorzr{ api_callsduration_seconds) rArUlogger exceptionrw__name__roundrQ _finalize)rmroexcrKrPrs r_workerz*dispatch_async_delegation.._workers!# 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): %srMP dispatchedrorKrr;)rrQlistrrrrr\rsubmitr rUrr`rinfo)rDrFrGrHrIrLrOrrNrrr9rexecutorrrrKrPs ` @@rdispatch_async_delegationrs`'((MIKKM&&.8DNNND" 4.&$F$ ))  ((      ( ( ($R:LRRR   ))))))))#))))))))))))))))"f/00H55555555&   3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 BSBB          KK={-g CRC/@#] C CCsZ ?B/ B//B36B3!"D E)E$D=1 E$=E E$E E$E)$E)roct5t|}| 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.Nrrorqr)rrrArQr_push_completion_eventr)rKrmror9 event_records rrr5sz  $ $m,, >  $ $ $ $ $ $ $ $(x!%~!%~F||  $ $ $ $ $ $ $ $ $ $ $ $ $ $ $<888 ""m,,  %F8 !!! """"""""""""""""""s(A./A..A25A20CCCc   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|dd d|dd|dd|dd|dp|dd|d|d|d|ddd|dt||z dd|d |d|di} 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: %srKNrzr{rPrqrwrxrLrMrNrOrDrFrGrHrIrorrr exit_reasonzHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryrrUrr{rArQrrrcompletion_queuer) r9rmrorrrzr{rPrqrs rrrKs;;;;;;;    JJ ' '    jj##G JJw  EJJ//>49;;M::n--<L " O44  vzz-44    +A2 F F   VZZ(;<<   62&&  6::i((  FJJz**   6""  G$$; 7(;(;  &  7    VZZ Q//! " FJJ l]&BA F F  # ( ) *  + , vzz-00-  C0V$$$ )--c22222       JJ ' '          s, A /AA #H?? J /I>>JrE List[str]c ttjt|} | dkr|dn$| ddd|Dz} | t |||rt |nd|||||dd| dd } t 5t d tD}|| krd d | d dcdddS| t<dddn #1swxYwYt| t| }dfd } | t|nh#t$r[}t 5tddddn #1swxYwYtd d|dcYd}~Sd}~wwxYwt d| |pdddS)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^rNrT)rKrDrErFrGrHrIrLrNrOrorPrqrrJc3LK|]}|ddkdV dSrr@rs 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) Nr{resultsc3DK|]}|ddvVdS)ro)rpsuccessNr@rs rrzCdispatch_async_delegation_batch.._worker..sF%%x(@@%%%%%%rrpz!Async delegation batch %s crashedrr)r r{total_duration_seconds) rAallrUrrrwrrrQ_finalize_batch)combinedro child_resultsrrKrPrs rrz0dispatch_async_delegation_batch.._workers5#% =vxx~2H$LL339rM %%%&%%%"" %!$      @- P P P II.77#77*/ m0KQ*O*OH FFFFFF  M8V < < < < ^>^ ^'e&.8DNNND" 4.&$F" ))  ((      ( ( ($N:LNNN ))))))))#)))))))))))))))) f/00H========2  3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 H3HH          KKKq+0#] C CCsZ?D* DDD2"E F:F5&F F5F F5F F5/F:5F:rcTt5t|}| 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|dd|ddd|dd|dd|dd|dd|dd|ddd|dpgd|dd|dd |d|i} 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.NrrorqrrrzVAsync delegation batch %s finished but process_registry import failed; result lost: %srPrwrxrKrLrMrNrOrDrErFrGrHrIrJTr r{rzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s)rrrArQrrrrUrr{rrrrr) rKrror9rrrrPrqrs rrrs $$m,, > $$$$$$$$(x!%~!%~F|| $$$$$$$$$$$$$$$;;;;;;;   & 3    !$$_55DM##N33Bty{{L "   |'' r::   0 01G L L   \--.ABB      ,,   !!'**  <##I..  L$$Z00     ((  !!'**  &  D  8<< **0b! " g&&# $ !(,,/G"H"H% & ' (  )  C,X&&& &)--c2222       3          & &\\-00F!#)x # % % %  & & & & & & & & & & & & & & & & & &] & &\\-00F!#)x # % % %  & & & & & & & & & & & & & & & &sA./A..A25A2:B B2 B--B28IK JI>9K>JK 0K  KKL'0L L'L L'"L #L'List[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 rrCz5list_async_delegations...Es( ? ? ?da1+>+>Q+>+>+>r)rrs rrz*list_async_delegations..Ds?    @ ?aggii ? ? ?   rN)rrrrrrlist_async_delegationsr>s     __&&                     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|Srr@rs rrz!interrupt_all..Ss3   AEE(OOy,H,HA,H,H,HrNrr!z&interrupt_all: %s interrupt failed: %srKz'Interrupted %d async delegation(s) (%s) rrrrAcallablerUrdebugr)r!counttargetsrfnrs r interrupt_allr*Jse 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)rorrNrMrLrO)rAr])r"rrNrOrLs 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: %srKz:Interrupted %d async delegation(s) for ending session (%s)r$) rLrNrOr!r'r(rr)rs ``` rinterrupt_for_sessionr.fs0 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)rrr rrrclearrrr_reset_for_testsr2s ""    E  * * * ! """"""""""""""" s';?? A44A8;A8)rr)r9r:rr;)rKr]rr;r)rlr:rmr:rr;)rr)rKr]rr~)rKr]rr]rr~)rr:rr]rr)rr:rr]rr;)rKr]rr)rrrr)rr])rDr]rFrrGrrHr]rIrrLr]rOrrrrNr]rrrrrr:)rKr]rmr:ror]rr;)r9r:rmr:ror]rr;)rErrFrrGrrHr]rIrrLr]rOrrrrNr]rrrrrr:)rKr]rr:ror]rr;)rr)r )r!r]rr)rMrMrMr+) rLr]rNr]rOr]r!r]rr)E__doc__ __future__rrWloggingr0 threadingrQrconcurrent.futuresrtypingrrrrr hermes_constantsr tools.daemon_poolr tools.thread_contextr getLoggerrrrr__annotations__Lockrrrr_DEFAULT_MAX_ASYNC_CHILDRENrerbrfrVrr8r\r`rYrrrtrrrrrrrrrrrrrrrrrrrrr*r.r2rrrr@sD!!!F#"""""  11111166666666666666,,,,,,666666<<<<<<  8 $ $ 5+/ ....!!   ')((((- 9>  ***%%%%P8^^^^ $$$$N         ,,,,^6 ! ! ! !!!!!(TTTT ! ! ! ! ! ! ! !TTTT SSSS "&aaaa ++++    4(, "159|D|D|D|D|D|D~"""",7 7 7 7 D(, "159vDvDvDvDvDvDr=&=&=&=&@     : " 55555p      r