fix(lik-ui): keep the scheduled-runs DB connection alive across a long run

The scheduled-runs cron has failed for three days with `PoolTimeout: couldn't
get a connection after 5.00 sec` — always at the terminal `complete_run`, and
always on a run long enough (33–40 min) to leave the pooled connection idle for
that whole stretch. The agent runs themselves completed; only recording the
outcome failed, which aborted the scan and left the row to be reclaimed as
`abandoned` (and the completed work repeated) on the next tick.

The path to the public Postgres endpoint silently drops a long-idle connection
without a FIN, so the checkout-time check can block on TCP retransmission far
past the 5s budget and then report a timeout with no attempt left to reconnect.

- Enable TCP keepalives (plus `connect_timeout`) on the conninfo: the connection
  now survives the idle stretch, and a dead one is reported in seconds instead
  of blocking.
- Make the pool's checkout timeout configurable; the scanner uses 30s so a
  stale connection can be replaced, while a web request still fails fast.
- Retry a run's terminal write, and if it still cannot be recorded, finish the
  remaining rows and exit non-zero instead of raising out of the scan.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: yoomlam

lik-ui/scripts/run_scheduled.py

@@ -37,6 +37,13 @@
 # the whole CI job. The job's timeout-minutes must exceed max_runtime_s + this margin.
 _HARD_TIMEOUT_MARGIN_S = 120
 
+# How long a DB connection checkout may take in this process (see ``main``).
+_DB_CHECKOUT_TIMEOUT_S = 30
+
+# Waits between attempts at a run's terminal DB write (see ``_record``). One entry per attempt; the
+# last entry's wait is never used. Short enough to stay well inside the job's timeout.
+_RECORD_RETRY_BACKOFF_S = (2, 5, 10, 0)
+
 
 class _HardRuntimeBackstop:
     """Process-level runtime guard: guarantees a claimed row terminates and is recorded even if the
@@ -69,6 +76,33 @@ def _chat_url(base_url: str, session_id: str) -> str:
     return f"{base}/chat/{session_id}" if base else session_id
 
 
+def _record(write, *args, **kwargs) -> bool:
+    """Perform a run's terminal DB write, retrying a transient connection failure. Returns whether
+    it succeeded.
+
+    This write happens after a long stretch of no DB traffic (the whole agent run), so it is the
+    one most exposed to a dropped connection to the public Postgres endpoint. A single blip here
+    used to raise out of the scan, aborting every remaining row and leaving this row's outcome
+    unrecorded (it is then reclaimed as ``abandoned`` on the next scan, and the completed run is
+    needlessly repeated). Retrying, and at worst continuing, keeps one blip from costing the scan."""
+    for attempt, backoff_s in enumerate(_RECORD_RETRY_BACKOFF_S, start=1):
+        try:
+            write(*args, **kwargs)
+            return True
+        except Exception as exc:  # noqa: BLE001 - any DB failure here is worth another try
+            last = attempt == len(_RECORD_RETRY_BACKOFF_S)
+            print(
+                f"[scheduled] recording the outcome failed (attempt {attempt}"
+                f"{'' if last else ', retrying'}): {exc}",
+                file=sys.stderr,
+                flush=True,
+            )
+            if last:
+                return False
+            time.sleep(backoff_s)
+    return False
+
+
 def run_due_schedules(store, sessions_client, vault_client, agents, base_url="") -> tuple[int, int]:
     """Claim and run every due schedule. Returns ``(ran, failed)``. One row's failure never
     aborts the scan — each is isolated and its outcome recorded on its own row. ``base_url`` is
@@ -91,7 +125,7 @@ def _log_session(session_id, run_id=run_id, agent=row["agent_name"]):
                 )
         except Exception as exc:  # noqa: BLE001 - never let one row (incl. a backstop trip) abort the scan
             duration_s = round(time.monotonic() - started)
-            store.complete_run(run_id, "failed", str(exc), None, duration_s=duration_s)
+            _record(store.complete_run, run_id, "failed", str(exc), None, duration_s=duration_s)
             failed += 1
             print(f"[scheduled] run {run_id} raised after {duration_s}s, recorded failed: {exc}", file=sys.stderr)
             continue
