߲jyz dZddlmZddlZddlZddlZddlZddlZddlm Z m Z m Z m Z m Z ddlmZmZmZmZddlmZddlmZddlmZdd lmZmZmZmZmZm Z m!Z!m"Z"ejFd Z$d Z%e&ejNd e%jQd ddjQdddhdzZ)dZ*dZ+dZ,dZ-de-zZ.dZ/dZ0dZ1dZ2dZ3dZ4dZ5dZ6e7Z8d'dZ9d(dZ:d)d Z;d*d!Zd.d$Z?Gd%d&Z@y)/zTranscript persistence for SessionDB: message append/replace/rewind, reactions, resume assembly, replayed-user dedupe. Mixin bound via the MRO, built on SessionDB's _read_ctx/_execute_write/_read_* primitives.) annotationsN)AnyDictListOptionalTuple)_DB_PERSISTED_MARKER_is_checkpoint_item_newest_checkpoint_carriersplit_user_originated_turn)sanitize_context)_sanitize_surrogates coerce_epoch)_COMPRESSION_LOCK_ROW_SQL_ENDED_ROW_SQL_RESET_END_REASONS_RESET_END_REASONS_SQL_ended_by_compression_legacy_reset_child_sql _placeholders_sql_json_extract hermes_stateaINSERT INTO messages (session_id, role, content, tool_call_id, tool_calls, tool_name, effect_disposition, timestamp, token_count, finish_reason, reasoning, reasoning_content, reasoning_details, codex_reasoning_items, codex_message_items, platform_message_id, observed, _compressed_summary, active, api_content, display_kind, display_metadata, display_identity) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)z\w+()>id compacted display_orderz INSERT INTO conversation_generations (source, session_key, generation) VALUES (?, ?, 1) ON CONFLICT(source, session_key) DO UPDATE SET generation = conversation_generations.generation + 1 zLSELECT holder, expires_at FROM session_turn_leases WHERE conversation_id = ?zADELETE FROM compression_locks WHERE session_id = ? AND holder = ?z" AND (active = 1 OR compacted = 1)zLSELECT display_metadata FROM messages WHERE id = ? AND session_id IN ({ids})zGSELECT id FROM messages WHERE session_id = ? AND active = 1 ORDER BY idzxSELECT id, role, content, tool_call_id, tool_calls FROM messages WHERE session_id = ? AND active = 1 ORDER BY id LIMIT ?z:UPDATE sessions SET message_count = ?, tool_call_count = ?zGUPDATE sessions SET message_count = 0, tool_call_count = 0 WHERE id = ?z5UPDATE messages SET display_metadata = ? WHERE id = ?zQUPDATE messages SET active = 0, compacted = 1 WHERE session_id = ? AND active = 1zSELECT id, codex_reasoning_items FROM messages WHERE session_id = ? AND active = 1 AND role = 'assistant' AND id < ? AND codex_reasoning_items LIKE '%"compaction"%'z:UPDATE messages SET codex_reasoning_items = ? WHERE id = ?c tj|S#tjtf$rtj ||cYSwxYw)zD``json.loads(raw)``; on failure log *warning* and return *fallback*.)jsonloadsJSONDecodeError TypeErrorloggerwarning)rawfallbackr&s :/home/ubuntu/.hermes/hermes-agent/hermes_state_messages.py_json_orr*>s@zz#  ) ,ws0A  A c6t|d}||S|dk(rdS|S)uExplicit message timestamp (datetime or number) or *default* when invalid or outside the sane epoch window — the write-side twin of the readers' ``coerce_epoch``: a bad row is never persisted.zmessage timestamp)fieldgr)valuedefaultresults r)_coerce_timestampr0Gs.%': ;F ~C-3+V+ct|ts|S tj|S#tjt f$rgcYSwxYw)zktool_calls is a list (live agent) or JSON string (import/export); parse so json.dumps never double-encodes.) isinstancestrr!r"r#r$ tool_callss r)_parse_tool_callsr7SsE j# &zz*%%  ) , s)AAcD|dSt|tr t|SdS)Nrr)r3listlenr5s r)_tool_calls_countr;]s&"1`JzSWZ8dd?Z9dd@Z:ddAZ;d dBddCZ
ddFZ?eddGZ@eddHZAe ddIZBddJZCeddKZDddLZEd dddM ddNZFddxdOZGddPZHddQZId dRddSZJdwdTZKddUZLddVZMd ddWddXZNy)SessionMessagesMixinzNMessage append/replace/rewind, reactions, resume conversations, replay dedupe.c|tvry|jd|fjyfddD\}}|r|r|jt||fyyy)a'Advance the peer's conversation generation past a boundary, in the txn that writes it. Only ``_RESET_END_REASONS`` count (compression continues one conversation). Never derived from session rows (deletes/prunes could re-emit a retired affinity identity); it only ever increments.Nz5SELECT source, session_key FROM sessions WHERE id = ?c3^K|]$}t|xsdj&yw)N)r4strip).0krHs r) zESessionMessagesMixin._bump_conversation_generation..~s'\Qs3q6|j|jr#t |t |jd|dS|S)z;Reverse :meth:`_encode_content`; returns scalars unchanged.NzCFailed to decode JSON-encoded message content; returning raw string)r3r4 startswithrar*r:rds r)_decode_contentz$SessionMessagesMixin._decode_contentsO gs #(:(:3;S;S(TGC(@(@$A$BCWUW Wr1cF|syt|tr@t|td}|turyt|tsPt j dyt|ts*t j dt|jytj|S)zSerialize ``display_metadata`` for its TEXT column; an already-serialized JSON string (import/replace paths) is not double-encoded.Nz+Ignoring non-JSON display metadata on writez-Ignoring non-object display metadata on writez6Ignoring unexpected display metadata type on write: %s) r3r4r*_INVALIDdictr%r&type__name__r!rbdisplay_metadatas r)_encode_display_metadataz-SessionMessagesMixin._encode_display_metadatas  & ,'(8(Dqr 8+.5NO,d3 NNSUYZjUkUtUt uzz*++r1c|y|}tdD]$}t|tst|td}&|turyt|t st jdy|S)zDecode a ``display_metadata`` column to a dict (never raw TEXT: the desktop does ``'task_count' in meta``). Pre-guard rows are double-encoded, so a second string layer is unwrapped.Nz0Ignoring invalid display metadata on message rowz3Ignoring non-object display metadata on message row)ranger3r4r*rlrmr%r&)r'meta_s r)_decode_display_metadataz-SessionMessagesMixin._decode_display_metadatask ;q dA$$h0bc d 8 $% NNP Q r1cX|sdSt|tr|Stj|S)aSerialize a structured reasoning field for its TEXT column. Strings are stored as-is: round-trips (get_messages -> replace_messages) hand back raw TEXT; re-dumping would double-encode it and reasoning-replay consumers (``isinstance(..., list)``) would drop it.N)r3r4r!rbrAs r)_reasoning_json_textz)SessionMessagesMixin._reasoning_json_texts) !t^ 5#0Fu^DJJW\L]^r1NFc &ddlm} ddlm} m} |ru|j t |fj} | Nt| tjr|j t|| dfn| d|k7r | d|d|s|r|j||} |j t| fj}tj}|rR||d|k7r | d|t|d |kr\|j d |td t|z| |fn0|.t||s | d ||j d | |dft|j t |fjr |s| |yy)u[Transcript-write admission checks, run INSIDE the write txn by every writer. Ordinary appends do NOT check compression_locks: the lock only stops two COMPRESSIONS colliding and archive_and_compact() commits against a watermark, so concurrent appends are safe (blocking them killed turns during slow summaries). Destructive user mutations opt in via ``reject_active_*`` so a compressor that captured its watermark cannot resurrect the removed turn. Shared by :meth:`append_message` and :meth:`append_messages_batch` so the two writers can never diverge on these correctness invariants (this guard has already needed targeted fixes — see the #74478 patience note below). User-initiated transcript mutations may opt in to rejecting an active unowned turn lease in that same transaction. r!SessionCompressionInProgressError)CompressionSessionClosedErrorSessionTurnLeaseLostErrorNrFzSession z& is being compressed by another writerz7Session turn lease lost; refusing transcript write for rEzVUPDATE session_turn_leases SET expires_at = ? WHERE conversation_id = ? AND holder = ?g?zCSession has an active turn lease; refusing transcript mutation for zHDELETE FROM session_turn_leases WHERE conversation_id = ? AND holder = ?)rr}hermes_state_errorsr~rrVrrWrJtime_DELETE_COMPRESSION_LOCK_SQL_session_turn_lease_key_on_conn_TURN_LEASE_ROW_SQLrGmaxrr)rYrZr[compression_lock_holderturn_lease_holderturn_lease_ttl_secondsreject_active_turn_leasereject_active_compression_lockallow_closed_compression_parentr}r~r active_lockconversation_idleaserIs r)_check_transcript_write_guardsz3SessionMessagesMixin._check_transcript_write_guardss C` *,,'@:-PYY[K& diik:LL!= KX`La?bc *.EE;":.0VWYY  8"BB4TOLL!46HIRRTE))+C =E(O7H$H3QR\Q_`bb|,-4LL"Cs3.D(EFFYjkm"$UC03]^h]klnn g$eHo68 nzm!L!U!U!W XbA/ ; ;bA Xr1c~ d}fd jjd}|rtj|nd} t jd} j jd} |||jd| | jd| d } |||jd| | jd |jd jd t  d t  dg fddDjdxsjdjdrdndjdrdndd|jd|jd| j j| S)zBind values for ``_INSERT_MESSAGE_SQL`` from one message dict (*tool_calls* already parsed; *keep_reasoning* False NULLs every reasoning column). ``platform_message_id`` falls back to ``message_id`` (yuanbao's message-dict convention).c<t|tr t|SdSN)r3r4rB)vs r)z:SessionMessagesMixin._message_row_params..sAs9K!21!5QUr1c.rj|SdSrget)keykeep_reasoningmsgs r)rz:SessionMessagesMixin._message_row_params..sTr1rfN tool_namerq tool_call_id display_kind)rolerf timestamprr6rrrqeffect_disposition token_count finish_reason reasoningreasoning_contentc3LK|]}j|ywr)rz)rQrR _reasoningrYs r)rSz;SessionMessagesMixin._message_row_params..s*^'' 1 6^s!$reasoning_detailscodex_reasoning_itemscodex_message_itemsplatform_message_id message_idobservedrr_compressed_summary api_content)rgrr!rbrBrr_display_identity_display_dedupe_key)rYr[rrr6message_timestampr _str_or_noneencoded_contentencoded_tool_callsencoded_tool_namerq identity_rowrs` ` ` @r)_message_row_paramsz(SessionMessagesMixin._message_row_paramss V I ..swwy/AB7ATZZ 3t-cggk.BC88AS9TU_CTGGN3CU*CGGN3J  1 GG( )+TVYV]V]^mVn j5 68I*UhJi8j ^^\^ ^ GG) * Ccggl.C ^$A! ^36'':O2PQVW ^Z[ ^ / 0 ^3?sww~?V2W ^  ^#44T5M5Ml5[\ ^ ^r1c|rdnd|ff\}}|dkDr|jd|dg|||y|dkDr|jd|dg||yy)zXBump sessions.* counters after an insert; *unit* bakes the ``+ 1`` literal into the SQL.)1?rz4UPDATE sessions SET message_count = message_count + zK, tool_call_count = tool_call_count + ? WHERE id = ? WHERE id = ?N)rV)rZr[insertedr6unitincparamss r)_bump_session_countersz+SessionMessagesMixin._bump_session_counterss$(icH;-? V > LLHNJM1&1*1j1 3\ LLFse=Y[p]c[peo[p rr1c.tt}j||d<tt |t j }j |||dfd}j|jS)aAppend one message; returns the row id and bumps the session counters. ``platform_message_id``: the platform's own id. ``api_content``: byte-fidelity sidecar, the exact string sent to the API when it differed from ``content``, stored as sent except lone surrogates.rqTrcj||jtj}j |dt d|S)NrrrTr)rrV_INSERT_MESSAGE_SQL lastrowidrr;) rZmsg_idrrrYr[r6rrs r)_doz0SessionMessagesMixin.append_message.._do8s`  / /jBY"3Lb 0 d\\"5v>HHF  ' 'j!=Nz=Zae ' fMr1 patience_s) rmlocalsrrr7r0rr_execute_write_TRANSCRIPT_WRITE_PATIENCE_S)rYr[rrfrr6rrrrrrrrrrrrrrrrqrrrrrrrs`` ` ``` @r)append_messagez#SessionMessagesMixin.append_message%s68n"&"?"?@P"Q &z2 -iE)) c:/@QU*W  ""343T3T"UUr1cjds td|jdrdndd}jd|dtjd fd }j |j S) aRecord a detached API result once, between client turns, including replay after rotation. The event's unit id, not its text or active flag, is the identity. Check and insert share the writer transaction, so independent gateway processes cannot duplicate it. delegation_idz3Delegation delivery requires a stable delegation_idpresentation_suppressedhiddenasync_delegation_complete)rfrrquserNTrc |jdjddfj}||dSj|dd|jtj }j |ddd|S) NaXWITH RECURSIVE lineage(id) AS ( SELECT ? UNION SELECT s.parent_session_id FROM sessions s JOIN lineage l ON s.id = l.id JOIN sessions p ON p.id = s.parent_session_id WHERE p.end_reason = 'compression' ) SELECT m.id FROM messages m JOIN lineage l ON m.session_id = l.id WHERE m.display_kind IN ('async_delegation_complete', 'hidden') AND json_extract(m.display_metadata, '$.delegation_id') = ? AND coalesce(json_extract(m.display_metadata, '$.delivery_notice'), '') = ? LIMIT 1delivery_noticerOrT)rrr)rVrrWrrrr)rZexistingrrmetadatarrYr[s r)rz._doPs||g]HLL9JB,OP RS[RZR\ #{"  / /j$ae / f\\"5v>HHF  ' 'j!QT ' JMr1r)rrcrrrr)rYr[rfrrrrrs`` ` @@r)append_delegation_deliveryz/SessionMessagesMixin.append_delegation_deliveryBs ! _5 RS S!+3<<8Q+RxXs#+-))*fc4ei)j  $""343T3T"UUr1c sy.msHB11*huUU_M_>`,CWh+A2CBs&)cj| ddlm}||jj}j ||\}}j |||d|S)Nrr)#resolve_and_repair_transcript_batch)encode_content_fndecode_content_fnFr)ragent.transcript_repairrrgrj_insert_message_rowsr) rZr inserted_rowsrtool_calls_totalrrrYr[rrs r)rz7SessionMessagesMixin.append_messages_batch.._doqs  / /jBY"3Lb 0 d S?jRZ"&"6"6$J^J^`M)-)B)B4Ub)c &H&  ' 'j(DT[` ' aOr1r)r:sumrurr)rYr[rrrrrrs``````` r)rz*SessionMessagesMixin.append_messages_batchdsw   !c(mj&@BB#1c(mZ@BB B  ""343T3T"UUr1rpcRrrsyfd}j|S)aStamp presentation metadata on this turn's freshly persisted row (newest active row by content, right after the serial turn flushed); the model still sees ``role``/``content`` unchanged, so producer provenance survives without classifying by content at render time.Fc|jdjfj}|y|jdtj |dfy)NzqSELECT id FROM messages WHERE session_id = ? AND role = ? AND content = ? AND active = 1 ORDER BY id DESC LIMIT 1FzGUPDATE messages SET display_kind = ?, display_metadata = ? WHERE id = ?rT)rVrgrWrBrr)rZrHrfrrqrrYr[s r)rzJSessionMessagesMixin.set_latest_matching_message_display_kind.._dosv,, JT4#7#7#@ACCK8: { LLb"<0$2O2OP`2acfghcij lr1r)rYr[rrfrrqrs`````` r)(set_latest_matching_message_display_kindz=SessionMessagesMixin.set_latest_matching_message_display_kind|s,    ""3''r1c|xsij|j}t|tr!|Dcgc]}t|ts|c}SgScc}w)zEWell-formed (dict) reactions stored under ``REACTIONS_METADATA_KEY``.)rREACTIONS_METADATA_KEYr3r9rm)rYrv reactionsrs r)_reaction_listz#SessionMessagesMixin._reaction_listsIZR$$T%@%@A @J9VZ@[9._do..s N1AEE(Ov4MQNs$$emoji)rrat) rVrWrxrrnextappendrBrrpop_SET_DISPLAY_META_SQLrr) rZrHrvrrrpreviousrrmessage_row_idrrYsqls r)rz6SessionMessagesMixin.set_message_reaction.._dos,,sF+446C{00Q8>BD**40H$,Jqh60IJIJNNPTUH(*hll7.Cu.L  +J V   ""3''r1c|r|gS|j|j||}|#|j|j|dSgS)zRSMP_t""4#@#@Q#HIdbddr1cn|j|}tjt||g|fS)unA reaction addresses a row the client can SEE, and a display resume materializes the whole compression lineage (active + compacted rows, with row ids) — so a row is "in this session" when its owner is any lineage segment, not only the tip, and a rewound row is not. Explicit ``/branch`` copies keep their own rows (``_resume_lineage_ids``).)ids)_resume_lineage_ids_DISPLAY_META_ROW_SQLformatr)rYr[rlineages r)rz(SessionMessagesMixin._reaction_row_querys; **:6$++ g0F+G.Ic[bIcccr1cb|sgSj|fd}j|S)a3Return *author*'s not-yet-surfaced reactions and mark them seen. Reactions are announced on the NEXT user turn (never by rewriting the reacted message: cache-safe); ``seen`` makes it exactly once. Include compaction-archived history that remains visible, but exclude rewound/superseded rows.c g}|jdt dtdtdd jzdt j D]} j|d}|r|j jnd}t|tsHd}|D]}t|tr%|jdk7s|jd r9d |d <d } j|d }|j|d |d |jdxsdt|tr|ndd|s|jt j!||d f|S)NzNSELECT id, role, content, display_metadata FROM messages WHERE session_id IN (rz AND rqz$.z IS NOT NULL ORDER BY idFrseenTrfrrrrO)row_idrrtext)rVr_DISPLAY_ACTIVE_CLAUSErrtuplefetchallrxrr3r9rmrjrr4rrr) rZpendingrHrvrchangedreactionrfrr rYs r)rz7SessionMessagesMixin.take_unseen_reactions.._dosG||%,,9',B+C1E[D\],-?HcHcAcdef"%"$)>34<8: j44S9K5LMEIDHHT%@%@At !)T2 )NH%h5h9OSY9Y]e]i]ijp]q '+HV$"G"223y>BGNN"%d)S[8<r assistantrNz0AND content IS NOT NULL AND TRIM(content) != '' rOzISELECT id FROM messages WHERE session_id = ? AND role = ? AND active = 1 z!ORDER BY id DESC LIMIT 1 OFFSET ?rr`)rYr[rrr text_filterrHs r)latest_message_row_idz*SessionMessagesMixin.latest_message_row_idseT)>>&1*LXH^` nn)]*KM s6{ +-s1v&$&r1c:|jd|f}|r|dSdS)aRole of the newest active user/assistant/tool row, or ``None``. ``session_meta`` / ``system`` rows are transcript bookkeeping stripped before the model sees history, so they must not hide an open user tail from the failed-turn boundary check.zSELECT role FROM messages WHERE session_id = ? AND active = 1 AND role NOT IN ('session_meta', 'system') ORDER BY id DESC LIMIT 1rNrrYr[rHs r)latest_conversation_rolez-SessionMessagesMixin.latest_conversation_roles2nn RT^S`bs1v&$&r1cT|sy|jdt||f}|r|dSdS)zDRole of the active message at *row_id* in *session_id*, or ``None``.NzHSELECT role FROM messages WHERE id = ? AND session_id = ? AND active = 1rr)rYr[rrHs r)get_message_rolez%SessionMessagesMixin.get_message_rolesBnngjmntjuxBjCDs1v&$&r1c Ntj}dx}}|D]}|jdd}t|jd} t|jd|} |j t |j |||| | |dk(} | |d<| j| j|d<|d z }|t| z }t|| d z}t|d } | dk\r;t|| jdtr|j|||| d||fS) zInsert *messages* as fresh active rows in the caller's txn -> ``(inserted, tool_call_count)``. Never touches sessions.* counters (callers reconcile differently); reasoning kept for assistant rows.rrunknownr6rrr_row_idrgư>r)rrr7r0rVrrrr;rr r3r`_drop_shadowed_checkpoint_rows) rYrZr[rnow_tsrrrrr6rcurcarriers r)rz)SessionMessagesMixin._insert_message_rowssD&''# ;C7769-D*377<+@AJ 1#''+2F O ,,2D4L4LD#z3DUY]hUh5M5jkC 1C }}(!$I MH  1* = = !23d:F ; -X7NO a.s5bTX6I$6O5bsr) rV_SHADOWED_CHECKPOINT_ROWS_SQLrr*r3r9anyr _SET_CODEX_REASONING_SQLrz) rYrZr[carrier_row_id rewrittenrr'itemsr/kepts r)r)z3SessionMessagesMixin._drop_shadowed_checkpoint_rowss  <<(E TbGcdmmo KFCS$(abEeT*#5b\a5b2b%*LT2Ed2KDLDL LL1D4M4Md4SU[3\ ] NI  Ms #B*4B*cVddlmfd}j|y)u\Atomically replace a session's messages (/retry, /undo, /compress). DESTRUCTIVE by default (rows DELETEd, leave FTS). ``active_only`` spares soft-archived rows (needed with in-place compaction). ``archive_dropped`` SOFT-archives live rows rewind-style: what rewind/edit/regenerate must use, since DELETE leaves nothing to recover. ``reject_active_turn_lease``: in-txn lease check for user rewrites. Pass ``archive_dropped=True`` to SOFT-archive the live rows instead of DELETEing them: the replaced turns stay on disk with ``active = 0``, ``compacted = 0`` — the same "the user took it back" marking :meth:`rewind_to_message` applies — and stay readable via :meth:`get_messages` with ``include_inactive=True``. This is the mode a rewind/edit/regenerate must use: those flows overwrite a transcript the user may not have meant to drop, and a plain DELETE also evicts the rows from the FTS index, leaving nothing to recover from (#82756). It implies active-only handling — already-archived rows are never touched — so ``active_only`` is redundant with it. Live rows that *messages* keeps in place (the common in-order prefix, matched on role/content/tool identity) are left untouched with their ids; only the divergent live suffix is archived and only the new suffix of *messages* is inserted (#82956: archiving and re-inserting the kept prefix grew the archive by the whole transcript on every rewind). The live view is identical either way; only the durability of the dropped turns differs. r)r~c^ r j| dddn6t|jt fj r dx}}r|jt t dzfj} j| }td|d|D}|t |kr5|jd ||dfn|jdrdnd  f j| |d\}}|jtd ||z||z fy) NTrrrrc3<K|]}t|ddyw)rr=Nr?rQrHs r)rSzESessionMessagesMixin.replace_messages.._do..Fs%_Coc!fQ&G&G%_zNUPDATE messages SET active = 0 WHERE session_id = ? AND active = 1 AND id >= ?)DELETE FROM messages WHERE session_id = ? AND active = 1rOr) rrrVrrW_LIVE_IDENTITY_SQLr:r_stamp_kept_live_prefixrr_SET_COUNTERS_SQL) rZr6kept_tool_callsliverinserted_tool_callsr~ active_onlyarchive_droppedrrrYr[s r)rz2SessionMessagesMixin.replace_messages.._do:s^'33*dTjn4p&t||NZM'R'['[']^3J??%& &D?||$6S]UVEV8WXaac33D(C"%%_SWX]Y]S^%_"_#d)#LL!q",d4jm!<> H^iIZoqHrswAvCD,0,E,EdJX`aeafXg,h )H) LL-.m</?=P+PR\] _r1N)rr~r)rYr[rrHrIrrr~s`````` @r)replace_messagesz%SessionMessagesMixin.replace_messages%s%( F _ _, C r1c|||j|j||||rtj|fSdfS)uThe (role, content, tool_call_id, tool_calls) columns a message writes — the identity the kept-prefix match compares. *content* is passed through the loader's lens first so a message read back from the DB matches the row it came from.N)rg_loaded_view_contentr!rb)rerrfrr6s r) _row_identityz"SessionMessagesMixin._row_identityRsM c))#*B*B4*QRT`*4 :&@ @:>@ @r1c d}t||D]\}}|jdd}|d|k7s|d|jdk7r|S|j||jd|jdt|jd}||j|d|j |d |dt|d k7r|S|d|d <|dz }|S) zLength of the in-order prefix of *messages* already present as the leading live rows; each matched message gets its existing ``_row_id`` stamped, as a fresh insert would set it.rrr'rrrfr6rtr;r()ziprrMr7rj)rYrFrr6rHrridentitys r)rCz,SessionMessagesMixin._stamp_kept_live_prefixZsD(+ HC7769-D1v~Q377>+B!B ))$ 0BCGGND[*;CGGLrr)r3r4r rP)rrfs r)rLz)SessionMessagesMixin._loaded_view_contentks/ ( (Z-E#G,224 4r1c,|jd|fduS)u}True if the session has any soft-archived (``active = 0``) rows (tests/diagnostics). Cheap existence probe — does not load rows. NOTE: production rewrite paths no longer branch on this (they pass ``active_only=True`` unconditionally — a probe can fail open or race a concurrent ``archive_and_compact``, #80216); kept for tests and diagnostics. zBSELECT 1 FROM messages WHERE session_id = ? AND active = 0 LIMIT 1Nr!rYr[s r)has_archived_messagesz*SessionMessagesMixin.has_archived_messagesss)~~ PS]R_ahlm mr1cF|syt|jd|fdS)zMAX(id) of the active rows (0 if none), captured at compression START: every active row above it arrived concurrently and must survive compaction verbatim.rzMSELECT COALESCE(MAX(id), 0) FROM messages WHERE session_id = ? AND active = 1)r`rrTs r)get_active_message_watermarkz1SessionMessagesMixin.get_active_message_watermark}s74>> [^h]jllmop pr1c|j||j}|Dcgc]}t|dc}td|DfScc}w)ze``(ids, tool_call_count)`` of the concurrent-tail rows selected by *sql* (``SELECT id, tool_calls``).rc38K|]}t|dyw)r6Nr=)rQrs r)rSzBSessionMessagesMixin._tail_rows_after_watermark..s0`VW<1Q0`s)rVrr`r)rYrZrrrowsrs r)_tail_rows_after_watermarkz/SessionMessagesMixin._tail_rows_after_watermarksH||C(113&*+AdG +S0`[_0`-```+sAr[c|du}d|rdndzdjfd|j|D}|jd|d|rdnd d |d|rd nd d t|d |r |g|y|y)zPure-SQL clone of *tail_ids* as fresh live rows (new id/display order, active=1, compacted=0; message payload columns stay byte-exact and FTS triggers index the clones), into *session_id* when given.N)ractiverrr\rz, c3,K|] }|vs| ywrr)rQcskips r)rSz;SessionMessagesMixin._clone_message_rows..sZ1AUYMQZs zINSERT INTO messages ( session_id, rOzactive, compacted) SELECT z?, z 1, 0 FROM messages WHERE id IN (z ) ORDER BY id)join_message_column_namesrVr)rYrZtail_idsr[retargetcol_listras @r)_clone_message_rowsz(SessionMessagesMixin._clone_message_rowssT)>T\bde99Z(B(B4(HZZ $XJb8QS0TUZr8%!<=)(34M C(0Z #( #  ?6>  ?r1cXddlmfd}j|S)uNon-destructive in-place compaction under ONE session id: soft-archive the active rows (``active=0, compacted=1``: summarized away, still searchable) and insert *compacted_messages* as fresh active rows, atomically; returns the new ACTIVE count (= ``message_count``). *watermark* (compression START): rows ``id > watermark`` arrived during the slow summary and are re-sequenced after the compacted set by a pure-SQL clone (fresh ids); ``None`` archives everything. *lock_holder*: verified in-txn so a reclaimed lease fails instead of clobbering the winner. *tail_count*: the LAST N compacted rows are the verbatim carried tail; their originals and the clones' originals are superseded duplicates and get rewind flags (``active=0, compacted=0``) so search doesn't return each carried message once per compaction. ``model_config_patch`` merges in the same txn (``None`` removes a key). Concurrent-append safety (#75316): when *watermark* is provided (the value of :meth:`get_active_message_watermark` captured at compression START), rows that arrived during the slow provider summary call (``id > watermark``) are NOT summarized away. They are re-sequenced after the compacted set by a pure-SQL column clone (every column except ``id`` — content, api_content, platform_message_id, token counts, reasoning sidecars all survive byte-exact, and the FTS triggers index the clones naturally), and the originals are archived. NOTE: re-sequencing assigns the tail rows fresh ids; consumers that reference durable row ids re-resolve by content (see 3e8ab0610). rr|c^|jtfj}|+|dk7s#t|dt jkr  dddu}|rj |dnd}gdfnj |dtf\}}g}dkDrbdu}|jd |rd nd d g|r tfnd tjDcgc]}t|d}}||z }|rDt|} |jd| dg||jtd| dg|n|jtfj| \} } |r%j||| t|z } | |z } |jt|rdnd d| | g|r|fnd | Scc}w)NrFrEzCompression lease for z; lost before commit; refusing to publish a stale compactionraise) on_missingrz^SELECT id, tool_calls FROM messages WHERE session_id = ? AND active = 1 AND id > ? ORDER BY idz;SELECT id FROM messages WHERE session_id = ? AND active = 1z AND id <= ?rOz ORDER BY id DESC LIMIT ?rrzNUPDATE messages SET active = 0, compacted = 0 WHERE session_id = ? AND id IN (rz AND id NOT IN (z, model_config = ?r)rVrrWrGr_merge_model_config_jsonr[r`rr_ARCHIVE_ACTIVE_SQLrrhr:rD)rZlock_rowpatchpatched_model_configretail_tool_calls rewind_idsboundrH placeholdersrrr}compacted_messages lock_holdermodel_config_patchrYr[ tail_count watermarks r)rz5SessionMessagesMixin.archive_and_compact.._dosz&<<(AJ=QZZ\#x'9['HERZ[gRhLimqmvmvmxLx;0>yz||&d2ENS$(#@#@j"4$A$JX\ !3<3DQ$JiJivS^,K. %Ho %'JA~!-8< QdiR`oqQrs//XC N#42XJX9Z[cZbZdfc#d)nf f ( "J,Z8  77CnAGISHaV`Hac 344D\NRSTWaVodnVop 0:-@)-)B)B4Ug)h &H&((x8CM) O3 LL-.u/CRT.UUbc+e%0D/FUWeZde gO'fs8G)rr}r) rYr[rvrxrzrwryrr}s ``````` @r)archive_and_compactz(SessionMessagesMixin.archive_and_compacts'* C% % L""3''r1ct|dds5|jdjDcgc]}|d c}|_|jScc}w)z>Column names of the messages table, cached per-connection era._message_columns_cacheNzPRAGMA table_info(messages)r)getattrrVrr})rYrZrs r)rdz*SessionMessagesMixin._message_column_namessKt5t<9=Fc9d9m9m9o*pA1Q4*pD '***+qs Ac\|jdt|||j|fS)ubBackfill the ``api_content`` sidecar onto the newest ACTIVE user row (0/1 rows). Preflight compaction inserts that row BEFORE the sidecar exists and the later persist identity-skips compacted dicts; without this a reload reopens the prompt-cache divergence. ``content`` match guards a racing rewrite. POSITIONAL, and only safe when the caller already knows the newest active user row IS the message it stamped. The content match is NOT sufficient on its own: repeated identical user turns ("ok", "y", "continue") make an OLDER row compare equal, so calling this before the current turn's row exists overwrites the previous turn's sidecar with this turn's bytes — durable wrong-bytes replay, a worse cache break than the missing sidecar. When the caller holds the durable row id (``_row_id``, synced onto the live dict by ``sync_flushed_message_markers`` and stamped by :meth:`_insert_message_rows`), use :meth:`set_message_api_content` instead — it addresses the exact row and cannot land on a neighbour. zUPDATE messages SET api_content = ? WHERE id = (SELECT id FROM messages WHERE session_id = ? AND role = 'user' AND active = 1 ORDER BY id DESC LIMIT 1) AND content IS ?)_write_rowcountrBrg)rYr[rfrs r)set_latest_user_api_contentz0SessionMessagesMixin.set_latest_user_api_contents:"## !{ +Z9M9Mg9V W Y Yr1c |r%t|tst|tr|dkry|jdt ||||j |fS)aeBackfill the ``api_content`` sidecar onto ONE known durable row. Row-addressed counterpart to :meth:`set_latest_user_api_content`: the caller passes the ``_row_id`` the write path stamped on the live message dict, so the update cannot drift onto a neighbouring row that merely carries the same text. Used by the turn prologue whenever the current turn's user row was already materialized before the sidecar could be composed (in-place preflight compaction, a close/early flush that raced the prologue). The crash persist then marker-skips that message, so this is the only way the stamped bytes reach the store. ``active = 1`` and the ``content`` match stay as defensive guards: a row the compaction archived, or one a racing rewrite changed, is left untouched. rzuUPDATE messages SET api_content = ? WHERE id = ? AND session_id = ? AND role = 'user' AND active = 1 AND content IS ?)r3boolr`rrBrg)rYr[rrfrs r)set_message_api_contentz,SessionMessagesMixin.set_message_api_contents^(Z5ZPS=TX^bcXc## @ { +VZAUAUV]A^ _a ar1c|r%t|tst|tr|dkry|jd|j |||fS)aRewrite the content of ONE known active user row. Used when a user turn was written at submit time (before the agent ran) and the turn prologue then rewrote the prompt it persists (@-file expansion, native image parts): the early row must show what the transcript will replay, not the raw keystrokes, and the turn must not append a second row for the same input.rz`UPDATE messages SET content = ? WHERE id = ? AND session_id = ? AND role = 'user' AND active = 1)r3rr`rrg)rYr[rrfs r)set_user_message_contentz-SessionMessagesMixin.set_user_message_contentsS Z5ZPS=TX^bcXc## n  ! !' *FJ ?A Ar1c |d}|ddk(r^td|j|d|d|j|dd\}}|"| |j|j d}|d||d|d|d |d fS) zZHistorical display identity, including normalized live content from user handoff carriers.rfrrrrq)rrfrrqrrr6r)r rjrxrgr)rYrHdedupe_contenthandoff live_views r)rz(SessionMessagesMixin._display_dedupe_keysY v;& !;4+?+?I+O #N 3$($A$A#FXBY$Z=\"] GY"y' ? AND (active = 1 OR compacted = 1) ORDER BY id LIMIT 1000rrrzHUPDATE messages SET display_order = ?, display_identity = ? WHERE id = ?)rVrWrr setdefaultrclose executemany) rZmissingrlast_idrZ batch_startupdatesrHrQorder missing_sqlrYr[s r)rz7SessionMessagesMixin._ensure_display_order.._doJsll; >GGIG)+HG||-  ) +& CC!$iG#55d6N6Ns6STH$//'BEO 44CHZD[8[x'AB C  k)  ^`gi'r1) _read_ctxsetrdrr~rr)rYr[rZcolumnsrrs`` @r)_ensure_display_orderz*SessionMessagesMixin._ensure_display_order<s ^^  <$44T:;G <!34? N  >>+ } 5 = 4e , :D'',--S < C#55d6N6Ns6STH-11(;G!$XD :I)g*=4=1 >  !/2 HNZ"-fg6v>ttDT^_e_fTghninTo MWX 9! <X X"1c,&7=0E(us{;EOOt||Em_U&&3E&:%;1>#,e,@.%/SY^%/00 8DZVvQYGY(Z&&LL,E" -" -. Y%/[&&LL,'C" -" -s[G!B F>0AF>4F/AF> F4 ( F>5 F9?F9F>G!/F>> GG!!G*ct|}|jdd|jdd|r|jddrd|d<|j|d|d<|jdrt |dgd |d |d<|jd |j |d |d <t |jD]6\}}|tvst|ttfs&|j|8|S) z``dict(row)`` with content/tool_calls/display_metadata decoded; *summary_flag* keeps ``_compressed_summary`` only as ``True``.rNrrrTrfr6z$Failed to deserialize tool_calls in z, falling back to []rq) rmrrjrr*rxr9r5_MESSAGE_SCHEMA_KEYSr3r_ bytearray)rYrH warn_context summary_flagrrr-s r)_row_to_message_dictz)SessionMessagesMixin._row_to_message_dicts3i "D) & CGG$91=)-C% &--c)n=I 77< (L!2)Ml^[o'p!rC  77% & 2&*&C&CCHZD[&\C" # syy{+ JC..:eeYEW3X   r1c|rdS|rtSdS)zcAudit: every row; display: active plus compaction-archived (never Undo/Rewind rows); default: live.rOrA)r)include_inactiveinclude_compacteds r)_active_clausez#SessionMessagesMixin._active_clauses&roFW,Bo]nor1c||s|r td| |r td|j||}|r:|s8|j|r'|rdnd} d| d} |j| ||dn|||g} nx|r|j |||||} n_d ||d nd d |rdnd} ||gn||g} ||r| d z } | j |dn||g|j| | } |r| j | D cgc]} |j| ddc} Scc} w)a0Load messages in insertion order (id, never timestamp: clocks regress). ``include_inactive``: rewind rows too; ``include_compacted``: compaction-archived display history (not rewind rows). ``latest`` pages back from the newest but returns chronological order; ``after_id``: keyset paging.z2after_id is incompatible with latest/offset pagingzYafter_id is incompatible with include_compacted (deduped display reads use offset paging)DESCASCzWITH page AS ( SELECT display_order FROM messages WHERE session_id = ? AND (active = 1 OR compacted = 1) GROUP BY display_order ORDER BY display_order a3 LIMIT ? OFFSET ? ) SELECT chosen.* FROM page JOIN messages AS chosen ON chosen.id = ( SELECT candidate.id FROM messages AS candidate WHERE candidate.session_id = ? AND candidate.display_order = page.display_order AND (candidate.active = 1 OR candidate.compacted = 1) ORDER BY candidate.active DESC, candidate.id DESC LIMIT 1 ) ORDER BY page.display_order ASCr)rrrrrz AND id > ?rOz ORDER BY id z LIMIT ? OFFSET ? get_messagesTrr)rcrr _read_allrextendreverser)rYr[rrrrrafter_idr directionrrZrrHs r)rz!SessionMessagesMixin.get_messagesss  VvQR R  $5xy y++,<>OP %5$:T:TU_:`"(eICDM+ N03C>># %-BUTZ\f'ghD ,,-uV\b-dDA$,$8=bAY_vejNkmC+3+;J<*hAWF F** U]rvFG>>#v.D jnocf))#NY])^ooos+D c |Dcgc]}|s| c}fdtdtdD}|Dcgc]1}|jdt|d|D] }|d|dd3c}}Scc}wcc}}w)ziTool results containing ``/pull/``: a deliberately loose scan, oldest-first so the caller takes the last.c3.K|] }||dzyw)rNr)rQrrs r)rSz.sPu#eeck*PsrrzRSELECT session_id, content FROM messages WHERE session_id IN (zi) AND role = 'tool' AND content LIKE '%/pull/%' ORDER BY id ASCr)r[rf)rur:rr)rY session_idsschunksrrHrs @r)find_pr_url_messagesz)SessionMessagesMixin.find_pr_url_messagess%+Qq+PaS38OPGMeY]YgYg**7*>)?@$' ZRUs1v#a&99 ,sA3A36A8c t|d}|j5}|jd||fjsgdddcdddS|jd|||dzfj }|jd|||fj }dddgt Dcgc]}|j |dd }}|tdt|dz t|dS#1swY]xYwcc}w) zUp to *window* messages either side of an anchor id (ascending). ``messages_before``/``_after`` count strictly around the anchor (fewer than *window* = session boundary). Empty for a foreign anchor.rz>SELECT 1 FROM messages WHERE id = ? AND session_id = ? LIMIT 1)windowmessages_beforemessages_afterNzPSELECT * FROM messages WHERE session_id = ? AND id <= ? ORDER BY id DESC LIMIT ?rzNSELECT * FROM messages WHERE session_id = ? AND id > ? ORDER BY id ASC LIMIT ?get_messages_aroundFr)rrrVrWrreversedrr:) rYr[around_message_idrrZ before_rows after_rowsr window_msgss r)rz(SessionMessagesMixin.get_messages_arounds6VQ ^^  D<< `!2J ?AAIL"$aP D D,,b. ;==EXZ `.799A  D!F(;"7 E* EG00AVej0kG G&#a[AQTUAU:Vjmnxjyzz D DGs(C4A C42D4C=cZ|s|S |j|xs|}|j5}|}|h}d}tdD]} |j d|fj |}|j dt dddt dddt dd d td td |fj }| |d r|d |vrn|d }|j|||n|cdddS#t$rYwxYw#t$r|cYccdddSwxYw#1swYyxYw)uRedirect a resume target to the descendant holding the messages: follow the compression chain to the live tip (lineage-aware, so delegate/branch children never hijack it), then walk ``parent_session_id`` forward to the DEEPEST node with messages (a continuation may hold newer turns), skipping branch/delegate/reset/tool children. Unchanged when nothing has messages; depth cap 32. Context compression ends the current session and forks a new child session (linked via ``parent_session_id``). The flush cursor is reset, so the child is where new messages actually land — the parent ends up with ``message_count = 0`` rows unless messages had already been flushed to it before compression. See #15000. N z3SELECT 1 FROM messages WHERE session_id = ? LIMIT 1zISELECT id FROM sessions AS child WHERE child.parent_session_id = ? AND zchild.model_configz$._branched_fromz IS NULL AND z$._delegate_fromz $._reset_fromz IS NULL AND NOT childza AND COALESCE(child.source, '') != 'tool' ORDER BY child.started_at DESC, child.id DESC LIMIT 1r) get_compression_tip ExceptionrrurVrWrrradd)rYr[rZrrbestrw child_rows r)resolve_resume_session_idz.SessionMessagesMixin.resolve_resume_session_ids  11*=KJ^^  < G9DD2Y " &||$Y\c[efooq}& $ !!23GI[!\ ]^!!23GI[!\ ]^!!23G!Y Z[%%@ @r1c|j|r|j|n|g|j||d}|r|j|}|j ||||||S)aLoad messages in OpenAI format. ``include_compacted`` (deduped display history) is for DISPLAY reads only: the model-fed restore must not regrow what compaction summarized away. ``repair_alternation`` repairs the loaded list for LIVE REPLAY callers (a durable ``user;user`` pair would re-trigger the per-request repair forever), preserving summary markers before repair so derivative context cannot merge with an original user turn; the stored transcript is never mutated.Frr[include_ancestorsrepair_alternationinclude_row_idsinclude_summary_markers)rr rr_rows_to_conversation)rYr[rrrrrrZs r)get_messages_as_conversationz1SessionMessagesMixin.get_messages_as_conversation,s,,4ED $ $Z 0J<    02C DV[-] 33D9D))$:Yj1?$6*8 8r1cV |j|d}|j|jd|}||j|nd d} $t fdt |Dd}||df}||j ||}|d|fS|\}} | r|j || |fS)aaAncestor-lineage dedupe of one decoded user *msg* -> ``(skip, exact_clone_key)``. Rotation column-clones the concurrent tail into the child, so copies need not be adjacent: the exact ``(timestamp, canonical content)`` clone index is checked first, then the adjacent heuristic. A rotated child carrier wins over the ancestor copy.rrNc32K|]\}}|us |ywrr)rQirprevious_exacts r)rSz=SessionMessagesMixin._dedupe_replayed_user..Hs"lIPY]kPk1"ls TF) _canonical_replayed_user_content_exact_replayed_user_clone_keyrr enumerate%_find_duplicate_replayed_user_messager) rYrrexact_user_clonescanonical_contentexact_clone_key duplicateprevious_indexduplicate_indexprefer_currentrs @r)_dedupe_replayed_userz*SessionMessagesMixin._dedupe_replayed_user>s !AA#FqI==cggk>RTefCRC^*..?dh  %!"l89L"lnrsN)+T2  BB8SQI  /) )*3'  LL )!!?22r1)rrcHddlm}m}g} i} |D]T|jd|j d} ddd| t di} |r dd| d<| j fd d Dd r|jd x} | | d <|r d rd| d <| j fd dDdrtdgd| d<drd| d<drd| d<ddk(rA| j fddDdd| d<| j fddD|r!|j| | | \}}|r=|| | |<| j| W||| } |r*| r(ddl m }|d| }|rtjd||| S)aDecode fetched rows (ordered by id, pre-filtered) into OpenAI format, stable key order. Every dict is stamped ``_DB_PERSISTED_MARKER_KEY`` (born durable) so an identity-losing handoff never re-appends the transcript on flush. ``_row_id`` is opt-in (gateway reactions); reasoning restored on assistant rows only; ``api_content`` VERBATIM (no sanitize/strip) so replay keeps the provider prompt cache byte-stable.r) _strip_background_review_harness_strip_stale_tool_call_markersrrfTrNr(c38K|]}|s ||fywrrrQcolrHs r)rSz=SessionMessagesMixin._rows_to_conversation..js"^3UXY\U]SX^  )rrrqrc38K|]}|s ||fywrrrs r)rSz=SessionMessagesMixin._rows_to_conversation..os)y$'orsvowc#hyr)rrrrr6zKFailed to deserialize tool_calls in conversation replay, falling back to []rrrrc38K|]}|s ||fywrrrs r)rSz=SessionMessagesMixin._rows_to_conversation..ys"asX[\_X`CS?ar)rrrc 3TK|]}|r|t|dd|df!yw)NzFailed to deserialize z, falling back to None)r*rs r)rSz=SessionMessagesMixin._rows_to_conversation..|sArhklohp(3s8T5KC5Pf3ghirs%(r)repair_message_sequenceuRepaired %d message-alternation violation(s) while restoring session %s — durable transcript kept them, see repair_message_sequence)rr r rLrj_DB_PERSISTED_MARKER_KEYrrxr*r ragent.agent_runtime_helpersrr%info)rYrZr[rrrrr r rrrfrdecodedrarrrepairedrHs @r)rz*SessionMessagesMixin._rows_to_conversationTs' bCE' !C//F T=Q=QRUV_R`=abG3v; 7L!:"&J6{k) a6Taa*+7/23F/GC+, rdrr!(,(B(B8SRc(d%o".9<%o6 OOC O' !T22RS[2\] ( K.tX>H 23;ZIr1c|j|j|td}|j|Dcgc]}|d|k(s |ds|c}|dddd}|j|j ||ddd}||fScc}w)u/``(model_history, display_history)`` for a resume from ONE SELECT; byte-identical to the separate reads. model: the tip's active rows, alternation-repaired, summary marker kept for pre-compress checkpointing. display: the full lineage (``/branch`` stands alone), compaction-archived rows deduped. The display projection also includes rows preserved by IN-PLACE compaction (``active=0, compacted=1``), deduped by :meth:`_dedupe_display_generations`. Without them a compacted conversation resumes showing only its summary plus the carried-forward tail — the user's own turns read as deleted even though every row is still on disk, and the REST transcript read (which has always included them) disagreed with this one about the same session (#92080). Trr[r^Frr[rrr)rr rrr)rYr[rZr model_historydisplay_historys r)get_resume_conversationsz-SessionMessagesMixin.get_resume_conversationss,,  $ $Z 02HZ^-`22 N1,: =!H+Q N[e#dlp3r 44  , ,T 2z"ud5To-- Os BB BcL|j|r|gS|j|S)zSession ids a display resume materializes: the compression lineage, or the session alone for an explicit ``/branch`` copy. Shared with the resume guard so it counts exactly what a resume loads.)_is_explicit_branch_session_session_lineage_root_to_tiprTs r)r z(SessionMessagesMixin._resume_lineage_idss+ $?? K |~QUQrQrs}Q~~r1c6|r|gdfS|j|dfS)zW``tip_only``: the tip's ACTIVE rows (model restore); else the full-lineage DISPLAY set.z active = 1z(active = 1 OR compacted = 1))r )rYr[tip_onlys r)_resume_count_scopez(SessionMessagesMixin._resume_count_scopes) <- -'' 35TTTr1)r$c|j||\}}t|jdt|d|t |dS)zHCount the rows a resume would materialize (see ``_resume_count_scope``).z3SELECT COUNT(*) FROM messages WHERE session_id IN () AND r)r%r`rrr)rYr[r$rrs r)get_resume_message_countz-SessionMessagesMixin.get_resume_message_counts[%)%=%=j(%S" ]4>>A-P[B\A]]cdqcr s +  !#$ $r1c ddlm}m}||}|dkr td|dk(ry|j ||\}}t |j dt|d|dg||dzd}||kDr||||rd d |S) a5Resume row count, or raise ``SessionResumeTooLargeError``. ``max_messages=None`` reads config; 0 disables the guard without counting. ``tip_only`` bounds only the tip's active rows for callers that never materialize the lineage: a heavily compressed conversation is a success, not a rejection.r)SessionResumeTooLargeErrorresolved_max_resume_messagesz!max_messages must be non-negativezBSELECT COUNT(*) FROM (SELECT 1 FROM messages WHERE session_id IN (r'z LIMIT ?)rzin its tip segmentzacross its lineage)scope)rr*r+rcr%r`rr) rYr[ max_messagesr$r*r+rr message_counts r)assert_resume_safez'SessionMessagesMixin.assert_resume_safes Z  79L ! @A A 1 %)%=%=j(%S" ]DNN,;;H;U:VW /,,-M{-ML1 C! C!C'!C'c@|j|}|r |dr|dS|S)zROOT id of the lineage: the stable conversation id across compression segments and delegate subagents (Nous Portal usage tagging). Unchanged when there is no recorded parent.r)r")rYr[chains r)get_conversation_rootz*SessionMessagesMixin.get_conversation_roots+11*= U1XuQx=:=r1c|jddk7ryt|\}}|duxr|du}|r|jd|fS|jd|fS)z=Return canonical live content and whether *msg* is composite.rr)NFNrf)rr )rrr is_composites r)rz5SessionMessagesMixin._canonical_replayed_user_contentse 776?f $7<d*Dy/D +7y}}Y'Q]]]SWWY=OQ]]]r1cx||ddgfvry |tj|dddfS#ttf$rYywxYw)z8Return a hashable key for a column-exact rotation clone.NrOFT),:) ensure_ascii sort_keys separators)r!rbr$rc)rrfs r)rz3SessionMessagesMixin._exact_replayed_user_clone_keysR  D"b> 9 djjuPTakll l:&  s '99c|jddk7rytj}||\}}|ddgfvrytt |dz ddD]}||}|jddk(r,||\}}||k(s-|s|st |t sB||fcS|jddk(s]|jds|jd syy) a;Adjacent replay duplicate ``(index, prefer_current)`` or None. Rotation may persist the current ask in the parent and again inside a composite child carrier: carriers compare by canonical live payload, ordinary users by exact string. The child carrier wins (it owns the durable row id and scaffold).rrNrOrrrrfr6)rrLrrur:r3r4) rr canonicalrfr indexprev prev_contentprev_is_composites r)rz:SessionMessagesMixin._find_duplicate_replayed_user_messages 776?f $(II "+C. tRn $3x=1,b"5 EE?Dxx6)2;D// /7*BSWabiknWo .00&![0dhhy6ITXXVbMc r1cn|jt|fDcgc]}t|dc}Scc}w)zbOrdered physical active ids for rewind CAS checks (includes legacy harness rows projections omit).r)r_ACTIVE_IDS_SQLr`r"s r)get_active_message_idsz+SessionMessagesMixin.get_active_message_idss+'+~~o }'UVCF VVVs2c~|jd|fj}t|td|DfS)z?Return active message/tool-call counts inside the caller's txn.zCSELECT tool_calls FROM messages WHERE session_id = ? AND active = 1c3<K|]}t|ddyw)rrr<Nr=r>s r)rSzASessionMessagesMixin._active_transcript_counts..sPCoc!fQ??Pr?)rVrr:r)rZr[rZs r)_active_transcript_countsz.SessionMessagesMixin._active_transcript_countss<||adncpqzz|4y#P4PPPPr1c |jds tdti||j|jd|j |jdd\}}| td|jd}t |t rt|j}|||k7r td|r | td |r|SdS) zValidate an active rewind target; return its handoff scaffold (or None). ``ValueError``: inactive / non-user-originated / missing composite carrier; ``RuntimeError``: live payload changed.r^zrewind target is not activerfrq)rfrqNz+rewind target is not a user-originated turnz2rewind target changed before it could be persistedz@preserve_compaction_handoff requires an active composite carrier) rrcr rjrxr3r4r rP RuntimeError)rY target_rowexpected_target_contentpreserve_compaction_handoffrr live_contents r)_split_rewind_targetz)SessionMessagesMixin._split_rewind_targets~~h':; ;79c9c%)%9%9*..:S%T $ = =jnnM_>` a9cd  JK K }}Y/ lC (+L9??AL " .= *target_message_id*, target included (the caller pre-fills it as the next prompt). Returns ``{"rewound_count", "target_message", "new_head_id"}``, plus ``replacement_message_id`` with ``preserve_compaction_handoff`` (archives a composite summary carrier, inserts its hidden handoff scaffold as the new head). ``ValueError``: target missing or not ``user``. ``expected_active_ids`` / ``expected_target_content`` pin the active set and canonical live payload in-txn before any mutation (presentation-only metadata changes don't invalidate a rewind). A live turn lease refuses; expired/dead holders are reclaimed. ``rewind_count`` always increments.c j|ddd O|jtfj}|Dcgc]}t |dc} k7r t d|jdfj }|tddt|}|jdd k7r!td |jdd d dx}} s j| }|jd fjDcgc]}|d }}|r|jdt|d ||?j||gt |jdj d}|jdfj|\}} |jtd|| f|jdfj d} ||| |fScc}wcc}w)NTr9rz>active transcript changed before the rewind could be persistedz6SELECT * FROM messages WHERE id = ? AND session_id = ?zmessage z not found in session rrz1rewind target must be a 'user' message (got role=z, id=rzGSELECT id FROM messages WHERE session_id = ? AND id >= ? AND active = 1z,UPDATE messages SET active = 0 WHERE id IN (zSELECT last_insert_rowid()zMUPDATE sessions SET rewind_count = COALESCE(rewind_count, 0) + 1 WHERE id = ?rz@SELECT MAX(id) FROM messages WHERE session_id = ? AND active = 1)rrVrFrr`rLrWrcrmrrQrrrJrD)rZ active_rowsrrHrMreplacement_message_id replacementrr.tool_call_counthead_idrRrNrOrYr[target_message_ids r)rz3SessionMessagesMixin.rewind_to_message.._do8sZ  / /j$fj 0 l"."ll?ZMJSSU '23!C!I37JJ&'ghh,,HK\^hJikksksku { 8,=+>>TU_T`!abbcJ~~f%/ G W]H^Gaafgxfyyz{}}37 7 "[*.E.Q"77 D[]xy !%.w.8:K-L"NNVhjZA1Q4ZCZ KMZ]L^K__`acfg&))$ [MJ),T\\:V-W-`-`-bcd-e)f& LL_blan p-1-K-KDR\-] *M? LL-.m<}o_i>j kllRU_TacckckcmnoqGsG-CC C54Zs G; Hrf) rewound_counttarget_message new_head_idrV)rrjrr:) rYr[rZrOrRrNrrMrewoundr]rVs `````` r)rewind_to_messagez&SessionMessagesMixin.rewind_to_message.s D D@DHCVCVWZC[@ G[*@ $ 4 4Z^^I5N O 9!$W\goId,.DEo ojlo or1cVd|rdndz}|j||r |fdSddS)z2Count messages, optionally for a specific session.zSELECT COUNT(*) FROM messagesrrOrrr!)rYr[rs r)r.z"SessionMessagesMixin.message_count^s9-J1H\^_~~cJJ=GJJBGJJr1c.|jd||fduS)zNProbe the accepted-input marker without allocating message bodies or archives.zSELECT 1 FROM messages WHERE session_id = ? AND role = 'user' AND observed = 0 AND (active = 1 OR compacted = 1) AND CASE WHEN json_valid(display_metadata) THEN json_extract(display_metadata, '$.gateway_input_owner') END = ? LIMIT 1Nr!)rYr[owners r)has_gateway_input_ownerz,SessionMessagesMixin.has_gateway_input_ownercs-~~ [  ! )- - -r1c.|jd||fduS)aTrue when *platform_message_id* exists (partial-index probe; the gateway's transient-failure dedupe). Uses the idx_messages_platform_msg_id partial index for efficient lookup. Used by the gateway's transient-failure dedupe guard (#47237) to skip re-persisting a user message that was already saved on a prior retry of the same inbound platform message. zOSELECT 1 FROM messages WHERE session_id = ? AND platform_message_id = ? LIMIT 1Nr!)rYr[rs r)has_platform_message_idz,SessionMessagesMixin.has_platform_message_idls)~~ ] , -/6:; ;r1) include_resetc|jddk(ry|jd}t|tr tj|}t|t sy|jd|jdf}|r||jdfz }|jd }|r||vStd |DS#tj $rYywxYw) agTrue when *session* is a branch, delegate, or tool child of its parent (``include_reset``: also a reset fork). Markers only count when they point at ``parent_session_id``: compression copies ``model_config`` onto the continuation, so presence-only matching would misclassify it (same binding as ``_NON_CONTINUATION_CHILD_FILTER_SQL``).rTtoolT model_configF_branched_from_delegate_from _reset_fromparent_session_idc3$K|]}|du ywrr)rQms r)rSzCSessionMessagesMixin._is_explicit_fork_child_row..s9YA!4-9Ys)rr3r4r!r"r#rmr1)rYsessionrfcfgmarkers parent_ids r)_is_explicit_fork_child_rowz0SessionMessagesMixin._is_explicit_fork_child_rowws ;;x F *kk.) c3  jjo#t$77+,cgg6F.GH   .0 0GKK 34 '0yG#Yc9YQX9Y6YY''  sCCCc`|j|}t|xr|j|S)zRead-only :meth:`_is_explicit_fork_child_row`; a missing row is not a fork (prompt_cache_scope keeps a declared conversation key from crossing the fork boundary).) get_sessionrrt)rYr[rps r)is_explicit_fork_childz+SessionMessagesMixin.is_explicit_fork_childs/"":.GI @ @ IJJr1ct|r|sy|jd||f}||dt|dnd}|dkDr|SdS)aConversation boundaries (``_RESET_END_REASONS`` ends) this peer has crossed, or ``None`` if never reset. The peer is ``(session_key, source)``, never the key alone (an API caller may legally reuse a Telegram row's key). Read from ``conversation_generations`` (advanced in each boundary's txn), not an aggregate over session rows: deletes/prunes would re-emit a retired pair. Rows are never GC'd (dropping one re-issues generation 1: the ABA this prevents). Wall-clock-free, so a backwards NTP correction cannot reorder it. DBs upgraded mid-conversation take their first generation from the next boundary written (a pre-upgrade reset shares its predecessor's scope once: costs a warm prompt-cache bucket, never crosses an identity).NzTSELECT generation FROM conversation_generations WHERE source = ? AND session_key = ? generationrr)rYrUrTrHrys r)latest_conversation_boundaryz1SessionMessagesMixin.latest_conversation_boundarysX&nn b [ !#033|CTC`S\*+fg '!^z55r1c2fd}|j|y)z9Delete all messages for a session and reset its counters.cZ|jdf|jtfy)Nr@)rV_RESET_COUNTERS_SQL)rZr[s r)rz0SessionMessagesMixin.clear_messages.._dos% LLDzm T LL,zm c}Scc}w)NziSELECT id, content FROM messages WHERE role = 'assistant' AND tool_calls IS NOT NULL AND tool_calls != ''rfr)rVrr3r4 fullmatchrP)rZcursorrHrs r)_find_affectedzJSessionMessagesMixin.purge_stale_tool_call_markers.._find_affectedsr\\#[\F)/):u#c)nc27Q7[7[\_`i\j\p\p\r7sIu uusAA)Nc$t|||dS)N)r rows_affectedrow_ids backup_path)r:)affectedrrs r)_resultzCSessionMessagesMixin.purge_stale_tool_call_markers.._results&XS[lwx xr1z %Y%m%d_%H%M%Sz.pre-clean-markers-backup-z VACUUM INTO ?z3Backed up state.db to %s before clean-markers writecZ|}|r|jdt|d||S)Nz.UPDATE messages SET content = '' WHERE id IN (r)rVr)rZrrs r)rz?SessionMessagesMixin.purge_stale_tool_call_markers.._dos6 &C Mm\_N`MaabcehiJr1zIPermanently cleared %d stale tool-call marker row(s) in state.db (#78148))return List[int]r)rrrdatetimerIstrftimer4db_path with_namename_lock_connrVr%rrr:) rYrrrrZ affected_idsrrstamprrrs ` @@r)purge_stale_tool_call_markersz2SessionMessagesMixin.purge_stale_tool_call_markerss- < u  y ^^  0)$/L 0 ,<( (%)  %%))+44_EEdll44 8I8I7JJdejdk5lmnK D ""?[NC D KKM{ [  **3/  KKcehiuev w|[11) 0 0 D Ds D9=E9EE)r[r4r\r4rNone)rfrrr)rqrr Optional[str])r'rrOptional[Dict[str, Any]])r-rrr)Nr@FFF)r[r4rrrrrrGrrrrrrrr)r[r4rr4rDict[str, Any]r6rrrGrrrr) r[r4rr`r6r`rrrr)NNNNNNNNNNNNFNFNNNNNNr)2r[r4rr4rfr4rr4r6rrr4rr`rr4rr4rr4rrrrrrrr4rrrrrrrrrrrrrqrrrrrrrGrr`)r[r4rfr4rrrr`)NNNr)r[r4rList[Dict[str, Any]]rrrrr Optional[int]rrGrr`) r[r4rr4rfr4rr4rqrrr)rvrrr) r[r4rr`rrrr4rzOptional[List[Dict[str, Any]]])r[r4rr`rr)r[r4rr`rzTuple[str, tuple])r[r4rr4rr) r[r4rr4rr`rrrr)r[r4rr)r[r4rr`rr)r[r4rrrtuple[int, int])r[r4r3r`rr`)FFF) r[r4rrrHrrIrrrrr) rr4rfrrrr6rrr)rFr9rrrr`)rr4rfrrr)r[r4rr)r[r4rr`)rr4rzTuple[List[int], int])rerr[rrr)NNNr)r[r4rvrrxrrzrrwrryr`rr`)r List[str])r[r4rfrrr4rr`) r[r4rr`rfrrr4rr`)r[r4rr`rfrrr`)rTuple[Any, ...])rrrr_) r[r4rr4rrrr`rrrz List[Any])rr4rrrr)rrrrrr4)FFNrFN)r[r4rrrrrrrr`rrrrrr)rrrr))r[r4rr`rr`rr)r[r4rr4)rrrr4rr)FFFFF)r[r4rrrrrrrrrrrr)rzTuple[bool, Any]) r[r4rrrrrrrrrr)r[r4rz1Tuple[List[Dict[str, Any]], List[Dict[str, Any]]])r[r4rr)r[r4r$rrzTuple[List[str], str])r[r4r$rrr`r)r[r4r-rr$rrr`)r[r4rr)rrrzTuple[Any, bool])rrrfrrzOptional[Tuple[Any, str]])rrrrrzOptional[Tuple[int, bool]])r[r4rr)r[r4rr)rMrrNrrOr) r[r4rZr`rOrrRzOptional[List[int]]rNrrr)r[r4rbr4rr)r[r4rr4rr)rprrfrrr)rUr4rTr4rr)r[r4rr)rrrrrr)Oro __module__ __qualname____doc__r] classmethodrgrj staticmethodrrrxrzrrrrrrrrrrrrrr#r%rr)rJrMrCrLrUrWr[rhr{rdrrrrrrrrrrrrrrrrr rrr r%r(r/r3r6rrrrGrJrQr_r.rcrertrwrzr~rrr1r)rLrLrs2X F 6 6,," __RWW\057<(7C^8 r rimgkcg[`fjIMdhQVVV%(V36VJMVbeVV/2VJMVadVV;>V^aV! V?B VUY V * V HL V ad V # V :G V3VUbV)VJOV\_V: VFimMQ(-VV)=VXeV(V=JV!&V36V4_c(?B(C[(gk(&d ,2(%((6T(4edGM (DEKZ[37 ',0 '<(4<(HU<("<(7:<(CF<(|+ Y.aa'*a58aGJa a6 A JSSM +.Z&-%)&-.7&-P0ppglRW/3-p)-p:=-pKO-p,-p8L-p^ {&'C</3<7;<H\<|., U MR$jo(N"> ^^3AF`0WQQ @&qvEI9=.o/B.o36.oBP.o`K - ;]bZ*K 6"!@EUY 2r1rL)r'rr(rr&r4rr)r-rr.rGrrG)r6rrr)r6rrr`)r)r'rr=r`rr`)r-rrr)rIrGrr)Ar __future__rrr!loggingrertypingrrrrragent.context_compressorr rr r r agent.memory_managerr agent.message_sanitizationrhermes_cli.timefmtrhermes_state_commonrrrrrrrr getLoggerr%r frozensetfindallsplitrrXrrrr rFrBrDr}rrnr0r2objectrlr*r0r7r;r?rBrJrLrr1r)rsHt#  33  2;+???   > *c!BJJv*00a8;AA#qI!LM()eb=fi[PP_Oi"ZW 8 ,aR L f X2X2r1