߲jw UdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl Z ddl m Z ddl mZmZmZmZmZddlmZddlmZddlmZej2eZdad ed <ej<Zda d ed <ej<Z!iZ"d ed<dZ#dZ$dZ%dZ&dZ'dZ(ej<Z)dZ*dZ+dZ,dZ-ej<Z.da/ded<ej`Z1hdZ2dZ3dZ4dZ5dZ6d Z7d\d!Z8d]d"Z9d#Z:d^d$Z;d_d%Zdbd(Z?dcd)Z@ddd*ZAddd+ZBded,ZCdfd-ZDdgd.ZEdhd/ZFdid0ZGdgd1ZHdgd2ZIdgd3ZJdgd4ZKdjd5ZLdjd6ZMdjd7ZNdkd8ZOdld9ZPddd:ZQddd;ZRdmd<ZSdndod>ZTdpd?ZUd`d@ZVdpdAZWdqdBZXdqdCZYdrdDZZd^dEZ[ddddF dsdGZ\dd=d=de#ddH dtdIZ]dd=d=de#ddddddJ dudKZ^dvdLZ_dwdMZ`dxdNZad`dOZbdydPZcdzdQZdd`dRZed{dSZfd|dTZgd}dUZhd~dVZiddWZjdddXZk d ddYZld`dZZmdd[Zny)a Async (background) delegation registry behind ``delegate_task(background=true)``. The parent dispatches a subagent on a module-level daemon executor and returns a handle immediately. On completion a ``type="async_delegation"`` event (self-contained task-source block) is pushed onto the SHARED ``process_registry.completion_queue`` the CLI/gateway drain while idle, so results surface as a NEW turn (never mid-turn) and inherit its de-dup and crash-recovery wiring. Only the async lifecycle lives here; the child run is an injected ``runner``.) annotationsN)ThreadPoolExecutor)AnyCallableDictListOptionalget_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_thread>runningstalling finalizing)rr)scope_iduser_id user_namestalled_after_quiet_secondsstall_threshold_seconds stall_phasestall_grace_seconds))_stall_quiet_secondsr)_stall_threshold_secondsr)_stall_in_tool stall_in_toolctdz S)Nzstate.dbr ;/home/ubuntu/.hermes/hermes-agent/tools/async_delegation.py_db_pathr)Qs  z ))r'cddlm}ddlm}t }|j j dd||d||ddd dt }|||S) Nr)open_db)_secure_state_db_filesT)parentsexist_ok) create_mainzstate.db (async_delegation)i'F)db_labelbusy_timeout_mswal row_factory initialize)hermes_cli.sqlite_utilr+ hermes_stater,r)parentmkdir_initialize_schema)r+r,pathconns r(_connectr<UsZ.4 :DKKdT24T2 4"?QW$;M OD4 Kr'c<ddlm}ddlm}||||y)Nr)apply_durability_barriers)reconcile_state_schema)hermes_state_repairr>hermes_state_schemar?)r;r>r?s r(r9r9fs=:d#4 r'c.ddlm}|tS)Nr transaction)r5rDr<rCs r( _transactionrEvs2 xz ""r'c ddlm}tDcic]"}|d|jdx}s ||$c}Scc}w#t$ricYSwxYw)zSnapshot scope_id/user_id/user_name on the PARENT thread (the daemon worker has no contextvars) so a restart-replayed completion can rebuild a SessionSource. Best-effort: empty values are omitted.rget_session_envHERMES_SESSION_)gateway.session_contextrH _ROUTING_KEYSupper Exception)rHkvs r(_capture_routing_originrQ|sZ;+i_WXW^W^W`VaEbdf5g0g0g1iii  s?!::?? A  A ctj} ddlm}|tj}ddddddd d d g t Dcic]}||vr||j|}} tj|d <t5t5}|jd |d|jdd|jdd|jd|d|tj|tj||jddf ddddddty#t $rd}Y wxYwcc}w#t$rYwxYw#1swYBxYw#1swYtyxYw)Nr)get_process_start_timegoalgoalscontexttoolsetsrolemodelis_batch task_indexestask_transcripts owner_cwda~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_keyrJorigin_ui_session_idparent_session_id dispatched_atorigin_session_id)timegateway.statusrSosgetpidrNrLgetgetcwdOSError_DB_LOCKrEexecutejsondumps_prune_durable_records)recordnowrSowner_started_atkey task_payloadr;s r(_persist_dispatchrus ))+C 91"))+> GY FGZYgi{MLM  &= VZZ_L $&IIK [!  M<> MT Q O $fjj&CVZZPfhjEk ZZ+ ,f_.EsBIIKYi ZZ %vzz2Er'J L  M M M)        M M MsN D4 E*E  E&BEE&4 EE EEE# E&&E9ctjtz }t5t5}|j d|f|j dj d}|t kDr|j d|t z f|j dj d}|tkDr|j d|tz 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_SECONDSrkrErlfetchone_MAX_RETAINED_COMPLETED_MAX_DURABLE_PENDING)cutoffr;terminal_count pending_counts r(roros YY[5 5F B<>BT _bhaj l `bbjbjblmnp 3 3 LL )+BBD  F  &[\\d\d\fghj / / LL(*>>@  BBBBBBBs# C(BC C(C% !C((C1c 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^)rdrkrErlrhrmrn)eventresultrqr;s r(_persist_completionrs ))+C M<>MT (YYx -uyy/Ms ZZ  6 2E/4J L MMMMMMMs# B%A"BB%B" B%%B.c  t5t5}|jd|fj}| ddddddyt j |dxsdxsi}|j dxsgDcgc](}|j d|j dk7s'|*}}|j||jdt j|dd tj|fddddddycc}w#1swYxYw#1swYyxYw#t$r+tjd ||j dd YywxYw) azDurably record ONE finished child of a still-running multi-child unit on the unit's own row, so a crash before the unit joins loses only the children that had not finished. Stored in ``result_json`` (overwritten by the real result at finalize); ``recover_abandoned_delegations`` replays it. Best-effort: a failed write costs recovery fidelity, never the live result.zSSELECT result_json FROM async_delegations WHERE delegation_id=? AND state='running'Nr{}results task_indexzbUPDATE async_delegations SET result_json=?, updated_at=? WHERE delegation_id=? AND state='running'T)rpartialz7Async delegation %s: could not record finished child %s)exc_info) rkrErlrxrmloadsrhappendrnrdrNloggerwarning)r^entryr;rowrrrs r(record_unit_childrs] I  j|~ j,,t -/119 { j j j jjQ406BG")++i"8">BqQ!%% BUY^YbYbcoYpBpqqGq NN5 ! LL}**T%JKTYY[Zgh j j j j r j j j j IPR_afajajkwaxDH IIsoD+ D%DDD+ 7D(D)D-AD=DD+DD DD($D+(D++1EEc tj|xsdxsi}|jdr"|jdr|jdsy|dDcic](}t|jdts#|d|*}}|jdxs0t t t|jdxsg}|Dcgc]}|j|xs|d d|d c}Scc}wcc}w) zaPer-task results for an abandoned unit: recorded children as they finished, the rest ``unknown``.rrZrrNrr[rUunknown)rrsummaryerror)rmrrh isinstancerlistrangelen)task result_jsonrrrrecordedindexesis r(_recovered_resultsrsjj,-3G HHZ W[[%; I@V,3I,>gq*QUUS_M`beBf,"gHghh~&S$uS'9J9Pb5Q/R*SGpw xklHLLO fa9QU`ef f xxh xs$C3>C3!C8cb ddlm}m}m}t j d}}t 5t5}|jdj}|D]}|\ }} } } } } }}}}}| r0|t| r||||t| xsdrDtj|xsd}d}t|||}|r$td|D}d|d t|d }||j!d xsid }dd lm}m}||d x}r||d<||j!dx}r||d<iddd|d| d| d|xsdd| d|j!ddd|j!dd|j!dd|j!dd|j!dd|j!ddt)|j!dd d!d"dd#|||rd$|ini| |d%t*Dcic]}|j!|s|||c}}d!d|d#d&||rd$|ini}|jd'||tj,|tj,||f|d(z } dddddd|S#t$rYywxYwcc}w#1swY'xYw#1swY|SxYw))zClassify records whose owning process disappeared as outcome unknown; children a multi-child unit had already recorded (``record_unit_child``) are replayed with their real results.r) _pid_existsrSstart_time_fingerprints_matcha%SELECT delegation_id, origin_session, origin_ui_session_id, parent_session_id, dispatched_at, owner_pid, owner_started_at, task_json, origin_session_id, result_json, state FROM async_delegations WHERE state IN ('running','finalizing')NrzLDelegation owner exited before recording a terminal result; outcome unknown.c3JK|]}|jddk7sdyw)rrNrh.0rs r( z0recover_abandoned_delegations..sX155?i;W1Xs##z2Delegation owner exited before the unit finished; /zJ child results were recorded and are included below, the rest are unknown.r\)last_known_statusr\)git_state_hinttranscript_tailsrr]rtypeasync_delegationr^r_r`rcrJrarTrUrVrWrXrYrZrrrrrrbr)rrrzUPDATE async_delegations SET state='unknown', completed_at=?, updated_at=?, event_json=?, result_json=?, delivery_state='pending' WHERE delegation_id=?r)rerrSrrNrdrkrErlfetchallrrmrrsumrrh%tools.async_delegation_recovery_hintsrrboolrLrn)rrSrrq recoveredr;rowsrr^r_ origin_ui parent_idrbpidstarted task_json origin_sidr last_staterrrecovered_resultsdone diagnosticsrrtailshintrOrrs r(recover_abandoned_delegationsrseeYY[!C )<>)T||QRS[RZR\ $ CMP JM; 9mSRY[dfpr}@J{3s8,#@J`adehaiJjJono#p::i/40DbE 24e L  X&7XXMdVSTUXYjUkTlm__0:PTPXPXYkPlPrprsK _(5G)HIIuI27 ./%dhh{&;<u F IT F6GI01B F"/ F(5D! 1d1g:D FE!*dU7^WWbW;L)%67RTWF LL,.13 58I4::V\K]_l-m o NII$  ))T [ LEG)))T sM J J$GJ6J J AJ2J$ JJJJ! J$$J.c 0ttjd}}t5t5}|j dj }|D]\}}}}|xs|} | rF|| z t kDr:|j d||ftjd||| z dz t dz Vtj|} t| trd| d<|j| |dz } d d d d d d |S#1swYxYw#1swY|SxYw) uEnqueue durable pending completions as fresh turns after process start. Restored events are stamped ``restored=True`` in memory only: they came from a PREVIOUS process, so drains without an ownership filter must leave them for a consumer that can prove ownership. Rows older than ``_MAX_COMPLETION_REPLAY_AGE_S`` are terminally dropped instead of replaying a turn nobody is waiting on. 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). 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 @TrestoredrN)rrdrkrErlr_MAX_COMPLETION_REPLAY_AGE_Srrrmrrdictput) target_queuerqrr;rr^payloadrrb age_basisevts r(restore_undelivered_completionsrs3"#IIKC <>T||789A  DH  ?M7L-$5 IcIo1MM MPSTaNbd i,sY&.HJfioJoq**W%C#t$"&J   S ! MH  * O+* Os# D B:C?-D ?D D  Dct5t5}|j||jdk(cdddcdddS#1swYnxYwdddy#1swYyxYw)z?Run one UPDATE on the ledger; True iff exactly one row changed.rN)rkrErlrowcount)sqlparamsr;s r(_update_deliveryrAsS 7<>7T||C(11Q67777777s! AA AA AA"cHtj}td|||fS)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'rdr)r^rqs r(mark_completion_deliveredrGs- ))+C  DFI3P]E^ ``r'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)rdrkrErlrxr)r^claim_idrqr;rcurs r(claim_completion_deliveryrOs ))+C  !<> !Tll PS`Rbddldldn ; ! ! ! llLsCc : < ||q  ! ! ! ! ! ! !s. B,%B B,(B B,B B,,B5cb|jddk(xrt|jdS)a An early per-task notice for a batch that is still running. It shares the batch's ``delegation_id`` but is NOT the durable completion: it must never claim, acknowledge or dedup against the final result's row (independent review reproduced exactly that loss).rrtask_failure_notice)rhr)rs r(is_interim_delegation_eventr_s, 776?0 0 YT#''BW:X5YYr'ct|ry|jddk(rt|jdxsdnd}|sy|dtjdt j j}t||r|SdS)zYClaim a durable delegation event; non-durable events (and interim notices) need no token.rJrrr^:N) rrhstrrfrguuiduuid4hexr)rconsumerr^rs r(claim_event_deliveryrfs{"3';>776?N`;`C06B7fhM 1RYY[M4::<+;+;*<=H0I8StSr'c tj}t5t5}|jd|||tf}|j dk(r-t jd|t ddddddy|jd|||f}|j dk(cdddcdddS#1swYnxYwdddy#1swYyxYw)zRelease a failed delivery claim so another consumer may retry. Attempts are counted at claim time; once the budget is exhausted the row converges to terminal ``dropped`` (only pending rows replay on restart).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>=?rznAsync 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=?)rdrkrErl_MAX_DELIVERY_ATTEMPTSrrr)r^rrqr;cappedrs r(release_completion_deliveryrqs ))+C !<>!TB-+A B D ??a  NNT(*@ B!!!ll),/ x*HJ||q !!!!!!!s/ CAB3,C=#B3 C3B< 8CCcDtdtj||fS)zOReturn an unadmitted completion to pending without spending a delivery attempt.zUPDATE async_delegations SET delivery_claim=NULL, delivery_claimed_at=NULL, delivery_attempts=MAX(0, delivery_attempts-1), updated_at=? WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?rrdr^rs r(defer_completion_deliveryrs' V mX.  00r'cDtdtj||fS)u8Terminally drop a claimed completion whose target is permanently gone (the spawning session ended at an explicit user boundary such as /new or reset). ``dropped`` — not ``delivered`` — keeps the ack honest; not ``pending`` keeps restart recovery from replaying it into a fail-closed drop forever.zUPDATE 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=?rrs r(drop_completion_deliveryrs* %(,yy{M8&L  NNr'cJtj}td||||fS)z;Acknowledge acceptance for the consumer holding this claim.aUPDATE async_delegations SET delivery_state='delivered', delivered_at=?, updated_at=?, delivery_claim=NULL, delivery_claimed_at=NULL WHERE delegation_id=? AND delivery_state='pending' AND delivery_claim=?r)r^rrqs r(complete_completion_deliveryrs/ ))+C %(+C&I  KKr'c&tt||yN)_event_deliveryrrrs r(complete_event_deliveryrs0#x@r'c&tt||yr)rrrs r(release_event_deliveryrs/h?r'c~|r;|jddk(r&|t|jdxsd|yyy)Nrrr^rJrhr)fnrrs r(rrs9CGGFO'99 3sww'-2 .9:xr'c Dt5t5}|jd|fj}dddddddS||d|d|d|d|drt j |dnd|d|d|d xsd d S#1swYWxYw#1swY[xYw) NzSELECT origin_session, state, dispatched_at, completed_at, result_json, delivery_state, delivery_attempts, origin_session_id FROM async_delegations WHERE delegation_id=?rrrrJ) r^origin_sessionstaterbrrdelivery_statedelivery_attemptsrc)rkrErlrxrmr)r^r;rs r(get_durable_delegationrs ]<>]Tll?BO@PRS[RZR\ ]] ;4H&#a&3q6dghidjAA$**SV*.sS155?l3R1S&&N _records_lockrrvaluesr&r'r( active_countrs3 TShoo/SSTTT$5>ct5tdtjDcdddS#1swYyxYw)uNumber of running child subagents (a batch of N contributes N; a batch with no goal list counts 1) — the truthful observability figure, unlike slots.c3K|]u}|jddvr`|jdrKt|jdttfr&|dr!t |jdxs|dndwyw)r>rrrZrUr[rN)rhrrtuplerrs r(rz$active_task_count..sxX!%%/=V*VuuZ Zgu %NSTU\S] n%37 4cd eXsA;A=Nrr&r'r(active_task_countrsA XX__&XXXXXrc2d|fd|fd|ffDcgc] \}}|r||f }}}|sgSt5tjDcgc],jd|vst fd|Dr.c}cdddScc}}wcc}w#1swYyxYw)uJRecords in ``statuses`` owned by a session: any non-empty selector claims the record — ``origin_ui_session_id`` (TUI tab), ``session_key`` (routing key at dispatch), or ``parent_session_id`` (spawner's durable id — the right one for gateway chats, whose session_key survives ``/new`` while the session id rotates).r`r_rarc3fK|](\}}tj|xsd|k(*yw)rJN)rrh)rfieldwantedrs r(rz#_session_records..s,YmeVAEE%L.B/69Ys.1N)rrrrhany)statusesr_r`rarr selectorsrs `r(_session_recordsr"s !56 8T /072=]UF5;=I=  [#??,[ah80KYyYY[[[ = [[[s(BB BB6B B  BrJc8ttt|||S)zOWhether a session still owns any live (running/stalling/finalizing) delegation.)rr"r)r_r`ras r(has_live_for_sessionr$s  {z)_prune_completed_locked..s+"Q%))N";"^r!uyy?Y"^]^r')rsNr) ritemsrhrsortmaxrrypop)ridrr_s r(_prune_completed_lockedr2s)1(8`fc1AEE(OS_<_#q`I` NN^N_Nc!S^6M%MNO Q S$ as BBch ddlm}|dddk(}|r |ddxsdSdS#t$rYywxYw)aRaw session id of the ORIGINATING api_server request, or ``""``. ``HERMES_SESSION_ID`` is unsafe here: building the child agent calls ``set_current_session_id(child.session_id)`` just before dispatch, so the wake would self-post into the subagent's own session. The request-scoped ``HERMES_SESSION_CHAT_ID`` (raw X-Hermes-Session-Id on api_server) survives child construction; on push platforms chat_id is a chat, not a session => ``""``.rrGHERMES_SESSION_PLATFORMrJ api_serverHERMES_SESSION_CHAT_ID)rKrHrN)rHis_apis r(_current_origin_session_idr8sI ; !:B?<OHN 8"=CVTVV s %% 11cdd|d|dS)Nrr)rrr api_callsduration_secondsr&rdurations r( _single_crashr>s$Qdl mmr'cg||dS)N)rrtotal_duration_secondsr&r<s r( _batch_crashrAsEX NNr'cf|jdxsg}d|rtfd|DrdSdS)zCBatch status: completed unless every child errored/was interrupted.r)rsuccessc3DK|]}|jdvyw)rNr)rroks r(rz _batch_status..s+]!AEE(O2,E+]s rr)rhall)combined child_resultsrEs @r( _batch_statusrIs6LL+1rM !B#+]}+](]7ncnnr'c ddlm}|j5}|sdddcdddStdi|cdddS#1swYyxYw)Nr retirementrejectedz*backend is retiring; reconnect to continuerrr&)hermes_cli.backend_retirementrLwork_dispatch_admitted)kwargsrLadmitteds r( _dispatchrTsI8  ,h(3_`,,"+F+,,,s < <A)slot_keyr[r\c x  !|du}|rdnd |rtnd|rtnttj|d|r dt |ini||r t |nd|||| | |dt dd| d |rd d ini||xsd |r d t |ini| dt |initjddd}t5tjDchc]2}|jdtvs|jdxs|d4}}|d|vrt|| k\rd|dcdddS|t<tdtjD}dddt!|t#t%| }d fd }ddlm!!j+ |j-t/|}|j1!fd| t?ddScc}w#1swYxYw#t2$r}!j5t5tj7ddddn #1swYnxYwt85t;5}|j=dfdddn #1swYnxYwdddn #1swYnxYwdd d|dcYd}~Sd}~wwxYw) a0Shared dispatch core for single (``goals is None``) and batch units. Capacity check + record insert happen under ONE lock hold so concurrent dispatches can't both pass the check and exceed the cap. At capacity the dispatch is REJECTED (never queued) so a runaway model can't pile up unbounded background work. ``slot_key`` names the pool slot the unit occupies (default: its own id); the units of one delegate_task call share the first unit's id so splitting a call into per-group completions never consumes more capacity than the call did.N batchrJc,|jdxsdS)Nrrr)rs r(r+z$_dispatch_admitted..9sx9WKr')r^rTrU)rVrWrXrYr_r`rcrar)rrbr interrupt_fnrZT) progress_fnrUr\r[)_context_progress_token _progress_ts_interrupted_atrrUr^rMrNc3PK|]}|jdtvsd ywrrrs r(rz%_dispatch_admitted..RsYqxL9XYrci}d}t5tj}|%|jdt jddd  xsi}|}t||y#1swY+xYw#t $rd}t jddt|jd|tt jz d}Yd}~d}~wwxYw#t||wxYw)NrT)_startedr]Async delegationz %s crashed: r) rrrhupdaterdrNr exceptionr__name__round _finalize) rrrecexcclassify crash_resultr^rblabelrunners r(_workerz#_dispatch_admitted.._workerXs!#  D,,}-C Dtyy{ C  D  5X^Ff%F mVV 4 D D i   /wkBM R!T#Y%7%7$83%"@% VcHcefBghF i mVV 4s6=A3A?3A<? C,AC'"C/'C,,C//C>rrKc$jSr)release)r1rLs r(r+z$_dispatch_admitted..ps:+=+=+?r'z3DELETE FROM async_delegations WHERE delegation_id=?z#Failed to schedule async delegationrc dispatched)rr^returnNone) rIrAr>rdrrQr contextvars copy_contextrrrrh_ACTIVE_STATESrrrur r.rOrLacquiresubmitr add_done_callbackrNrqr/rkrErl_ensure_stale_monitor)"r^rTrUrVrWrXrYr_rarnr`rcrYmax_async_childrenrZcapacity_errorrUr[r\rZrpr active_slots live_unitsexecutorrofuturerjr;rkrlrbrmrLs"` ` @@@@@r(rQrQ(sD H HbE (}/WH#+<LIIKM&YS['4;9OacY(XVZej"BZZ#)YHOO$5YY Z fS!3Z@AH55"9 e!'G9G9G>-G>:.H 9G>>H J9J4)I  J4 I  J4 J&J: JJ J J4J! J4.J94J9)rar`rcrYr}rZc t} tdid| d|ddd|d|d|d|d |d |d |d |d | d| d| d| dd| d}|ddk(r#tjd| |xsd|xsddd|S)aSpawn ``runner`` on the daemon executor and return a handle immediately. ``session_key``/``parent_session_id`` are captured on the parent thread (the worker carries no contextvars) and route the completion back to the spawning session. ``progress_fn() -> (token, in_tool)`` enables stale monitoring; omitted = unmonitored. Returns ``{"status": "dispatched", "delegation_id"}`` or ``{"status": "rejected", "error"}``.r^rTrUNrVrWrXrYr_rarnr`rcrYr}rZr~#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.rrrz3Dispatched async delegation %s (session_key=%s): %srJPr&)r'rTrinfo)rTrVrWrXrYr_rarnr`rcrYr}rZr^handles r(dispatch_async_delegationr}s '(M  w# w*. w6: wDK w w $ w,1 w?J w, w5; w2 wFW w " w 7I w Wb w22D1EFu u wFh<' I!;#9'DJB;L N Mr') rar`rcrYr}r^rZrUr[r\c| xs t} ||Dcgc]}|| c}n t|}t|}|dk(r|dn|ddjd|Dz}t did| d|d|d |d |d |d |d |d|d|d|d| d| d| d| d|d|d|dd| d}|ddk(rt j d| ||xsd|Scc}w) uDispatch a fan-out unit (a whole batch, or one ``group`` of a delegate_task call) as ONE background unit: ``runner`` runs its tasks and returns the combined ``{"results": [...], "total_duration_seconds": N}`` dict. The unit occupies ONE async slot — or joins the slot named by ``slot_key`` (in-unit parallelism is bounded separately) — and produces a SINGLE completion event carrying per-task ``results``.rrz parallel subagents: z; c3&K|] }|dd yw)N(r&)rgs r(rz2dispatch_async_delegation_batch..sXtdeYZ[^\^Y_Xtsr^rTrUrVrWrXrYr_rarnr`rcrYr}rZrUr[r\r~rz 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.rrrzADispatched async delegation batch %s (%d task(s), session_key=%s)rr&)r'rrjoinrTrr)rUrVrWrXrYr_rarnr`rcrYr}r^rZrUr[r\r unit_goalsn combined_goalrs r(dispatch_async_delegation_batchrs"9%7%9M5A5ML1q%(1SWX]S^J JA%&!VJqMA36K1LtyyXtisXtOt1tM  G# G*7 G?D GNU G G $ G,1 G?J G, G5; G2 GFW G " G 7I G Wb G mu G " G 5E G22D1EFE E GFh<' W!1k& M#2s Cct5tj|}||jdtvr dddyd|d<t j|d<d|d<d|d<t |}dddt t|r||n||t5|tvr |t|d<tdddy#1swYWxYw#1swYyxYw)aAtomically claim terminal delivery, push the completion event, then mark ``status``. ``result`` is a dict or a callable receiving the record snapshot (stall path). The record stays active ("finalizing") until durable persistence and queue publication finish; otherwise process shutdown can kill this daemon worker after status flips but before SQLite commits. A second call for the same id (late runner return after a forced stall) is a no-op.NrrrrYrZ) rrrhrxrdr_push_completion_eventcallabler2)r^rrrpsnapshots r(rhrhs  m, >VZZ1G   (x!%~!%~ $}< 8&9IVH%5vW]^ " H $06H] #H -!""  ""s0C1CCCCc t|jd}|rdnd} ddlm}|jd xstj}|jd xstj}|rcd |jd xsg|jd|jd|jdd|jdd|dini} nR|jd|jd|jdd|jdt||z dd} d|jd|jdd|jdd|jdd|jd|jddd|rd|jdini|jd |jd!|jd"|r|jd#n#|jd#xs|jd#|d$| ||d%|rind&|jd&itD cic]} |j| s| || c} tD cic] } | |vs| || c} } t| | |jj| y #t$r4}t j d|d|jd|Yd }~y d }~wwxYwcc} wcc} w#t$r4}t j d|d'|jd|Yd }~d }~wwxYw#t$r4}t j d|d(|jd|Yd }~y d }~wwxYw))agPush a type='async_delegation' event onto the shared completion queue. Batch records (``is_batch``) carry the per-task ``results`` list (plus live transcript paths, the full-fidelity record of each child's run) instead of a single summary. Best-effort: failure must not crash the worker, but it WOULD mean a silently-lost result, so we log loudly.rZrWrJrprocess_registryrbz@ %s finished but process_registry import failed; result lost: %sr^NrbrTrlive_transcriptsrr@)rZrrrr@grouprr:r;r)rrr:r;rr_r`rcrarT)rr^r_r`rcrarTrUrVrWrXrY)rVrWrXrYrr exit_reasonzo %s: durable completion write failed; delivering in-memory only (a restart may report this unit as unknown): %sz8 %s: failed to enqueue completion event; result lost: %s)rrhtools.process_registryrrNrrrdrgrL_STALL_META_KEYSrcompletion_queuer) rprrrZrmrrjrbrrrOrs r(rrss FJJz*+H HbE; JJ/>499;M::n-<LI)>)D" & += >T[I\&,jj1I&JW.4ZZ-@-L)RT Wzz),vzz'7JY_YcYcdoqrYs & +=u\TaEacd?e fh#VZZ5Pzz-4 & +A2 F#ZZ(;R@#ZZ(;< 62& CNVGVZZ5H+I[] C::i(fjj6LV\V`V`agVh(0G$vzz'7J7afjjY`Na C$ C7DUa C2 vzz-/HI C"/ @A&**Q-1fQi< @ C"2 AAQ&[1fQi< A CCoC(J))--c2I  'w/''(. ?(CS J8 A A o 'w/LLMSZZXgMhjm o oo J 'w/''(. ?(CS J JJs_J%;K%K%% K*/K*< K/ L/% K".*KK"/ L,8*L''L,/ M,8*M''M,c2t5tj|}||jdtvr dddyt |}ddd ddlm}idddd d d d |d |d |gdjddd|jddd|jddd|jdd|jddd|jdd|jdd|jdd|jdd|jddd|jdxstjtjdtDcic]}|j|s|||c}} |jj|y#1swYdxYw#t$r!}tjd||Yd}~yd}~wwxYwcc}w#t$r!}tjd||Yd}~yd}~wwxYw)aSurface ONE failed child of a still-running detached batch to the parent now, instead of when the slowest sibling finishes. In a 1,393-agent run every wave-1 child died in a 401 storm at 08:29 and the parent learned of it at 09:36, when the batch's "unknown outcome" block finally arrived: 66 minutes of a dead wave with nothing running. The notice rides the same ``type="async_delegation"`` event shape as the batch result (so every drain/route/format path treats it identically) with ``task_failure_notice=True`` and a single-entry ``results`` list; the batch record is NOT finalized and its consolidated result still arrives as before.NrrrzTAsync delegation batch %s: task failure notice dropped (process_registry import): %srrrTrZn_tasksr^rr_rJr`rcrarTrUrVrWrXrYrrbrzDAsync delegation batch %s: failed to enqueue task failure notice: %s)rrrhrxrrrrNrrrdrLrr) r^rrrprrrjrOrs r(push_task_failure_noticerss  m, >VZZ1G   <  ; F" F$94 FAKT FS\^e F F(1E7 F x||M26 F  -CR H F X\\*=rB F X\\*=> F  VR( F+28<<3H FKTU]UaUabkUl F HLL, F/5hll66J FMTU]UaUabiUj F ) F/7ll?.K.Ztyy{lplululw F$1 DaHLLO1hqk> D FCq))--c2-    kmz| A E q []jloppqsM0F- F-F:-G'G'G,-F7: G$GG$, H5HHc t5ttjr dddytj t j tddatjdddy#1swYyxYw)zStart (once) the stale-delegation monitor thread. One daemon thread serves every dispatch; it exits when no monitorable records remain and is restarted by the next dispatch with a ``progress_fn``.Nzasync-delegate-stale-monitorT)targetnamedaemon) _monitor_lockris_alive _monitor_stopclear threadingThread_stale_monitor_loopstartr&r'r(r|r|+sn   &?+C+C+E    #**&-KTXZ    sA9AA99Bc ggd}}}tjD]}|jd}|dk(r6d}||jdxs|z tk\r|j |dP|jd}|dk7s|id}|jd s} |\}}||jd k7r|j || ||jd xs|z } |rtnt} | | k\s|j d|t| d | t||j |d| |f|||fS#t $r|jd d}}YwxYw)aOne monitor pass over ``_records``; caller holds ``_records_lock``. Returns ``(stalled, expired, any_monitorable)``: newly-stalling ``(delegation_id, quiet_for, in_tool)`` tuples, stalling ids past the grace window, and whether anything is left to monitor.FrrTr^r^rZrrar\)r\r]r]r)rr^r!r"r#) rrrh_STALL_GRACE_SECONDSrrNrd_STALE_IN_TOOL_SECONDS_STALE_IDLE_SECONDSrgr) rqstalledexpiredany_monitorablerprrZtokenin_tool quiet_forlimits r(_sweep_stale_lockedr9s)+BoWG//#JH% Z "Ofjj!23:s;?SSvo67 jj/ Y +"5 zz*%  B(]NE7 FJJ01 1 MM%cM B 6::n5<= *1&7J   MM!3US\^_M`).tG}  N NNF?3YH I;J< G_ ,, B$ZZ(9:E7E Bs D??EEct|sy |y#t$r"}tj|g||Yd}~yd}~wwxYw)zLInvoke an ``interrupt_fn``; True on success, else debug-log ``msg`` (+ exc).FTN)rrNrdebug)rmsgargsrjs r(_call_interruptr_sB B<   S%4%%s A<Actjts,tj}t5t |\}}}dddD]g\}}}t jd|||tt5tj|xsijd}dddtd|iD]j}t5tj|xsijdxstj}dddjt||fddlsytjts+yy#1swYxYw#1swYxYw#1swY_xYw)aSSweep running delegations for stalled progress. A changed progress token refreshes the record's timestamp; a frozen token past the idle/in-tool threshold marks the record ``stalling`` and calls ``interrupt_fn``; a ``stalling`` record still unreturned after the grace window is force-finalized with a terminal ``stalled`` event.Nu`Async delegation %s made no progress for %.0fs (in_tool=%s) — interrupting; grace window %.0fsrYz.Async delegation %s stall interrupt failed: %sr[ct||Sr)_stalled_result)rids r(r+z%_stale_monitor_loop..~s?[\^aKbr'r)rwait_STALE_CHECK_INTERVALrdrrrrrrrhrrvrwrunrh) rqrrrr^rrrctxs r(rrksW   !67iik  I0CC0H -GWo I18 a -M9g NNO()W>R T Mll=17R<<^L M B PR_ `  a% oM h||M28b==jIg[MeMeMg h GGI}M.bdm n o !  !677 I I  M M h hs#E9)E?E!EE !E* cdjt|jdxsdj}|r d|dddnd}t |jd xsd }|d k\rd t |d z d nd}|d|dS)uHuman wording for a force-finalized stall. This string reaches the user (CLI timeline, Desktop async-result card), so it names the task, how long it was silent, and what to do — no issue numbers or worker internals (those stay in the log line and the stall_* metadata). rTrJzBackground task "Nx"zThe background taskr!r<z after z min of no progressz stopped respondingz[ and was cancelled. Nothing else was affected; ask me to run it again if you still need it.)rrrhsplitfloatrg) event_recordrTrmquietsilences r(_stalled_error_textrs 88C ((06B7==? @D15Tc |1 -;PE ,""#9:?a @EBG2+ebj)**=>SUGg( 2; ;)r^rrr=rr$ stall_metas r(rrs##N3Btyy{L\\%5%5o%F%V,WYZ[H  -E LLM}^fg!$$%56M'3'7'78N'O#/#3#34N#O$1yAZv`d3 5J  #>,uh/>:>> jmE8, j R[ j_i jjr'c  t|}g}|D]}t|ttfrt |dk\s|j d9|d|dd}t |dk\rAt|dt tfr(ttd|t|dz d|d<|j ||S#t$rYywxYw) zParse a progress token into per-child activity dicts (best-effort): delegate_tool emits one ``(api_call_count, current_tool, last_activity_ts)`` tuple per child; foreign token shapes degrade to ``None`` entries.Nrrr)r: current_toolrgseconds_since_activity) r TypeErrorrrrrrrrgr.)rrqpartsoutpartrs r(_children_activity_from_tokenrsU +-C4$/CIN JJt  .21gtAw O t9>ja3,?.3CS5a>=Q4RTU.VE* + 5 J s B99 CCc Dtj}i}t5g}tjD]}|j Dcic]\}}|dvs |j dr||!}}}|j d}|tvrK|j drt||dz d|d<t|j dr |d||d<|d vr4tD]+\}} |j ||j ||| <-|j| d d d D]Q}|j |j d} | & | \} } t| |} | | |d <t| |d <S|Scc}}w#1swYgxYw#t$rYswxYw) aISnapshot of async delegations (running + recently completed) without callables or private monitor bookkeeping; adds computed live fields for UIs (``seconds_since_progress``, ``children_activity``/``in_tool`` sampled from ``progress_fn``) and stall context once tripped. Safe to call from any thread. See #51690. >rZrYr1rr]rseconds_since_progressrZr^)rrNchildren_activityr)rdrrrr, startswithrhrxrgr_STALL_FIELD_MAPrrNrr)rqsamplersr,rrOrPitemrsrcdstrrractivitys r(list_async_delegationsrs ))+C$&H " A%&WWYuTQ!;Z2ZcdcocopsctAqDuDuUU8_F'55(5:3>AR;RTU5VD12AEE-0134]3CHQ/000 0/HCuuSz-$%EE#JS / LL  " ( \\$((?3 4 :  TNE71<  (0D$ %w-Y ( L5v.   sB-F F F )F .BF7)F FFF FFcbtfd|D}|rtj||||S)zSCall ``interrupt_fn`` on each record; log ``msg`` once; returns how many succeeded.c 3vK|]0}t|jdd|jd2yw)rYz%s: %s interrupt failed: %sr^N)rrh)rrcallers r(rz%_interrupt_records..s;  n-/LfVWV[V[\kVlms69)rrr)targetsrreasonrcounts ` r(_interrupt_recordsrs4  E  C' Lr'ct5tjDcgc]}|jdtvs|}}dddt d|dScc}w#1swYxYw)zSignal every running async delegation to stop (``/stop``, shutdown). Returns how many. The child still emits a completion event (status='interrupted') via the normal finalize path.rN interrupt_allz'Interrupted %d async delegation(s) (%s))rrrrhrxr)rrrs r(rrsa V&oo/U155?n3T1UUV g@i jjVVVs AAAAAA%cBtt|||}t|d|dS)zSignal running async delegations owned by ONE ending session to stop (any selector matches, see ``_session_records``). Returns how many.interrupt_for_sessionz:Interrupted %d async delegation(s) for ending session (%s))r"rxr)r_r`rarrs r(rrs0 ~{rrrr`rJrNr)rrr`s r(rz%active_for_session..sI uuX"GGAEE017R8#$  sAANr)r`s`r(active_for_sessionrs=    __&     s &;A)rtsqlite3.Connection)r;rrtru)rtDict[str, Any])rprrtrurs)rrrrrtru)r^rrrrtru)rrr Optional[str]rrrtzOptional[List[Dict[str, Any]]])rtr)rrrrrtr)r^rrtr)r^rrrrtr)rrrtr)rrrrrtr)rrrrrtru)r^rrtzOptional[Dict[str, Any]])rrrtr)r_rr`rrarrtr)rJrJrJ)r_rr`rrarrtr)rtr)rrr=rrtr)rGrrtr)(r^rrTrrUOptional[List[str]]rVrrWrrXrrYrr_rrarrnCallable[[], Dict[str, Any]]r`rrcrrYOptional[Callable[[], None]]r}rrZOptional[Callable[[], tuple]]r~rrUrr[Optional[List[int]]r\Optional[Dict[str, str]]rtr)rTrrVrrWrrXrrYrr_rrarrnrr`rrcrrYrr}rrZrrtr)$rUz List[str]rVrrWrrXrrYrr_rrarrnrr`rrcrrYrr}rr^rrZrrUrr[rr\rrtr)r^rrrrrrtru)rprrrrrrtru)r^rrrrrrtru)rqr)rrrtr)rrrtr)r^rrrrtr)rrrqrrtzOptional[List])rtList[Dict[str, Any]]) rrrrrrrrrtr)r)rrrtr)rJrJrJ session_end) r_rr`rrarrrrtr)r`rrtr)o__doc__ __future__rrvrmloggingrfsqlite3rrdrconcurrent.futuresrtypingrrrrr hermes_constantsr tools.daemon_poolr tools.thread_contextr getLoggerrfrr__annotations__Lockr rrr_DEFAULT_MAX_ASYNC_CHILDRENryrwrzrrrkrrrrrrEventrrrxrLrrr)r<r9rErQrurorrrrrrrrrrrrrrrrrrr rrr"r$r'r2r8r>rArIrTrQrrrhrrr|rrrrrrrrrrrrr&r'r(rsh#   166,6<   8 $ +/ '.!s  &( #(- + 9>>   .2+2 ! 4 (5 sr *" ! # 8B2MI(y2j%P7 ` ! ZT!.0 NKA@: H T X [f +  nOo,`d(,15RDRD!$RD-@RDKXRD!RD),RD5BRDQTRD%RD/KRDcfRD RD+G RD]` RD / RD AD RD P] RD & RD/RDRDn:> "Rnr9hl  (4GOR[h)6Gc8;Pl If   <:> "Rnr9Z^15QU(,15!!"/!;N!VY!bo!!)6!Gc!!8;!Pl! !KX ! / ! BO ! & !/!!J",0Jf qH  #-L 0 <k((&Rkgtpp14pNQp`cpp,  r'