(GjO|UdZddlmZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z mZddlmZddlmZejeZeZdaded <ejZdad ed <ejZiZd ed <dZdZd:dZ d;dZ!dd(Z$d?d,Z%d@d.Z&deddAd1Z'dBd3Z(dCd5Z)dDdEd8Z*d=d9Z+dS)Fu 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)AnyCallableDictListOptional)DaemonThreadPoolExecutor)propagate_context_to_threadzOptional[ThreadPoolExecutor] _executorint_executor_max_workerszDict[str, Dict[str, Any]]_records2 max_workersreturnrct5t |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_lockr r_DaemonThreadPoolExecutor)rs zactive_count..cs6PP155??i3O3O13O3O3O3OPPN) _records_locksumrvaluesr&r active_countr+`s QQPPhoo//PPPPPQQQQQQQQQQQQQQQQQQs0AA  A strcHdtjjddS)Ndeleg_)uuiduuid4hexr*r&r_new_delegation_idr3fs" *DJLL$RaR( * **r&Nonec.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#Srrr )r#ridr$s r z+_prune_completed_locked..os? C 55??i ' ' a ' ' 'r&Ncr|ddp|ddpdS)Nr completed_at dispatched_atrr )kvs rz)_prune_completed_locked..ws1"Q%))N";";"^r!uyy?Y?Y"^]^r&)key)ritemslen_MAX_RETAINED_COMPLETEDsortpop) completedr8_s r_prune_completed_lockedrGjs nn&&I  9~~000 NN^^N___Fc)nn/FFFG  Q S$  r&) interrupt_fnmax_async_childrengoalcontext Optional[str]toolsetsOptional[List[str]]rolemodel session_keyrunnerCallable[[], Dict[str, Any]]rHOptional[Callable[[], None]]rIDict[str, Any]c ttj|||rt|nd|||dd|d } t5t dt D} | |krdd|ddcdddS| t <dddn #1swxYwYt|} dfd } | t| nY#t$rL} t5t ddddn #1swxYwYdd | dcYd} ~ Sd} ~ wwxYwt 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. 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) delegation_idrJrKrMrOrPrQrr<r;rHc3LK|]}|ddkdV dSrr r"s rr%z,dispatch_async_delegation..?  AEE(OOy,H,HA,H,H,H,H  r&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.rerrorrr4c i}d} pi}|dpd}nu#t$rh}tdddt |jd|dt tjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nr]rrEzAsync delegation %s crashed: r)rsummaryr] api_callsduration_seconds) r! Exceptionlogger exceptiontype__name__roundtime _finalize)resultrexcrWr<rRs 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): %sP dispatchedrrWrr4)r3rjlistr'r(rr)rsubmitr rdrDreinfo)rJrKrMrOrPrQrRrHrIrecordrexecutorrnrmrWr<s ` @@rdispatch_async_delegationrz|sR'((MIKKM&&.8DNNND"&$  F ))  ((      ( ( ($R:LRRR   ))))))))#))))))))))))))))"/00H55555555&   3G<<====     . . LL - - - . . . . . . . . . . . . . . .!BSBB          KK={-g CRC/@#] C CCsZ ?B- B--B14B1"C33 E =ED, E,D0 0E3D0 4 E>E E rWrlrc4t5t|}| ddddS||d<tj|d<d|d<t |}t dddn #1swxYwYt |||dS)zDMark a record complete and push the completion event onto the queue.Nrr;rH)r'rr!rjdictrG_push_completion_event)rWrlrrx event_records rrkrks  " "m,, >  " " " " " " " ""x!%~!%~F|| !!! " " " " " " " " " " " " " " "<88888sA<=A<<BBrxc  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|dd|dp|dd|d|d|d|ddd|dt||z dd|d |d|d} |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: %srWNrar]r<r;rgasync_delegationrQrprJrKrMrOrPrrbrcr` exit_reasonzHAsync delegation %s: failed to enqueue completion event; result lost: %s) tools.process_registryrrdrer]r!rjricompletion_queueput) rxrlrrrmrar]r<r;evts rr}r}s;;;;;;;    JJ ' '    jj##G JJw  EJJ//>49;;M::n--<L " O44  vzz-44    62&&   6::i((  FJJz**   6""  G$$; 7(;(;  &  7    VZZ Q//  FJJ l]&BA F F   $ % &  ' ( vzz-00) C, )--c22222       JJ ' '          s, A /AA %H I /IIgoals List[str]c ttjt|} | dkr|dn$| ddd|Dz} | t |||rt |nd|||dd|dd } t 5t d tD} | |krd d |d dcdddS| t<dddn #1swxYwYt|} dfd } | t|nY#t$rL}t 5t ddddn #1swxYwYd d|dcYd}~Sd}~wwxYwtd| |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. rrz parallel subagents: z; c3*K|]}|ddVdS)N(r*)r#gs rr%z2dispatch_async_delegation_batch..\s+G^G^ST#2#G^G^G^G^G^G^r&NrT) rWrJrrKrMrOrPrQrr<r;rHis_batchc3LK|]}|ddkdV dSrr r"s rr%z2dispatch_async_delegation_batch..nrYr&rZr[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\rr4ci}d} pi}|dpg}|rtd|Drd}nd}ns#t$rf}tdgt |jd|ttjz dd}d}Yd}~nd}~wwxYwt||dS#t||wxYw) Nr]resultsc3DK|]}|ddvVdS)r)rEsuccessNr r"s rr%zCdispatch_async_delegation_batch.._worker..sF%%x(@@%%%%%%r&rEz!Async delegation batch %s crashedr_r`)rr]total_duration_seconds) r!allrdrerfrgrhrirj_finalize_batch)combinedr child_resultsrmrWr<rRs rrnz0dispatch_async_delegation_batch.._workers5#% =vxx~2H$LL339rM %%%&%%%"" %!$      @- P P P II.77#77*/ m0KQ*O*OH FFFFFF  M8V < < < < ^>^ ^'e&.8DNNND"&$F ))  ((      ( ( ($N:LNNN ))))))))#)))))))))))))))) /00H========2  3G<<====     . . LL - - - . . . . . . . . . . . . . . .!H3HH          KKKq+0#] C CCsZ?C>( C>>DD!"E FFE=1 F=F FF  FFFrct5t|}| ddddS||d<tj|d<d|d<t |}t dddn #1swxYwY ddlm}n4#t$r'}t d||Yd}~dSd}~wwxYw|dptj}|dptj}id d d |d |d d d|dd d|dd|dd|dd|dd|dd|ddd|dpgd|dd|dd|d|} |j | dS#t$r'}t d||Yd}~dSd}~wwxYw)zDMark a batch record complete and push ONE combined completion event.Nrr;rHrrzVAsync delegation batch %s finished but process_registry import failed; result lost: %sr<rgrrWrQrprJrrKrMrOrPrTrr]rzNAsync delegation batch %s: failed to enqueue completion event; result lost: %s) r'rr!rjr|rGrrrdrer]rr) rWrrrxr~rrmr<r;rs rrrsV ""m,, > """""""""x!%~!%~F|| !!!""""""""""""""";;;;;;;   & 3    !$$_55DM##N33Bty{{L "   |'' r::     ,,   !!'**   <##I..   L$$Z00     ((  !!'**  &  D  8<< **0b  g&&  !(,,/G"H"H! " # $  % C( )--c22222       3          sFA<=A<<BBB CB;;CH$$ I.IIList[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)rHr*)r#kvs r z5list_async_delegations...s( ? ? ?da1+>+>Q+>+>+>r&)r@r"s rr9z*list_async_delegations..s?    @ ?aggii ? ? ?   r&N)r'rr)r*r&rlist_async_delegationsrs     __&&                     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|Sr7r r"s rr9z!interrupt_all..s3   AEE(OOy,H,HA,H,H,Hr&NrHrz&interrupt_all: %s interrupt failed: %srWz'Interrupted %d async delegation(s) (%s)) r'rr)r!callablerdredebugrw)rcounttargetsr$fnrms r interrupt_allrse E     ((                     UU> " " B<<       <EE/**C    N =ufMMM Ls%$:>>.A>> C/B<<Cct5ttddadadddn #1swxYwYt5t ddddS#1swxYwYdS)z6Test-only: clear all state and tear down the executor.NF)waitr)rr rrr'rclearr*r&r_reset_for_testsr s ""    E  * * * ! """"""""""""""" s';?? A44A8;A8)rr rr)rr )rr,rt)rJr,rKrLrMrNrOr,rPrLrQr,rRrSrHrTrIr rrU)rWr,rlrUrr,rr4)rxrUrlrUrr,rr4)rrrKrLrMrNrOr,rPrLrQr,rRrSrHrTrIr rrU)rWr,rrUrr,rr4)rr)r)rr,rr ),__doc__ __future__rlogging threadingrjr0concurrent.futuresrtypingrrrrr tools.daemon_poolr tools.thread_contextr getLoggerrhrerr __annotations__Lockrrr'r_DEFAULT_MAX_ASYNC_CHILDRENrBrr+r3rGrzrkr}rrrrrr*r&rrso!!!F#""""" 11111166666666666666666666<<<<<<  8 $ $ 5+/ ....!!   ')((((&QQQQ ++++    6269qDqDqDqDqDqDh 9 9 9 9 4 4 4 4 @269pDpDpDpDpDpDf5 5 5 5 p     8      r&