mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
Teknium
e1de0c2665
Agent cron runs hand their send to the durable delivery queue and book last_status=delivery_queued / ledger delivery_outcome=queued. Nothing ever wrote the drain's result back, so after a gateway sent (or refused) the message the job stayed "delivery still in progress" forever and a refused send never surfaced as delivery_failed + last_delivery_error (review M1; main records ok/delivered and delivery_failed/failed + error). cron/delivery_outcome.py settles a terminal queue row onto: - the execution ledger: only a terminal attempt whose delivery_outcome is still 'queued' (new executions.settle_delivery_outcome), and - the job record: only while the execution is both the ledger's latest attempt for the job and the job's latest canonical completion, so a later run's status is never rewritten. It runs after the drain's _finish, after the run's own bookkeeping (a drain can win the race before mark_job_run lands; _finish_completed_run also books the settled outcome instead of 'queued'), and after receipt recovery. Write-back failures are logged, never undo the at-most-once send. Restores main's assertions in the rewritten W1 tests, checked after an explicit drain: test_warning_execution_outcome (ok/delivered|suppressed), test_cron_failure_deliver (delivered / failed outcome after drain), test_fire_claim_lost_after_delivery (last_status ok), test_script_claim_heartbeat self-removal (delivery_outcome delivered). New invariant test: test_delivery_outcome_writeback (sent/refused x drain after run / drain before bookkeeping). (cherry picked from commit 0402eff3d91050bbfa68b83ba6f3d7169700da3f)
95 lines
4.0 KiB
Python
95 lines
4.0 KiB
Python
"""Write a queued cron delivery's terminal outcome back onto the run that queued it.
|
|
|
|
An agent run hands its send to ``cron.delivery_queue`` and books ``last_status =
|
|
"delivery_queued"`` / ledger ``delivery_outcome = "queued"``. Once a gateway drain sends,
|
|
refuses, suppresses or fences the send ``unknown``, the job record and the execution
|
|
ledger must carry the real result (main's inline ``ok``/``delivered`` and
|
|
``delivery_failed``/``failed`` + ``last_delivery_error``); otherwise ``hermes cron list``
|
|
says "delivery still in progress" forever and refused sends never surface.
|
|
|
|
Both sides call :func:`settle_queued_delivery`: the drain right after it terminalizes a
|
|
row, and the run right after its own bookkeeping (a drain can finish before the run's
|
|
``mark_job_run`` lands). Every write is fenced on the execution id, so the call is
|
|
idempotent and never rewrites a later run's status.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Queue terminal status -> ledger delivery_outcome. ``unknown`` (owner died mid-send, never
|
|
# retried) is surfaced as a failure: the user cannot be told the result arrived.
|
|
_LEDGER_OUTCOME = {
|
|
"delivered": "delivered", "suppressed": "suppressed", "failed": "failed", "unknown": "failed",
|
|
}
|
|
|
|
|
|
def _settle_job_record(job_id: str, execution_id: str, delivery_error: Optional[str]) -> bool:
|
|
"""Stamp the outcome onto the job only while *execution_id* is still its current run:
|
|
the ledger's latest attempt for the job AND the job's latest recorded completion."""
|
|
from cron.executions import latest_execution
|
|
from cron.jobs import _with_job, save_jobs
|
|
|
|
latest = latest_execution(job_id)
|
|
if latest is None or latest.get("id") != execution_id:
|
|
return False
|
|
|
|
def apply(jobs, _i, job):
|
|
completions = job.get("canonical_completions") or []
|
|
if not completions or completions[-1] != execution_id:
|
|
return False
|
|
status = job.get("last_status")
|
|
if status == "delivery_queued":
|
|
status = "delivery_failed" if delivery_error else "ok"
|
|
if (status, delivery_error) == (job.get("last_status"), job.get("last_delivery_error")):
|
|
return True
|
|
job["last_status"] = status
|
|
job["last_delivery_error"] = delivery_error
|
|
save_jobs(jobs)
|
|
return True
|
|
|
|
return bool(_with_job(job_id, apply, missing=False))
|
|
|
|
|
|
def _terminal(execution_id: str) -> Optional[tuple[str, Optional[str]]]:
|
|
"""``(ledger_outcome, delivery_error)`` once the queue row is terminal, else None."""
|
|
from cron.delivery_queue import get_status
|
|
|
|
row = get_status(str(execution_id))
|
|
if not row or row["status"] not in _LEDGER_OUTCOME:
|
|
return None
|
|
delivery_error = None
|
|
if row["status"] in ("failed", "unknown"):
|
|
delivery_error = str(row.get("error") or f"delivery {row['status']}")
|
|
return _LEDGER_OUTCOME[row["status"]], delivery_error
|
|
|
|
|
|
def settled_outcome(execution_id: str) -> Optional[str]:
|
|
"""The ledger outcome a drain already reached for *execution_id* (None while queued)."""
|
|
terminal = _terminal(execution_id)
|
|
return terminal[0] if terminal else None
|
|
|
|
|
|
def settle_queued_delivery(job_id: Optional[str], execution_id: Optional[str]) -> bool:
|
|
"""Apply the queue's terminal outcome for *execution_id*; False while still in flight."""
|
|
if not job_id or not execution_id:
|
|
return False
|
|
terminal = _terminal(str(execution_id))
|
|
if terminal is None:
|
|
return False
|
|
from cron.executions import settle_delivery_outcome
|
|
|
|
settle_delivery_outcome(str(execution_id), terminal[0])
|
|
_settle_job_record(str(job_id), str(execution_id), terminal[1])
|
|
return True
|
|
|
|
|
|
def settle_quietly(job_id: Optional[str], execution_id: Optional[str]) -> None:
|
|
"""Bookkeeping failure must not undo a terminal at-most-once send or fail its run."""
|
|
try:
|
|
settle_queued_delivery(job_id, execution_id)
|
|
except Exception:
|
|
logger.warning("Cron delivery %s: outcome write-back failed", execution_id, exc_info=True)
|