`gj2rdZddlmZddlZddlZddlZddlZddlZddlZddl Z ddl m Z m Z m Z mZddlmZej"eZej(ZdZdZdZd Zd Zd Zdd Zdd ZddZddZ ddZ ddZ!ddZ"d d!dZ#d d"dZ$ d#dd d$dZ%d#d%dZ&d#d&dZ'd'd(dZ(y))uDurable 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: ctdz S)Nzstate.dbrJ/root/.hermes/venv/lib/python3.12/site-packages/gateway/delivery_ledger.py_db_pathrIs  z ))rct}|jjddtj|d}|j d|j d|S)NT)parentsexist_ok )timeoutzPRAGMA journal_mode=WALaCREATE 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 ))rparentmkdirsqlite3connectexecute)pathconns r_connectrMsU :DKKdT2 ??4 ,DLL*+LL " Krcrtj} ddlm}|||fS#t$r|dfcYSwxYw)Nrget_process_start_time)osgetpidgateway.statusr Exception)pidr s r _owner_stampr&fs? ))+C9*3/// Dys & 66cd|sy t|} ddlm}||}| t j|dy|y t|t|k(S#ttf$rYywxYw#t $rd}YWwxYw#t$rYyt$rYyt$rYywxYw#ttf$rYywxYw)zFTrue when the recorded owning process still exists (pid + start time).FrrNT) int TypeError ValueErrorr#r r$r!killProcessLookupErrorPermissionErrorOSError)r% started_atr current_starts r _owner_aliver1ps #h9.s3   GGCO=!S_44/ z "  "     z "sP AA)A:BA&%A&) A76A7: BBBBB/.B/c|d|d|}tj|jddjddS)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) session_key message_refcontentpayloads rcompute_obligation_idr>sD  Q{m1WI6G >>'..)< = G G I#2 NNrcFtj}t\}}t5t5} | j d|||t ||r t |nd|||||f ddddddt y#1swYxYw#1swYt yxYw)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_LOCKrrstr_prune) obligation_idr:platformchat_id thread_idr<nowr%startedrs rrecord_obligationrJs ))+C>LC   8:    F K3w<(S^dGS# '      H     Hs" B 3B&B B B  B ct|dy)N attempting _update_staterDs rmark_attemptingrPs -.rct|dy)N deliveredrMrOs rmark_deliveredrSs --rc t|d|y)Nfailed)errorrM)rDrVs r mark_failedrWs-7rc t5t5}|jd|tj|r|ddnd|fddddddy#1swYxYw#1swYyxYw)NzwUPDATE delivery_obligations SET state=?, updated_at=?, last_error=? WHERE obligation_id=?r )rArrr@)rDstaterVrs rrNrNsd  8:   (DIIKt4 O        s" A0AAA AA()deliverable_platformsc6||ntj}t\}}g}t5t5}|j dj }|D]\ }}} } } } } }}}}t ||r|tk\s ||z tkDr|j d||fF|| |vrM|j d||||||f}|jsr|j||| | | | | dk7|dzd dddddd|S#1swYxYw#1swY|SxYw)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)rDr:rErFrGr< needs_markerattempts) r@r&rArrfetchallr1 MAX_ATTEMPTSSTALE_AFTER_SECONDSrowcountappend)rHrZr%rIclaimedrrowsoidr:rErFrGr<rYr_ created_at owner_pidowner_started_atcursors rsweep_recoverablerlsk*#diikC>LC$&G -8:-|| E (* DH%  @S+x)We z9.>I'78<'C*,<@S+S T#J %1$99\\PgsCI> F%(#. (&!*&%*Y$6 (1   5% --\ N]---\ Ns) DBD "D0DD DDc||ntj}|tz } t5}|jd|f|jdj d}t d|t z }|r|jd|fdddy#1swYyxYw#t$rtjddYywxYw)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) r@_RETENTION_SECONDSrrfetchonemax _MAX_ROWSr$loggerdebug)rHcutoffrtotalexcesss rrCrCs#diikC % %FD Z 4 LLS   LL;hjEEI-.F %I    , D 3d CDs/ B AB B BB B CCc | ddlm}|}|jdxsi}|jdd}t|tr |j j dvSt|S#t$rYywxYw)z>Read the ``gateway.delivery_ledger`` config gate (default on).r) load_configgatewaydelivery_ledgerT>0noofffalse) hermes_cli.configryget isinstancerBstriplowerboolr$)configrygwvalues rledger_enabledr2s{  > 5 ]F ZZ " (b($/ eS !;;=&&(0KK KE{ sA%A3( A33 A?>A?c Ft5t5}|jd|fj}ddddddt j Dcgc]!}|d|d|d|d|d|d|d d #c}d S#1swYSxYw#1swYWxYwcc}w) 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 ?Nrr]r )idsessionrYr_rh updated_at last_error)indent)rArrr`jsondumps)limitrrfrs r debug_rowsrBs 8:|| 3H  (*  ::   dqtad!d!A$ad     s' B"BB&BB BB)returnzsqlite3.Connection)rztuple[int, Optional[int]])r%rr/rrr)r:rBr;rBr<rBrrB)rDrBr:rBrErBrFrBrGz Optional[str]r<rBrNone)rDrBrr))rDrBrVrBrr)rDrBrYrBrVrBrr)N)rHOptional[float]rZz Optional[set]rzList[Dict[str, Any]])rHrrr)rzOptional[Dict[str, Any]]rr))rr(rrB))__doc__ __future__rr6rloggingr!r threadingr@typingrrrrhermes_constantsr getLogger__name__rsLockrArarbrorrRECOVERED_MARKERrrr&r1r>rJrPrSrWrNrlrCrrr rrrs.%N#   ,,,   8 $ 9>>   "%  & *2DO            4/.8  F,0F F)F FRD<  r