߲jldZddlmZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z ddlmZe rddlmZej$dZeZd Zd d ZGd d Zy)aSessionStore storage plumbing: per-profile SessionDB handle resolution and the routing-index load/save paths (state.db gateway_routing primary, sessions.json legacy mirror). Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.) annotationsN)Path) TYPE_CHECKINGAnyDictOptional)atomic_json_write SessionEntryzgateway.sessionaLEGACY MIRROR of the gateway routing index (the primary copy lives in the gateway_routing table in ~/.hermes/state.db). Maps messaging session keys (agent:main::...) to active session IDs. This is NOT the session list. ALL sessions (CLI, TUI, and gateway) live in ~/.hermes/state.db and are shown by `hermes sessions list` and `/sessions`. Disable this file with `gateway.write_sessions_json: false` in config.yaml.c@t|txr dt|vS)zCTest-isolation guard: must stay a loud failure and is never cached.zlive-system guard) isinstance RuntimeErrorstr)excs @/home/ubuntu/.hermes/hermes-agent/gateway/session_persistence.py_is_live_system_guardr#s c< ( L-@CH-LLcNeZdZdZd#d$dZdZedZejd%dZedZ d&dZ d'd Z d(d Z d)d Z d*d Zd%d Zd%dZd+dZd,dZd-dZd.dZed/dZd%dZd0dZd%dZd1dZd2dZd%dZd3dZd%dZd4dZd5dZ d6dZ!d%dZ"dd d! d7d"Z#y)8SessionPersistenceMixinzWSessionStore storage plumbing: SessionDB handle resolution and routing-index load/save.Ncddlm}ddlm| t |nt |fd}|j j |tS)a)SessionDB for the active profile scope. ``db_path`` pins the store; otherwise ``_default_db_path()`` follows the context-local HERMES_HOME (resolved per call so multiplexed profiles reach their own store). Handles are cached per path; failed opens enter a bounded backoff during which callers keep using the JSONL fallback. Resolving here rather than once in ``__init__`` is the whole fix for #88532: it lets the scoping that the multiplexed inbound path already performs actually reach session storage. r)_default_db_path)acquirecl S#t$r}t|std|d}~wwxYw)NzL[gateway] Warning: SQLite session store unavailable, falling back to JSONL: ) Exceptionrprint)erpaths r_openzHSessionPersistenceMixin._open_session_db_for_active_scope.._open9s? t}$ ,Q/hijhklm s 3.3) non_cacheable) hermes_staterhermes_state_registryrr_db_handle_cachegetr)selfdb_pathrrrrs @@r!_open_session_db_for_active_scopez9SessionPersistenceMixin._open_session_db_for_active_scope+sJ 21 ' 3tG}>N>P9Q $$((uDY(ZZrc$t|dtS)zKReturn the explicitly pinned DB (``store._db = x``), else ``_DB_UNPINNED``. _db_pinned)getattr _DB_UNPINNEDr$s r _pinned_dbz"SessionPersistenceMixin._pinned_dbCst\<88rcV|j}|tur|jS|S)a=The SessionDB for the active profile scope, or a pinned override. Assigning ``store._db`` pins that value for every subsequent read (tests install a fake or disable the DB with ``store._db = None``); unpinned, each read resolves the scope so a multiplexed profile's writes reach its own store.)r,r*r&)r$pinneds r_dbzSessionPersistenceMixin._dbGs, ";A\;Qt557]W]]rc||_yN)r()r$values rr/zSessionPersistenceMixin._dbPs rc|j}|tur|St|dd} | |jS|j |dz S#t $rYywxYw)uQThe one store that owns the routing index, whatever scope is active. ``_entries`` is one flat dict holding every profile's keys, so it must persist to ONE file (``_routing_home``), not whichever profile is scoped — otherwise a mid-turn rewrite and the unscoped startup load see different copies and crash markers under a secondary profile go unrecovered. A pinned handle still wins; bare test instances lacking the handle cache report no DB. Reading it through ``_db`` made that file whichever profile happened to be scoped at the time: a whole-index rewrite during one profile's turn copied every other profile's routing rows into that profile's store, and startup — which runs unscoped — then loaded a different copy than the one the last writer produced. See #66887. _routing_homeNstate.dbr%)r,r*r)r/r&r)r$r.homes r _routing_dbz#SessionPersistenceMixin._routing_dbTsi"  %Mt_d3 |xx99$BS9T T  s A A AAcjt|jddsy|j|}|r|dk(rdS|S)uThe non-default profile that owns *session_key*, or None (ambient store is authoritative: multiplexing off, or legacy ``agent:main``). Deliberately does NOT cover "that profile has no directory" — ownership and resolvability are separate questions for ``_db_for_key``.multiplex_profilesFNdefault)r)config_profile_from_session_key)r$ session_keyprofiles r_named_profile_for_keyz.SessionPersistenceMixin._named_profile_for_keyls<t{{$8%@00="g&:tGGrc |j|}|y|j}||vr||Sd} ddlm}m}||rt ||}||||<|S#t $r#}tjd||d}Yd}~0d}~wwxYw)zbHERMES_HOME of the profile owning *session_key*, or None (no named owner or unresolvable).Nr)get_profile_dirprofile_existsz)Could not resolve profile home for %r: %s) r@_profile_home_cachehermes_cli.profilesrBrCrrloggerdebug)r$r>r?cacher7rBrCrs r_profile_home_for_keyz-SessionPersistenceMixin._profile_home_for_keyus--k: ?(( e > !#  Kg&OG45  !E'N   LLDkSV WD s!A B A>>Bc|j}|tur|S|j|}|Ht|dd}|t|jdds |j S |j |dz S|j|}|tjd||y |j |dz S#t$rYywxYw#t$rYywxYw)upThe SessionDB holding *session_key*'s rows, whatever scope is active (the owning profile is encoded in the key). ``_db`` follows the ambient HERMES_HOME that only the inbound message path installs; unscoped background work (expiry watcher) would otherwise write profile rows into the ROOT store until the stale-route self-heal drops a live conversation. Background work runs unscoped while operating on every profile's keys out of the single process-wide ``_entries`` dict — ``_session_housekeeping_watcher`` is the clearest case — so it reads and writes the ROOT store for rows that actually live under ``profiles//state.db``. The two writers then drift apart on the same logical session until the routing index disagrees with the row and the #54878 self-heal drops a live conversation (#66887). Nr4r:Fr5r6zggateway.session: profile %r has no resolvable home (key %r); refusing to fall back to the ambient store) r,r*r@r)r<r/r&rrIrFwarning)r$r>r.r? routing_homer7s r _db_for_keyz#SessionPersistenceMixin._db_for_keys"  %M--k: ?#4$?L#74;;@TV[+\xx ==lU_F_=``))+6 < NN,-4k C 99$BS9T T    s$!B7"C7 CC CCc|sy t|jjD]}|j|k(s|jcS t |ddxsij|S#t $rY+wxYw)a The routing key that owns *session_id*, or None. The published index is authoritative; ``_session_owner_hints`` covers the window where ownership is proven but routing not yet published. Deliberately lock-free: several callers already hold ``_lock``.N_session_owner_hints)list_entriesvalues session_idr>rr)r#)r$rSentrys r_owner_key_for_session_idz1SessionPersistenceMixin._owner_key_for_session_ids dmm2245 -##z1 ,,, - 4d;ArFFzRR   s5A+ A+ A++ A76A7c^|s |jS|j|j|S)zThe SessionDB holding *session_id*'s row (owner from the index or a pre-published hint; unknown ids fall back to the ambient store).)r/rMrU)r$rSs r_db_for_session_idz*SessionPersistenceMixin._db_for_session_ids,88O > >z JKKrcBdd}|jj|y)aClose every SessionDB handle this store opened (one per path). Closing only ``store._db`` would strand secondary profiles' handles with their WAL lock held ('database is locked' on restart). Drained under the lock, closed outside it; a pinned handle is the pinner's.czddlm} ||y#t$r }tj d|Yd}~yd}~wwxYw)Nr)release_or_closez-SessionDB close error during handle sweep: %s)r!rZrrFrG)dbrZrs r_closez._closes7 > S $ S LcRR Ss :5:NreturnNone)r" close_all)r$r\s rclose_all_db_handlesz,SessionPersistenceMixin.close_all_db_handless S ''/rcf|j5|jdddy#1swYyxYw)z4Load sessions index from disk if not already loaded.N)_lock_ensure_loaded_lockedr+s r_ensure_loadedz&SessionPersistenceMixin._ensure_loadeds* ZZ )  & & ( ) ) )s'0cX|j|jj|S)zALoad the index and return the entry for *session_key*. Lock held.)rdrQr#)r$r>s r _entry_lockedz%SessionPersistenceMixin._entry_lockeds# ""$}}  --rc tt|jjS#t$rt|jcYSwxYw)zNamespace for this store's gateway_routing rows: the resolved sessions_dir, so stores with different dirs never share entries.)rr sessions_dirresolverr+s r_routing_scopez&SessionPersistenceMixin._routing_scopesF *tD--.6689 9 *t(() ) *s+.AAcVt|jxsd|d}t|r|SdS)zHBound ``_routing_db.`` if the handle exists and has it, else None.N)r)r8callable)r$namemethods r_routing_db_methodz*SessionPersistenceMixin._routing_db_methods-))1T4>!&)v3t3rc(|jd}|y ||jjD])\}}|j||}|||j|<+y#t $r }t jd|Yd}~yd}~wwxYw)z~Load state.db routing entries into ``_entries``; False when there is no loader or the load failed (warned). Lock held.load_gateway_routing_entriesNFscopeTz1gateway.session: state.db routing load failed: %s)rprkitems_routing_entry_from_jsonrQrrFrK)r$loaderkey entry_jsonrTrs r_load_routing_rows_lockedz1SessionPersistenceMixin._load_routing_rows_lockeds(()GH > #)0C0C0E#F#L#L#N /Z55c:F$).DMM#& /  NNNPQ R s?A(A(( B1B  Bcddlm} tj|}t |t r|j |S y#tttf$r!}tjd||Yd}~yd}~wwxYw)zBParse one gateway_routing row; None (with a warning) when invalid.rr z%Skipping invalid routing entry %r: %sN) gateway.sessionr jsonloadsr dict from_dict ValueErrorKeyError TypeErrorrFrK)rxryr entry_datars rrvz0SessionPersistenceMixin._routing_entry_from_jsonst 1 LJ/J*d+-|--j99,Hi0 L NNBC K K Ls6AA6A11A6cZ|jr|jy|jjdd|j }|xrt |j }|j|d|_||_|rdn|j|_ |jy)aLoad the routing index (lock held). state.db ``gateway_routing`` is primary; sessions.json is the legacy import for keys the DB lacks (persisted on the next _save). Read order (#9006 follow-up): the ``gateway_routing`` table in state.db is the primary source; sessions.json is the legacy import path for pre-migration installs (its entries are folded in for keys the DB doesn't have, then persisted to the DB on the next _save). NT)parentsexist_ok) _loaded#_reconcile_recovered_routing_lockedrimkdirrzboolrQ_import_legacy_sessions_json_routing_db_loaded_entries_as_dicts_routing_fallback_baseline_prune_stale_sessions_locked)r$db_load_succeededdb_had_entriess rrdz-SessionPersistenceMixin._ensure_loaded_locked s <<  4 4 6  t< ::<*BtDMM/B )).9 "32C$I_I_Ia' ))+rcddlm}|jdz }|jsy t |dd5}t j |}dddd}jD]\}}|jds||jvr&t|ts+tjd |t|ja |j ||j|<|d z }|r"|rtj)d ||d k(rd ndyyy#1swYxYw#t"t$t&f$r!} tjd || Yd} ~ d} ~ wwxYw#t*$r} t-d| Yd} ~ yd} ~ wwxYw)zELegacy import: sessions.json fills only keys the DB lacks. Lock held.rr sessions.jsonNrzutf-8)encoding_z8Skipping invalid session entry %r: expected dict, got %sz%Skipping invalid session entry %r: %sz\gateway.session: imported %d legacy sessions.json entr%s missing from state.db routing tableyiesz,[gateway] Warning: Failed to load sessions: )r|r riexistsopenr}loadru startswithrQr rrFrKtype__name__rrrrinforr) r$rr sessions_filefdataimportedrxrrs rrz4SessionPersistenceMixin._import_legacy_sessions_jsonsp0))O; ##%  FmS7; $qyy| $H#'::< TZ>>#&#*>!*d3NNRTWZ(113T)?)?)? )KDMM#&MH TN -.6x1}RWY+x# $ $#Hi8TNN#JCQRSST  F @D E E FsYED A;E$D,$EDEE3E EEE E8 E33E8c |jsyg}d} |jjD]\}}|j|}||j|j}||j dI|j |||}|dk(r|j|s|v||j|<|dz } |D]}|j|=|s|r|jyy#t$r }tjd|Yd}~yd}~wwxYw)zRemove routing entries whose session has ended in state.db (startup, lock held). Stale == ``end_reason IS NOT NULL``; rows absent from the DB are kept; a ``None`` DB handle is a no-op; DB errors are non-fatal.Nr end_reasonprunerz@gateway.session: stale-entry pruning skipped due to DB error: %s) rQrurM get_sessionrSr#_stale_entry_verdictappendrrFrK_save) r$ stale_keysrecovered_keysrxrTr[rowverdictrs rrz4SessionPersistenceMixin._prune_stale_sessions_locked=s}}   "mm113 ( U%%c*:nnU%5%56;#'',"7"?33CDg%%%c*()0DMM#&"a'N ($ #C c" #  JJL(   NN]_b c  sBC$#C$$ D -DD cddlm}d}|j% |j||j|d}|K|j|jk7r2t jd||j|d|j|S|&t jd |j||dyt jd ||j|dy #t$r,}t j d||j|Yd}~yd}~wwxYw) zrFor a routing entry whose row has ended: ``"prune"``, a replacement entry (repoint), or None (keep as-is).r)_nowNT)r>sourcenowraise_on_lookup_errorzRgateway.session: recovery lookup failed for stale sessions.json entry %r -> %s: %szfgateway.session: repointing stale sessions.json entry %r from ended %s (end_reason=%r) to recovered %srzdgateway.session: reopened ended session %s for sessions.json entry %r (end_reason=%r); keeping routezfgateway.session: pruning stale sessions.json entry %r -> %s (end_reason=%r); left by a crashed gatewayr) gateway.session_lifecyclerorigin_recover_session_from_dbrrFrGrSrKr)r$rxrTrrrecovered_entryrs rrz,SessionPersistenceMixin._stale_entry_verdict]s  3 << # "&"?"? #ELLdf\`#@#b  &?+E+EIYIY+Y NN2368H8H#lJ[** ,# "  & KK1272B2BC\IZ \ !$e&6&6L8I K7  !5#3#3S:  s$C D "DD c|jjDcic]\}}||jc}}Scc}}w)z1Serializable snapshot of ``_entries``. Lock held.)rQruto_dict)r$rxrTs rrz)SessionPersistenceMixin._entries_as_dictss27;}}7J7J7LMeU]]_$MMMs=c>|j|jy)z;Persist the routing index while the caller holds ``_lock``.N)_persist_routing_data_snapshot_routing_lockedr+s rrzSessionPersistenceMixin._saves"""D$A$A$CDrcDt|dddz|_|jS)zBump and return the shared routing counter (lock held). Full snapshots AND single-entry fast saves MUST allocate from this one counter: the stale-write protection is a total order over serialization times and silently breaks otherwise._routing_generationrr)r)rr+s r_next_routing_generation_lockedz7SessionPersistenceMixin._next_routing_generation_lockeds'$+41F#JQ#N '''rct|dd}t|dds|y|jd}|y ||j}|j }|jD][\}}|j||}|||vr|jj||<||vrA||||k(sM||j|<]d|_ d|_ y#t$r }tj d|Yd}~yd}~wwxYw) z  4#6#6#89G((*&}} 3OC 99#zJM$(" ((m<G##.%2 c" 3#'*.'#  NNXZ] ^  sC D C;;Dcb|j|j|jfS)z:Capture immutable routing data and a monotonic generation.)rrrr+s rrz0SessionPersistenceMixin._snapshot_routing_lockeds+ 002%%')M)M)OOOrc |jdtj5|t|ddkr dddyt|dd}|r9|j D]&\}\}}||kDst j |||<(d}|jd}|P ||j D cic]\} } | t j| c} } |jd }t|d d s|s |j|||_|r6|j D cgc]\} \} } | |ks| c} } } D]}||=dddycc} } w#t$r } tjd | Yd} ~ d} ~ wwxYw#t$r#} |stjd | Yd} ~ d} ~ wwxYwcc} } } w#1swYyxYw) zASerialize all whole-index writers through one durable write lock. _save_lock_persisted_routing_generationrN_fast_persisted_entriesFreplace_gateway_routing_entriesrsTz1gateway.session: state.db routing save failed: %s_write_sessions_jsonzKgateway.session: sessions.json mirror save failed after state.db commit: %s)_lazy threadingLockr)rur}r~rpdumpsrkrrFrK_save_sessions_jsonr)r$r generationfast_persistedrxrevisionrydb_savedreplacerkvrrevrs rrz-SessionPersistenceMixin._persist_routing_datas ZZ inn 5" ,WT+JANN" ," , %T+DdKN3A3G3G3I;/C/(J*,$(JJz$:S ;H../PQH#]4::<H41aaA.HPTPcPcPef#Ht3T:( #,,T22 D* ? ""44 ? ?s <AF)r lock_heldc|rtjn |j}|5|jj |}| dddy| t |n|j }tj||j}}ddd|jd} | |jdtj5t|ddk\r dddy|jdt } | j |} | | d|k\r dddy| ||j||f| |<dddy|<|5|j'} dddt | |<|j)| y|j+y#1swY xYw#1swYyxYw#t $r!} t"j%d|| Yd} ~ d} ~ wwxYw#1swYxYw) uPersist ONE routing entry via UPSERT — the per-turn fast path (a full rewrite fsyncs a multi-MB sessions.json). The key -> session_id mapping never changes here: structural transitions use the full rewrite (which also refreshes the sessions.json mirror; it may lag in metadata only). The revision comes from the shared routing generation counter; under ``_save_lock`` the upsert is skipped if a full snapshot or a newer fast save of this key already persisted (the reverse case lives in ``_persist_routing_data``). No DB or a failed upsert falls back to the full rewrite. ``entry_data`` persists a candidate BEFORE it is published to the live entry (failure-atomic transitions); the fallback carries it too.Nsave_gateway_routing_entryrrrrrszagateway.session: single-entry routing save failed for %r (%s); falling back to full index rewrite) contextlib nullcontextrcrQr#rrr}rrrprrrr)rkrrFrKrrr)r$r>rrguardrT serializedryrsaverr persistedr fallback_datas r _save_entryz#SessionPersistenceMixin._save_entrys-6 &&(4::  bMM%%k2E} b b.8-Cj)J#'::j#94;_;_;aJ  b''(DE   ?ZZ inn=It%DaHHTII&*ZZ0I4%PN . 2 2; ?I ,11I II+z9L9L9NO3;Z2HN;/I  ! 9 $ 6 6 8  9)-j)9M+ &  & &}h ?    ? b bI ?,-8#?? ?  9 9slF AF+ F2 F&F2&2F&F2! F&F2GF#&F/+F2/F22 G;GGG(r1)r%Optional[Path]r])r> Optional[str]r^r)r>rr^r)r>r)rSrr^r)rSr)r>rr^Optional[SessionEntry])r^r)rnr)r^r)rxrryrr^r)rrr^r_)rxr)r^Dict[str, Any])r^int)r^ztuple[Dict[str, Any], int])rrrrr^r_)rrr^r_)r>rrzOptional[Dict[str, Any]]rrr^r_)$r __module__ __qualname____doc__r&r,propertyr/setterr8r@rIrMrUrWrarergrkrprz staticmethodrvrdrrrrrrrrrrrrrrrr(sa[09^^ ZZ  .H.&P SL 0) . *4    ,*F>@$LNE(/:P $,Lw5KO+!+!/G+!+!$(+!rr)r BaseExceptionr^r)r __future__rrloggingr}rpathlibrtypingrrrrutilsr r|r getLoggerrFobjectr*rrrrrrrsiD# 55#,   , -x EM g!g!r