߲jadZddlmZddlZddlZddlZddlZddlZddlZddl m Z m Z m Z m Z mZejdZdZddZed Zed Zd Zed ZGd dZy)zToken/usage accounting mixin for SessionDB: the coalescing background token writer, per-model usage rows, and billing-route columns. Writer thread state lives on the instance.) annotationsN)AnyDictListOptionalTuple hermes_state) input_tokens output_tokenscache_read_tokenscache_write_tokensreasoning_tokensc dfd dfd }djfdtD}rdnd}d|zd|d |d d |d d zS)z``UPDATE sessions`` for one usage report: *delta* adds to the stored counters (CLI per-call path), otherwise sets them (gateway cumulative path). Cost/route columns COALESCE-fill either way (statement text is pinned by the SQL trace harness).cr|dSdS)Nz + ??coldeltas 7/home/ubuntu/.hermes/hermes-agent/hermes_state_usage.pyaddz_token_update_sql..adds$#d|-#-crd|dSdS)Nz COALESCE(z, 0) + ?rrrs radd0z_token_update_sql..add0s,13%x(:s:rc3<K|]}d|d|dyw)z z = z, Nr).0crs r z$_token_update_sql..s%X1,QCs3q6(#>Xsz0COALESCE(estimated_cost_usd, 0) + COALESCE(?, 0)zCOALESCE(?, 0)zUPDATE sessions SET z( estimated_cost_usd = z, actual_cost_usd = CASE WHEN ? IS NULL THEN actual_cost_usd ELSE actual_cost_usda END, cost_status = COALESCE(?, cost_status), cost_source = COALESCE(?, cost_source), pricing_version = COALESCE(?, pricing_version), billing_provider = COALESCE(billing_provider, ?), billing_base_url = COALESCE(billing_base_url, ?), billing_mode = COALESCE(billing_mode, ?), model = COALESCE(model, ?), api_call_count = api_call_countz WHERE id = ?)rstrreturnr")join_TOKEN_COUNTERS)rrcounters estimatedrs` @r_token_update_sqlr(sv.;wwXXXHFKBQaI(*6ykB""345 6%&**:%;$<=  # #rF)rTaINSERT INTO session_model_usage ( session_id, model, billing_provider, billing_base_url, billing_mode, task, api_call_count, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, reasoning_tokens, estimated_cost_usd, actual_cost_usd, cost_status, cost_source, first_seen, last_seen ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(session_id, model, billing_provider, billing_base_url, billing_mode, task) DO UPDATE SET api_call_count = api_call_count + excluded.api_call_count, input_tokens = input_tokens + excluded.input_tokens, output_tokens = output_tokens + excluded.output_tokens, cache_read_tokens = cache_read_tokens + excluded.cache_read_tokens, cache_write_tokens = cache_write_tokens + excluded.cache_write_tokens, reasoning_tokens = reasoning_tokens + excluded.reasoning_tokens, estimated_cost_usd = estimated_cost_usd + excluded.estimated_cost_usd, actual_cost_usd = actual_cost_usd + excluded.actual_cost_usd, cost_status = COALESCE(excluded.cost_status, cost_status), cost_source = COALESCE(excluded.cost_source, cost_source), last_seen = excluded.last_seen)modelbilling_providerbilling_base_url billing_moder r r r restimated_cost_usdr cost_status cost_sourcer!ceZdZdZdd ddZddZddZdddZddZdd Z dd Z d d!d Z dd Z d" d#dZ ddddd d d d d ddddd dd d$dZdddd d d d d ddd d%dZd&dZdddd'dZy)(SessionUsageMixinz._doas3 LL# &.xz$R  T  4 4T :rN)flush_token_counts_execute_write)r9r:r8r7r,r;s````` rupdate_session_billing_routez.SessionUsageMixin.update_session_billing_routeUs+ ! ; ; C rc H|j5|j}|duxr|j}|jxr| }|s|jj ||f|s|t j|jdd}||_|j|j7tj|dfd }||_ tj||jjdddr|j |fi|yy#1swY xYw)zEnqueue a token/cost delta for the background writer (same kwargs as :meth:`update_token_counts`). After close() stopped the writer, falls back to the synchronous path and may raise.Nzsession-db-token-writerT)targetnamedaemonc:}||jyyN)_drain_token_queue_at_exit)dbself_refs r_drain_at_exitz._drain_at_exits !)B!~ " = = ? .rr#None)_token_queue_cond_token_writer_threadis_alive_token_writer_stop _token_queueappend threadingThread_token_writer_loopstart_token_atexit_hookweakrefrefatexitregister notify_allupdate_token_counts)r9r:kwargsthread writer_alivewriter_stoppedrHrGs @rqueue_token_countsz$SessionUsageMixin.queue_token_countsls # # 4..F!-C&//2CL!44I\9IN!!!((*f)=>#'--#66=V_ceF06D-LLN..6#*;;t#4@ 3A/7&&113/ 40  $D $ $Z :6 : 1 4 4s C+DD!c: |j||j5d|_|jjdddy#1swYyxYw#|j5d|_|jjdddw#1swYwxYwxYw)zKApply a batch whose ``busy`` flag the caller already claimed, then release.FN)_apply_token_batchrK_token_writer_busyrZ)r9batchs r_apply_claimed_batchz&SessionUsageMixin._apply_claimed_batchs 4  # #E *'' 4*/'&&113 4 4 4'' 4*/'&&113 4 4 4s.A"A  A B#"B BB Bcx|js |jsyd}|j5tj|z}|js |jr|j }||j sC|js7d|_t|j}|jjn[|tjz }|dkr dddy|jj||jr|jrddd|r|j|y#1swYxYw)zBlock until every queued token delta has been applied. False on timeout (callers then read totals stale by the queued deltas). Never raises.TNrF) rOrcrKtime monotonicrLrMlistclearwaitre)r9timeoutrddeadliner] remainings rr<z$SessionUsageMixin.flush_token_countss  )@)@  # # 7~~''1H##t'>'> 22N&//*;TE\E\.2D+ !2!23E%%++-$t~~'77 >  7 7 &&++I6##t'>'> 7"   % %e ,' 7 7sB.D0'D0 D00D9cF |j5tj|jz}|jsm|j sa|tjz }|dkrd|_ dddy|jj||js |j sa|jsd|_ dddyd|_t|j}|jjddd|j#1swYxYw)NTr) rKrgrh_TOKEN_WRITER_IDLE_SECONDSrOrNrLrkrcrirjre)r9 idle_deadlinernrds rrSz$SessionUsageMixin._token_writer_loops'' * $ 043R3R R ++D4K4K -0@ @I A~591 * ***// :++D4K4K((04D- * *+/'T../!!'')# *$  % %e ,' * *sAD53D)D6DD c |j|}|D]\}} |j|fi|y#t$r"}tjd||}Yd}~Ed}~wwxYw#t$r!}tjd||Yd}~pd}~wwxYw)zBApply queued deltas in order, coalescing where safe. Never raises.z?async token accounting: coalesce failed, applying raw batch: %sNz5async token accounting: apply failed (session=%s): %s)_coalesce_token_deltas Exceptionloggerwarningr[)r9rd coalescedexcr:r\s rrbz$SessionUsageMixin._apply_token_batchs 33E:I #, i J i(((>v> i   NN\^a bI  iVXbdghh is+2A  AAA B )BB c g}|D]\} d} jds|g fd|jD}|r||dd|k(r|dd}|jD]*}|j|d j|dz||<,|jD]1} j|}||j|xsd|z||<3|j ||t f|D cgc] \}} } | | f c} } }Scc} } }w)zMerge adjacent incremental deltas with an identical route, so ordering across sessions and /model switches is preserved exactly. absolute=True never merges.Nabsolutec3@K|]}j|ywrD)get)rfr\s rrz;SessionUsageMixin._coalesce_token_deltas..s$[qVZZ]$[sr)r|_TOKEN_DELTA_ROUTE_FIELDS_TOKEN_DELTA_SUM_FIELDS_TOKEN_DELTA_COST_FIELDSrPdict) r9rdgroupsr:keymergedr}value_sidkwr\s @rrsz(SessionUsageMixin._coalesce_token_deltass)EG"' ? JC::j)!\$[D ?-333jabb 333s.Dcp|j5d|_|jj|j}ddd]|j rM|j ||j r+t jd|t|jy|j5tj|z}|jrx|tjz }|dkr4t jd|t|j dddy|jj||jrxt|j}|r!d|_ |jjdddr|j!|yy#1swYfxYw#1swY+xYw)z@Stop the writer thread and drain remaining deltas. Never raises.TN)rlzZasync token accounting: writer did not stop within %.0fs; %d queued delta(s) not persistedrzfasync token accounting: concurrent drain did not finish within %.0fs; %d queued delta(s) not persisted)rKrNrZrLrMr$rurvlenrOrgrhrcrkrirjre)r9 join_timeoutr]rmrnrds r_stop_token_writerz$SessionUsageMixin._stop_token_writers  # # /&*D #  " " - - /..F /  &//"3 KK K - 78Dc$J[J[F\^ # # *~~',6H))$t~~'77 >NNP$c$*;*;&<> * *&&++I6))**+E*.'!!'') *   % %e , ? / / * *s$.F.A+F,"'F, 8F,F),F5ctjt5|jdddy#1swYyxYwrD) contextlibsuppressrtr)r9s rrEz,SessionUsageMixin._drain_token_queue_at_exits1   + &  # # % & & &s4=rFc tjDcic]\}}|tvs||c}}j|xsd|rtnt t |xs|xs|xs|xs |xs|xs|}t |xs| ||||||| | | | | r ndrndrndrnd|f| xr| f d}j|ycc}}w)a]Update token counters and backfill model if unset. *absolute*=False increments (per-API-call deltas, CLI path); *absolute*=True sets directly (gateway path, where the cached agent holds cumulative totals). ``source`` is the session's real surface for the row-existence guard; callers that don't know it leave the placeholder.unknown)r)Nc |jd fj}| t|ni}t|j dxsddk(xrHxrDt xr7t xr*|j dk7xs|j dk7}|r|jd f|j  r j | fiyy)NzISELECT model, billing_provider, api_call_count FROM sessions WHERE id = ?r!rr)r*zUPDATE sessions SET model = ?, billing_provider = ?, billing_base_url = ?, billing_mode = ? WHERE id = ?)r4fetchonerintr|bool_record_model_usage)r6rowexistingfirst_accounted_router+r,r*has_accounted_usager)paramsrecord_model_usager9r:sqlusages rr;z2SessionUsageMixin.update_token_counts.._do<s,,[^h]jhj %(OtCyH HLL!127a8A=mBUmZ^_dZem)*m\\'*e3kx||DV7W[k7k " % '*/0@BRT`bl(mo LLf %!(((zCUC"r)localsitems_MODEL_USAGE_FIELDS_insert_session_row_TOKEN_UPDATE_ABSOLUTE_SQL_TOKEN_UPDATE_DELTA_SQLrr=)r9r:r r r)r r rr-r r.r/pricing_versionr*r+r,r!rzsourcekv has_usager;rrrrrs`` ` ``` @@@@@rr[z%SessionUsageMixin.update_token_countss!#)(.."2O$!Qa;N6NAO   V-@y N,4(:QAA:KAOaAeuA*A.@B "9#?@ -):z:SessionUsageMixin.record_auxiliary_usage..s#)A)A)A$ )gY])gaf)gr)rrrrrr=)r9r:rr)r*r+r r r r rr-r!rrrs``` @rrecord_auxiliary_usagez(SessionUsageMixin.record_auxiliary_usageqst#)(.."2O$!Qa;N6NAO '5'=!3~CV   Y7 ghPs A;A;c |siS|j|}tt|dz ddD]}|j||s||d}n|j ddj dt|zd|}|Dcic].}|d|j Dcic]}|dk7s |||c}0c}}Scc}wcc}}w) aPer-task auxiliary usage (``task != ''``: vision, compression, title_generation, ...) summed over the session's compression lineage. Aux calls bill to the id the turn STARTED with while compression mints child ids mid-turn, so a single-id read misses rows (#112848).rr~NaySELECT task, COALESCE(SUM(api_call_count), 0) AS api_calls, COALESCE(SUM(input_tokens), 0) AS input_tokens, COALESCE(SUM(output_tokens), 0) AS output_tokens, COALESCE(SUM(cache_read_tokens), 0) AS cache_read_tokens, COALESCE(SUM(cache_write_tokens), 0) AS cache_write_tokens, COALESCE(SUM(reasoning_tokens), 0) AS reasoning_tokens, COALESCE(SUM(estimated_cost_usd), 0) AS estimated_cost_usd FROM session_model_usage WHERE session_id IN (,rz/) AND task != '' GROUP BY taskr)_session_lineage_root_to_tipranger_is_explicit_branch_session _read_allr$keys)r9r:chainirowsrrs rauxiliary_usage_by_taskz)SessionUsageMixin.auxiliary_usage_by_tasksI11*=s5zA~r2. A//a9ab  ~~ '(+xxc%j0@'A&BC "   X\\PSF KAqF{aQiKK\\K\sC B<+B<3C<C)min_message_countinclude_archivedcddg}|g}|s|jd|jddj|d|}t|dxsdt |dxsd d S) zTokens and spend across the whole store (one scan), so the sidebar total does not shrink with paging. Spend prefers the billed figure over the estimate.zparent_session_id IS NULLzmessage_count >= ?zCOALESCE(archived, 0) = 0z SELECT COALESCE(SUM(COALESCE(input_tokens, 0) + COALESCE(output_tokens, 0)), 0), COALESCE(SUM(COALESCE(actual_cost_usd, estimated_cost_usd, 0)), 0) FROM sessions WHERE z AND z rrr)tokenscost_usd)rP _read_oner$rr)r9rrwhererrs r usage_totalszSessionUsageMixin.usage_totalss-.BC./ LL4 5nn"LL'()    c!fk*c!fm8LMMr) r:r"r8r"r7r"r, Optional[str]r#rJ)r:r"r#rJrI)g@)rlrr#r)rd List[Tuple[str, Dict[str, Any]]]r#rJ)rdrr#r)g$@)rrr#rJ)rrNrrrNNNNNNNNrFN)&r:r"r rr rr)r"r rr rrrr-Optional[float]r rr.rr/rrrr*rr+rr,rr!rrzrrrr#rJ)"r:r"r)rr*rr+rr,rr rr rr rr rrrr-rr rr.rr/rr!rrr"r#rJ)r:r"rr"r)rr*rr+rr rr rr rr rrrr-rr!rr#rJ)r:r"r#zDict[str, Dict[str, float]])rrrrr#zDict[str, float])__name__ __module__ __qualname____doc__r>r`rer<rSrbrsrrEr[rrrrrrrr1r1RsF_c!!,/!;>!N[! !.;@48-, i4*"-H& st`dkoswPU" >!>!-0>!CF>!QT>!nq>!>!58>!P_>!)>!=J>!^k>!' >!@M >!fs >! $ >! ;> >! LP >!  >! >!B>Bcg(,$bcQPQij,0SW#'D^_km r #r/<rUbr'rCGhl(,VWpq !`d ii%(i4AiZgi'irs_#  33   > *r:/U;+$750 !GH aNaNr