+ jHjea0t$Rt^RIHt^RIt^RIt^RIt^RIt^RIt^RI t ^RI t ^RI t ^RI t ^RI Ht^RIHtHtHt^RIHt]P*!]4tRtRtRt]P6!4t]!4tR ]R &R t R R lt!RRlt"RR/RRllt#RR/RRllt$RRlt%RR/RRllt&RRlt'RRlt(R R!lt)R"R#lt*R$R%lt+R&R'lt,R(R)lt-R*R+lt.R7R,R/R-R.lllt/R7R,R/R/R0lllt0R,R/R1R2llt1R3R4lt2R5R6lt3R#)8aFlush pending messages and agent transcripts to disk before shutdown to prevent data loss. When FTS5 corruption blocks ``INSERT INTO messages``, ``_pending_messages`` and the live ``agent._session_messages`` are the only surviving copies; shutdown ``.clear()`` would drop them. All hooks write atomic JSON payloads under ``/pending_messages/``: ``flush_pending_to_file`` / ``flush_overflow_to_file`` (queue head / FIFO tail, before clear), ``recover_pending_to_db`` (at startup, before resume turns; replays via ``SessionDB.append_message``, deletes each file on success), ``flush_agent_history_to_file`` (DB flush raised), ``spool_dropped_transcript_message`` / ``drain_transcript_spool``. ) annotationsN)Path)AnyDictOptional) make_user_dirtranscript_cap_dropz'shutdown-with-unpersisted-agent-historyz.badz set[Path]_REPLAYED_UNREMOVABLEc^RIHpV!4R, p^RIHpV!V4\VRRRR7\P R8Xd\P !VR4V#)zIReturn the pending-messages flush directory under the active HERMES_HOME.)get_hermes_homepending_messages)assert_named_profile_home_liveTi)parentsexist_okmodeposix)hermes_constantsr r rosnamechmod)r flush_dirr s ;/home/ubuntu/.hermes/hermes-agent/gateway/shutdown_flush.py_get_flush_dirr,sL0!$66I?"9-)TDuE ww' E" c$V^8dQhRRRRRR/#)rrpayloaddict[str, Any]return)formats"r __annotate__r!8s!d^rcH^RIHpVR\P!4P R2, pV!W1R\ R7\ PR8XdV\ P!V\ P4p\ P!V4\ P!V4V#V# \d!p\PRT4Rp?L@Rp?ii;i \ P!T4i;i \d"p\PRT4Rp?T#Rp?ii;i) zOAtomically write one private, uniquely named recovery payload; return its path.)atomic_json_writezpending-z.jsoni)rdefaultrz-Failed to fsync pending-message directory: %sN)utilsr#uuiduuid4hexstrrropenO_RDONLYfsyncOSErrorloggerdebugclose)rrr# final_path directory_fdexcs&& r_write_payloadr48s'x (8(8'9??JjsC ww' '779bkk:L '&& :  S LcRR S& O LLH# N N  OsB%C5<B,, C7C CCCC25 D!DD!c 0V^8dQhRRRRRRRRRRR R /#) rrrkindr) session_keyvaluerextrarboolr)r s"rr!r!Ns8  D  # c TW \` rc \V4pVfR#\VRV/VCRV/C4R# \d#p\P RYT4Rp?R#Rp?ii;i)zNSerialise and write one pending value; return True when a payload was written.NFr7dataTz%Failed to flush %s message for %s: %s)_serialise_valuer4 Exceptionr.r/)rr6r7r8r9 serialisedr3s&&&&, r _flush_valuer@Ns^%e,  y=+"["[PZ"[\   c SV rc &V'g^#\4\\P!44^rCp\VP 44F!wrVVfK V\ VRWVWR7, pK# V'd\ PRWBV4V#)zXSerialise non-empty ``_pending_messages`` slots (``MessageEvent`` or str); return count.rD)rAtsz/Flushed %d pending message(s) to %s (reason=%s))rrEtimelistitemsr@r.info)rDrArrGflushedr7r8s&$ rflush_pending_to_filerM[sq +-s499;/?7I"7==?3   |Iy+U[c cG4 Ew[ab Nrc$V^8dQhRRRRRR/#)roverflow_by_sessionrrAr)rrEr)r s"rr!r!hs"3`crc V'g^#\4\\P!44^rCp\VP 44FQwrVV'd V'gK\ \V44F"wrxVfK V\ VRWXVW7R7, pK$ KS V'd\PRWBV4V#)ajSerialise the FIFO overflow tails (``queued_events``) to disk; return events flushed. The adapter slot holds the queue head and ``SessionState.conversation.queued_events`` the tail; both must survive restart. Each event is its own payload in the slot-flush shape so ``recover_pending_to_db`` replays them unchanged; ``seq`` preserves arrival order per session. overflow)rArGseqz7Flushed %d queued overflow message(s) to %s (reason=%s)) rrErHrIrJ enumerater@r.rK) rOrArrGrLr7eventsrRr8s &$ rflush_overflow_to_filerUhs +-s499;/?7I#$7$=$=$?@ & #DL1JC < :{Z`+-882 A Mw  Nrc$V^8dQhRRRRRR/#)r session_idr)messagerrzOptional[Path]r)r s"rr!r!s"~R`rc\\4RVR\R\\P!44R\ \ 4RRVRV//4# \d"p\PRY4R p?R #R p?ii;i) aWSpool a cap-evicted transcript message; ``None`` on failure (callers degrade to drop+log). Uses the same on-disk pending spool as :func:`flush_pending_to_file` (one atomic JSON payload per message under ``/pending_messages/``), so a runtime cap rotation no longer silently discards user data while the process stays up (#78182). r7rArGrRr<rWrXz9Failed to spool cap-dropped transcript message for %s: %sN) r4rTRANSCRIPT_CAP_DROP_REASONrErHnext_TRANSCRIPT_SPOOL_SEQr>r.r/)rWrXr3s&& r spool_dropped_transcript_messager]szn. :x1KTSVW[W`W`WbSc 4-. \:y'B1    PR\bsAA A=A88A=db_known_failingFc$V^8dQhRRRRRR/#)rrWr)r^r:rztuple[int, int]r)r s"rr!r!s"55s55Zi5rc\\4PR44p.pTEFpT\ 9dK\P!TPRR74p\T\4'd1TPR4\8wgTPR4T8wdKTPR4;'g/PR 4p\T\4'g+\PR T4TPR R 7KTP!\#YvP$4Yh34EK \'TR R7^^rp \)T 4F&wp wrpT!T4\1Y`4T ^, p K( T 'd\P3RT T4Y3# \d#p\P RT4RuRp?#Rp?ii;i \dEKi;i \drp\+T4'd\-Y`T4Rp?KT'd\P M\P!RY`T4\/T 4T , p Rp?KRp?ii;i)aReplay cap-dropped transcript messages spooled for *session_id*; return ``(replayed, remaining)``. ``replay(message_dict)`` runs per message in drop order; a spool file is deleted only after its replay succeeds. The first failure stops the drain (the DB is likely still unhealthy) and keeps the rest for retry, unless the DB rejected that row's own values (:func:`is_row_rejection`): it is quarantined and the drain goes on. With ``db_known_failing`` (the caller's last write already failed and is being logged/escalated) a replay failure is expected and logs at DEBUG, so a stalled session does not add one WARNING per append on top of its ERROR (#114266). zpending-*.jsonz Cannot scan transcript spool: %sN utf-8-sigencodingrAr7r<rXz6Removing structurally invalid transcript spool file %sT missing_okcV^,#r)es&r(drain_transcript_spool..s1rkeyzWReplay of spooled transcript message %s for %s failed; keeping spool file for retry: %szBReplayed %d spooled transcript message(s) for %s after DB recovery)rhrh)rIrglobr>r.r/r jsonloads read_text isinstancedictgetrZwarningunlinkappend_spool_sort_keyrsortedrSis_row_rejection_quarantine_spool_filelen_remove_replayedrK)rWreplayr^ candidatesr3entriespathrrXorderedreplayed remainingidx_keys&&$ rdrain_transcript_spoolrs.*//0@AB G ( (  jj!EFG 7D));;x(,FF;;}-; ;;v&,,"11)<'4(( NNSUY Z KK4K ( ;TKL%&$*'~#F1yG&/&8" "d'  7O *A '9 XZb  S  7=    $ $$&t=-V\\6>> A348c KG s*I  sR"F%&G#G(% G0G G G G%$G%( I$3II:II$c V^8dQhRRRR/#)rr8rrr:r)r s"rr!r!scdrcd\P!V4R# \\3dR#i;i)TF)rodumps TypeError ValueError)r8s&r _json_safers- 5 z "s //c V^8dQhRRRR/#)rr8rrzOptional[dict]r)r s"rr!r!s  C N rc^\VR4'dMR\VRR4/pRF5p\WR4pVfK\V4'dTM \V4W&K7 V#\ V\4'dRV/#\ V\ 4'd\V4'dV#R\V4/#)z>+;; = cIz>=Q RRrc$V^8dQhRRRRRR/#)rrrrWr)rNoner)r s"rr!r! s& j j4 jS jT jrcVPRR7R# \d8p\PT4\P RYT4Rp?R#Rp?ii;i)aDelete a spool file whose row was just written. If it cannot be deleted, this process never replays it again, and replay goes on with the files after it, which are still older than any live row. Nothing on disk marks it, so a restart writes the row a second time.TrdzSpooled transcript row for %s was written, but %s could not be deleted (%s); delete it by hand, or the next start writes that row againN)rvr-r addr.error)rrWr3s&& rr}r} sUj t $ j!!$' RS]eh j jjs A,AAc(V^8dQhRRRRRRRR/#) rrrrWr)r3rrrr)r s"rr!r!s( , , ,3 ,] ,t ,rcVPVP\,4pVPV4\ PRYT4R# \d#p\ P RYY$4Rp?R#Rp?ii;i)zMove a spool file whose row the database rejected out of every replay, so it cannot hold its session's later rows back on every write and every restart. It stays on disk for manual recovery.zfSpooled transcript row for %s in %s was rejected by the database (%s) and could not be quarantined: %sNzSpooled transcript row for %s was rejected by the database (%s); moved it to %s for manual recovery and continuing with the session's later rows) with_namerQUARANTINE_SUFFIXreplacer-r.rru)rrWr3targetmove_excs&&& rr{r{sr^^DII(99 :F V  NNVF,  ./9 PsA A?A::A?cV^8dQhRR/#)rrz list[Path]r)r s"rr!r!&s,,,rc.pVFp\P!VPRR74p\ V\4'd-VP^.\W2P4O5V34KlVP^RRVP3V34K VPRR7VUUu.uFwrBVNK upp# \\\ 3dRpLi;iuuppi)aOrder recovery payloads by :func:`_spool_sort_key`, as :func:`drain_transcript_spool` does. ``SessionDB`` restores a conversation by AUTOINCREMENT id and spool files have random names, so any other order permanently scrambles the recovered transcript. Keeps only the key and path, not decoded payloads; the caller re-reads each file to replay it. Unparseable files sort last by name. rarbNrcV^,#rgr)entrys&rrj$_order_flush_files..9s58rrl) rorprqr-rRecursionErrorrrrsrwrxrsort)pathsrrrrs& r_order_flush_filesr&sG jj!EFG gt $ $ NNQE))!DEtL M NNQS$))4d; < LL+L,#* +7ZTD7 ++^4 G  ,s&B?. C?CCsession_resolvercV^8dQhRR/#rrrEr)r s"rr!r!=sSSSrc(\WR7^,#)zReplay flush-dir ``*.json`` files into state.db; return the number of messages recovered. See :func:`recover_pending_spool`, which also reports the sessions it held back.r)recover_pending_spool) session_dbrs&$rrecover_pending_to_dbr=s ! OPQ RRrcV^8dQhRR/#)rrztuple[int, set[str]]r)r s"rr!r!Cs8,8,H\8,rc \\4PR44pV'g ^\43#VRJpV'd^RIHpV!4p^p/pVFpV\ 9dK\P!VPRR74pVPR4\8XdKS\WVVVR7'd'V^, p\WxPRR 44KK V'd;\"P$!\4;_uu_4^R IHp V !V4RRR4MV'd\P)R V4V'dS\P)R \+VP-44RP/RVP14444V\V43# \d#p \P!R Yy4Rp ? EKRp ? ii;i +'giL;i T'dN\"P$!\4;_uu_4^R IHp T !T4RRR4i +'gii;ii;i)a4Replay flush-dir ``*.json`` files via ``SessionDB.append_message``, deleting each on success. ``session_db=None`` opens (and afterwards releases) the shared default ``state.db``. ``session_resolver`` (optional ``(session_key, not_after=ts) -> (session_id, db) | None``, e.g. ``SessionStore.resolve_session_id_for_key``) is required for real flush files: adapter ``MessageEvent`` objects carry no ``session_id``, so without it every recovery lands in the skip branch. A returned ``db`` routes the append to the profile store owning the key (multiplexed gateways); ``None`` falls back to ``session_db``. Returns ``(recovered, held_back)``: the number of messages recovered, and the session ids whose spool files this pass held back after a failed replay, so the live writer can drain them before that session's next row. z*.jsonN)acquirerarbrA)rblocked_sessionsr7rz-Failed to recover pending message from %s: %s)release_or_closez3Recovered %d pending message(s) from shutdown flushzTHeld back %d spooled transcript file(s) for the next start after a failed replay: %sz, c36"TFwrV RV R2xK R#5i)z ()Nr).0sidcounts& r (recover_pending_spool..zs!\C[ZSRwa0C[s)rrrnsethermes_state_registryrr rorprqrtAGENT_HISTORY_REASON_recover_one_payloadr}r>r.ru contextlibsuppressrrKsumvaluesjoinrJ) rr flush_filesown_dbr recoveredrrrr3rs &$ rrrCs%^%5%:%:8%DEK #%x 4 F 1Y I(*-D,, [**T^^[^%IJ;;x(,@@' '9I9IKKNI$T;;}b+IJ K * $$Y//B ,0/  I9U j(//12II\CSCYCYC[\\ ^ c*+ ++ [NPTZZ [0/ $$Y//B ,0/// sgG4(AF1(G4*9F1#G4G!1 G<GG4GG4! G1 4+I H7 . I 7 I  I c(V^8dQhRRRRRRRR/#) rrrrrrzdict[str, int]rr:r)r s"rr!r!~s*::4:.:BP:UY:rc VPR4\8Xd^RIHpVPR/4;'g/pVPRR4VPR4rV'd\ V\ 4'g\ PRV4R#Wt9dWG;;,^, uu&R#VP!R/V!WxVPR 4R 7BR #VPR R4VPR/4rjVPRR4p V 'd V 'g\ PRV4R#VPRR4TrV 'g5Ve1V!WPR 4R7pV'dV^,eVwrV 'g\ PRW4R#V PV RV VPR \\P!444R7R # \d.p \T 4'd\YT 4R p ? R#^YG&hR p ? ii;i \d#p \ PRY4R pR p ? LR p ? ii;i)zAppend one flush payload to ``session_db``; False when it was not replayed (the file is kept, or quarantined when the database rejected the row itself).rA)transcript_append_kwargsr<rWrrXz]Cannot recover structurally invalid transcript spool file %s; preserved for manual inspectionFrG) fallback_tsNTr7rz^Cannot recover structurally invalid pending message from %s; the flush file has been preserved) not_afterz,Session key->id resolution failed for %s: %szCannot recover pending message for %s: no session_id in flush file and session_key-to-id resolution failed. The message text is preserved in %suser)rWrolecontent timestampr)rtrZgateway.session_transcriptrrrrsr.ruappend_messager>rzr{r/rErH)rrrrrrr< spooled_sidrXr3r7rrW target_dbresolveds&&&$$ rrr~s {{8 ::G{{62&,,"#xx b9488I;NW*Wd";"; NNFGK M  *  )Q . )   % % v(@cjcncnosct(u v M26 FB8O 88FB D {;<@ B !HH\26 *6 ' {{4?PQH  /$, !J =>I Q  '.{{4TYY[9I'JL I $$&t#>-.  )  0  LLG ZH s62)G,H', H$7HHH$' I2IIcV^8dQhRR/#rr)r s"rr!r!ssrc^RIHp^RIHpHpHpVP p\VPR7wrgVPV4\VPRR4'gV#\V!44P4pV!VP4Fwr\V 4P4V8Xg'\V 4R, P4'gKKV!\V 44p \VPR7wrT!T 4Yl, pTPT4K V# \ d%\"P%RT RR 7T!T 4Ki;i T!T 4i;i) a7Replay every ``pending_messages`` spool this gateway owns into state.db; return the count. ``_get_flush_dir`` follows the active HERMES_HOME, so a routed turn on a multiplexed gateway spools its stalled transcript backlog under ``profiles//`` and the runtime drain forgets it on restart. After the launch home, replay each served profile inside its own home so the default store ``recover_pending_spool`` opens is that profile's state.db (#123584). The sessions a pass holds back are marked on the session store, so their next live row drains the spool first. )_multiplex_profile_homes)r reset_hermes_home_overrideset_hermes_home_overridermultiplex_profilesFr z.Pending-message recovery failed for profile %sT)exc_info) gateway.runrrr rr session_storerresolve_session_id_for_keymark_spooled_drop_sessionsrconfigrresolveis_dirr)r>r.ru) runnerrr rrstorer held_back launch_homerhometokenrs & rrecover_gateway_pendingr s95ff  E0%BbBbcI $$Y/ 6=="6 > >()113K.v}}=  :   ; .tDzDV7V6^6^6`6` (T3 .4eFfFfg E9 'u -  ((3>   NNKT\`N a  &u -   'u -s$2D22#E!E$ E!!E$$ E.c$V^8dQhRRRRRR/#)rrWz Optional[str]historyrIrrr)r s"rr!r!s!MDTrc V'gR#\4p.pVF\p\V\\\\ \ \\R434pTPV'dTM \ V44K^ \VR\RRRVR\V4RV/4\PR\V4V4R# \dKi;i \d#p\PR TT4Rp?R#Rp?ii;i) uKBest-effort dump of an agent's in-memory transcript before teardown. Used when ``_flush_messages_to_session_db`` raises (e.g. FTS/SQLite corruption): the transcript is written outside the broken DB so an operator can salvage it after repairing state.db. Failures are swallowed — shutdown must never block on a best-effort backup.NrAissuez#72680rWrmessagesuoPreserved %d in-memory message(s) for session %s (possible FTS corruption — recover after repairing state.db)z=Agent-history shutdown preservation failed for session %s: %s)rrrrsrIr)rErr:typerwr>r4rr|r.ru)rWr rsnapshot_mplain_es&& rflush_agent_history_to_filers "$ B "2dCeT4PT:'VWeR9 y *GX *gs8}j(#   X8}j 2    VXb  s<CAB?6AC? C C CC C>C99C>)N)4__conditional_annotations____doc__ __future__rr itertoolsrologgingrrrrHr&pathlibrtypingrrrhermes_dir_aclr getLogger__name__r.rZrrrr\rr __annotations__rr4r@rMrUr]rrr=rrxrzr}r{rrrrr r)rs@rr!s #   &&(   8 $ 3@!)$'5y( ,  J Q[.$5PU5p "4&W S j ,,.StSS 8,t8,8,v:*.:zDr