Show last run timestamp on ops automation status
All checks were successful
CI Smoke / host-smoke (push) Successful in 1s
CI Smoke / container-smoke (push) Successful in 5s
Build and Publish Container Image / build-and-push (push) Successful in 1m9s

Expose last_run_at / last_run on the status contract and display Last run
and Tasks columns on the /ops/ui status page for the selected window.
This commit is contained in:
tegwick 2026-07-22 00:16:25 +02:00
parent a304ad7397
commit 6c34e2c1f1
3 changed files with 75 additions and 4 deletions

View file

@ -654,6 +654,11 @@ def classify_activity(definition: dict[str, Any], window: dict[str, Any], runs:
else:
status = "no_due"
last_run = latest_run_summary(runs)
temporal_last = None
if temporal and temporal.get("last_fired_at"):
temporal_last = temporal.get("last_fired_at")
return {
"id": definition["id"],
"name": definition["name"],
@ -666,6 +671,9 @@ def classify_activity(definition: dict[str, Any], window: dict[str, Any], runs:
"expected_fires": expected,
"expected_fire_count": len(expected),
"observed_run_count": len(runs),
"last_run_at": last_run.get("fired_at") if last_run else None,
"last_run": last_run,
"temporal_last_fired_at": temporal_last,
"runs": runs,
"evidence": evidence,
"temporal": temporal,
@ -674,6 +682,26 @@ def classify_activity(definition: dict[str, Any], window: dict[str, Any], runs:
}
def latest_run_summary(runs: list[dict[str, Any]]) -> dict[str, Any] | None:
"""Return the newest run by fired_at (then scheduled_for) within the provided list."""
best: dict[str, Any] | None = None
best_ts: datetime | None = None
for run in runs:
ts = coerce_datetime(run.get("fired_at")) or coerce_datetime(run.get("scheduled_for"))
if ts is None:
continue
if best_ts is None or ts > best_ts:
best_ts = ts
best = {
"run_id": run.get("run_id"),
"fired_at": run.get("fired_at") or iso(ts),
"scheduled_for": run.get("scheduled_for"),
"tasks_spawned": run.get("tasks_spawned"),
"version_used": run.get("version_used"),
}
return best
def workflow_status_matches(workflow: dict[str, Any], names: set[str]) -> bool:
value = str(workflow.get("status") or "").upper()
return any(name in value for name in names)