feat(scheduler): a channel's "get full history" backfill is a recorded run, and a row knows its last three runs

The one-channel deep backfill ran as a FastAPI BackgroundTask: what it fetched, what it spent and
how it failed reached only the log. It is now a one-shot run on the scheduler (`channel_backfill`,
trigger ad_hoc, actor = the user, target = the channel) with its quota tagged to the run. Its
per-channel exclusion with the scheduled backfill and the two-run cap are unchanged, and a run
they turn away says why in its result instead of returning nothing. A failure is no longer
swallowed: the run records it.

The scheduler snapshot carries each job's last three runs (one windowed query), so the page can
show an interrupted or skipped run instead of the latest outcome alone.

backend/tests/test_scheduler.py

@@ -146,6 +146,28 @@ def test_the_last_run_survives_a_restart(db):
     assert plex["last_finished"] is not None
 
 
+def test_a_row_shows_its_last_three_runs_newest_first_an_interrupted_one_included(db):
+    """The row said only how the LAST run went; an interrupted run read as "no run recorded"."""
+    oldest = jobruns.start("shorts", RunTrigger.SCHEDULED, None)
+    jobruns.interrupt_leftovers()
+    scheduler._job("shorts", lambda _db: {"probed": 1})
+    scheduler._job("shorts", lambda _db: {"probed": 2})
+    scheduler._job("shorts", lambda _db: 1 / 0)
+    scheduler._job("rss_poll", lambda _db: {"new": 0})  # another job's run stays on its own row
+
+    recent = _snapshot_job(db, "shorts")["recent"]
+    assert [r["status"] for r in recent] == ["error", "ok", "ok"]
+    assert oldest not in [r["id"] for r in recent]  # the fourth-newest is not shown
+    assert recent[0]["trigger"] == "scheduled" and recent[0]["finished_at"] is not None
+    assert [r["status"] for r in _snapshot_job(db, "rss_poll")["recent"]] == ["ok"]
+
+    jobruns.start("shorts", RunTrigger.SCHEDULED, None)
+    jobruns.interrupt_leftovers()
+    shorts = _snapshot_job(db, "shorts")
+    assert shorts["status"] == "interrupted"
+    assert [r["status"] for r in shorts["recent"]] == ["interrupted", "error", "ok"]
+
+
 def test_a_paused_sync_is_a_skipped_run(db):
     state.set_sync_paused(db, True)
     db.commit()
@@ -286,6 +308,51 @@ def test_run_now_goes_through_the_scheduler_even_with_interval_jobs_off(
     assert (run.trigger, run.actor_id, run.result) == ("manual", admin.id, {"probed": 3})
 
 
+def test_an_ad_hoc_run_is_the_users_run_of_its_target_with_its_spend_and_failure(
+    db, make_user, live_scheduler
+):
+    """The one-channel deep backfill ran as a BackgroundTask nothing recorded: its spend counted as
+    nobody's run and a failure lived only in the log."""
+    user = make_user("adhoc@example.com")
+
+    def work():
+        quota.record_usage(3)
+        return {"videos_added": 50}
+
+    scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, work, actor_id=user.id, target="UC_a")
+    run = _wait_finished(db, "channel_backfill")
+    assert (run.trigger, run.actor_id, run.target) == ("ad_hoc", user.id, "UC_a")
+    assert (run.status, run.result, run.quota_units) == ("ok", {"videos_added": 50}, 3)
+
+    def fail():
+        raise RuntimeError("quota exceeded")
+
+    scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, fail, actor_id=user.id, target="UC_b")
+    deadline = time.monotonic() + 10
+    while time.monotonic() < deadline and len(_finished(db, "channel_backfill")) < 2:
+        time.sleep(0.05)
+    failed = next(r for r in _runs(db, "channel_backfill") if r.target == "UC_b")
+    assert (failed.status, failed.error) == ("error", "quota exceeded")
+
+
+def _finished(db, job) -> list[JobRun]:
+    return [r for r in _runs(db, job) if r.status != "running"]
+
+
+def test_a_second_ad_hoc_request_before_the_first_starts_is_that_same_run(
+    db, make_user, live_scheduler
+):
+    user = make_user("twice@example.com")
+    live_scheduler.pause()  # hold the one-shots so the second request finds the first pending
+    for target in ("UC_x", "UC_x", "UC_y"):
+        scheduler.run_ad_hoc(scheduler.CHANNEL_BACKFILL, dict, actor_id=user.id, target=target)
+    assert sorted(j.id for j in live_scheduler.get_jobs()) == [
+        "channel_backfill@UC_x",
+        "channel_backfill@UC_y",
+    ]
+    live_scheduler.remove_all_jobs()  # never run: a run finishing after the test would land in the next
+
+
 def test_a_manual_run_leaves_no_actor_behind_on_its_thread(db, make_user):
     """The scheduler's pool threads are reused: an actor a manual run left set made the next
     scheduled run on that thread an admin's "manual" run, with a notice in their inbox."""