epjpUdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZmZmZmZmZmZddlmZddlmZdd lmZejeZeZdad ed <ej Z!da"d ed <ej Z#iZ$ded<dZ%dZ&dZ'dZ(dZ)ej Z*dZ+dZ,dZ-dZ.ej Z/da0ded<ej1Z2dZ3ddZ4dd!Z5e dd#Z6dd&Z7dd)Z8dd*Z9dd-Z:dd.Z;dd/Zdd4Z?dd8Z@dd9ZAdd:ZBdd;ZCdd<ZDdd=ZEdd?ZFddBZGddCZHddEZIddFZJdGdGdGdHddKZK dddLZLddMZMddNZNddOZOddGdGde%ddPdd_ZPddaZQddcZRdddZSddeZTddGdGde%dddfddiZUddkZVddmZWddnZXddoZYddpZZddvZ[ddxZ\ddd{Z] ddd}Z^dd~Z_dS)u 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)contextmanager)AnyCallableDictIteratorListOptionalget_hermes_home)DaemonThreadPoolExecutor)propagate_context_to_threadzOptional[ThreadPoolExecutor] _executorint_executor_max_workerszDict[str, Dict[str, Any]]_records2i: ig>@g |@g@g^@zOptional[threading.Thread]_monitor_threadc$tdz S)Nzstate.dbr z%_initialize_schema..sVVV#s1vVVVrz$PRAGMA table_info(async_delegations))) owner_pidINTEGER)owner_started_atr9) task_jsonTEXT)delivery_claimr<)delivery_claimed_atREAL)origin_session_idr<z)ALTER TABLE async_delegations ADD COLUMN  ) hermes_stater1execute)r-r1columnsnamesql_types rr)r)s444444D+HIIIILL .WV.T!U!UVVVG  X Xh w   LLVTVVHVV W W W X XrIterator[sqlite3.Connection]c#Kt} |5|Vdddn #1swxYwY|dS#|wxYw)uOpen a connection, commit/rollback on exit, and ALWAYS close it. ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the transaction; they do not close the connection. Using ``with _connect()`` alone therefore leaks a connection — and its WAL/SHM file descriptors — on every durable dispatch, completion, and delivery-claim, deferring the close to the garbage collector. On a long-running gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of this bug was #69567 / PR #69594). N)r.r+)r-s r _transactionrIs ::D   JJJ                 s&A% A)A)AArecordDict[str, Any]ctj} 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)r5keyrJs 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_atr@) timegateway.statusrM __import__getpidr*_DB_LOCKrIrCrQjsondumps_prune_durable_records)rJnowrMr: task_payloadr-s` r_persist_dispatchrks )++C 99999911*T2B2B2I2I2K2KLL     XL    <>>  T  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  EEr[strct5t5}|d|fdddn #1swxYwYddddS#1swxYwYdS)Nz3DELETE FROM async_delegations WHERE delegation_id=?)rerIrCr[r-s r_delete_durable_delegationros ^^<>>^T 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) ra_DURABLE_RETENTION_SECONDSrerIrCfetchonemax_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)ricutoffr-terminal_countexcess pending_countoverflows rrhrhs )++C - -F   <>> T _ 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_atr[)rarerIrCrQrfrg)r{r|rir-s r_persist_completionrs8 )++C   <>> T  (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=?)rerIrCrarns r_note_delivery_attemptrs   <>> T 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_existsrMaSELECT 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_delegationr[r\r^r@r]r_rTrUrVrWrXrYrZr~unknownsummaryerrorzLDelegation owner exited before recording a terminal result; outcome unknown.r`r)r~rrzUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r4)rbrrMr*rarerIrCfetchallrrfloadsrQboolrg)rrMri recoveredr-rowsr6r[r\ origin_ui parent_idr`pidstartedr;r@livetaskr{r|s rrecover_abandoned_delegationsr%sFFFFFFFFF qq )++CI ''<>>'T|| 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) rrerIrCrrfr isinstancedictputlen) target_queuer-r_delegation_idpayloadevts rrestore_undelivered_completionsrXsV"###  " "<>> "T|| 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'r4NrarerIrCrowcount)r[rir-curs rmark_completion_deliveredrss) )++C !!<>>!Tll 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,r4)rarerIrCrrr)r[rrir-r6rs rclaim_completion_deliveryrs )++C !!<>>!Tll 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.rrr]r[:rNN)rQrlrcrduuiduuid4hexr)rrr[rs rclaim_event_deliveryrs 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>=?r4znAsync 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=?)rarerIrC_MAX_DELIVERY_ATTEMPTSrloggerwarning)r[rrir-cappedrs rrelease_completion_deliveryrs )++C !!<>>!T 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=?r4Nrr[rrir-rs rdrop_completion_deliveryrs, )++C  ! !<>> !Tll ) - *   |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=?r4Nrrs rcomplete_completion_deliveryrs, )++C  ! !<>> !Tll ) #}h /   |q  ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s4B%A( B(A, ,B/A, 0BBBc|rM|ddkr6tt|dpd|dSdSdSNrrr[r])rQrrlrrs rcomplete_event_deliveryrsaTCGGFOO'999$S)A)A)GR%H%H(SSSSSTT99rc|rM|ddkr6tt|dpd|dSdSdSr)rQrrlrs 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=?rr4rr]) r[origin_sessionstater`rr|delivery_statedelivery_attemptsr@)rerIrCrrrfr)r[r-r6s rget_durable_delegationrsV <>>Tll ?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)a~Number of async delegation UNITS currently running. A unit is one dispatch: a single subagent OR a whole fan-out batch. A batch counts as ONE here because it occupies one async-pool slot (the capacity semantics ``dispatch_async_delegation_batch`` relies on). For the count of actual concurrent child subagents (batch expanded), use ``active_task_count()``. c3HK|]}|ddvdVdS)r~>runningstalling finalizingr4NrPr5rs r zactive_count..!sD  uuX"GGG GGGG  rN _records_locksumrvaluesrrr active_countrs     ((                       0AA  A r^csdSt5tfdtDcdddS#1swxYwYdS)z9Number of live async delegations owned by one UI session.rc3K|]E}|ddvr,t|dpdkAdVFdS)r~>rrrr^r]r4N)rQrl)r5rr^s rrz%active_for_session..,sw  uuX"GGGAEE0117R88#$$ $$$$   rNr)r^s`ractive_for_sessionr's q       __&&                       s2A  AAcxt5d}tD]|}|ddvr|drH|d}|t |t t fr|rt|ndz }w|dz }}|cdddS#1swxYwYdS)aNumber of async delegation TASKS (child subagents) currently running. Unlike ``active_count()`` (units/slots), this expands a batch to its child count: a running batch of N tasks contributes N, a single subagent contributes 1. This is the truthful "how many subagents are actually working right now" figure for observability, where a 3-task batch shown as "1" undercounts real concurrent work. Falls back to counting a batch as 1 if its goal list is missing. rr~>rrrZrUr4N)rrrrQrlisttupler)totalrrUs ractive_task_countr5s   ""  AuuX&???uuZ   gz%$'G'GXEXUWXX                   sBB//B36B3r]r\r^r_r\r_c|r(t|dpd|kpS|o't|dpd|kp)|o't|dpd|kS)Nr^r]r\r_)rlrQ)rJr\r^r_s r_matches_session_selectorsrLs  g#fjj1G&H&H&NB"O"OSg"g c  OC = 9 9 ?R@@KO c  a#fjj1D&E&E&K"L"LPa"arcsssdSt5tfdtDcdddS#1swxYwYdS)uWhether a session still owns any live async delegation. Live = running / stalling / finalizing — the same states the reapers' keepalive treats as active work. Fc3lK|].}|ddvot|V/dS)r~>rrrrNrQrr5rr^r_r\s rrz'has_live_for_session..gsl    EE(OOD D *'%9"3        rN)ranyrrrs```rhas_live_for_sessionrZs 3z+_prune_completed_locked..|s? C 55??i ' ' a ' ' 'rNcr|ddp|ddpdS)Nr4rr`rrP)kvs rz)_prune_completed_locked..s1"Q%))N";";"^r!uyy?Y?Y"^]^r)rR)ritemsrrtsortpop)rr_s r_prune_completed_lockedrws 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_PLATFORMr] api_serverHERMES_SESSION_CHAT_ID)gateway.session_contextrr*rs r_current_origin_session_idr sp$;;;;;; ?4b 9 9\ I I27<<BB rrs( ( 66)r_r^r@ interrupt_fnmax_async_children progress_fnrTrVrWOptional[List[str]]rXrYrunnerCallable[[], Dict[str, Any]]r@r Optional[Callable[[], None]]r rOptional[Callable[[], tuple]]c ttjidd|d|d|rt|ndd|d|d|d |d | d |d d dddd| d| dddddi} t5t dt D}|| krdd| ddcdddS| t <dddn #1swxYwYt| t| }d$fd } | t|nh#t$r[}t5t ddddn #1swxYwYtdd|dcYd}~Sd}~wwxYw| ttd|pd|pd dd!d"d#S)%aSpawn ``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). progress_fn Optional zero-arg callable returning ``(token, in_tool)`` where ``token`` is any comparable snapshot of the child's progress (api call count + current tool) and ``in_tool`` says whether the child is currently inside a tool call. Sampled by the stale monitor; a frozen token past the stale threshold marks the delegation stuck (see the stale-detection block at the top of this module). When omitted, the delegation is not monitored. 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. r[rTrVrWNrXrYr\r^r@r_r~rr`rr r_progress_token _progress_ts_interrupted_atc3HK|]}|ddvdVdSr~rrr4NrPrs rrz,dispatch_async_delegation..D  uuX"999 9999  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.r~rrr/c i}d} pi}|dpd}nu#t$rh}tdddt |jd|dt tjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nrr~rzAsync delegation %s crashed: rr)r~rr api_callsduration_seconds) rQr*r exceptionr__name__roundra _finalize)r|r~excr[r`rs 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): %sr]P dispatchedr~r[rr/)rrarrrrrrkrsubmitrr*rro_ensure_stale_monitorrinfo)rTrVrWrXrYr\r_rr^r@r r rrJrexecutorr(r'r[r`s ` @@rdispatch_async_delegationr2st'((MIKKM 7 h8DNNND      {  4 . . )     {" 4#$  %& 4'F0 ))  ((      ( ( ($R:LRRR   )))))))) #)!)))))))))))))))$f/00H55555555&   3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 BSBB          KK={-g CRC/@#] C CCsZ6?C CCC "D-- F7F >E& F &E* *F -E* .F F Fr~cxt|}|dS|\}}t|||t||dS)zDMark a record complete and push the completion event onto the queue.N)_begin_finalization_push_completion_event_finish_finalization)r[r|r~claimed event_record _interrupt_fns rr&r&7sK!-00G")L-<888/////r=Optional[tuple[Dict[str, Any], Optional[Callable[[], None]]]]c\t5t|}||ddvr ddddSd|d<tj|d<|d}d|d<d|d<t |}dddn #1swxYwY||fS)zCAtomically claim terminal delivery while keeping the record active.Nr~rrrr r)rrrQrar)r[rJr r8s rr4r4Bs  $ $m,, >VZZ119PPP  $ $ $ $ $ $ $ $(x!%~zz.11 !%~ $}F||  $ $ $ $ $ $ $ $ $ $ $ $ $ $ $  %%s5B A BB#&B#ct5t|}|||d<tddddS#1swxYwYdS)Nr~)rrrQr)r[r~rJs rr6r6Ws ""m,,  %F8 !!! """"""""""""""""""rc ^ 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} dD]} | |vr || | | <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: %sr[Nrrr`rrrr\r]r^r@r_rTrVrWrXrYr~r!r"r exit_reason)rr@stalled_after_quiet_secondsstall_threshold_seconds stall_phasestall_grace_secondszHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryr?r*rrrQrar%rcompletion_queuer) rJr|r~r?r'rrr`rr_ks rr5r5_sN;;;;;;;    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/   C6!! <<RjCGV$$$ )--c22222       JJ ' '          s, A /AA  I(( J,2/J''J,)r_r^r@r r r[rrU List[str]c8  p t tjt|}|dkr|dn$|ddd|Dz}id d|dt |d |d |rt |nd d |d |d|d|d| d|ddddd d| ddd| d d d}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 }~wwxYw| t!t"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. r4rz parallel subagents: z; c3*K|]}|ddVdS)N(r)r5gs rrz2dispatch_async_delegation_batch..s+G^G^ST#2#G^G^G^G^G^G^rr[rTrUrVrWNrXrYr\r^r@r_r~rr`rr rZTr)rrrc3HK|]}|ddvdVdSrrPrs 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)r~)rsuccessNrPrs rrzCdispatch_async_delegation_batch.._worker..sF%%x(@@%%%%%%rrz!Async delegation batch %s crashedr rrPrtotal_duration_seconds) rQallr*rr#rr$r%ra_finalize_batch)combinedr~ child_resultsr'r[r`rs rr(z0dispatch_async_delegation_batch.._workers5#% =vxx~2H$LL339rM %%%&%%%"" %!$      @- P P P II.77#77*/ m0KQ*O*OH FFFFFF  M8V < < < < ^>^ ^  e 7  h8DNNND     {  4 . . )     D!" {#$ %)F, ))  ((      ( ( ($N:LNNN ))))))))#))))))))))))))))"f/00H========2  3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 H3HH          KKKq+0#] C CCsZ ?D+ D++D/2D/"F G% G F9- G 9F= =G F= G G% G%rWcxt|}|dS|\}}t|||t||dS)zDMark a batch record complete and push ONE combined completion event.N)r4_push_batch_completion_eventr6)r[rWr~r7r8r9s rrVrV(sM"-00G")L- x@@@/////rr8c  ddlm}nG#t$r:}td|d|Yd}~dSd}~wwxYw|dpt j}|dpt j}idd d|dd |d d d |d d d |d d d|dd|dd d|dd|dd|dd|dd|dd|ddd|dpgd|dd|d|d||d}dD]}||vr ||||<t|| |j |dS#t$r:}td|d|Yd}~dSd}~wwxYw)z8Push a combined async-delegation batch completion event.rr>zVAsync delegation batch %s finished but process_registry import failed; result lost: %sr[Nr`rrrr\r]r^r@r_rTrUrVrWrXrYr~rZTrPlive_transcriptsrrT)rTr`rrAzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s) rFr?r*rrrQrarrGr) r8rWr~r?r'r`rrrHs rr^r^5sC;;;;;;;   &   _ - -s    !$$_55DM##N33Bty{{L " ))/::  |'' r::   0 01G L L   \--.A2FF   \--.ABB      ,,  !!'**  <##I..  L$$Z00     ((  !!'**  &  D " 8<< **0b# * HLL);<<+ , g&&- .#+,,/G"H"H&$3   C:## >>rlCGX&&& )--c22222         _ - -s          s, A /AA #H?? J /I>>Jc>t5t'tr ddddStt jtddatddddS#1swxYwYdS)zStart (once) the module-level stale-delegation monitor thread. One daemon thread serves every dispatch; it exits on its own when no monitorable records remain, and is restarted by the next dispatch that carries a ``progress_fn``. Nzasync-delegate-stale-monitorT)targetrEdaemon) _monitor_lockris_alive _monitor_stopclear threadingThread_stale_monitor_loopstartrrrr/r/us     &?+C+C+E+E &           #*&/                       s"BABBBc ttstj}g}g}d}t5t D]}|d}|dkrCd}|dp|}||z tkr| |da|dkrh|d}|d} |\}} n'#t$r|d d} }YnwxYw||d kr ||d <||d <||d p|z } | rtnt} | | krsd|d<||d<t| d |d <| |d<t| |d<| |dt|d| | f d d d n #1swxYwY|D]\} } } } td| | | tt5t | }|r|dnd }d d d n #1swxYwYt#|r? |#t$r&}td| |Yd }~d }~wwxYw|D]} t'| |sd Sttd Sd S)uYSweep running delegations for stalled progress. Per sweep, for every running record with a ``progress_fn``: - Sample ``(token, in_tool)``. A changed token refreshes the record's progress timestamp — a child that keeps advancing is never touched, no matter how long it runs. - A frozen token past the idle/in-tool threshold marks the record ``stalling``: we call ``interrupt_fn`` so a responsive-but-slow child can unwind and deliver its (partial) result through the normal ``_finalize`` path with full fidelity. - A ``stalling`` record whose runner still hasn't returned after the grace window is force-finalized with one terminal ``stalled`` event so the owning session hears an outcome and the async slot frees. A late runner return after that is ignored by ``_begin_finalization``. Fr~rTrr[rrNrrr_stall_quiet_seconds_stall_threshold_seconds_stall_in_toolrZu`Async delegation %s made no progress for %.0fs (in_tool=%s) — interrupting; grace window %.0fsr z.Async delegation %s stall interrupt failed: %s)rfwait_STALE_CHECK_INTERVALrarrrrQ_STALL_GRACE_SECONDSappendr*_STALE_IN_TOOL_SECONDS_STALE_IDLE_SECONDSr%rrrcallabledebug_finalize_stalled)ristalledexpiredany_monitorablerJr~interrupted_atrtokenin_tool quiet_forlimitr[ _is_batchfnr's rrjrjsD"  !677Hikk! . . "//++- - H--Z''&*O%+ZZ0A%B%B%IcN^+/CCCvo'>???Y&&$jj77 &"&J%0[]]NE77 JJJ&,ZZ0A%B%BE7EEEJFJJ'8999905F,--0F>*6::n#=#=#DE .5N**;N%%'1F8$03F,- 6;9a5H5HF129>F56/3G}}F+,NN"?3 J!7!788%# M- . . . . . . . . . . . . . . . ^=D   8M9i NNDy'3G     D D!m4439CVZZ///t D D D D D D D D D D D D D D D|| BDDDD LLH%s % - -M m , , , ,  FQ  !677HHHHHsbBG) C,+G),!D G)DC G))G-0G-'4I''I+ .I+  J  J=J88J=c nt|}|dS|\}}|dptj}t||dp|z d}|d}|d}|d}d|d } td |||||rd n|d ndt d } |drt|g| |d| dnt|dd| d|dd| dt|ddS)zAForce-finalize a stalling delegation whose runner never returned.Nrr`rrmrnrozAsync delegation u^ stalled: the detached subagent stopped making progress (no new API calls, tool activity, or streamed tokens), did not respond to interruption, and never produced a completion event. The worker may be wedged inside a model API call — this is a known failure mode of long-lived gateway processes (#60203). Re-dispatch the task if it is still needed.z:Async delegation %s force-finalized as stalled after %.0fsr~idlerArZrSryr)r~rrr!r"r@) r4rQrar%rrrrr^r5r6) r[r7r8r9rduration quiet_secondsthreshold_seconds stall_in_toolr stall_metas rrxrxs!-00G")L-##N33Bty{{L ((99I\J H!$$%;<>122A33./ JJu     JJt     Js   List[Dict[str, Any]]c^tj}i}t5g}tD]}d|D}|d}|dvr\|d}|rt ||z d|d<|d}t|r |||d<|d vr5d D]2\}} |||||| <3|| d d d n #1swxYwY|D]u}||d}|- |\} } n#t$rYHwxYwt| |} | | |d <t| |d <v|S)aSnapshot of async delegations (running + recently completed). Safe to call from any thread. Excludes the non-serialisable callables and private monitor bookkeeping, but exposes computed live-status fields for UIs (#51690): - ``seconds_since_progress``: how long the stale monitor has seen a frozen progress token (running/stalling records). - ``children_activity``: per-child ``{api_calls, current_tool, seconds_since_activity}`` sampled live from the dispatch's ``progress_fn``. - ``stalled_after_quiet_seconds`` / ``stall_threshold_seconds`` / ``stall_in_tool``: stall context once the monitor has tripped. cLi|]!\}}|dv |d||"S)>rr r) startswith)r5kvs rrSz*list_async_delegations..\sHAq;;; S))<1;;;rr~rrr4seconds_since_progressrr[)rry))rmrB)rnrC)rorNchildren_activityr~) rarrrrrQr%rvrsr*rr) risamplersrritemr~tsrsrcdstr}r~activitys rlist_async_delegationsrHs4 )++C$&H ""  AGGIID UU8__F000UU>**H5:38Q5G5GD12UU=))B<<635HQ/0000!//HC uuSzz-$%EE#JJS LL    / < ( ( \\$((?33 4 4 :  RTTNE77    H 0<<  (0D$ %w--Y Ls$DD**D.1D.% E33 F?Fshutdownreasoncd}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. rc@g|]}|ddv|S)r~rrPrs rrz!interrupt_all..s8   uuX"999 999rNr r4z&interrupt_all: %s interrupt failed: %sr[z'Interrupted %d async delegation(s) (%s) rrrrQrvr*rrwr0)rcounttargetsrrr's r interrupt_allrse 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. rchg|].}|ddvt|,|/S)r~rrrrs rrz)interrupt_for_session..s\    uuX"999*'%9"3 : 999rNr r4z.interrupt_for_session: %s interrupt failed: %sr[z:Interrupted %d async delegation(s) for ending session (%s)r) r\r^r_rrrrrr's ``` rinterrupt_for_sessionrs0 3 " " B<<       DEE/**C      H 6    Ls)(A  A A =B  C/C  Cct5ttddadadddn #1swxYwYtt 5t}dadddn #1swxYwY|*|r| dt5t ddddS#1swxYwYdS)z@Test-only: clear all state and tear down the executor + monitor.NF)rprrr#) rrrrrfsetrdrrerYrrrg)threads r_reset_for_testsrs ""    E  * * * ! """""""""""""""   foo// A  s1';??& A<<BB:C!!C%(C%)rr)r-rrr/)rrG)rJrKrr/)r[rlrr/r-)r{rKr|rKrr/)rr)r[rlrr)r[rlrrlrr)rrKrrlrr)rrKrrlrr/)r[rlrr)rrrr)r^rlrr) rJrKr\rlr^rlr_rlrr)r]r]r])r\rlr^rlr_rlrr)rrl)rTrlrVrrWrrXrlrYrr\rlr_rrrr^rlr@rlr rr rrrrrK)r[rlr|rKr~rlrr/)r[rlrr:)r[rlr~rlrr/)rJrKr|rKr~rlrr/)rUrIrVrrWrrXrlrYrr\rlr_rrrr^rlr@rlr rr rr[rrrrrK)r[rlrWrKr~rlrr/)r8rKrWrKr~rlrr/)r}rrirrr)rr)r)rrlrr)r]r]r]r) r\rlr^rlr_rlrrlrr)`__doc__ __future__rrfloggingr'rhrarconcurrent.futuresr contextlibrtypingrrrr r r hermes_constantsr tools.daemon_poolrtools.thread_contextr getLoggerr$rrr__annotations__Lockrrrr_DEFAULT_MAX_ASYNC_CHILDRENrtrqrurrerqrurtrrrdrEventrfrr.r)rIrkrorhrrrrrrrrrrrrrrrrrrrrrr r2r&r4r6r5r\rVr^r/rjrxrrrrrrrrrs*!!!F#"""""  111111%%%%%%@@@@@@@@@@@@@@@@,,,,,,666666<<<<<<  8 $ $ 5+/ ....!!   ')((((-  9>  .   .22222 !! ***    )X)X)X)XX$:^^^^ $$$$N         0000f6 ! ! ! !!!!!(TTTT!!!!!!!!H!!!!, ! ! ! !TTTT SSSS &&         4 "       "     2++++    $H(, "15915ODODODODODODd0000&&&&*""""B B B B Z(, "159#'15ADADADADADADH 0 0 0 0= = = = @    (YYYYxC3C3C3C3L:;;;;|< " 66666rr