@@ -100,14 +134,23 @@ def _log_session(session_id, run_id=run_id, agent=row["agent_name"]):
         if outcome.status == AUTH_LAPSED:
             # Pause instead of advancing — re-running every cadence would just re-fail until the
             # owner re-authenticates interactively. The Settings badge shows "needs re-auth".
-            store.pause_and_flag(run_id, "needs_reauth", error=outcome.error, duration_s=duration_s)
+            recorded = _record(
+                store.pause_and_flag, run_id, "needs_reauth", error=outcome.error, duration_s=duration_s
+            )
         else:
-            store.complete_run(run_id, outcome.status, outcome.error, outcome.skipped or None, duration_s=duration_s)
+            recorded = _record(
+                store.complete_run, run_id, outcome.status, outcome.error, outcome.skipped or None,
+                duration_s=duration_s,
+            )
 
         ran += 1
-        if outcome.status in _FAILURE_STATUSES:
+        # An unrecorded outcome is a job-level failure even when the run itself succeeded: the row
+        # stays in flight and the next scan will re-run work that already completed.
+        if outcome.status in _FAILURE_STATUSES or not recorded:
             failed += 1
         note = f" skipped={len(outcome.skipped)}" if outcome.skipped else ""
+        if not recorded:
+            note += " (outcome NOT recorded — the row will be reclaimed next scan)"
         where = f" {_chat_url(base_url, outcome.session_id)}" if outcome.session_id else ""
         # Duration is logged and persisted (last_duration_s) so max_runtime can be tuned per agent
         # from real run times instead of guessed.
@@ -128,7 +171,10 @@ def main() -> int:
             file=sys.stderr,
         )
         return 1
-    store = Store(Database(settings.conninfo))
+    # A generous checkout timeout (vs. the web app's default): this process leaves its connection
+    # idle for a whole agent run, so re-establishing one before the terminal write is worth the
+    # wait — no user is waiting on a page here, and losing the write costs a repeated run.
+    store = Store(Database(settings.conninfo, checkout_timeout=_DB_CHECKOUT_TIMEOUT_S))
     agents_client = build_agents_client(settings)
     agents = resolve_agent_options(settings, agents_client)
     sessions_client = build_sessions_client(settings)

lik-ui/src/lik_ui/db.py

@@ -18,20 +18,28 @@ class Database:
     """Owns the Postgres connection pool. The app holds one; call sites borrow
     connections through ``connection()`` and never open their own."""
 
-    def __init__(self, conninfo: str, *, min_size: int = 1, max_size: int = 4):
+    def __init__(self, conninfo: str, *, min_size: int = 1, max_size: int = 4, checkout_timeout: float = 5):
         # check on checkout: the scheduled-runs scanner borrows a connection for claim_due_runs,
         # then holds the pool idle for a whole multi-minute agent run (no DB traffic) before the
         # terminal complete_run. The public Postgres/network silently drops that idle connection,
         # so an unchecked pool would hand back a dead socket and the final write would fail with
         # "SSL error: unexpected eof while reading" — losing the run's recorded outcome.
         # check_connection validates (and the pool reconnects) on checkout, so a stale connection
-        # is replaced before use instead of erroring mid-write.
+        # is replaced before use instead of erroring mid-write. Keepalives in the conninfo bound
+        # how long that check can block on a dead socket (see Settings.conninfo).
+        #
+        # ``checkout_timeout`` must leave room for the check to fail AND a replacement connection to
+        # be opened (the pool backs off ~1s between check attempts, and a fresh TLS connection to
+        # the public endpoint takes a moment): too small a budget turns a merely stale connection
+        # into a PoolTimeout. The default suits serving a web request (waiting longer than that just
+        # stalls a page); the scheduled-runs scanner raises it, because there its terminal write is
+        # worth waiting out — losing it means re-running a whole completed agent run.
         self.pool = ConnectionPool(
             conninfo,
             min_size=min_size,
             max_size=max_size,
             open=True,
-            timeout=5,
+            timeout=checkout_timeout,
             check=ConnectionPool.check_connection,
         )
 

