`gjmUdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z m Z mZmZmZddlmZddlmZddlmZej.eZeZdad ed <ej:Zdad ed <ej:Z iZ!d ed<dZ"dZ#dZ$dZ%ej:Z&dZ'd4dZ(d5dZ)d6dZ*d7dZ+d8dZ,d6dZ-d9dZ.d9dZ/d:dZ0d;dZ1dd#Z7d?d$Z8d9d%Z9d@d&Z:d7d'Z;dd(de"d) dAd*Zdd(de"dd- dDd.Z? dEd/Z@dFd0ZAdGdHd1ZB dI dJd2ZCd7d3ZDy)Ku 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: ictdz S)Nzstate.dbr I/root/.hermes/venv/lib/python3.12/site-packages/tools/async_delegation.py_db_pathrSs  z ))rc^t}|jjddtj|d}|j d|j d|j dDchc]}|d }}d D]!\}}||vs |j d |d |#|Scc}w) 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 )z$PRAGMA table_info(async_delegations))) owner_pidINTEGER)owner_started_atr ) task_jsonTEXT)delivery_claimr#)delivery_claimed_atREALz)ALTER TABLE async_delegations ADD COLUMN  )rparentmkdirsqlite3connectexecute)pathconnrowcolumnsnamesql_types r_connectr3Ws :DKKdT2 ??4 ,DLL*+LL ,"&.T!UV#s1vVGVXh w  LLDTF!H:V WX KWs4 B*cztj} ddlm}|tdj }dDcic]}||vr||j |}}t5t5}|jd|d|j dd|j d d|j d |d |tdj |tj|f ddddddty#t $rd}YwxYwcc}w#1swY2xYw#1swYtyxYw) Nr)get_process_start_timeos)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.statusr5 __import__getpid Exceptionget_DB_LOCKr3r,jsondumps_prune_durable_records)recordnowr5r!key task_payloadr.s r_persist_dispatchrRs. ))+C 91*T2B2I2I2KL Y  &= VZZ_L   8:    N O $fjj&C ZZ. 3VZZ@S5T O $c:d+;+B+B+D tzz,7 9    )       s;%DD$ D'/A;D*D' DDD$ D''D:ct5t5}|jd|fddddddy#1swYxYw#1swYyxYw)Nz3DELETE FROM async_delegations WHERE delegation_id=?)rJr3r,r>r.s r_delete_durable_delegationrUsO ^8:^ J]L\]^^^^^^s A7AA AA ctj}|tz }t5t5}|j d|f|j dj d}t d|tz }|r|j d|f|j dj d}t d|tz }|r|j d|fddddddy#1swYxYw#1swYyxYw)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) rD_DURABLE_RETENTION_SECONDSrJr3r,fetchonemax_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)rOcutoffr.terminal_countexcess pending_countoverflows rrMrMs ))+C - -F  8:  _ I  ` (*QQ)@@A  LL     [  (*Q q-*>>?  LL   3      s# C4B(C(C4(C1 -C44C=c btj}t5t5}|jd|j dd|j d||t j |t j ||dfddddddy#1swYxYw#1swYyxYw)NzUPDATE async_delegations SET state=?, completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?status completed completed_atr>)rDrJr3r,rIrKrL)eventresultrOr.s r_persist_completionrgs ))+C  8:   (YYx -uyy/Ms ZZ  6 2E/4J L        s# B%A"BB%B" B%%B.ct5t5}|jdtj|fddddddy#1swYxYw#1swYyxYw)NzfUPDATE async_delegations SET delivery_attempts=delivery_attempts+1, updated_at=? WHERE delegation_id=?)rJr3r,rDrTs r_note_delivery_attemptrisR  8:  t YY[- (       s! A'A A A AAc  ddlm}m}t j}d}t 5t 5}|jdj}|D]S}|\}}} } } } } }d}| r2|t| }|r| |t| t| k(}|rHtj|xsd}iddd |d |d | d | d |jd dd|jdd|jdd|jdd|jdd|jddt|jdddddddd| d|}dd|dd}|jd||tj|tj||f|dz }V dddddd|S#t$rYywxYw#1swY"xYw#1swY|SxYw)zEClassify records whose owning process disappeared as outcome unknown.r) _pid_existsr5zSELECT 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_delegationr>r?rArBr7r@r8r9r:r;r<r=rbunknownsummaryerrorzLDelegation owner exited before recording a terminal result; outcome unknown.rCrd)rbrorpzUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r)rErkr5rHrDrJr3r,fetchallrrKloadsrIboolrL)rkr5rO recoveredr.rowsr/r>r? origin_ui parent_idrCpidstartedr"livetaskrerfs rrecover_abandoned_delegationsr|skF ))+CI #8:#|| Q (*  Cgj dM; 9mSRY[dD"3s8,G/1#c(;s7|KD::i/40D * ,;] { ,BI $Y 178L '* -6txx 7J  DHHZ0  39$((6:J  '*  -7TXXj=Q8R ) &/ g   1? E!*dU7^TF LL,c4::e,djj.@-P   NI9 ##H Q ###H s4F= GE9G +G= G G  G GG"c~tt5t5}|jdj }|D]@\}}t j |}t|trd|d<|j|B ddddddtS#1swYxYw#1swYtSxYw)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) r|rJr3r,rqrKrr isinstancedictputlen) target_queuer.ru_delegation_idpayloadevts rrestore_undelivered_completionsr s"#  "8: "|| 7  (* (, " #NG**W%C#t$"&J   S !  " " " t9 " " " t9s# B)A%BB)B& "B))B<ctj}t5t5}|jd|||f}|jdk(cdddcdddS#1swYnxYwdddy#1swYyxYw)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'rNrDrJr3r,rowcount)r>rOr.curs rmark_completion_deliveredr%sq ))+C !8:!ll H #} %  ||q !!!!!!!s" A2$A A2A& "A22A;c ptj}t5t5}|jd|fj }| ddddddy|jd|||||dz f}|j dk(cdddcdddS#1swYnxYwdddy#1swYyxYw)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)rDrJr3r,rXr)r>claim_idrOr.r/rs rclaim_completion_deliveryr1s ))+C !8:!ll P    (*  ; !!!ll LsCc :  ||q !!!!!!!s. B,%B B,(B B,B B,,B5c|jddk7ryt|jdxsd}|sy|dtdjdt j j }t||r|SdS)zCClaim a durable delegation event; non-durable events need no token.rlrmr@r>:r6N)rIstrrFrGuuiduuid4hexr)rconsumerr>rs rclaim_event_deliveryrEsz wwv,,06B7M 1Z-4467q9I9I8JKH0I8StSrct5t5}|jdtj||f}|jdk(cdddcdddS#1swYnxYwdddy#1swYyxYw)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=?rN)rJr3r,rDr)r>rr.rs rrelease_completion_deliveryrPsm !8:!ll )YY[- 2  ||q !!!!!!!s" A06A A0A$ A00A9c tj}t5t5}|jd||||f}|jdk(cdddcdddS#1swYnxYwdddy#1swYyxYw)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=?rNr)r>rrOr.rs rcomplete_completion_deliveryr]st ))+C  !8: !ll ) #}h /  ||q  ! ! ! ! ! ! !s" A3%A A3A' #A33A<c|r>|jddk(r)tt|jdxsd|yyyNrlrmr>r@)rIrrrrs rcomplete_event_deliveryrls9CGGFO'99$S)A)GR%H(S:xrc|r>|jddk(r)tt|jdxsd|yyyr)rIrrrs rrelease_event_deliveryrqs9CGGFO'99#C(@(FB$GR:xrc 2t5t5}|jd|fj}ddddddy||d|d|d|d|drt j |dnd|d|dd S#1swYNxYw#1swYRxYw) NzSELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts FROM async_delegations WHERE delegation_id=?rrr)r>origin_sessionstaterCrdrfdelivery_statedelivery_attempts)rJr3r,rXrKrr)r>r.r/s rget_durable_delegationrvs 8:ll ?AN@P  (*   {&#a&3q6QQ(+A$**SV$Da&s1v  s! B "BB B B  Bct5t |tkDrt|da|atcdddS#1swYyxYw)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) max_workersthread_name_prefix)_executor_lockrr_DaemonThreadPoolExecutor)rs r _get_executorrsE    .C C1'#3I%0 !$5>ct5tdtjDcdddS#1swYyxYw)z.Number of async delegations currently running.c3HK|]}|jddvsdyw)rb>running finalizingrNrI.0rs r zactive_count..s `155?F_3_1`s""N) _records_locksumrvaluesrrr active_countrs3 a`hoo/``aaarcJdtjjddS)Ndeleg_)rrrrrr_new_delegation_idrs# DJJL$$Ra() **rc:tjDcgc]\}}|jddk7r||f}}}t|tkry|j d|dt|tz D]\}}tj |dycc}}w)ziDrop the oldest completed records beyond the retention cap. Caller must hold ``_records_lock``. rbrNc^|djdxs|djdxsdS)NrrdrCrr)kvs rz)_prune_completed_locked..s+"Q%))N";"^r!uyy?Y"^]^r)rP)ritemsrIrrZsortpop)ridrrc_s r_prune_completed_lockedrsnn& C 55?i ' aI  9~00 NN^N_Fc)n/FFG Q S$ s"Br@)rBrA interrupt_fnmax_async_childrenc ttj|||r t|nd|||||dd| d } t5t dt j D} | | k\rdd| ddcdddS| t <dddt| t| } dfd } | jt|tjd |xsd |xsd ddddS#1swYjxYw#t$rO}t5t jddddn #1swYnxYwtdd |dcYd}~Sd}~wwxYw)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) r>r7r9r:r;r<r?rArBrbrCrdrc3JK|]}|jddk(sdywrbrrNrrs rrz,dispatch_async_delegation..$ AEE(Oy,HA ##rejected#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.rbrpc hi}d} xsi}|jdxsd}t||y#t$ra}tjdddt |j d|dt tjz dd}d}Yd}~sd}~wwxYw#t||wxYw) NrprbrczAsync delegation %s crashed: rr)rbrorp api_callsduration_seconds) rIrHlogger exceptionrl__name__roundrD _finalize)rfrbexcr>rCrunners r_workerz*dispatch_async_delegation.._workers!# 5X^FZZ)8[F mVV 4    :M J! I../r#7$)$))+ *Eq$I FF  mVV 4s' 5 BABB"BB""B1z%Failed to schedule async delegation: z3Dispatched async delegation %s (session_key=%s): %sr@P dispatchedrbr>returnNone)rrDlistrrrrrRrsubmitr rHrrUrinfo)r7r9r:r;r<r?rBrrArrrNrexecutorrrr>rCs ` @@rdispatch_async_delegationrs`'(MIIKM&&.DND" 4.&$F$ ) (   ( ($9:L9MNQR  ))#))"f/0H5&   3G<= KK={-g CR/@#] CCq))V   . LL - . . ."=1 rfrbrN event_records rrr5s  $m, >  $ $(x!%~!%~F|  $<8 "m,  %F8 ! "" $ $""sB ,B 0'B, B),B5c V ddlm}|j d}|j d}|j dxst j }|j d xst j }id d d|j dd |j d d d|j dd d|j dd|j dd d|j dd|j dd|j dd|j dxs|j dd|d|d|d|j ddd|j dt||z dd|d |d|j di} t| | |jj| y#t$r0}tj d|j d|Yd}~yd}~wwxYw#t$r0}tj d|j d|Yd}~yd}~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: %sr>NrorprCrdrlrmr?r@rArBr7r9r:r;r<rbrrr exit_reasonzHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryrrHrrprIrDrrgcompletion_queuer) rNrfrbrrrorprCrdrs rrrKs];jj#G JJw EJJ/>499;M::n-<L " O4  vzz-4    +A2 F   VZZ(;<   62&  6::i(  FJJz*   6"  G$; 7(;  &  7    VZZ Q/! " FJJ l]&BA F # ( ) *  + , vzz-0- C0V$ ))--c2O    JJ '  P     JJ '   s/F3G/3 G,<&G''G,/ H(8&H##H()rBrArrr>c *  xs t tjt|} | dk(r|dn| ddjd|Dz} | t |||r t |nd|||||dd| dd }t 5t d tjD}|| k\rd d | d dcdddS|t <dddt|t| }d fd } |jt|t j#d | |xsdd dS#1swYcxYw#t$rO}t 5tj ddddn #1swYnxYwt d d|dcYd}~Sd}~wwxYw)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. rrz parallel subagents: z; c3&K|] }|dd yw)N(r)rgs rrz2dispatch_async_delegation_batch..sG^ST#2G^sNrT)r>r7r8r9r:r;r<r?rArBrbrCrdrr=c3JK|]}|jddk(sdywrrrs 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.rci}d} xsi}|jdxsg}|rtd|Drd}nd}t||y#t$r_}tj dgt |j d|ttjz dd}d}Yd}~qd}~wwxYw#t||wxYw) Nrpresultsc3BK|]}|jddvyw)rb)rcsuccessNrrs rrzCdispatch_async_delegation_batch.._worker..s&%x(@@%srcz!Async delegation batch %s crashedrr)rrptotal_duration_seconds) rIallrHrrrlrrrD_finalize_batch)combinedrb child_resultsrr>rCrs rrz0dispatch_async_delegation_batch.._workers#% =x~2H$LL39rM%&%"!$ M8V <    @- P I../r#7*/ m0KQ*OH F  M8V n combined_goalrNrrrrrCs ` ` @rdispatch_async_delegation_batchr sF"9%7%9MIIKM E AFa1#%: ;diiG^X]G^>^ ^'e&.DND" 4.&$F" ) (   ( ($9:L9MNMN ))#)) f/0H=2  3G<= KKKq+0#] CCy))^   . LL - . . ."=1 B3%H   sH2D. D.3D:.D7: FF  E) F )E2 .F F Fct5tj|}| dddyd|d<tj|d<d|d<t |}ddd ddlm}jd xstj}|jdxstj}id d d |d |jd dd|jddd|jdd|jddd|jdd|jdd|jdd|jdd|jdd|ddd|jdxsgd|jdd|jdd|jd||d} t| | |jj| t5tj|}|||d<tdddy#1swYxYw#t$r!}tjd||Yd}~yd}~wwxYw#t$r!}tjd||Yd}~d}~wwxYw#1swYyxYw#t5tj|}|||d<tdddw#1swYwxYwxYw)zDMark a batch record complete and push ONE combined completion event.NrrbrdrrrzVAsync delegation batch %s finished but process_registry import failed; result lost: %srCrlrmr>r?r@rArBr7r8r9r:r;r<r=Trlive_transcriptsrpr)rCrdzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s)rrrIrDrrrrHrrprgrrr) r>rrbrNrrrrCrdrs rrrs $m, > $$(x!%~!%~F| $;!$$_5DM##N3Btyy{L "   |'' r:   0 01G L   \--.AB      ,   !!'*  <##I.  L$$Z0     (  !!'*  &  D  8<< *0b! ( HLL);<) * g&+ , !(,,/G"H- .'$1 C4X& &))--c2 &\\-0F!#)x # %  & &s$$   & 3  N     3    & &] &\\-0F!#)x # %  & & &svH,HH!I$'I;H! I *II  I8I3.J3I88J;JK 'J>5 K >K K c t5tjDcgc],}|jDcic]\}}|dk7s ||c}}.c}}}cdddScc}}wcc}}}w#1swYyxYw)zSnapshot of async delegations (running + recently completed). Safe to call from any thread. Excludes the non-serialisable interrupt_fn. rN)rrrr)rkvs rlist_async_delegationsrCsm  __&   ggi ?da1+>QT ?   ?   s3A,A% AA A% A,A% %A,,A5cd}t5tjDcgc]}|jddk(s|}}dddD],}|jd}t |s ||dz }.|rt jd |||Scc}w#1swYZxYw#t $r0}t jd|jd|Yd}~d}~wwxYw) 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. rrbrNrrz&interrupt_all: %s interrupt failed: %sr>z'Interrupted %d async delegation(s) (%s)) rrrrIcallablerHrdebugr)reasoncountrtargetsfnrs r interrupt_allrOs E  ( AEE(Oy,HA     UU> " B<     =ufM L!     <EE/*C s8BBBB+ B%BB"% C.&CCc |s|s|syd}t5tjDcgc]}|jddk(rk|r!t |jdxsd|k(sF|r!t |jdxsd|k(s#|r#t |jdxsd|k(r|}}dddD],}|jd }t |s ||d z }.|rtjd |||Scc}w#1swYZxYw#t $r0}tjd |jd |Yd}~d}~wwxYw)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. rrbrrAr@r?rBNrrz.interrupt_for_session: %s interrupt failed: %sr>z:Interrupted %d async delegation(s) for ending session (%s)) rrrrIrrrHrrr) r?rArBrrrrrrs rinterrupt_for_sessionrksM0 3 " B<     H 6  L3      DEE/*C s5DBD-D DDD E &EE ct5ttjddadadddt5t j dddy#1swY-xYw#1swYyxYw)z6Test-only: clear all state and tear down the executor.NF)waitr)rrshutdownrrrclearrrr_reset_for_testsrsg "    E  * ! "  "" s!AA A A))rzsqlite3.Connection)rNDict[str, Any]rr)r>rrrr)rer rfr rr)rr)r>rrrs)r>rrrrrs)rr rrr Optional[str])rr rrrr)r>rrzOptional[Dict[str, Any]])rrrr)rr)r7rr9r!r:Optional[List[str]]r;rr<r!r?rrBr!rCallable[[], Dict[str, Any]]rArrOptional[Callable[[], None]]rrrr )r>rrfr rbrrr)rNr rfr rbrrr)r8z List[str]r9r!r:r"r;rr<r!r?rrBr!rr#rArrr$rrr>r!rr )r>rrr rbrrr)rzList[Dict[str, Any]])r)rrrr)r@r@r@ session_end) r?rrArrBrrrrr)E__doc__ __future__rrKloggingr* threadingrDrconcurrent.futuresrtypingrrrrr hermes_constantsr tools.daemon_poolr tools.thread_contextr getLoggerrrrr__annotations__Lockrrrr_DEFAULT_MAX_ASYNC_CHILDRENrZrWr[rJrr3rRrUrMrgrir|rrrrrrrrrrrrrrrrr rrrrrrrrr3s$!F#  166,6<   8 $ 5+/ '.!s  ') #(- 9>> *%P8^ $N   ,^6 !!(T ! !T S "&a + 4(, "159|D |D|D" |D  |D  |D|D%|D )|D|D/|D|D|D~",7 7 $27