Tdjd7vdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl m Z mZmZmZmZddlmZejeZejZdZdZd Zd Zd Zd Zd=dZd>dZ e d?dZ!d@dZ"dAdZ#dBd Z$dCd&Z%dDd'Z&dDd(Z'dEdFd+Z(dEdGd-Z) dHdd.dId4Z*dHdJd5Z+dHdKd8Z,dLdMd<Z-dS)NuDurable delivery-obligation ledger for gateway final responses. A final agent response that was generated but not yet confirmed-delivered to the messaging platform is the one artifact the gateway can lose without a trace: the turn already burned its tokens, the text exists only in a Python local, and a crash / planned restart between finalize and platform ACK drops it silently (#58818, #41696, #63695). This module records a small durable row per outbound final response in the shared ``state.db`` (same file and conventions as ``tools.async_delegation`` — WAL, owner pid + process-start-time liveness, bounded retention). The gateway writes three checkpoints around the send: record_obligation() state='pending' before any send attempt mark_attempting() state='attempting' immediately before the await mark_delivered() / state='delivered' only on SendResult.success mark_failed() state='failed' on a definitive rejection On startup, ``sweep_recoverable()`` claims rows whose owning process is dead and hands them to the gateway for redelivery. Crash semantics are explicit about ambiguity (the contract review of the earlier delivery-outbox attempt, #61790, closed it for silently resending ambiguous sends): - ``pending`` — the send never started: redeliver plainly, no dup risk. - ``attempting`` — crashed mid-await: the platform MAY already have the message. Redelivered WITH a visible recovered-reply marker so the contract is honest at-least-once, never a silent duplicate. - ``failed`` — definitively rejected once; the restart is a natural retry boundary. Also carries the marker. - ``delivered`` — nothing to do; retention prunes. Poison rows cannot spin: attempts are capped, stale rows expire, and both transition to ``abandoned`` (kept briefly for inspection, then pruned). Everything here is best-effort by design: ledger failures must never block or delay an actual send. Callers wrap every call in try/except. ) annotationsN)contextmanager)AnyDictIteratorListOptionalget_hermes_homeiQi: u_♻️ Recovered reply — the gateway restarted during delivery, so this may be a duplicate: c$tdz S)Nzstate.dbr =/home/thesage/.hermes/hermes-agent/gateway/delivery_ledger.py_db_pathrJs   z ))rreturnsqlite3.Connectionct}|jddtj|d} t |n##t $r|wxYw|S)NT)parentsexist_ok )timeout)rparentmkdirsqlite3connect_initialize_schema Exceptionclose)pathconns r_connectr#Ns ::DKdT222 ?4 , , ,D4       Ks A A2r"NonecVddlm}||d|ddS)Nr)apply_wal_with_fallbackzstate.db (delivery_ledger))db_labelaCREATE TABLE IF NOT EXISTS delivery_obligations ( obligation_id TEXT PRIMARY KEY, session_key TEXT NOT NULL, platform TEXT NOT NULL, chat_id TEXT NOT NULL, thread_id TEXT, content TEXT NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL, owner_pid INTEGER, owner_started_at INTEGER, last_error TEXT )) hermes_stater&execute)r"r&s rrr\sP444444D+GHHHHLL rIterator[sqlite3.Connection]c#Kt} |5|Vdddn #1swxYwY|dS#|wxYw)udOpen a connection, commit/rollback on exit, and ALWAYS close it. ``sqlite3.Connection.__enter__``/``__exit__`` only commit or roll back the transaction; they do not close the connection. Using ``with _connect()`` alone therefore leaks a connection — and its WAL/SHM file descriptors — on every call, deferring the close to the garbage collector. On a long-running gateway that exhausts ``RLIMIT_NOFILE`` (the cron-ledger sibling of this bug was #69567 / PR #69594). ``record_obligation`` runs on every outbound final response, so this ledger is the highest-frequency leaker. N)r#r )r"s r _transactionr,ss ::D   JJJ                 s&A% A)A)AAtuple[int, Optional[int]]cxtj} ddlm}|||fS#t$r|dfcYSwxYw)Nrget_process_start_time)osgetpidgateway.statusr0r)pidr0s r _owner_stampr5sf )++C999999**3//// Dys (99r4r started_atboolc|sdS t|}n#ttf$rYdSwxYw ddlm}||}n#t $rd}YnwxYw|B t j|dn)#t$rYdSt$rYdSt$rYdSwxYwdS|dS t|t|kS#ttf$rYdSwxYw)zFTrue when the recorded owning process still exists (pid + start time).Frr/NT) int TypeError ValueErrorr3r0rr1killProcessLookupErrorPermissionErrorOSError)r4r6r0 current_starts r _owner_aliverAsZ u#hh z "uu999999..s33    GCOOOO!   55   44   55 tt=!!S__44 z "ttsM++A AAA,, B9 B BBB<<CC session_keystr message_refcontentc|d|d|}tj|ddddS)aStable id: same turn + same content re-records idempotently, while distinct threads/topics on the same chat can never collide (the session_key carries platform, chat and thread; ``message_ref`` is the triggering inbound message id, distinguishing turns in one session).|zutf-8replaceN)hashlibsha256encode hexdigest)rBrDrEpayloads rcompute_obligation_idrOsR 66{66W66G >'..)<< = = G G I I#2# NNr obligation_idplatformchat_id thread_id Optional[str]cptj}t\}}t5t5} | d|||t ||rt |nd|||||f dddn #1swxYwYdddn #1swxYwYt dS)zBRecord a final response as owed to the platform (state='pending').a'INSERT OR REPLACE INTO delivery_obligations (obligation_id, session_key, platform, chat_id, thread_id, content, state, attempts, created_at, updated_at, owner_pid, owner_started_at) VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?)N)timer5_DB_LOCKr,r)rC_prune) rPrBrQrRrSrEnowr4startedr"s rrecord_obligationr[s0 )++C>>LC    <>>  T  F K3w<<( 2S^^^dGS# '                                  HHHHHs4B?B: BB B B BB!$B!c&t|ddS)N attempting _update_staterPs rmark_attemptingras-.....rc&t|ddS)N deliveredr^r`s rmark_deliveredrds------rerrorc*t|d|dS)Nfailed)rfr^)rPrfs r mark_failedris-777777rstatec t5t5}|d|tj|r |ddnd|fdddn #1swxYwYddddS#1swxYwYdS)NzwUPDATE delivery_obligations SET state=?, updated_at=?, last_error=? WHERE obligation_id=?r )rWr,r)rV)rPrjrfr"s rr_r_s   <>> T  (DIKK!?tt4 O                                     s4A38A A3A A3"A #A33A7:A7)deliverable_platformsrYOptional[float]rl Optional[set]List[Dict[str, Any]]cz||ntj}t\}}g}t5t5}|d}|D]\ }}} } } } } }}}}t ||r|tks||z tkr|d||fQ|| |vrX|d||||||f}|j r%| ||| | | | | dk|dzd dddn #1swxYwYdddn #1swxYwY|S)aeClaim undelivered rows owned by dead processes; return them for redelivery. Claiming atomically re-stamps the owner to THIS process and increments ``attempts``, so a second gateway racing the same sweep cannot double-claim (the UPDATE is guarded on the previous owner stamp). Rows over the attempts cap or older than the stale cutoff transition to 'abandoned' instead of being returned. ``deliverable_platforms`` (platform value strings) restricts claiming to platforms the caller can actually send on this boot. ``attempts`` is the redelivery budget, so it must only be spent on a real send: a platform that failed to connect would otherwise burn one attempt per boot and hit the cap having never been sent once. Rows for absent platforms are left untouched for a later boot; the stale cutoff still bounds them. NaSELECT obligation_id, session_key, platform, chat_id, thread_id, content, state, attempts, created_at, owner_pid, owner_started_at FROM delivery_obligations WHERE state IN ('pending', 'attempting', 'failed')zlUPDATE delivery_obligations SET state='abandoned', updated_at=? WHERE obligation_id=?zUPDATE delivery_obligations SET owner_pid=?, owner_started_at=?, attempts=attempts+1, updated_at=? WHERE obligation_id=? AND (owner_pid IS ? OR owner_pid=?)pending)rPrBrQrRrSrE needs_markerattempts) rVr5rWr,r)fetchallrA MAX_ATTEMPTSSTALE_AFTER_SECONDSrowcountappend)rYrlr4rZclaimedr"rowsoidrBrQrRrSrErjrt created_at owner_pidowner_started_atcursors rsweep_recoverablers-*##dikkC>>LC$&G --<>>-T|| E  (** DH% %  @S+x)We z9.>I'788 <''C*,<@S+S+S T#J %1$999\\PgsCI> F %(#. (&!*&%*Y$6 (1      5% ------------------------------\ Ns6D0C D D0D D0 D !D00D47D4c||ntj}|tz } t5}|d|f|dd}t d|t z }|r|d|fddddS#1swxYwYdS#t$r t ddYdSwxYw)NzpDELETE FROM delivery_obligations WHERE state IN ('delivered', 'abandoned') AND updated_at < ?z)SELECT COUNT(*) FROM delivery_obligationsraDELETE FROM delivery_obligations WHERE obligation_id IN ( SELECT obligation_id FROM delivery_obligations ORDER BY CASE state WHEN 'delivered' THEN 0 WHEN 'abandoned' THEN 1 ELSE 2 END, updated_at ASC LIMIT ?)zdelivery ledger prune failedT)exc_info) rV_RETENTION_SECONDSr,r)fetchonemax _MAX_ROWSrloggerdebug)rYcutoffr"totalexcesss rrXrX5s[##dikkC % %FD ^^ t LLS     LL;hjjEEI-..F  %I                     , DDD 3d CCCCCCDs5CA6B4' C4B88C;B8<C&C+*C+configOptional[Dict[str, Any]]c< |ddlm}|}|dpi}|dd}t|tr(|dvSt|S#t$rYdSwxYw)z>Read the ``gateway.delivery_ledger`` config gate (default on).Nr) load_configgatewaydelivery_ledgerT>0noofffalse) hermes_cli.configrget isinstancerCstriplowerr7r)rrgwvalues rledger_enabledrSs  > 5 5 5 5 5 5 []]F ZZ " " (b($// eS ! ! L;;==&&((0KK KE{{ ttsA;B >B BBlimitr9ct5t5}|d|f}dddn #1swxYwYdddn #1swxYwYt jd|DdS)z>Human-readable dump for ad-hoc inspection (sqlite3-free path).zSELECT obligation_id, session_key, state, attempts, created_at, updated_at, last_error FROM delivery_obligations ORDER BY updated_at DESC LIMIT ?Nc vg|]6}|d|d|d|d|d|d|dd7S)rrrr )idsessionrjrtr} updated_at last_errorr).0rs r zdebug_rows..ns]   dqtad!d!A$ad     rr)indent)rWr,r)rujsondumps)rr"r{s r debug_rowsrcs# <>>T|| 3H   (**  :          s4A$*A  A$ A A$A A$$A(+A()rr)r"rrr$)rr*)rr-)r4rr6rrr7)rBrCrDrCrErCrrC)rPrCrBrCrQrCrRrCrSrTrErCrr$)rPrCrr$)re)rPrCrfrCrr$)rPrCrjrCrfrCrr$)N)rYrmrlrnrro)rYrmrr$)rrrr7)r)rr9rrC).__doc__ __future__rrJrloggingr1r threadingrV contextlibrtypingrrrrr hermes_constantsr getLogger__name__rLockrWrvrwrrRECOVERED_MARKERrr#rr,r5rArOr[rardrir_rrXrrrrrrs%%N#"""""   %%%%%%66666666666666,,,,,,  8 $ $ 9>    "%  & ***    .&DOOOO    4////....88888      F,0FFFFFFRDDDDD<      r