߲j|dZddlmZddlZddlZddlZddlZddlmZddl m Z m Z m Z m Z mZe rddlmZej"dZGdd eZdd Zd ZGd d Zy)zSessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute following, FTS corruption recovery, rewrite/rewind/load. Mixin split out of ``gateway/session.py``; bound onto ``SessionStore`` via the MRO.) annotationsN)extract_api_content_sidecar) TYPE_CHECKINGAnyDictListOptional) SessionEntryzgateway.sessionc$eZdZdZdfd ZxZS)TranscriptReadErrorz4Raised when persisted history cannot be read safely.c8||_t| d|y)Nz#transcript read failed for session ) session_idsuper__init__)selfr __class__s ?/home/ubuntu/.hermes/hermes-agent/gateway/session_transcript.pyrzTranscriptReadError.__init__s$ >zlKLrstrreturnNone)__name__ __module__ __qualname____doc__r __classcell__)rs@rr r s>MMrr c@ ddlm}|||S#t$rYywxYw)zSpool one evicted/undeliverable message to disk (same machinery as the shutdown flush, so it is replayed after DB recovery); path or None.r) spool_dropped_transcript_messageN)gateway.shutdown_flushr Exception)rmessagers r_spool_droppedr#s)K/ GDD s  ) reasoningreasoning_contentreasoning_detailscodex_reasoning_itemscodex_message_itemscZeZdZdZdZdZdZd dZ d!dZ d"dZ dZ d#d$d Z d%d Z d&d Z d'd Zd%dZ d(dZd)dZd*dZd+dZd,dZd*dZed-dZd.dZd,dZd/dZd0dZ d1 d2dZd3dZd4dZ d5d d d6dZy)7SessionTranscriptMixinzSessionStore transcript I/O: SQLite append with a per-session retry queue, compression-reroute following, FTS corruption recovery, rewrite/rewind/load.i,c|r|j|nd}||S |j|xs|S#t$rtj d|d|cYSwxYw)zLatest compression continuation for *session_id* (heals a mapping left pointing at a compressed parent by a restart or failed send).Nz,Compression-tip lookup failed for session %sTexc_info)_db_for_session_idget_compression_tipr!loggerdebugrrdbs r_compression_tip_for_session_idz6SessionTranscriptMixin._compression_tip_for_session_id<sg5?T $ $Z 0D :  ))*5C C  LLG^bL c  s0#AAc|r|sy|j|k7s||k(rytjd|j|||_y)zDRewrite *entry* to the compression continuation if stale. Lock held.Fz8SessionStore healed compressed session mapping: %s -> %sT)rr2info)rentryoriginal_session_idcanonical_session_ids r_heal_compression_tip_lockedz3SessionTranscriptMixin._heal_compression_tip_lockedHsO#*>   2 26JNa6a FHXHX  "0rc |r|r|sy|j5|j|}| dddy|j|k(r |cdddS|j|||s dddy|j |cdddS#1swYyxYw)u'CAS-advance one route along an already-verified compression lineage. Unlike ``switch_session`` this never ends/reopens SQLite rows (the compression transaction owns that). ``None`` means the route moved after the caller's snapshot (e.g. /new) — caller must fail closed.N)_lock _entry_lockedrr<_save)r session_keyexpected_session_idtarget_session_idr9s radvance_compression_sessionz2SessionTranscriptMixin.advance_compression_sessionVs"5=N ZZ &&{3E}  #44   44Upending fallback failed for replaced state.db transcript on %sTr.r)divert_session_transcript_jsonlz:JSONL divert failed for replaced state.db transcript on %s)r2errortyper_transcript_retry_locklistrYgetr^_transcript_append_failuresr#r_ hermes_staterdr!)rrqueue_session_idexc remainingdroppedrds r$_divert_transcript_after_db_replacedz;SessionTranscriptMixin._divert_transcript_after_db_replaceds    LMW I   % ( ( CT44889I2NOI  # # ' '(8$ ?  , , 0 0T B C! #Gj'2:TV`!# #   D +J B C C  NNLj   sAC C'C$'!D  D c|j|}||j|nd}|r4||k7r/|j|}||jd t |Sy)a,Transitive compression tip of *session_id* if it is a different, still-live row, else "" (a depth-1 lookup misses multi-hop lineages). Uses the PARENT's proven owner handle: the child's id is unpublished until its write succeeds, so a by-id lookup would hit the ambient store.Nended_at)r0r1 get_sessionrir)rrowner_dbtiptip_rows r_live_compression_childz.SessionTranscriptMixin._live_compression_childsf **:6:B:Nh**:6TX 3*$**3/G"w{{:'>'F3xrc|r|d|ur|jd|jj|g}|r!|j|||j|<n|r|}|jj|d|jj|d}|r3t ||jj|d|j|<||j |<|S)abMove the retry queue + failure counter from parent to child and record the reroute (retry lock held); returns the child's pending list. Older parent backlog must precede messages already queued directly on the child; routing is published only AFTER the queue moved (caller), so new child writes cannot bypass older parent backlog.rN)r^rYriextendrjmaxrQ)rrrlchild_idr`msgexisting_child_pendingprevious_failuress r"_migrate_transcript_queue_to_childz9SessionTranscriptMixin._migrate_transcript_queue_to_childs wqzS( KKN!%!8!8!_entriesvaluesrr@getattrr^)rrr|r9s r_publish_transcript_reroutez2SessionTranscriptMixin._publish_transcript_reroutes| ZZ --/ 0##z1'/E$ 0 JJL   -t 4 :??$O   s-A>A>>Bc rj5j|ddddd fd } j jj5|}|sddddryL#1swYaxYw#1swYxYw#t$r}ddlm}m}ddlm }t|||frj|Yd}~yt||r j} j} | r| r| jdt| < j| j5j!| | dddn #1swYnxYwj#| sYd}~yd| Yd}~`#t$r } | }Yd} ~ nLd} ~ wwxYwj5|dddn #1swYnxYwt$j'dYd}~yj)|rjj+rZ jj5|dddn #1swYnxYwYd}~#t$r } | }Yd} ~ nd} ~ wwxYwj5j,j/ddz} | j,<dddn #1swYnxYw j0k\r6j3}t$j'd | t5||n"t$j7d | t5|Yd}~yd}~wwxYw) a>Append a message to a session's transcript (SQLite), draining the per-session retry queue. Args: skip_db: When True, skip the SQLite write. Used when the agent already persisted messages to SQLite via its own _flush_messages_to_session_db(), preventing the duplicate-write bug (#860). rNcrdurjds9jjdjjdyy)zCPop the acknowledged head (retry lock held). True if queue drained.rNTF)r^rYrj)r}r`rlrrsr _ack_headzJSessionTranscriptMixin._append_to_transcript_serialized.._ack_headsR71:, A''++,A.L11L6c|j5|jj|g}t|}dddd}D]}t ||n|dz }|r_|j dt j||j5d|=|s|jj|dddd|S|S#1swYxYw#1swY|SxYw)auMove a stalled session's in-memory backlog (oldest first) to the on-disk pending spool so a crash/restart during the outage no longer loses it (#114266): ``recover_pending_to_db`` replays it at boot, ``_drain_spooled_drops`` before the next live write. Stops at the first spool failure so order holds; whatever stays in memory remains under the cap.NrrrX) rgrYrirhr#rGrSrTr^)rrrlr`backlogrr"s rrz-SessionTranscriptMixin._spool_stalled_backlogCs  ( ( $--112BBGG7mG $ Gj'2: qLG   JJ/ 5 9 9* E,, HHWH%++//0@$G Hw $ $ Hs(B< $C<CCcntdd}|r|vry ddlm}j5t j j }ddd|fd\}}|s|jyy#1swY0xYw#t$r!}tjd|Yd}~yd}~wwxYw)zReplay cap-dropped spooled transcript messages after DB recovery. Best-effort: replay failures keep the spool files for the next successful flush; nothing here may raise.rXNr)drain_transcript_spoolc(j|S)N)r)r"rrs rz=SessionTranscriptMixin._drain_spooled_drops..esD,K,KJX_,`r)db_known_failingz+Failed to drain transcript spool for %s: %s) rr rrgrrjridiscardr!r2r_)rrspooled_sessionsr known_failing _replayedrnrms`` rrz+SessionTranscriptMixin._drain_spooled_dropsXs#4)A4H:5E#E  [ E,, W $T%E%E%I%I*%U V  W#9`!.$ Iy ((4  W W [ NNH*VY Z Z [s.B %A>-B >BB B4B//B4c |j|}|td|d|jddk(}|jd||jdd|jd|jd|jd |jd d tDcic]}||r|j|ndc}|jd xs|jd t |jd|jdt ||jd|jddycc}w)z7Write one transcript row. Caller handles retry queuing.Nzno owning session store for z; deferring transcript writerole assistantunknowncontent tool_name tool_calls tool_call_id)rrrrrrplatform_message_id message_idobserved timestamp display_kinddisplay_metadata)rrr api_contentrr)r0 RuntimeErrorriappend_message_ASSISTANT_ONLY_KEYSrr)rrr"_db is_assistantks rrz1SessionTranscriptMixin._append_transcript_messagems1%%j1 ;.zl:VWY Y{{6*k9  !VY/KK *kk+.{{<0 ^4  EYYqqL'++a.d:Y ")-B!C!`w{{S_G`'++j12kk+.4G<!^4$[[);<% Zs D< cdt|jvryddl}ddlm}t ||j xr|j|S)uTrue only when the failure is provably scoped to the FTS index. A bare SQLITE_CORRUPT can mean structural B-tree damage; only errors naming ``messages_fts`` or carrying FTS provenance (``SessionDB._is_fts_write_corruption_error``) may authorize the one-shot rebuild-and-retry; everything else takes the retry path. A generic ``database disk image is malformed`` (bare SQLITE_CORRUPT) can mean structural damage to canonical B-trees, not just the FTS shadow tables — treating it as FTS-only here made the store rebuild the index and retry transcript writes against a structurally corrupt database (#97940). messages_ftsTrN) SessionDB)rlowersqlite3rkrr DatabaseError_is_fts_write_corruption_error)rmrrs rrz/SessionTranscriptMixin._is_fts_corruption_errorsD SX^^- -*#w445g):b:bcf:ggrctj}|j}|||z |jkry|j}| t |dsy||_d}t |dr|j }|rtjd|y |j}|rtjd||dkDS#t$r }tjd|Yd}~yd}~wwxYw) aGAttempt FTS5 ``rebuild``, at most once per ``_FTS_REBUILD_COOLDOWN_SECONDS`` window; True if any index was rebuilt. A permanent one-shot flag meant a single rebuild failure (e.g. a transient WAL split-brain guard hit) permanently disabled recovery for the life of the process, even though later corruption on the same store could be fixable. Retrying on a cooldown lets the store try again after the underlying condition (e.g. a foreign holder) has likely cleared, without hammering a database that is corrupt on every write. NF rebuild_fts_foreign_state_db_holderszSkipping Session DB FTS rebuild while foreign processes hold the database or WAL sidecars (%s); canonical transcript writes remain available.z!Session DB FTS rebuild failed: %sz;Rebuilt %d Session DB FTS index(es) after append corruptionr) time monotonic_fts_rebuild_last_attempt_at_FTS_REBUILD_COOLDOWN_SECONDSrhasattrrr2r_rr!)rnow last_attemptr5foreign_holdersrebuiltrms rrz(SessionTranscriptMixin._rebuild_fts_oncesnn88  #|);t?a?a(a XX :WR7,/) 22 3 :: D sB== C&C!!C&c|j5|jj|d|jj|ddddy#1swYyxYw)zHDrop queued pending messages so a rewrite/rewind doesn't re-insert them.N)rgrYr^rj)rrs r_clear_dirty_transcriptz.SessionTranscriptMixin._clear_dirty_transcriptsP  ( ( C  # # ' ' D 9  , , 0 0T B C C Cs 9AAc|j|}|sy |j||S#t$rtj ddYywxYw)aWhether a message with this platform_message_id is persisted (False without a DB). Thin wrapper over SessionDB.has_platform_message_id(). Returns False when no DB is available (in-memory sessions). Used by the gateway's transient-failure dedupe guard (#47237). Fz%has_platform_message_id lookup failedTr.)r0has_platform_message_idr!r2r3)rrrr5s rrz.SessionTranscriptMixin.has_platform_message_idsV  $ $Z 0 --j:MN N  LL@4L P s( A  A c|j|j|}|j|}|sy |j|S#t$rt j d|dYywxYw)uRole of the newest live conversation row on the route ``load_transcript`` reads (``None`` when empty, no DB, or the read fails — the boundary write would fail the same way).Nz$transcript tail lookup failed for %sTr.)r6rMr0latest_conversation_roler!r2r3r4s rtranscript_tail_rolez+SessionTranscriptMixin.transcript_tail_rolesr99$:O:OPZ:[\  $ $Z 0 ..z: :  LL?VZL [ sA!A+*A+c.|j|}|sy|j5 |j|||||j | dddy#t$r(}tj d|Yd}~dddyd}~wwxYw#1swYyxYw)u Replace a session's transcript (/retry, /compress). DESTRUCTIVE by default: ``active_only=False`` DELETEs every row incl. soft-archived compaction history (pass ``active_only=True`` for sessions that may carry archived rows). True when the write lands or there is no DB, False on failure — callers committing a destructive change on top (/compress repointing) must check it. ``reject_active_turn_lease`` is for user-initiated rewrites that do not own the cross-process turn lease.T) active_onlyreject_active_turn_leasez&Failed to rewrite transcript in DB: %sNF)r0rJreplace_messagesr!r2r3r)rrmessagesrrr5es rrewrite_transcriptz)SessionTranscriptMixin.rewrite_transcripts $ $Z 0  , , .  ##k-E$G  ( ( 4)     EqI%    !  s3B AB  B B6B BB  Bcz |j|}|j|}|j|xs|}t}|rZ||vrV|j ||j ||ry|j |}|r|j|s y|d}|r||vrVy#t$r}t||d}~wwxYw)aFind this accepted input on the canonical live continuation and its ancestors. Content and unrelated writers cannot establish ownership. Query only existence; compaction archives can contain many megabytes that replay never needs to load. Tparent_session_idFN) rMr0r1rSrThas_gateway_input_ownerrt_is_compression_child_rowr!r )rrownercurrentr5rVrowrs rhas_input_ownerz&SessionTranscriptMixin.has_input_owners  9++J7G((1B,,W5@G5DgT1!--gu=nnW-"">">s"C12gT1 9%j1q 8 9s$A*B -%B  B B:) B55B:c|j|sgS|j|}tjt5|j|}|j |xs|}ddd |j|j |dS#1swY,xYw#t$r*}tjd||dt||d}~wwxYw)uLoad all messages from a session's transcript (state.db is canonical). Reads follow the same routing writes use — the in-memory reroute map, then the durable compression tip — otherwise the transcript "vanishes" while every message sits under the child.NT)repair_alternationzVTranscript read failed for session %s; refusing to treat the conversation as empty: %sr.) r0rM contextlibsuppressr!r1get_messages_as_conversationr2rer )rrr5rs rload_transcriptz&SessionTranscriptMixin.load_transcripts&&z2I**:6   + J((4B// ;IzJ J 9**:6SStT5 5  J J 9 LL'T  ;&j1q 8  9s#'B.!BB C%%C  C)require_retryable_compositec|j|}|syddlm}|j5 |j |t |d ||}|j||j|j|jdcdddS#|$r(}t jd|Yd}~dddyd}~wt$rt$r(}t jd|Yd}~dddyd}~wwxYw#1swYyxYw) u%Back up ``n`` user turns via soft-delete (``active=0``), mirroring CLI ``/undo [N]``. Returns ``{"rewound_count", "turns_undone", "target_text"}`` or ``None`` (no DB / no rewindable turn / persistence failure); ``n`` clamps to the oldest user turn. ``require_retryable_composite`` is the gateway ``/retry`` guard: the selected turn must be a composite carrier whose live payload is losslessly replayable as text — that replay-policy ``ValueError`` propagates so /retry can explain why the carrier is unsafe.Nr)RewindTargetUnavailableErrorr)require_retryablerequire_compositezrewind_session: %sz!rewind_session: rewind failed: %s) rewound_count turns_undone target_text)r0hermes_state_rewindrrJrewind_user_turnr{r2r3 ValueErrorr!rrr live_text)rrnrr5routcomers rrewind_sessionz%SessionTranscriptMixin.rewind_session2s $ $Z 0D  , , . 6 --Q >Y&A.C  ( ( 4%,%:%:GL`L`#*#4#46 6 6 0  115 6 6   @!D 6 6  6 6sFC3 B  4C3 C0B3&C33C0C+C3+C00C33C<N)r Optional[str]rr)r9z'SessionEntry'r:rr;rrr)rArrBrrCrrzOptional[SessionEntry])F)rrr"Dict[str, Any]rNrrr)rrrr)rrr"rrrh)rrrlrrmr!rr) rrrlrr|rr`rhrrh)rrr|rrr)rrr"rrr)rrrlrrintr)rmr!rrr)rrrrrr)rrrr)FF) rrrList[Dict[str, Any]]rrrrrr)rrrrrr)rrrr)r)rrrrrrrzOptional[Dict[str, Any]]) rrrrr]rrr6r<rDrJrOrMrbrprxrrrLrrr staticmethodrrrrrrrrrrrrr*r*.sT #%(!783  # :G + 04 58MP (E^.14;D 4 14@CNR .PbH*[* <hh "HC  TY).  )= LP "& 37 D9.90)*6RW66"%6KO6 !6rr*)rrr"r)r __future__rrloggingrHragent.turn_contextrtypingrrrrr gateway.sessionr getLoggerr2rr r#rr*rrrr sf,# :;;,   , -M,M `6`6r