.cj2FdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z m Z mZddlmZejeZejZdZdZdZd Zd Zd Zd8dZd9dZd:dZd;dZdd?d&Z#d>d@d(Z$ dAdd)dBd/Z%dAdCd0Z&dAdDd3Z'dEdFd7Z(dS)GuDurable 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)AnyDictListOptionalget_hermes_homeiQi: u_♻️ Recovered reply — the gateway restarted during delivery, so this may be a duplicate: c$tdz S)Nzstate.dbr9/home/ice/.hermes/hermes-agent/gateway/delivery_ledger.py_db_pathrIs   z ))rreturnsqlite3.Connectioncddlm}t}|jddt j|d}||d|d |S) Nr)apply_wal_with_fallbackT)parentsexist_ok )timeoutzstate.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_staterrparentmkdirsqlite3connectexecute)rpathconns r_connectr"Ms444444 ::DKdT222 ?4 , , ,DD+GHHHHLL " Krtuple[int, Optional[int]]cxtj} ddlm}|||fS#t$r|dfcYSwxYw)Nrget_process_start_time)osgetpidgateway.statusr& Exception)pidr&s r _owner_stampr,hsf )++C999999**3//// Dys (99r+r 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 ValueErrorr)r&r*r'killProcessLookupErrorPermissionErrorOSError)r+r-r& current_starts r _owner_aliver8rsZ 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)r9r;r<payloads rcompute_obligation_idrFsR 66{66W66G >'..)<< = = G G I I#2# NNr obligation_idplatformchat_id thread_id Optional[str]Nonecptj}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)timer,_DB_LOCKr"rr:_prune) rGr9rHrIrJr<nowr+startedr!s rrecord_obligationrSs0 )++C>>LC    8::    F K3w<<( 2S^^^dGS# '                                  HHHHHs4B?B: BB B B BB!$B!c&t|ddS)N attempting _update_staterGs rmark_attemptingrYs-.....rc&t|ddS)N deliveredrVrXs rmark_deliveredr\s------rerrorc*t|d|dS)Nfailed)r^rV)rGr^s r mark_failedras-777777rstatec t5t5}|d|tj|r |ddnd|fdddn #1swxYwYddddS#1swxYwYdS)NzwUPDATE delivery_obligations SET state=?, updated_at=?, last_error=? WHERE obligation_id=?r )rOr"rrN)rGrbr^r!s rrWrWs   8::   (DIKK!?tt4 O                                     s4A38A A3A A3"A #A33A7:A7)deliverable_platformsrQOptional[float]rd 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)rGr9rHrIrJr< needs_markerattempts) rNr,rOr"rfetchallr8 MAX_ATTEMPTSSTALE_AFTER_SECONDSrowcountappend)rQrdr+rRclaimedr!rowsoidr9rHrIrJr<rbrl created_at owner_pidowner_started_atcursors rsweep_recoverablerys-*##dikkC>>LC$&G --8::-|| 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) rN_RETENTION_SECONDSr"rfetchonemax _MAX_ROWSr*loggerdebug)rQcutoffr!totalexcesss rrPrPs[##dikkC % %FD ZZ 4 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 isinstancer:striplowerr.r*)rrgwvalues rledger_enabledr4s  > 5 5 5 5 5 5 []]F ZZ " " (b($// eS ! ! L;;==&&((0KK KE{{ ttsA;B >B BBlimitr0ct5t5}|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)rrjr )idsessionrbrlru updated_at last_errorr ).0rs r zdebug_rows..Os]   dqtad!d!A$ad     rr)indent)rOr"rrmjsondumps)rr!rss r debug_rowsrDs# 8::|| 3H   (**  :          s4A$*A  A$ A A$A A$$A(+A()rr)rr#)r+rr-rrr.)r9r:r;r:r<r:rr:)rGr:r9r:rHr:rIr:rJrKr<r:rrL)rGr:rrL)r])rGr:r^r:rrL)rGr:rbr:r^r:rrL)N)rQrerdrfrrg)rQrerrL)rrrr.)r)rr0rr:))__doc__ __future__rrArloggingr'r threadingrNtypingrrrrhermes_constantsr getLogger__name__rLockrOrnror|rRECOVERED_MARKERrr"r,r8rFrSrYr\rarWryrPrrr rrrs@%%N#"""""   ,,,,,,,,,,,,,,,,,,  8 $ $ 9>    "%  & ***6DOOOO    4////....88888      F,0FFFFFFRDDDDD<      r