lik-ui/src/lik_ui/settings.py

@@ -269,9 +269,16 @@ def agent_roster(self) -> list[AgentRosterEntry]:
 
     @property
     def conninfo(self) -> str:
+        # TCP keepalives are not optional here: the scheduled-runs scanner leaves its pooled
+        # connection idle for a whole multi-minute agent run before the terminal write, and the
+        # path to the public Postgres endpoint silently drops a long-idle connection (no FIN). With
+        # keepalives the connection stays alive across that idle, and if it does die the kernel
+        # reports it in ~keepalives_idle + count*interval instead of letting the next query block
+        # on TCP retransmission for minutes. connect_timeout bounds re-connecting the same way.
         return (
             f"host={self.db_host} port={self.db_port} dbname={self.db_name} "
-            f"user={self.db_user} password={self.db_password} sslmode={self.db_sslmode}"
+            f"user={self.db_user} password={self.db_password} sslmode={self.db_sslmode} "
+            "connect_timeout=10 keepalives=1 keepalives_idle=60 keepalives_interval=10 keepalives_count=3"
         )
 
     @property

lik-ui/tests/test_run_scheduled_script.py

@@ -108,6 +108,48 @@ def _blocks(*_args, **_kwargs):
     assert row["last_duration_s"] is not None  # recorded even for the backstop path
 
 
+def test_terminal_write_is_retried_after_a_transient_db_failure(store, monkeypatch):
+    """The terminal write happens after a long stretch of no DB traffic, so it is the write most
+    exposed to a dropped connection. A blip must be retried, not lost."""
+    import scripts.run_scheduled as rs
+
+    monkeypatch.setattr(rs, "_RECORD_RETRY_BACKOFF_S", (0, 0, 0))
+    a = store.upsert_user("a@navapbc.com")
+    store.create_scheduled_run(a["id"], AGENT, "sync", timedelta(hours=1))
+    real_complete = store.complete_run
+    calls = {"n": 0}
+
+    def _flaky(*args, **kwargs):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            raise RuntimeError("couldn't get a connection after 5.00 sec")
+        return real_complete(*args, **kwargs)
+
+    monkeypatch.setattr(store, "complete_run", _flaky)
+    ran, failed = rs.run_due_schedules(store, FakeSessions([[{"type": "done"}]]), FakeVault(), _agents())
+    assert (ran, failed) == (1, 0)  # the retry recorded it, so this is not a job failure
+    row = store.list_scheduled_runs(a["id"])[0]
+    assert row["last_status"] == "success"
+    assert row["started_at"] is None  # advanced, not left in flight
+
+
+def test_unrecordable_outcome_counts_as_failure_without_aborting_the_scan(store, monkeypatch):
+    """When the outcome cannot be persisted at all, the scan still finishes the remaining rows and
+    the job exits non-zero (the row stays in flight and is reclaimed next scan)."""
+    import scripts.run_scheduled as rs
+
+    monkeypatch.setattr(rs, "_RECORD_RETRY_BACKOFF_S", (0, 0))
+    a = store.upsert_user("a@navapbc.com")
+    store.create_scheduled_run(a["id"], AGENT, "sync", timedelta(hours=1))
+
+    def _always_fails(*_args, **_kwargs):
+        raise RuntimeError("couldn't get a connection after 5.00 sec")
+
+    monkeypatch.setattr(store, "complete_run", _always_fails)
+    ran, failed = rs.run_due_schedules(store, FakeSessions([[{"type": "done"}]]), FakeVault(), _agents())
+    assert (ran, failed) == (1, 1)
+
+
 def test_run_due_schedules_auth_lapse_pauses_and_counts_failure(store):
     a = store.upsert_user("a@navapbc.com")
     store.create_scheduled_run(a["id"], AGENT, "sync", timedelta(hours=1))