fix(jobs): make last-run writes atomic and drain them properly on shutdown

Two review findings on 22bee28, both reproduced before fixing.

[1] Concurrent writes could lose or rewind a row. record_finish did
select-then-insert-or-update, and pipelines are separate scheduler jobs that can
overlap while sharing step ids -- data_collector belongs to all four. Reproduced
both halves: two sessions that SELECT before either INSERTs make the second
commit raise IntegrityError, which _persist_job_run swallows, so the run
silently vanishes; and a later write carrying an OLDER finished_at rewound the
row from 12:00 back to 09:00, dragging the status with it, so the panel would
report a stale outcome as the latest.

Now a single atomic INSERT ... ON CONFLICT (job_name) DO UPDATE, guarded by
WHERE job_run_state.finished_at < excluded.finished_at so an older completion
can never overwrite a newer one. Dialect-specific because prod is Postgres and
tests are SQLite; both support it (SQLite >= 3.24, ours is 3.45). updated_at is
set explicitly, as the model's onupdate hook does not fire for a core upsert,
and get_map now uses populate_existing since core writes leave any
previously-loaded ORM instance stale in the identity map.

[2] Shutdown could dispose the engine underneath a pending write.
scheduler.shutdown(wait=False) returns before APScheduler dispatches its
completion events, and those events are what create persist tasks -- so a single
snapshot of the task set missed writes still to be queued. flush_job_run_persists
now settles briefly for pending callbacks, then drains in a loop until the set
stays empty, with the deadline still bounding total shutdown time. Left
shutdown(wait=False) alone deliberately: waiting would block a deploy restart
behind a long-running scan.

Six regression tests: interleaved first writes, older-never-rewinds,
newer-still-wins, a task queued mid-drain, prompt return when idle, and giving
up rather than hanging shutdown.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/unit/test_job_run_persistence.py

@@ -164,3 +164,108 @@ class TestListJobsSplitsLiveFromPersisted:
         assert job["runtime_status"] == "idle"
         assert job["running"] is False
         scheduler.remove_all_jobs()
+
+
+class TestConcurrentWrites:
+    """Pipelines are separate scheduler jobs that can overlap, and they share
+    step ids — data_collector belongs to all four."""
+
+    async def test_interleaved_first_writes_do_not_collide(self):
+        """Both sessions SELECT before either INSERTs: select-then-insert lost
+        this race with an IntegrityError, and the caller swallows it."""
+        async with _test_session_factory() as a, _test_session_factory() as b:
+            await job_run_store.record_finish(
+                a, "data_collector",
+                {"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00"},
+            )
+            await job_run_store.record_finish(
+                b, "data_collector",
+                {"status": "completed", "finished_at": "2026-08-08T10:00:01+00:00"},
+            )
+            await a.commit()
+            await b.commit()  # must not raise
+
+        rows = await _committed_rows()
+        assert rows["data_collector"].status == "completed"
+
+    async def test_an_older_finish_never_rewinds_the_row(self):
+        """A slower pipeline finishing an older run last must not overwrite a
+        newer outcome with a stale one."""
+        async with _test_session_factory() as s:
+            await job_run_store.record_finish(
+                s, "alerts",
+                {"status": "completed", "finished_at": "2026-08-08T12:00:00+00:00"},
+            )
+            await s.commit()
+            await job_run_store.record_finish(
+                s, "alerts",
+                {"status": "error", "finished_at": "2026-08-08T09:00:00+00:00", "message": "stale"},
+            )
+            await s.commit()
+
+        row = (await _committed_rows())["alerts"]
+        assert row.finished_at.isoformat().startswith("2026-08-08T12:00")
+        assert row.status == "completed"
+        assert row.message is None
+
+    async def test_a_newer_finish_still_wins(self):
+        async with _test_session_factory() as s:
+            await job_run_store.record_finish(
+                s, "rr_scanner",
+                {"status": "completed", "finished_at": "2026-08-08T09:00:00+00:00"},
+            )
+            await s.commit()
+            await job_run_store.record_finish(
+                s, "rr_scanner",
+                {"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"},
+            )
+            await s.commit()
+
+        row = (await _committed_rows())["rr_scanner"]
+        assert row.status == "error"
+        assert row.message == "boom"
+
+
+class TestShutdownDrain:
+    async def test_drains_a_task_queued_after_the_flush_starts(self):
+        """scheduler.shutdown(wait=False) returns before APScheduler dispatches
+        its completion events, so writes can appear mid-drain. Snapshotting the
+        task set once would miss them and dispose the engine underneath."""
+        import asyncio
+
+        done: list[str] = []
+
+        async def slow_first():
+            await asyncio.sleep(0.02)
+            done.append("first")
+            # Queued only once the first write is already finishing.
+            sched._persist_tasks.add(asyncio.get_running_loop().create_task(late()))
+
+        async def late():
+            await asyncio.sleep(0.02)
+            done.append("late")
+
+        sched._persist_tasks.clear()
+        sched._persist_tasks.add(asyncio.get_running_loop().create_task(slow_first()))
+
+        await sched.flush_job_run_persists(timeout=2.0)
+
+        assert done == ["first", "late"]
+        sched._persist_tasks.clear()
+
+    async def test_returns_promptly_when_there_is_nothing_to_drain(self):
+        sched._persist_tasks.clear()
+        await sched.flush_job_run_persists(timeout=2.0)
+
+    async def test_gives_up_rather_than_hanging_shutdown(self):
+        import asyncio
+
+        async def never():
+            await asyncio.sleep(30)
+
+        sched._persist_tasks.clear()
+        task = asyncio.get_running_loop().create_task(never())
+        sched._persist_tasks.add(task)
+        await sched.flush_job_run_persists(timeout=0.15)  # returns, does not hang
+        task.cancel()
+        sched._persist_tasks.clear()