test(cron): two invariant tests for the live-owner stale-claim bound

Rename tests/cron/test_stale_running_recovery_115692.py to a behaviour name
(#115692 is cited in the docstring). Seed claimed_at with an aware ISO
timestamp via the module's own now-helper instead of a local-naive
time.strftime, which the code reads as UTC and so failed in zones >= 2 h west
of UTC. Assert the bound tracks HERMES_CRON_TIMEOUT, and fold the 0/nan/inf
fail-closed idea from #116253 into the not-recovered test as parametrised
cases rather than a third test.

tests/cron/test_stale_running_recovery.py

@@ -0,0 +1,83 @@
+"""Live-but-deadlocked cron workers (#115692).
+
+A worker wedged on a lock passes ``_owner_is_live()`` forever, so its ``running`` row was never
+reclaimed and every future fire of the job was skipped. ``recover_interrupted_executions`` now
+also releases a live-owned claim once it outlives a bound derived from the existing timeouts
+(``max(3 × HERMES_CRON_TIMEOUT, script timeout, 7200)``) and fails closed when the inactivity
+timeout is unlimited or not a finite positive number.
+"""
+
+from __future__ import annotations
+
+import time
+import uuid
+from datetime import timedelta
+
+import pytest
+
+import cron.executions as executions_mod
+from cron.executions import _hermes_now, _transaction, recover_interrupted_executions
+
+
+@pytest.fixture(autouse=True)
+def _ledger(monkeypatch, tmp_path):
+    monkeypatch.setattr(executions_mod, "EXECUTIONS_FILE", tmp_path / "cron" / "executions.db")
+    # The owner is alive for every row in this module; the bound alone decides.
+    monkeypatch.setattr(executions_mod, "_owner_is_live", lambda pid, started_at: True)
+
+
+def _seed_running(job_id: str, *, age_seconds: float, pid: int = 99999) -> str:
+    claimed_at = (_hermes_now() - timedelta(seconds=age_seconds)).isoformat()
+    with _transaction() as conn:
+        eid = uuid.uuid4().hex
+        conn.execute(
+            """INSERT INTO executions
+               (id, job_id, status, source, process_id, pid, process_started_at, claimed_at)
+               VALUES (?, ?, 'running', 'cron', ?, ?, ?, ?)""",
+            (eid, job_id, "ext-proc", pid, int(time.time()), claimed_at),
+        )
+    return eid
+
+
+def _status(eid: str) -> str:
+    with _transaction() as conn:
+        return conn.execute("SELECT status FROM executions WHERE id=?", (eid,)).fetchone()["status"]
+
+
+def test_live_owner_stale_claim_gets_recovered(monkeypatch):
+    """A live-owned claim older than the derived bound is released as ``unknown``; the bound
+    follows the configured timeouts rather than a fixed wall-clock constant."""
+    monkeypatch.setenv("HERMES_CRON_TIMEOUT", "600")
+    bound = executions_mod._live_owner_stale_after_seconds()
+    assert bound is not None
+    assert bound == pytest.approx(7200.0)  # max(3×600, 3600 script default, 7200 floor)
+
+    eid = _seed_running("job-stale", age_seconds=bound + 60)
+    assert recover_interrupted_executions() == 1
+    assert _status(eid) == "unknown"
+
+    # A larger inactivity timeout widens the bound: the same age is no longer stale.
+    monkeypatch.setenv("HERMES_CRON_TIMEOUT", "3600")
+    assert executions_mod._live_owner_stale_after_seconds() == pytest.approx(10800.0)
+    eid2 = _seed_running("job-long", age_seconds=bound + 60)
+    assert recover_interrupted_executions() == 0
+    assert _status(eid2) == "running"
+
+
+@pytest.mark.parametrize("timeout", ["600", "0", "nan", "inf"])
+def test_live_owner_recent_claim_not_recovered(monkeypatch, timeout):
+    """A live owner with a recent claim is left alone; with an unlimited (0) or non-finite
+    inactivity timeout there is no bound to derive, so even an ancient live-owned claim is
+    left alone (fail closed)."""
+    monkeypatch.setenv("HERMES_CRON_TIMEOUT", timeout)
+    recent = _seed_running("job-recent", age_seconds=60, pid=88888)
+    ancient = _seed_running("job-ancient", age_seconds=30 * 24 * 3600, pid=77777)
+
+    recovered = recover_interrupted_executions()
+
+    assert _status(recent) == "running"
+    if timeout == "600":
+        assert recovered == 1 and _status(ancient) == "unknown"
+    else:
+        assert executions_mod._live_owner_stale_after_seconds() is None
+        assert recovered == 0 and _status(ancient) == "running"

tests/cron/test_stale_running_recovery_115692.py

@@ -1,63 +0,0 @@
-"""A live-but-deadlocked worker passes ``_owner_is_live()`` and never gets
-reclaimed. The wall-clock guard catches that: a claimed/running execution
-older than ``STALE_RUNNING_CLAIM_TIMEOUT_SECONDS`` is marked unknown even
-though the owner PID exists (#115692)."""
-import os
-import sqlite3
-import time
-import uuid
-
-import pytest
-
-from cron.executions import (
-    create_execution,
-    recover_interrupted_executions,
-    _transaction,
-)
-
-
-def _seed_running(job_id, claimed_at, pid=99999, proc="ext-proc"):
-    with _transaction() as conn:
-        eid = uuid.uuid4().hex
-        conn.execute(
-            """INSERT INTO executions
-               (id, job_id, status, source, process_id, pid, process_started_at, claimed_at)
-               VALUES (?, ?, 'running', 'cron', ?, ?, ?, ?)""",
-            (eid, job_id, proc, pid, int(time.time()), claimed_at),
-        )
-        return eid
-
-
-def test_live_owner_stale_claim_gets_recovered(tmp_path, monkeypatch):
-    """A claimed/running row whose owner is alive but stale past the wall-clock
-    guard must be marked unknown — a live-but-deadlocked worker otherwise
-    blocks every future fire forever (#115692)."""
-    monkeypatch.setattr("cron.executions._owner_is_live", lambda pid, sat: True)
-
-    # claimed_at = 3h ago (well past the 2h stale-claim timeout)
-    old = "2026-09-17T00:00:00"
-    eid = _seed_running("job-stale", claimed_at=old, pid=99999)
-
-    assert recover_interrupted_executions() >= 1
-
-    with _transaction() as conn:
-        row = conn.execute(
-            "SELECT status FROM executions WHERE id=?", (eid,)
-        ).fetchone()
-        assert row["status"] == "unknown"
-
-
-def test_live_owner_recent_claim_not_recovered(tmp_path, monkeypatch):
-    """A live owner with a recent claim must NOT be recovered (normal running)."""
-    monkeypatch.setattr("cron.executions._owner_is_live", lambda pid, sat: True)
-
-    now = time.strftime("%Y-%m-%dT%H:%M:%S")
-    eid = _seed_running("job-recent", claimed_at=now, pid=88888)
-
-    assert recover_interrupted_executions() == 0
-
-    with _transaction() as conn:
-        row = conn.execute(
-            "SELECT status FROM executions WHERE id=?", (eid,)
-        ).fetchone()
-        assert row["status"] == "running"