߲jE0UdZddlmZddlZddlZddlZddlZddlZddlZddl m Z ddl m Z ddl m Z ddlmZmZmZmZmZddlmZdd lmZdd lmZdZd ed <d ZdZdZdZ ejBZ"ejFjHZ%d-dZ&d.dZ'e d/dZ(d0dZ)dd d1dZ*d2dZ+d3dZ,d4dZ-d5dZ.d.dZ/dd d6dZ0d7dZ1d8dZ2d8d Z3d8d!Z4ddd" d9d#Z5d$Z6d%Z7d:d&Z8dd'dd( d;d)Z9d8d*Z:ducProfile-local durable audit ledger for cron execution attempts. The ledger records what is known about each attempt; it is not a retry queue. Interrupted attempts become ``unknown`` only after their owner process is proved gone — a start-time reading that fails to match the claim-time fingerprint is not proof of death. Terminal states are immutable. ) annotationsN)contextmanager)datetime)Path)AnyDictIteratorListOptional)get_hermes_home)now)CLAIM_TTL_INACTIVITY_HEADROOMzOptional[Path]EXECUTIONS_FILEig>@g @) completedfailedunknowncddlm}ddlm}txst j dz dz }||j||ddtS) Nr)_ensure_cron_dir)open_dbcronz executions.dbzcron/executions.dbT)db_labelsynchronous_full initialize) cron.jobsrhermes_cli.sqlite_utilrrr resolveparent_initialize_schema)rrpaths 4/home/ubuntu/.hermes/hermes-agent/cron/executions.py_connectr!(sI+.  V088:VCoUDT[[! 4"6Zl mmcddlm}|jd||ddd||ddd|jd |jd ||dd d ||dd d|jdy)Nr)add_column_if_missingabCREATE TABLE IF NOT EXISTS executions ( id TEXT PRIMARY KEY, job_id TEXT NOT NULL, source TEXT NOT NULL, process_id TEXT NOT NULL, pid INTEGER NOT NULL, process_started_at INTEGER, status TEXT NOT NULL CHECK(status IN ('claimed','running','completed','failed','unknown')), handoff_pending INTEGER NOT NULL DEFAULT 0, handoff_started_at REAL, claimed_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, error TEXT ) executionshandoff_pendingz*handoff_pending INTEGER NOT NULL DEFAULT 0handoff_started_atzhandoff_started_at REALzeCREATE INDEX IF NOT EXISTS idx_executions_job_claimed ON executions(job_id, claimed_at DESC, id DESC)zhCREATE INDEX IF NOT EXISTS idx_executions_status_claimed ON executions(status, claimed_at DESC, id DESC)delivery_outcomezdelivery_outcome TEXTscheduled_instantzscheduled_instant TEXTzvCREATE INDEX IF NOT EXISTS idx_executions_occurrence ON executions(job_id, scheduled_instant) WHERE status='completed')rr$execute)connr$s r rr4s<LL $ l-4 l02K LL : LL :$ .@BYZ$ .AC[\LL Lr"c#Kddlm}t5|t5}|ddddddy#1swYxYw#1swYyxYww)Nr) transaction)rr-_lockr!)r-r+s r _transactionr/`sE2  HJ'4 s- AA6A A? AA Acb|jd|fj}| t|SdS)N#SELECT * FROM executions WHERE id=?)r*fetchonedict)r+ execution_idrows r _fetchr6hs1 ,,<|o N W W YC491T1r"r(cD ddlm}|||y#t$rYywxYw)zFProject durable state to monitoring without affecting ledger behavior.r)emit_execution_stater7N)agent.monitoring.cron_healthr9 Exception)recordr(r9s r _emit_execution_stater=ms' EV6FG    s  c> ddlm}||S#t$rYywxYw)Nr)get_process_start_time)gateway.statusr?r;)pidr?s r _process_start_timerBys'9%c** s  c ddlm}||sy ||tjk(St |}|yddlm}|||S#t$rYywxYw)Nr) _pid_existsFT)start_time_fingerprints_match)r@rDr;osgetpidrBrE)rA started_atrDcurrentrEs r _owner_is_liverJsk.3 biik!!!#&G= (W == sA AAcddlm}ddlm}t |}t j |r|dkryt|tzt |tS)uAge past which a claimed/running row with a LIVE owner is treated as wedged. Derived from the existing knobs, never a bare wall-clock constant: ``max(3 × HERMES_CRON_TIMEOUT, cron script timeout, 7200)``. Returns ``None`` (never reclaim live owners — today's behaviour) when the inactivity timeout is 0/unlimited or not a finite positive number: with no bound to derive from, fail closed. r)_cron_inactivity_seconds)_get_script_timeoutN) cron.schedulerrLcron.scheduler_scriptrMfloatmathisfinitemaxr$LIVE_OWNER_STALE_CLAIM_FLOOR_SECONDS)rLrM inactivitys r _live_owner_stale_after_secondsrVsR89/12J == $ a 22 !#$, r"c^ttj|z jS)zZSeconds since ``claimed_at`` (NOT NULL, always the aware ISO string from hermes_time.now).) _hermes_nowr fromisoformat total_seconds) claimed_ats r _claim_age_secondsr\s# MH22:> > M M OOr"c X|jdtdttfy)NzDELETE FROM executions WHERE id IN ( SELECT id FROM executions WHERE status IN ('completed','failed','unknown') ORDER BY finished_at DESC, claimed_at DESC, id DESC LIMIT -1 OFFSET ? )r)r*rSintMAX_TERMINAL_EXECUTIONS)r+s r _prune_unlockedr`s(LL  Q+, -/ r"r)cddlm}tj}t j j }tj}t5}|jd|t|t|t|t||||ft||}dddt|S#1swYxYw)zO,P R  l+ ,&! M , ,s A B<<Cc ddlm}t5}|jd|||tt j f}|jdk7r td dddy#1swYyxYw)zGBind the store-claimed snapshot before a provider hands it to a worker.rrazzUPDATE executions SET scheduled_instant=? WHERE id=? AND status='claimed' AND handoff_pending=0 AND process_id=? AND pid=?z2Cron occurrence could not be bound before dispatchN) rcr)r/r*rirFrGrowcount RuntimeError)r4instantr)r+curs r set_execution_occurrencertsn2 U4ll ? w '{BIIK P  <<1 ST T  UUUs A A((A1c  t5}|jdtj|tt j f}|j dk7r dddyt||}dddt|S#1swYxYw)zDFence restart recovery while an external worker is adopting a claim.zUPDATE executions SET handoff_pending=1, handoff_started_at=? WHERE id=? AND status='claimed' AND process_id=? AND pid=?roN) r/r*timerirFrGrpr6r=)r4r+rsr<s r mark_execution_handoff_pendingrws  ,4ll /YY[, RYY[ A   <<1  , ,l+ ,&! M , ,sAB# BB c Htj}t|}tj }t 5}|j dt||||f}|jdk7r dddyt||}dddt|S#1swYxYw)uAtomically transfer and start an attempt in its worker process. The dispatching gateway creates the row before spawning a restart-safe worker. Adoption is the single ``claimed`` → ``running`` gate: only the winner may acknowledge ownership or run side effects. aUPDATE executions SET process_id=?, pid=?, process_started_at=?, status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL WHERE id=? AND status='claimed' AND handoff_pending=1roN) rFrGrBrXrdr/r*rirpr6r=)r4rAprocess_started_atr r+rsr<s r adopt_claimed_executionrzs ))+C,S1 - ! ! #C  ,4ll H #13 E   <<1  , ,l+ ,&! M , ,s,B7 BB!c ,tj}t5}|jd||tt j f}|jdk7r dddyt||}dddt|S#1swYxYw)z7Transition one claimed attempt to running exactly once.zUPDATE executions SET status='running', started_at=?, handoff_pending=0, handoff_started_at=NULL WHERE id=? AND status='claimed' AND handoff_pending=0 AND process_id=? AND pid=?roN) rXrdr/r*rirFrGrpr6r=)r4r r+rsr<s r mark_execution_runningr|s - ! ! #C  ,4ll / , RYY[ 9   <<1  , ,l+ ,&! M , ,s=B ) B  B)errorr(c ~tj}|rdnd}|rdn|r t|nd}t5}|j d|||||t t jf}|jdk7r dddyt|t||} dddt || S#1swYxYw)zDWrite a terminal result once; terminal attempts cannot be rewritten.rrNzunknown failureaUPDATE executions SET status=?, finished_at=?, error=?, handoff_pending=0, handoff_started_at=NULL, delivery_outcome=? WHERE id=? AND status IN ('claimed','running') AND process_id=? AND pid=?ror7) rXrdrhr/r*rirFrGrpr`r6r=) r4successr}r(r statusdetailr+rsr<s r finish_executionrs - ! ! #C#[FT3u:z$latest_executions..s/AC/s z@SELECT e.* FROM executions e WHERE e.job_id IN (z) AND e.id=(SELECT e2.id FROM executions e2 WHERE e2.job_id=e.job_id ORDER BY e2.claimed_at DESC, e2.id DESC LIMIT 1)Nrj)r3fromkeysrhrr/r*r)job_idsrjclean placeholdersr+rr5s r latest_executionsrs'+}}W'= HVS[ HE H  88///L 4||$$0>2M P    (* 15 5CM49 $ 55 I 6sB B%B$B0$B-)returnsqlite3.Connection)r+rrNone)rzIterator[sqlite3.Connection])r+rr4rhrOptional[Dict[str, Any]])r<rr( Optional[str]rr)rAr^r Optional[int])rAr^rHrrbool)rzOptional[float])r[rhrrP)rjrhrkrhr)rrzDict[str, Any])r4rhrrrrr)r4rhrr) r4rhrrr}rr(rrr)rr^)rjrrr^rrrzList[Dict[str, Any]])rjrhrr)rz List[str]rzDict[str, Dict[str, Any]])=__doc__ __future__rrQrFsqlite3 threadingrvre contextlibrrpathlibrtypingrrr r r hermes_constantsr hermes_timer rXcron.constantsrr__annotations__r_rrT_TERMINAL_STATESRLockr.rfrgrir!rr/r6r=rBrJrVr\r`rmrtrwrzr|rrrrrrrrrr"r rs#  %66,*8 #'&!%'-$5 djjl n)X2 LP  $  ;H     >$*P EI 4A. U"4(AE&*#'0=#4; @H $"Y]'',/'IV''.2% 6r"