sj|UdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl mZmZmZmZmZmZddlmZddlmZdd lmZejeZeZdad ed <ej Z!da"d ed <ej Z#iZ$ded<dZ%dZ&dZ'dZ(dZ)dZ*ej Z+dZ,dZ-dZ.dZ/ej Z0da1ded<ej2Z3dZ4ddZ5dd"Z6e dd$Z7dd&Z8dd(Z9dd+Z:dd,Z;dd/Zdd2Z?dd4Z@dd6ZAdd:ZBdd;ZCdd<ZDdd=ZEdd>ZFdd?ZGddAZHddDZIddEZJddGZKddHZLdIdIdIdJddMZM dddNZNddOZOddPZPddQZQddIdIde%ddRddaZRddcZSddeZTddfZUddgZVddIdIde%dddhddkZWddmZXddoZYddpZZddqZ[ddrZ\ddxZ]ddzZ^ddd}Z_ dddZ`ddZadS)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: igAg>@g |@g@g^@zOptional[threading.Thread]_monitor_threadc$tdz S)Nzstate.dbr :/home/agent/.hermes/hermes-agent/tools/async_delegation.py_db_pathr|s   z ))rreturnsqlite3.Connectionct}|jddtj|d} t |n##t $r|wxYw|S)NT)parentsexist_ok timeout)rparentmkdirsqlite3connect_initialize_schema Exceptionclose)pathconns r_connectr.s ::DKdT222 ?4 , , ,D4       Ks A A2r-Nonecddlm}||d|dd|dD}dD]$\}}||vr|d |d |%dS) Nr)apply_wal_with_fallbackzstate.db (async_delegation))db_labela+CREATE 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, origin_session_id TEXT NOT NULL DEFAULT '' )ch|] }|d S)r).0rows r 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)AADict[str, Any]cpi} ddlm}dD]\}}||d}|r|||<n#t$rYnwxYw|S)u&Snapshot the dispatching turn's routing origin for the completion event. Captured on the PARENT thread at dispatch time (the daemon worker doesn't carry the contextvars) and persisted with the durable record, so a completion replayed after a restart can reconstruct a full SessionSource even when the session-store origin and in-memory source cache are gone. scope_id matters most: on a relay-fronted deployment the connector's fail-closed egress guard needs the tenant discriminator (or a user binding) to route a scoped reply; without it, post-restart scoped completions bounce with "target not routed to an onboarded tenant" (staging 2026-08-09 defect #4). Best-effort — empty values are simply omitted so CLI/contextvar-unaware paths persist nothing new. rget_session_env))scope_idHERMES_SESSION_SCOPE_ID)user_idHERMES_SESSION_USER_ID) user_nameHERMES_SESSION_USER_NAMEgateway.session_contextrMr*)originrMevt_keyenv_namevalues r_capture_routing_originr[s F  ;;;;;;"  ( ( GX $OHb11E ("'w (      Ms !& 33recordctj} 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)r5keyr\s r z%_persist_dispatch..s:     &== VZZ__ ==r) goalgoalscontexttoolsetsrolemodelis_batchrNrPrRa~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_keyrTorigin_ui_session_idparent_session_id dispatched_atr@) timegateway.statusr^ __import__getpidr*_DB_LOCKrIrCrbjsondumps_prune_durable_records)r\nowr^r: task_payloadr-s` r_persist_dispatchr{s )++C 99999911*T2B2B2I2I2K2KLL             L    <>>  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  EErlstrct5t5}|d|fdddn #1swxYwYddddS#1swxYwYdS)Nz3DELETE FROM async_delegations WHERE delegation_id=?)rurIrCrlr-s r_delete_durable_delegationrs ^^<>>^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) rq_DURABLE_RETENTION_SECONDSrurIrCfetchonemax_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)rycutoffr-terminal_countexcess pending_countoverflows rrxrxs )++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_atrl)rqrurIrCrbrvrw)rrryr-s r_persist_completionr;s8 )++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=?)rurIrCrqr~s r_note_delivery_attemptrGs   <>> T t Y[[- (                                    s4A%*A  A% A A%A A%%A),A)c  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]"}| |r ||||<#dd|dd}|d||tj|tj||f|d z } dddn #1swxYwYdddn #1swxYwY|S)!zEClassify records whose owning process disappeared as outcome unknown.r) _pid_existsr^aSELECT 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_delegationrlrmrnr@rTrorerfrgrhrirjrkrunknownsummaryerrorzLDelegation owner exited before recording a terminal result; outcome unknown.rprrNrPrR)rrrzUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r4)rrrr^r*rqrurIrCfetchallrrvloadsrbboolrw)rr^ry recoveredr-rowsr6rlrm origin_ui parent_idrppidstartedr;r@livetaskr_krs rrecover_abandoned_delegationsrOsFFFFFFFFF 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"; ) )88B<<) $RE"I )dU7^TTF LL,c4:e,,dj.@.@-P     NIIM& ------------------------------\ sB I G H4( I 4H8 8I ;H8 <I  IIc zttj}d}t5t5}|d}|D]\}}}}|p|} | rR|| z t krD|d||ftd||| z dz t dz _tj |} t| trd| d<| | |dz } d d d n #1swxYwYd d d n #1swxYwY|S) uEnqueue 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). Staleness cap: a pending completion older than ``_MAX_COMPLETION_REPLAY_AGE_S`` is terminally dropped instead of replayed. Replaying a weeks-old completion re-runs its parent session as a full-context turn (a July session replayed in August burned a 102K-token context on the staging fleet) for a result nobody is waiting on anymore; the payload stays queryable on the dropped row. rzSELECT delegation_id, event_json, completed_at, dispatched_at FROM async_delegations WHERE state != 'running' AND delivery_state='pending' AND event_json IS NOT NULL ORDER BY completed_at, delegation_idzUPDATE async_delegations SET delivery_state='dropped', delivery_claim=NULL, delivery_claimed_at=NULL, updated_at=? WHERE delegation_id=? AND delivery_state='pending'z|Async delegation %s: pending completion is %.1fh old (cap %.1fh); terminally dropping the replay (result remains queryable).g @Trestoredr4N)rrqrurIrCr_MAX_COMPLETION_REPLAY_AGE_Sloggerwarningrvr isinstancedictput) target_queueryrr-rrlpayloadrrp age_basisevts rrestore_undelivered_completionsrs&"### )++CH <>>T|| 7  (** DH   ?M7L-$5 I cIo1MMM M-( *"C)Ov#=069 *W%%C#t$$ '"&J   S ! ! ! MHH- < Os5D0CD D0D D0 D !D00D47D4rctj}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'r4NrqrurIrCrowcount)rlryr-curs rmark_completion_deliveredrs) )++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)rqrurIrCrr)rlrryr-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.rrrTrl:r_N)rbr|rsrtuuiduuid4hexr)rrrlrs 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=?)rqrurIrC_MAX_DELIVERY_ATTEMPTSrrr)rlrryr-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=?r4Nrrlrryr-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_deliveryr$s, )++C  ! !<>> !Tll ) #}h /   |q  ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! ! !s4B%A( B(A, ,B/A, 0BBBc|rM|ddkr6tt|dpd|dSdSdSNrrrlrT)rbrr|rrs rcomplete_event_deliveryr3saTCGGFOO'999$S)A)A)GR%H%H(SSSSSTT99rc|rM|ddkr6tt|dpd|dSdSdSr)rbrr|rs rrelease_event_deliveryr8saSCGGFOO'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=?rr4rrT) rlorigin_sessionstaterprrdelivery_statedelivery_attemptsr@)rurIrCrrvr)rlr-r6s rget_durable_delegationr=sV <>>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_executorrPs    .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 finalizingr4Nrar5rs r zactive_count..msD  uuX"GGG GGGG  rN _records_locksumrvaluesrrr active_countrcs     ((                       0AA  A rncsdSt5tfdtDcdddS#1swxYwYdS)z9Number of live async delegations owned by one UI session.rc3K|]E}|ddvr,t|dpdkAdVFdS)r>rrrrnrTr4N)rbr|)r5rrns rrz%active_for_session..xsw  uuX"GGGAEE0117R88#$$ $$$$   rNr)rns`ractive_for_sessionrss 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>rrrkrfr4N)rrrrbrlisttuplelen)totalrrfs ractive_task_countrs   ""  AuuX&???uuZ   gz%$'G'GXEXUWXX                   sBB//B36B3rTrmrnrormroc|r(t|dpd|kpS|o't|dpd|kp)|o't|dpd|kS)NrnrTrmro)r|rb)r\rmrnros r_matches_session_selectorsrs  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>rrrrNrbrr5rrnrorms rrz'has_live_for_session..sl    EE(OOD D *'%9"3        rN)ranyrrrs```rhas_live_for_sessionrs 3z+_prune_completed_locked..s? C 55??i ' ' a ' ' 'rNcr|ddp|ddpdS)Nr4rrprra)kvs rz)_prune_completed_locked..s1"Q%))N";";"^r!uyy?Y?Y"^]^r)rc)ritemsrrsortpop)rr _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. rrLHERMES_SESSION_PLATFORMrT api_serverHERMES_SESSION_CHAT_IDrUrLs r_current_origin_session_idrsp$;;;;;; ?4b 9 9\ I I27<<BB rrs( ( 66)rornr@ interrupt_fnmax_async_children progress_fnrergrhOptional[List[str]]rirjrunnerCallable[[], Dict[str, Any]]r@rOptional[Callable[[], None]]rrOptional[Callable[[], tuple]]c ttj|||rt|nd||||| |d tdd| | ddd} t5t dt D}|| krdd| dd cdddS| t <dddn #1swxYwYt| t| }dfd } | t|nh#t$r[}t5t ddddn #1swxYwYtdd |d cYd}~Sd}~wwxYw| tt d|pd|pdddddS)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. N) rlrergrhrirjrmrnr@ror)rrprrr_progress_token _progress_ts_interrupted_atc3HK|]}|ddvdVdSrrrr4Nrars rrz,dispatch_async_delegation..GD  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.rrrr/c i}d} pi}|dpd}nu#t$rh}tdddt |jd|dt tjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) NrrrzAsync delegation %s crashed: rr)rrr api_callsduration_seconds) rbr*r exceptionr__name__roundrq _finalize)rrexcrlrpr s r_workerz*dispatch_async_delegation.._worker[s!# 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): %srTP dispatchedrrlrr/)r rqrr[rrrrr{rsubmitrr*rr_ensure_stale_monitorrinfo)rergrhrirjrmror rnr@rrrr\rexecutorr8r7rlrps ` @@rdispatch_async_delegationrBst'((MIKKM&&.8DNNND" 4.. " # #&$"%)F2 ))  ((      ( ( ($R:LRRR   )))))))) #)!)))))))))))))))$f/00H55555555&   3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 BSBB          KK={-g CRC/@#] C CCsZ#?C/ CC  C 7"D E?$E:+E E:E E:E E:4E?:E?rcxt|}|dS|\}}t|||t||dS)zDMark a record complete and push the completion event onto the queue.N)_begin_finalization_push_completion_event_finish_finalization)rlrrclaimed event_record _interrupt_fns rr6r6sK!-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.Nrr*rrrr)rrrbrqr)rlr\rrHs rrDrDs  $ $m,, >VZZ119PPP  $ $ $ $ $ $ $ $(x!%~zz.11 !%~ $}F||  $ $ $ $ $ $ $ $ $ $ $ $ $ $ $  %%s5B A BB#&B#ct5t|}|||d<tddddS#1swxYwYdS)Nr)rrrbr)rlrr\s rrFrFs ""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]"} || r || | | <#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: %srlNrrrprrrrmrTrnr@rorergrhrirjrr1r2r exit_reason)rrPrstalled_after_quiet_secondsstall_threshold_seconds stall_phasestall_grace_secondszHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryrOr*rrrbrqr5rcompletion_queuer) r\rrrOr7rrrprrrs rrErEs;;;;;;;    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/   C83!! ::b>> !RjCG!! <<RjCGV$$$ )--c22222       JJ ' '          s, A /AA 1J K/K  K)rornr@rrrlrrf List[str]c   p t tjt|}|dkr|dn$|ddd|Dz} |t |||rt |nd||||| |d t dd| d | ddd }t 5td t D}|| krd d | ddcdddS|t <dddn #1swxYwYt|t| }d fd } | t|nh#t$r[}t 5t ddddn #1swxYwYt! d d|dcYd}~Sd}~wwxYw| t#t$d ||pdd dS)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^rN) rlrerfrgrhrirjrmrnr@rorT) rrprrrkrr%r&r'c3HK|]}|ddvdVdSr)rars rrz2dispatch_async_delegation_batch..;r+rr,r-z 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.r.rr/ci}d} pi}|dpg}|rtd|Drd}nd}ns#t$rf}tdgt |jd|ttjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nrresultsc3DK|]}|ddvVdS)r)rsuccessNrars rrzCdispatch_async_delegation_batch.._worker..UsF%%x(@@%%%%%%rrz!Async delegation batch %s crashedr0rr_rtotal_duration_seconds) rballr*rr3rr4r5rq_finalize_batch)combinedr child_resultsr7rlrpr s rr8z0dispatch_async_delegation_batch.._workerNs5#% =vxx~2H$LL339rM %%%&%%%"" %!$      @- P P P II.77#77*/ m0KQ*O*OH FFFFFF  M8V < < < < rr*rrr?rr@)rfrgrhrirjrmror rnr@rrrlrn combined_goalr\rrAr8r7rps ` ` @rdispatch_async_delegation_batchrkssJ"9%7%9%9MIKKM E AFFa1 ; ; ;diiG^G^X]G^G^G^>^>^ ^'e&.8DNNND" 4.. " # #&$"%+F. ))  ((      ( ( ($N:LNNN ))))))))#))))))))))))))))"f/00H========2  3G<<====     . . LL - - - . . . . . . . . . . . . . . ."=111 H3HH          KKKq+0#] C CCsZ6?D DDD "E-- G7G >F& G &F* *G -F* .G G Grfcxt|}|dS|\}}t|||t||dS)zDMark a batch record complete and push ONE combined completion event.N)rD_push_batch_completion_eventrF)rlrfrrGrHrIs rrere|sM"-00G")L- x@@@/////rrHcV 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]"}||r ||||<#dD]}||vr ||||<t|| |j |dS#t$r:}td|d|Yd}~dSd}~wwxYw) z8Push a combined async-delegation batch completion event.rrNzVAsync delegation batch %s finished but process_registry import failed; result lost: %srlNrprrrrmrTrnr@rorerfrgrhrirjrrkTr_live_transcriptsrrc)rcrprrrQzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s) rVrOr*rrrbrqrrWr) rHrfrrOr7rprrrs rrmrmsy;;;;;;;   &   _ - -s    !$$_55DM##N33Bty{{L " ))/::  |'' r::   0 01G L L   \--.A2FF   \--.ABB      ,,  !!'**  <##I..  L$$Z00     ((  !!'**  &  D " 8<< **0b# * HLL);<<+ , g&&- .#+,,/G"H"H&$3   C83''   B   '"2&CG## >>rlCGX&&& )--c22222         _ - -s          s, A /AA I$$ J(./J##J(c>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?s     &?+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``. FrrTr'rlrrNr%r&r_stall_quiet_seconds_stall_threshold_seconds_stall_in_toolrku`Async delegation %s made no progress for %.0fs (in_tool=%s) — interrupting; grace window %.0fsrz.Async delegation %s stall interrupt failed: %s)ruwait_STALE_CHECK_INTERVALrqrrrrb_STALL_GRACE_SECONDSappendr*_STALE_IN_TOOL_SECONDS_STALE_IDLE_SECONDSr5rrrcallabledebug_finalize_stalled)rystalledexpiredany_monitorabler\rinterrupted_atrtokenin_tool quiet_forlimitrl _is_batchfnr7s rryrysD"  !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.Nrrprr|r}r~zAsync 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 %.0fsridlerQrkrbrr)rrrr1r2rP) rDrbrqr5rrrrmrErF) rlrGrHrIrduration quiet_secondsthreshold_seconds stall_in_toolr stall_metas rrr=s!-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)>rrr) startswith)r5kvs rrdz*list_async_delegations..sHAq;;; S))<1;;;rrr*r&r4seconds_since_progressrrl)rr))r|rR)r}rS)r~rNchildren_activityr) rqrrrrrbr5rrr*rr) rysamplersrritemrtsrsrcdstrractivitys rlist_async_delegationsrs4 )++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)rr*rars rrz!interrupt_all..s8   uuX"999 999rNrr4z&interrupt_all: %s interrupt failed: %srlz'Interrupted %d async delegation(s) (%s) rrrrbrr*rrr@)rcounttargetsrrr7s 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)rr*rrrs rrz)interrupt_for_session..s\    uuX"999*'%9"3 : 999rNrr4z.interrupt_for_session: %s interrupt failed: %srlz:Interrupted %d async delegation(s) for ending session (%s)r) rmrnrorrrrrr7s ``` 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)rrrr#) rrrrrusetrsrrtrhrrrv)threads r_reset_for_testsr4s ""    E  * * * ! """""""""""""""   foo// A  s1';??& A<<BB:C!!C%(C%)rr)r-rrr/)rrG)rrJ)r\rJrr/)rlr|rr/r=)rrJrrJrr/)rr)rlr|rr)rlr|rr|rr)rrJrr|rr)rrJrr|rr/)rlr|rr)rrrr)rnr|rr) r\rJrmr|rnr|ror|rr)rTrTrT)rmr|rnr|ror|rr)rr|)rer|rgrrhrrir|rjrrmr|rorr r!rnr|r@r|rr"rrrr#rrJ)rlr|rrJrr|rr/)rlr|rrJ)rlr|rr|rr/)r\rJrrJrr|rr/)rfrXrgrrhrrir|rjrrmr|rorr r!rnr|r@r|rr"rrrlrrr#rrJ)rlr|rfrJrr|rr/)rHrJrfrJrr|rr/)rrryrrr)rr)r)rr|rr)rTrTrTr) rmr|rnr|ror|rr|rr)b__doc__ __future__rrvloggingr'rwrqrconcurrent.futuresr contextlibrtypingrrrr r r hermes_constantsr tools.daemon_poolrtools.thread_contextr getLoggerr4rrr__annotations__Lockrrrr_DEFAULT_MAX_ASYNC_CHILDRENrrrrrrurrrrrsrEventrurr.r)rIr[r{rrxrrrrrrrrrrrrrrrrrrrr rrrBr6rDrFrErkrermr?ryrrrrrrrrrrsL!!!F#"""""  111111%%%%%%@@@@@@@@@@@@@@@@,,,,,,666666<<<<<<  8 $ $ 5+/ ....!!   ')((((-  + 9>  .   .22222 !! ***    )X)X)X)XX$>    F^^^^ $$$$N         6666r4444n ! ! ! !!!!!(TTTT!!!!!!!!H!!!!, ! ! ! !TTTT SSSS &&         4 "       "     2++++    $H(, "15915PDPDPDPDPDPDf0000&&&&*""""H H H H f(, "159#'15BDBDBDBDBDBDJ 0 0 0 0A A A A H    (YYYYxC3C3C3C3L:;;;;|< " 66666rr