߲j dZddlmZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z mZmZddlmZmZmZmZmZej,dZded d d ed d d ej0ddZdZedededZedededZdZdeddedd dedd dZded Zded!ed"d#ed"d$ed"d% Z d&Z!d+d'Z"d(Z#Gd)d*Z$y),zGateway-facing SessionDB persistence: routing index, peers, orphans, heartbeats, handoffs. Mixin bound onto ``SessionDB`` via the MRO; built on its ``_read_ctx`` / ``_execute_write`` / ``_write_sql`` / ``_read_all`` primitives.) annotationsN)Path)AnyDictListOptionalSetTuple)_RECOVERABLE_END_REASONS_SQL_RESET_CHILD_SQL_RESET_END_REASONS_SQL_sql_json_extract_sql_session_last_active hermes_statea WITH RECURSIVE compression_lineage(id) AS ( SELECT ? UNION SELECT parent.id FROM compression_lineage lineage JOIN sessions child ON child.id = lineage.id JOIN sessions parent ON parent.id = child.parent_session_id WHERE parent.end_reason = 'compression' AND zchild.model_configz$._branched_fromz' IS NULL AND z$._delegate_fromz, IS NULL AND NOT (child)azk) AND COALESCE(child.source, '') != 'tool' ) a SELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved, (COALESCE(s.message_count, 0) > 0 OR EXISTS ( SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1 )) AS _has_messages FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash z WHERE s.session_key = ? AND s.source = ? AND (s.ended_at IS NULL OR s.end_reason IN (a)) AND NOT EXISTS ( SELECT 1 FROM sessions b WHERE b.session_key = s.session_key AND b.source = s.source AND b.ended_at IS NOT NULL AND b.end_reason IN (a$) AND b.ended_at > COALESCE(s.last_activity_at, s.started_at) ) ORDER BY _has_messages DESC, COALESCE(s.last_activity_at, s.started_at) DESC LIMIT 1 a WHERE s.source = ? AND COALESCE(s.user_id, '') = COALESCE(?, '') AND COALESCE(s.chat_id, '') = COALESCE(?, '') AND COALESCE(s.chat_type, '') = COALESCE(?, '') AND COALESCE(s.thread_id, '') = COALESCE(?, '') AND (? IS NULL OR COALESCE(s.profile_name, ?) = ?) AND (s.ended_at IS NULL OR s.end_reason IN (a)) AND (COALESCE(s.message_count, 0) > 0 OR EXISTS ( SELECT 1 FROM messages WHERE messages.session_id = s.id LIMIT 1 )) AND NOT EXISTS ( SELECT 1 FROM sessions b WHERE b.source = s.source AND COALESCE(b.user_id, '') = COALESCE(s.user_id, '') AND COALESCE(b.chat_id, '') = COALESCE(s.chat_id, '') AND COALESCE(b.chat_type, '') = COALESCE(s.chat_type, '') AND COALESCE(b.thread_id, '') = COALESCE(s.thread_id, '') AND b.ended_at IS NOT NULL AND b.end_reason IN (z) AND b.ended_at > COALESCE(s.last_activity_at, s.started_at) ) ORDER BY COALESCE(s.last_activity_at, s.started_at) DESC LIMIT 1 zpd.id, d.session_key, d.chat_id, d.chat_type, d.thread_id, d.user_id, d.origin_json, d.display_name, d.end_reasonz SELECT o.id, o.source, o.user_id, o.started_at, o.parent_session_id, oa AS last_active, (SELECT COUNT(*) FROM messages m WHERE m.session_id = o.id) AS message_count FROM sessions o WHERE o.session_key IS NULL AND EXISTS (SELECT 1 FROM messages m WHERE m.session_id = o.id) AND COALESCE(o.source, '') != 'tool' AND zo.model_configz IS NULL AND zC IS NULL ORDER BY o.started_at ASC z SELECT z FROM sessions d WHERE d.id = ? AND d.session_key IS NOT NULL AND COALESCE(d.source, '') = COALESCE(?, '') , da AS last_active FROM sessions d WHERE d.session_key IS NOT NULL AND d.id != ? AND COALESCE(d.source, '') = COALESCE(?, '') AND (COALESCE(d.user_id, '') = '' OR COALESCE(?, '') = '' OR d.user_id = ?) AND z/ BETWEEN ? AND ? AND zo < ? ORDER BY last_active DESC LIMIT 2 )telegram_dm_topic_modetelegram_dm_topic_bindingsdelivery_obligationsc |jdtDchc]}|d }}tDcic]-}||vr'||jd|dDchc]}|d c}/c}}Scc}wcc}wcc}}w)aLive column sets of the lazily-created tables that exist (``{}`` when none do). ``apply_telegram_topic_migration`` runs only on explicit ``/topic`` opt-in, so a store can hold supported v1/v2 tables without ``profile_name`` for its whole life; likewise the delivery ledger adds ``delivery_obligations.adapter_profile`` only when a gateway opens it. Identity settlement must gate its column SQL on the column actually being there rather than on table existence, and must not force those migrations (#113757). zGSELECT name FROM sqlite_master WHERE type='table' AND name IN (?, ?, ?)rzPRAGMA table_info('z'))execute_OPTIONAL_TABLE_NAMES)connrowexistingtables 9/home/ubuntu/.hermes/hermes-agent/hermes_state_gateway.py_optional_table_columnsr"s#',,Q# 3A H +  ex.? $,,1DUG2/N"OP3APP  Q s A&!A0 A+A0+A0zFUPDATE sessions SET handoff_state = 'failed', handoff_error = ? WHERE c eZdZdZ d,dZdddddddddd d-dZddd.d Zddd/d Zddd0d Zd1d Z d2d Z d3dZ dd d4dZ ddd d5dZ dddddd d6dZddd7dZd8dZd9dZd:dZd;dZddZd?d Zd@d!ZdAd"ZdBd#Zd2d$ZdAd%ZdCd&ZdDd'ZdEd(Z dd) dFd*Z!dGd+Z"y)HSessionGatewayMixinzLRouting index, session peers/orphans, hygiene streaks, heartbeats, handoffs.c ddlm}ddlm}m}t j jdr|gS ddlm }tj}g}||j|D]} |j| } | jdD cgc]} | j} } || j!|| j#z ||| j%| s |j'| g} |D].} | j)| j'| j*0| sgS |j1|d \}}|D]} | j3|r |j1|d | S| S#t$rgcYSwxYwcc} w#t$rY7wxYw#|j,t.f$rYwxYw#t$rg}YwxYw#|j,t.f$rYwxYw#t$rY| SwxYw) zTerminate old PPID-1 Desktop ephemeral backends with no client. Fails closed: anything whose parent, age, argv, or network connections cannot be proved safe remains a repair-blocking holder.r)psutil)_concrete_state_db_holder_pids"_is_inactive_orphan_desktop_holderlinux)_is_ephemeral_port_zero_backendinet)kind)ppid age_secondsmin_age_secondsephemeral_backendconnection_statusesg?)timeout)rr&hermes_state_dbfiler'r(sysplatform startswithhermes_cli.dashboard_procsr* Exceptiontimedb_pathProcessnet_connectionsstatusr- create_timecmdlineappend terminatepidErrorOSError wait_procskill)selfholdersr/r&r'r(r*now candidatesrBprocessrstatuses signalled_gonealives r!%_reap_inactive_orphan_desktop_holdersz9SessionGatewayMixin._reap_inactive_orphan_desktop_holderss" (j||&&w/6>I  Riik 1$,,H 'C  ..-4;4K4KQW4K4XYDDKKYY9 S7;N;N;P5P$3&EgooFW&X(0 2  2   g & ' " ! G !!#  -  I !,,Z,ELE5 G     !!%!5yQ I Z  LL'*    E  LL'*     s~E7&%F  FAF =+F/F8 G  G$7 FFF  FFF54F58 GG G! G!$ G10G1NF) user_id session_keychat_id chat_type thread_id display_name origin_jsoninclude_compression_ancestorstransport_profilec  rsy f } | rg| ng|   f d} j| y)aPersist the gateway routing peer for an existing session row. ``display_name`` / ``origin_json``: ``None`` leaves the stored value untouched (consumers read routing data from state.db, not sessions.json). ``include_compression_ancestors`` keeps a compression lineage on one routing peer when an explicit resume moves its tip to another lane; per-turn refreshes update only the supplied row. Self-healing: a missing target row (deferred ``create_session`` write, or crash between routing publication and row creation) is INSERTed with full identity rather than no-opped, so a gateway row is never first-created by the identity-less lazy writer (``update_token_counts``) and left unroutable forever. See #9006. Nc |jrtnddrdndry|jdfj?|jd  j t jf yy)Na UPDATE sessions SET session_key = ?, source = ?, user_id = ?, chat_id = ?, chat_type = ?, thread_id = ?, display_name = COALESCE(?, display_name), origin_json = COALESCE(?, origin_json), transport_profile = COALESCE(?, transport_profile) z0WHERE id IN (SELECT id FROM compression_lineage)z WHERE id = ?z+SELECT 1 FROM sessions WHERE id = ? LIMIT 1aIINSERT INTO sessions ( id, source, user_id, session_key, chat_id, chat_type, thread_id, display_name, origin_json, profile_name, transport_profile, started_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET session_key = COALESCE(sessions.session_key, excluded.session_key), chat_id = COALESCE(sessions.chat_id, excluded.chat_id), chat_type = COALESCE(sessions.chat_type, excluded.chat_type), thread_id = COALESCE(sessions.thread_id, excluded.thread_id), display_name = COALESCE(sessions.display_name, excluded.display_name), origin_json = COALESCE(sessions.origin_json, excluded.origin_json), transport_profile = COALESCE(sessions.transport_profile, excluded.transport_profile))r_COMPRESSION_LINEAGE_CTEfetchone_own_profile_namer9)r ancestorsrSrTrVrW query_paramsrG session_idrRsourcerUrYrQs r!_doz._dos LL1:-CDKTFYghl ||IJ=Ybbdl  w  +w S\^j $"8"8": SessionEntry JSON); ``scope`` namespaces the index per sessions_dir so two stores never share routing state.NaINSERT INTO gateway_routing (scope, session_key, entry_json, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(scope, session_key) DO UPDATE SET entry_json = excluded.entry_json, updated_at = excluded.updated_at) _write_sqlr9)rGrR entry_jsonrjs r!save_gateway_routing_entryz.SessionGatewayMixin.save_gateway_routing_entrys1*   7 KTYY[ 9  recbtjfd}|j|y)zzAtomically replace the routing index for *scope* (keys absent from *entries* are removed); other scopes untouched.c |jdfr>|jdjDcgc]\}}|s |s ||fc}}yycc}}w)Nz+DELETE FROM gateway_routing WHERE scope = ?z\INSERT INTO gateway_routing (scope, session_key, entry_json, updated_at) VALUES (?, ?, ?, ?))r executemanyitems)rkventriesrIrjs r!rdz@SessionGatewayMixin.replace_gateway_routing_entries.._do's[ LLF Q  *4;MMOODAqqQeQ3'OQPs A A A N)r9rg)rGrurjrdrIs `` @r!replace_gateway_routing_entriesz3SessionGatewayMixin.replace_gateway_routing_entries#s&iik Q C rec`|jd|f}|Dcic] }|d|d c}Scc}w)z>Load routing entries for *scope* as {session_key: entry_json}.zCSELECT session_key, entry_json FROM gateway_routing WHERE scope = ?rRrm) _read_all)rGrjrowsrs r!load_gateway_routing_entriesz0SessionGatewayMixin.load_gateway_routing_entries0s9~~cfkemn;?@a- !L/1@@@s+c|dkrtd|dtjt|dzz }|jd|f}|Dcgc] }t |c}Scc}w)u*Keyed, still-open rows with no evidence of a single turn (no messages, tokens, tool/API calls, activity, or title): leaked fixtures or chats routed but never answered. Safe to drop — the gateway mints a fresh session on the next message. Needs its own selector because ``bulk prune``/``archive`` are pinned to ``ended_at IS NOT NULL``. ``pinned``/``archived`` = explicit keep intent. That is exactly the shape of a leaked test fixture (#82770) — and also of a chat that was routed but never answered. rz"older_than_days must be >= 0, got zX: a negative retention builds a future cutoff that matches every never-active keyed row.g@a SELECT s.id, s.session_key, s.source, s.chat_id, s.chat_type, s.user_id, s.started_at FROM sessions s WHERE s.session_key IS NOT NULL AND s.ended_at IS NULL AND s.title IS NULL AND s.last_activity_at IS NULL AND COALESCE(s.message_count, 0) = 0 AND COALESCE(s.tool_call_count, 0) = 0 AND COALESCE(s.api_call_count, 0) = 0 AND COALESCE(s.input_tokens, 0) = 0 AND COALESCE(s.output_tokens, 0) = 0 AND COALESCE(s.pinned, 0) = 0 AND COALESCE(s.archived, 0) = 0 AND s.started_at IS NOT NULL AND s.started_at < ? AND NOT EXISTS ( SELECT 1 FROM messages m WHERE m.session_id = s.id ) ORDER BY s.started_at ) ValueErrorr9floatrxdict)rGolder_than_dayscutoffryrzs r! list_never_active_keyed_sessionsz4SessionGatewayMixin.list_never_active_keyed_sessions5s Q 4_4GH^^_ _o 6 @A~~ ,I/ 2"&&AQ&&&sA$c|jdD]H} tj|dxsd}t |t s1|j d|k(sF|cSy#t$rYWwxYw)zThe routing entry (any scope) whose current owner is *session_id*, or None. The id lives only inside ``entry_json``, so matching is done in Python; an archived/rotated row has none.z&SELECT entry_json FROM gateway_routingrm{}rbN)rxjsonloadsr8 isinstancerget)rGrbrentrys r!!gateway_routing_entry_for_sessionz5SessionGatewayMixin.gateway_routing_entry_for_session^st>>"JK C  3|#4#<=%&599\+Bj+P     sA A*)A*cP|syg}|jdD]]} tj|dxsd}t |t s1|j d|vsE|j|d|df_|sy|jd|d t|S#t$rYwxYw) zDrop ``gateway_routing`` rows pointing at any of *session_ids*; the target id lives only inside ``entry_json``, so matching is done in Python over all scopes.rz:SELECT scope, session_key, entry_json FROM gateway_routingrmrrbrjrRz?DELETE FROM gateway_routing WHERE scope = ? AND session_key = ?T)many) rxrrr8rrrr@rllen)rG session_idsdoomedrrs r!$_delete_routing_entries_for_sessionsz8SessionGatewayMixin._delete_routing_entries_for_sessionsjs(*>>"^_ BC  3|#4#<=%&599\+Bk+Q s7|S-?@A  B Y[ahlm6{  sB B%$B% sessions_dircj|}|sy|Dchc]}t|d}}j|}tfd|D}||fScc}w)aTDelete never-active keyed rows and the routing entries naming them; returns ``(sessions_deleted, routing_entries_deleted)``. Routing entries go first: a stale entry outliving its target would have the gateway resume a nonexistent id. Deletion goes through :meth:`delete_session` (delegate cascade, FTS, transcripts).)r)rridc3JK|]}j|sdyw)rrN)delete_session).0sidrGrs r! zHSessionGatewayMixin.prune_never_active_keyed_sessions..s$^C$*=*=cP\*=*]a^s##)rstrrsum)rGrrrJridsrouting_deleteddeleteds` ` r!!prune_never_active_keyed_sessionsz5SessionGatewayMixin.prune_never_active_keyed_sessions|si ::?:[ )34#s3t9~44CCCH^3^^))5sAT)r5 active_onlyc|jdtdd}|r|gng}||rdnd|rdndzdzz }|j||Dcgc]}|j|c}Scc}w)zList gateway sessions (rows with a session_key): newest row per key, one live mapping per routing key. ``platform`` filters on ``source``.z SELECT sessions.*, COALESCE(sp.prompt, sessions.system_prompt) AS _system_prompt_resolved, sessionsag AS last_active FROM sessions LEFT JOIN system_prompts sp ON sp.hash = sessions.system_prompt_hash WHERE session_key IS NOT NULL AND started_at = ( SELECT MAX(s2.started_at) FROM sessions s2 WHERE s2.session_key = sessions.session_key ) z AND LOWER(source) = LOWER(?)r\z AND ended_at IS NULLz ORDER BY last_active DESC)flush_token_countsrrx_session_row_dict)rGr5rqueryparamsrzs r!list_gateway_sessionsz)SessionGatewayMixin.list_gateway_sessionss !-Z89 :  &.z2 X12'2 #<>Z[ [37>>%3PQa&&q)QQQsA))rQrRrSrTrUc |sy|j5}|jt||fj}||j |cdddS|| dddy|j } |jt |||||| | | fj}dddr|j |SdS#1swYxYw)umFind the latest recoverable gateway session for a routing peer. The durable ``session_key`` on the row rebuilds a missing/pruned ``sessions.json`` mapping. Rows ended only by the old ``agent_close`` bug or a mistaken TUI ``ws_orphan_reap`` are recoverable; explicit boundaries (/new, /resume switches, compression splits) are not. Ranked by ``last_activity_at`` (fallback ``started_at`` — alone it resurrected days-old zombies); rows with messages win, but an empty keyed row still beats ``None`` (which mints a new id while the transcript may live under a compression child). Reset fence: a candidate is rejected when a peer boundary row ended *after* its last activity, or the has-messages ranking could reach behind a /new. The exact-key fallback requires the complete peer tuple (never cross chats/threads/users) plus a profile fence: a Telegram DM's tuple is identical for every bot, so a sibling profile's legacy row would otherwise be adopted. Ours = profile_name is the owner or NULL; stores outside the profile tree derive no owner and stay unfenced. See #60609. Reset boundaries fence recovery (#68539): an intentional boundary such as ``session_reset`` (or any explicit non-recoverable end_reason) must block fallback to an *older* row for the same peer. Each candidate is therefore rejected when a boundary row for the peer ended *after* the candidate's last activity — if the conversation's most recent event is an intentional reset, recovery returns nothing rather than reaching behind it. N) _read_ctxr_PEER_BY_KEY_SQLr^rr__PEER_BY_TUPLE_SQL) rGrcrQrRrSrTrUrrowners r!$find_latest_gateway_session_for_peerz8SessionGatewayMixin.find_latest_gateway_session_for_peers. ^^  ,,/+v1FGPPRC--c2  )"3   **,E,,"VWgy)UZ\ach$ihj  /2t%%c*;t;  s9B>B>%>C) max_gap_sc@ | |jn t|}g}|j5}|jtj D]}d}d}|dr3d}|jt |d|dfj}|nd}nkd}|dxsd } |jt|d |d|d |d | |z | |z|d fj } | sd |dd}nt| dkDrd}n| d }|j|d |d|d|d|d |r|d nd|r|dnd|r|nd|du|d  ddd|D chc]# drt fd|DdkDr d%} } |D]} | d| vs d| d<d| d<|S#1swYSxYwcc} w)uReport message-bearing rows that lost their routing identity (messages, no ``session_key``). Adoptable only when exactly one keyed predecessor can be named: ``lineage`` (``parent_session_id`` is a keyed row of the same source; no time window) or ``contiguity`` (exactly one keyed same-source row with compatible ``user_id`` fell quiet within *max_gap_s* of the orphan's start and is older than its last activity). Ambiguity is reported ``adoptable=False`` with a reason, never guessed — mis-adopting splices one person's conversation into another's chat. Branch/delegate/tool rows are excluded: unkeyed by design, not damage.Nr\parent_session_idlineagercz9parent session carries no gateway identity of this source contiguity started_atrrrQ last_activez'no keyed predecessor fell quiet within z.0fzs of this session's startrz>ambiguous: more than one keyed predecessor matches this window message_countrR) orphan_idrcrrrdonor_idrRevidence adoptablereasonrc3:K|]}|ddk(sdyw)rrN)rxrzs r!rzESessionGatewayMixin.find_orphaned_gateway_sessions..s %[AAjMQz] sDF (FFcHrrk(ryfd}|j|S)zStamp *orphan_id* with *donor_id*'s routing identity, retire *donor_id*. Re-verifies the pair inside the write txn so a concurrent gateway that healed either row makes this a no-op. Non-NULL orphan columns are preserved.Fc z|jdfj}|jdfj}|||dr|ds|dxsd|dxsdk7ry|jd|d|d|d |d |d |d |d f |jdtjfy)NzxSELECT session_key, chat_id, chat_type, thread_id, user_id, origin_json, display_name, source FROM sessions WHERE id = ?z5SELECT session_key, source FROM sessions WHERE id = ?rRrcr\Fa6UPDATE sessions SET session_key = ?, chat_id = COALESCE(chat_id, ?), chat_type = COALESCE(chat_type, ?), thread_id = COALESCE(thread_id, ?), user_id = COALESCE(user_id, ?), origin_json = COALESCE(origin_json, ?), display_name = COALESCE(display_name, ?), parent_session_id = COALESCE(parent_session_id, ?) WHERE id = ? AND session_key IS NULLrSrTrUrQrWrVzfUPDATE sessions SET ended_at = COALESCE(ended_at, ?), end_reason = 'superseded_by_repair' WHERE id = ?T)rr^r9)rrrrrs r!rdz?SessionGatewayMixin.adopt_orphaned_gateway_session.._dosLLO hj  \\G)WW_W_Wa  u]7KvVcOdh-26(3C3IrJ LL <}%uY'7{9KUS^M_y!5#7~9NPXZce  LLCh' )rerf)rGrrrds `` r!adopt_orphaned_gateway_sessionz2SessionGatewayMixin.adopt_orphaned_gateway_sessions-I,A& N""3''rec6syfd}|j|S)zEAtomically increment the session-hygiene failure streak for one chat.rc|jdf|jdfj}t|dS)NzINSERT INTO gateway_hygiene_state (session_key, failure_streak) VALUES (?, 1) ON CONFLICT(session_key) DO UPDATE SET failure_streak = gateway_hygiene_state.failure_streak + 1zFSELECT failure_streak FROM gateway_hygiene_state WHERE session_key = ?r)rr^int)rrrRs r!rdzASessionGatewayMixin.increment_hygiene_failure_streak.._do3sO LLT  ,,X[fZhhj s1v; rerf)rGrRrds ` r! increment_hygiene_failure_streakz4SessionGatewayMixin.increment_hygiene_failure_streak/s! ""3''rec0|sy|jd|fy)z@Clear the persisted session-hygiene failure streak for one chat.Nz7DELETE FROM gateway_hygiene_state WHERE session_key = ?rl)rGrRs r!reset_hygiene_failure_streakz0SessionGatewayMixin.reset_hygiene_failure_streakAs  QT_Sabrec |xsdj|xsdjcirrk(rSddddc t  fd}|j|S)zAAtomically rewrite exact profile identity in this state database.r\agent::c z|jdjDchc]}|d }}t|}|jddzdzfj}|t d|dd|ddD]v\}}d|j |t vr#d jd |D}|jd |d |d |dfj}|jt d||jdfjd<|jdfjd<|jddzfjd<d}|jdjD]t\} } tj| } t| ts-| j dk(sB| d<|jdtj| d| f|dz }v|d<d|vrcd|j dt vr!|jdfjd<|jd dzfjd!<d"D]G}d|j |t vs |jd#|d$fj|d%<Id&|j d't vr&|jd(dzfjd)<|jd*fj} | D]\} }}|dz}|}|r tj|} t| trt| j d&tr"| d&j!r| d&dz| d&<| j d+}t|tr|j dk(r|d<tj| d}|jd,||| ft#| d-<ycc}w#ttf$rYwxYw#ttf$rd} YwxYw).N1SELECT name FROM sqlite_master WHERE type='table'rzSELECT old.scope, ? || substr(old.session_key, ?) FROM gateway_routing AS old JOIN gateway_routing AS target ON target.scope = old.scope AND target.session_key = ? || substr(old.session_key, ?) WHERE substr(old.session_key, 1, ?) = ? LIMIT 1rz#profile routing collision in scope z: ))r)rS)r)rSrU profile_namez AND c3,K|] }d|d|yw)ztarget.z = old.Nr)rcolumns r!rzGSessionGatewayMixin.rekey_profile_state.._do..es$(M:@gfXWVH5(MszSELECT 1 FROM z AS old JOIN z* AS target ON target.profile_name = ? AND z# WHERE old.profile_name = ? LIMIT 1zprofile identity collision in z;UPDATE sessions SET profile_name = ? WHERE profile_name = ?sessions_profile_namez;UPDATE gateway_heartbeats SET profile = ? WHERE profile = ?gateway_heartbeats_profilezaUPDATE sessions SET session_key = ? || substr(session_key, ?) WHERE substr(session_key, 1, ?) = ?sessions_session_keyzBSELECT id, origin_json FROM sessions WHERE origin_json IS NOT NULLprofilez0UPDATE sessions SET origin_json = ? WHERE id = ?F) ensure_asciisessions_origin_jsonradapter_profilezMUPDATE delivery_obligations SET adapter_profile = ? WHERE adapter_profile = ?$delivery_obligations_adapter_profilezmUPDATE delivery_obligations SET session_key = ? || substr(session_key, ?) WHERE substr(session_key, 1, ?) = ? delivery_obligations_session_key)rrzUPDATE z, SET profile_name = ? WHERE profile_name = ? _profile_namerRrzsUPDATE telegram_dm_topic_bindings SET session_key = ? || substr(session_key, ?) WHERE substr(session_key, 1, ?) = ?&telegram_dm_topic_bindings_session_keyz^SELECT rowid, session_key, entry_json FROM gateway_routing WHERE substr(session_key, 1, ?) = ?originzJUPDATE gateway_routing SET session_key = ?, entry_json = ? WHERE rowid = ?gateway_routing)rrr"r^r}rsetjoinrowcountrr TypeErrorrrdumpsrr6r)rrr topic_columns collisionr columnsequality origin_countrbrWpayloadroutingrowidrRrmnew_session_keynew_jsonrcountsnewnew_nsns_lenoldold_nss r!rdz4SessionGatewayMixin.rekey_profile_state.._doPs"*.,,C+EEMXZQ3AQHQ3D9M B !VVaZH hj $ 9)A,9IIVWLK[\^^# Ow"):):5#%)HH"<<(MDK(MM LL$UG=@66>Z@99;>*FGOhj($'EeW%MNN O/3llMPSUXz/[[c[c * +37<<MPSUXz4[[c[c / 0-1\\6!VV4.67?h ) * L+/<<X,ZZbZbZd &' K"jj5Ggt,Y1G31N),GI&LL!S"&**W5"I:!VX A%L &.:F) *%1$ (9(9:PRURW(XXEI\\gc F$$,HAB>B\\:VaZ8>:;C(9:R -!]%6%6uce%DD6:ll!%(TUc 7$$,HeWM23 -  1 12NPSPU VVCG<<:VaZ8 D:;C( ?@ ll68>7GIIQ 3: 8.{J"(;vw+?"?%'"&**Z"8"'40%gkk-&@#F7S`KaKlKlmsKt5;gm>TU[U\>]5]GM2!(X!6%fd3 98MQT8T03F9-#'::gE#J `$h68 8$),G F$ %}QV#I.N' 2'"&'s) P  P.P&P#"P#&P:9P:striprrg) rGold_namenew_namerdrrrrrrs @@@@@@r!rekey_profile_statez'SessionGatewayMixin.rekey_profile_stateGsN))+hn"-C-C-ES!##M!#aF3%q/V_ 5_ 5B C  rec|xsdjisSddtddcfd}|j|S)u Delete exact profile identity from this state database (#111926, delete side). The mirror of :meth:`rekey_profile_state`: a rename must rekey a profile's identity, a delete must purge it. ``agent::*`` routing keys, ``gateway_heartbeats.profile``, ``delivery_obligations`` and the telegram topic tables are bookkeeping for a profile that no longer exists — left behind, every inbound event on a chat keyed to the deleted name enters the routing index, resolves a profile whose directory is gone, and logs ``Profile '' does not exist`` on each event for the life of the store. What each store gets, and why: * Routing keys and heartbeat rows are hard-deleted — pure bookkeeping for a dead name. * ``delivery_obligations`` rows are **terminalized** (``state='abandoned'``), not deleted: a pending obligation is delivery state that should not vanish silently, and the ledger's own retention prunes abandoned rows. Delivered history is left as it was. * ``sessions`` rows are not deleted here: this helper settles identity, not history, and it does not decide what a delete leaves of a profile's conversation record — ``delete_profile`` removes the profile's own home, ``state.db`` included, with the directory. Rows in a shared store keep whatever ownership they had; re-binding or archiving them belongs to the flow that recreates the name, not to this purge. Idempotent. r\rrc(|jdjDchc]}|d }}t|}d|vr!|jd fjd<d|vr |jdfjd<d|vrtd|j dt vrd nd }|rt j  fnt j f}|jd |d |jd<d |j dt vr |jdfjd<|j dt }d|vr7d |vrdnd }|r fn f}|jd|d|jd<yycc}w)Nrrrz?DELETE FROM gateway_routing WHERE substr(session_key, 1, ?) = ?gateway_heartbeatsz0DELETE FROM gateway_heartbeats WHERE profile = ?rrzadapter_profile = ? OR r\zGUPDATE delivery_obligations SET state='abandoned', updated_at=? WHERE (zJsubstr(session_key, 1, ?) = ?) AND state NOT IN ('delivered', 'abandoned')rrz9DELETE FROM telegram_dm_topic_mode WHERE profile_name = ?rrRzprofile_name = ? OR z-DELETE FROM telegram_dm_topic_bindings WHERE zsubstr(session_key, 1, ?) = ?)rrr"rrrr9) rrrr by_profilerbinding_columnsrnamensrs r!rdz4SessionGatewayMixin.purge_profile_state.._dos*.,,C+EEMXZQ3AQHQ3D9M H,,0LLURL-""*(()$x//3||F0QQYQY+,%1 "3m6G6GH^`c`e6f!f8#%=G$))+tVR8TYY[Z`bdLe15(\*BBCI2KLT8-.!2!23KSU!SS37<<ORVQX4ZZbZb/0+//0LceTO/ 8F7X3^` /9$+|7;||'L(EGHN8PPXPX3407Qs Fr)rGrrdrr r rs @@@@r!purge_profile_statez'SessionGatewayMixin.purge_profile_statesb0 2$$&!#MdV1%sVD6+;'< F% YN C  rec ddlm}|xsijd}t|tr t j |}t|tsi}|jd}t|tr8|jdr'|jDcic] \}}| || c}}SdDcic]&}|j|s||j|(}}|r|St |xsijdxsdj}|r|j|vrd|iSt|tsiS|jDcic] \}}| || c}}S#t$ri}Y/wxYwcc}}wcc}wcc}}w) uRead the persisted runtime route off a session row dict (``model_config`` as JSON string or parsed dict). Precedence: nested ``gateway_runtime`` (gateway sync / CLI ``/model``), then top-level ``provider``/``base_url``/``api_mode`` (TUI), then ``billing_provider`` so sessions that never ran ``/model`` still restore the provider that served them. Empty dict on parse failure — resume uses ambient config.r)_BARE_BILLING_PROVIDERS model_configgateway_runtimeprovider)rbase_urlapi_modebilling_providerr\) rr rrrrrr8rrrrlower) session_metar rawruntimersrtkey top_levelrs r!session_gateway_runtimez+SessionGatewayMixin.session_gateway_runtimesm 9!r&&~6 c3  jjo#t$C''+, gt $Z)@%,]]_FTQ AqDF F2Vg3Z]ZaZabeZfS#''#,&g g   2778JKQrRXXZ  0 6 6 8@W W 01 1'4(I!(BAAM1BB)  GgCs5E# E5$E51E;E; FF# E21E2)last_heartbeatrhostc |sy|tjn t|}|jdt|t |t||t|t|fy)aTUpsert this backend's liveness row. ``backend_id`` MUST be stable for the process lifetime (e.g. ``f"{profile}@{host}:{pid}"``) so a respawn cannot inherit a dead predecessor's heartbeat; ``started_at`` is when THIS process started, so a backend whose previous run died is not mistaken for a freshly-spawned sibling.Na,INSERT INTO gateway_heartbeats (backend_id, pid, started_at, last_heartbeat, profile, host) VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT(backend_id) DO UPDATE SET pid = excluded.pid, started_at = excluded.started_at, last_heartbeat = excluded.last_heartbeat, profile = excluded.profile, host = excluded.host)r9r~rlrr)rG backend_idrBrrrrtss r!register_backend_heartbeatz.SessionGatewayMixin.register_backend_heartbeats\ *2TYY[n8M  @_c#hj(92s7|SQUY W  YrecF|sy|jdt|fdkDS)zRemove this backend's heartbeat row (from ``atexit``); True if removed. A crashed backend's row is reclaimed later by ``prune_stale_heartbeats``.Fz3DELETE FROM gateway_heartbeats WHERE backend_id = ?r)_write_rowcountr)rGrs r!clear_backend_heartbeatz+SessionGatewayMixin.clear_backend_heartbeat/s4## AC OCUWYZ[ [rec|dkrgStj|z fd}t|j|xsgS)uDrop heartbeat rows older than the staleness window; return removed backend ids. Safe from any process — only stale rows are touched.rc|jdf}|jDcgc]}t|dc}Scc}w)NzLDELETE FROM gateway_heartbeats WHERE last_heartbeat < ? RETURNING backend_idr)rrr)rcurrzrs r!rdz7SessionGatewayMixin.prune_stale_heartbeats.._do=s=,,^ C(+||~6!C!I6 66s?)r9listrg)rGmax_age_secondsrdrs @r!prune_stale_heartbeatsz*SessionGatewayMixin.prune_stale_heartbeats7sD a I. 7 D'',233rec`|jd}|Dcgc] }t|c}Scc}w)zQSnapshot of every backend heartbeat (diagnostics/tests); fields mirror the table.zvSELECT backend_id, pid, started_at, last_heartbeat, profile, host FROM gateway_heartbeats ORDER BY last_heartbeat DESC)rxrrGryrzs r!list_backend_heartbeatsz+SessionGatewayMixin.list_backend_heartbeatsDs/~~ ,-"&&AQ&&&s+c0|jd||fdkDS)zVMark a session pending handoff to *platform*; False if a handoff is already in flight.zUPDATE sessions SET handoff_state = 'pending', handoff_platform = ?, handoff_error = NULL WHERE id = ? AND (handoff_state IS NULL OR handoff_state IN ('completed', 'failed'))rr")rGrbr5s r!request_handoffz#SessionGatewayMixin.request_handoffKs/## Mz "    recl |jd|f}|sy|d|d|ddS#t$rYywxYw)zWReturn ``{"state", "platform", "error"}`` or None if the session has no handoff record.zPSELECT handoff_state, handoff_platform, handoff_error FROM sessions WHERE id = ?N handoff_statehandoff_platform handoff_error)stater5error) _read_oner8)rGrbrs r!get_handoff_statez%SessionGatewayMixin.get_handoff_stateTsY ..b C 1sCU?V 13 3  s'' 33c |jd}|Dcgc]}|j|c}Scc}w#t$rgcYSwxYw)zPAll sessions in handoff_state='pending', oldest first (gateway handoff watcher).zSELECT s.*, COALESCE(sp.prompt, s.system_prompt) AS _system_prompt_resolved FROM sessions s LEFT JOIN system_prompts sp ON sp.hash = s.system_prompt_hash WHERE s.handoff_state = 'pending' ORDER BY s.started_at ASC)rxrr8r+s r!list_pending_handoffsz)SessionGatewayMixin.list_pending_handoffsasO >>,-D 8<SELECT 1 FROM sessions WHERE handoff_state = 'pending' LIMIT 1N)r6)rGs r!has_pending_handoffsz(SessionGatewayMixin.has_pending_handoffsts~~^_gkkkrec*|jd|fy)zMark a handoff as completed.zRUPDATE sessions SET handoff_state = 'completed', handoff_error = NULL WHERE id = ?Nrr;s r!complete_handoffz$SessionGatewayMixin.complete_handoffxs  ` M re) only_statesc|r t|nd}tdz|rddjd|Ddndz}|j||dd |g|d kDS) uMark a handoff failed and record the reason; True when a row transitioned. ``only_states`` makes the write a compare-and-swap on ``handoff_state``. Waiters that give up (CLI 60s poll, Desktop bounded poll) MUST pass ``only_states=("pending",)``: once the watcher has claimed the row (``running``) it owns the terminal state, and an unconditional waiter-side fail races the dispatch — the gateway later overwrites ``failed`` → ``completed`` after the user was told the gateway is down (split-brain: the handoff delivered and ``switch_session`` re-pointed the session). The watcher fails its OWN claimed row unconditionally.rzid = ?z AND handoff_state IN (rc3 K|]}dyw)?Nr)r_s r!rz3SessionGatewayMixin.fail_handoff..s/D/Ds )r\Nr)tuple_HANDOFF_FAIL_SQLrr")rGrbr5rAstatessqls r! fail_handoffz SessionGatewayMixin.fail_handoff~sl(3{#(*KQ%dii/DV/D&D%EQ GWY[##C%+z)KF)KLqPPrecfd} |j|xsgS#t$rtjddgcYSwxYw)uFail every handoff stuck in ``running``; returns the ids reclaimed. Only the gateway watcher sets ``running``, for one in-process dispatch — so a ``running`` row at watcher startup belongs to a PREVIOUS gateway that died mid-dispatch. It is poisonous: ``request_handoff`` only accepts NULL/``completed``/``failed``, so the session could never hand off again, with no error surfaced. Failing rather than re-queueing is deliberate: the dead gateway may already have switched the session key and dispatched the synthetic turn, so a blind retry risks double delivery; a clean terminal state the user can retry from is right.c|jd}|jDcgc]}|d }}|r|jtdzddf|Scc}w)Nz7SELECT id FROM sessions WHERE handoff_state = 'running'rzhandoff_state = 'running'rG)rrrI)rr&rzrr5s r!rdz?SessionGatewayMixin.reclaim_stale_running_handoffs.._dos[,,XYC!$0A1Q40C0 .1LLuUYVY{n]J1s Azbreclaim_stale_running_handoffs failed; stranded 'running' handoff rows (if any) were left in placeT)exc_info)rgr8loggerwarning)rGr5rds ` r!reclaim_stale_running_handoffsz2SessionGatewayMixin.reclaim_stale_running_handoffssU   &&s+1r 1  NN;EI  KI  s"AA)rHzList[Tuple[int, str]]r/r~returnz List[int])rbrrcrrQrrRrrSrrTrrUrrVrrWrrXboolrYrrSNone)rRrrmrrjrrSrU)ruDict[str, str]rjrrSrU)rjrrSrV)rr~rSList[Dict[str, Any]])rbrrSOptional[Dict[str, Any]])rzSet[str]rSr)rr~rzOptional[Path]rSzTuple[int, int])r5 Optional[str]rrTrSrW)rcrrQrYrRrYrSrYrTrYrUrYrSrX)rOptional[float]rSrW)rrrrrSrT)rRrrSr)rRrrSrU)rrrrrSDict[str, int])rrrSr[)rrXrSzDict[str, Any])rrrBrrr~rrZrrrrrSrU)rrrSrT)r(r~rS List[str])rSrW)rbrr5rrSrT)rbrrSrT)rSrT)rbrrSrU)rbrr5rrAzOptional[Tuple[str, ...]]rSrT)r5rrSr\)#__name__ __module__ __qualname____doc__rPrirnrvr{rrrrrrrrrrrr  staticmethodrr r#r)r,r/r7r9r<r>r@rLrRrrer!r$r$svV4,4BG4LU4n?CW[dT_cu!% ;!;!*-;!8;;!QT;!;!(+;!?B;!Y\;!;!AE;! ;!+/ ;!z]_  XZ !<>A ''R &IM *"' *7E *Q` *,0TR(R>BRNbR48rksC #  88   > *  11EGYZ[\01EGYZ[\$$;$4$;$;g$F#GH" **+??[>\]. /E-EF ,,-? @\>\ ]./E-EF%6=1567))9;MNOP()9;MNOP ! 567% 56b9QRU9V8WX 8<=>7<=> !T$]C C re