fix: an idle GPU agent could not check in, so the roster called it stopped

Operator, 2026-09-23: *"I'm running the gpu agent on my device and it
currently reads as 'offline' but it's running and has checked in recently."*

It had checked in — twelve minutes ago. Two cadences that never agreed:

    idle lease poll ceiling   900s   agent/fc_agent/worker.py (sleep mode)
    heartbeat while idle      never  gated on holding leases
    roster "stopped" after    300s   api/system_health.py

The roster records an agent check-in on `lease` and `heartbeat`. The heartbeat
loop was gated on `if ids:`, so an agent holding no leases sent nothing at
all — leaving the lease poll as the only check-in, and sleep mode backs that
off exponentially to a 900s ceiling. 900 against 300: an IDLE agent was
structurally guaranteed to read as stopped. Nothing was broken; nothing was
misconfigured; the two halves simply disagreed.

Not a recent regression. Sleep mode landed 2026-07-02; the roster adopted the
lease as its check-in on 2026-09-02 — *"A lease IS the check-in … Recorded on
the call that was already happening"* — without noticing that the call it was
piggybacking on had been deliberately slowed ten weeks earlier.

The heartbeat now sends whether or not it holds leases. An empty one extends
nothing (`id.in_([])` matches no rows) and costs one small POST every 45s —
against the 6/min lease poll sleep mode exists to avoid, that is not a cadence
worth protecting, and it is what makes "is the agent alive" answerable at all.

Still gated on `self._running`: a worker that has been stopped is not checking
in for work, and reporting it as present would be a different lie.

Two things I could NOT determine from the code, both needing the live table:
whether a stale `agent:agent` row exists from an older build that omitted
`agent_id` (the server defaults it), and whether changing `AGENT_ID` has ever
stranded an abandoned row — nothing prunes `service_seen`, so either would sit
there reading "stopped" forever.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

agent/fc_agent/worker.py

@@ -342,15 +342,44 @@ class Worker:
 
     # --- background loops ---------------------------------------------------
     def _heartbeat_loop(self) -> None:
-        """Keep every held lease alive so buffered jobs waiting on the GPU aren't
-        reclaimed by curator's 180s TTL. Errors are swallowed by client.heartbeat;
-        a reclaimed lease just re-leases elsewhere — never fatal."""
+        """Keep every held lease alive, and say we are here even when holding none.
+
+        Leases: buffered jobs waiting on the GPU would otherwise be reclaimed by
+        curator's 180s TTL. Errors are swallowed by client.heartbeat; a reclaimed
+        lease just re-leases elsewhere — never fatal.
+
+        ## Why this sends with an EMPTY list rather than skipping
+
+        Curator's roster records a check-in on this call (and on `lease`), and
+        calls an agent stopped after 300s of silence. This loop used to be
+        gated on `if ids:` — so an agent holding no leases sent nothing at all,
+        and the only check-in left was the lease poll, which sleep mode backs
+        off exponentially to a 900s ceiling (see IDLE_POLL_MAX_SECONDS).
+
+        900 against 300: an IDLE agent was structurally guaranteed to read as
+        stopped. Operator, 2026-09-23: *"I'm running the gpu agent on my device
+        and it currently reads as 'offline' but it's running and has checked in
+        recently."* It had — twelve minutes ago, partway up the backoff ladder.
+
+        The two halves were written ten weeks apart and never reconciled: sleep
+        mode landed 2026-07-02, and the roster adopted the lease as its
+        check-in on 2026-09-02 without noticing the call it was piggybacking on
+        had been deliberately slowed.
+
+        An empty heartbeat extends nothing (`id.in_([])` matches no rows) and
+        costs one small POST every 45s — against the 6/min lease poll sleep
+        mode exists to avoid, that is not a cadence worth protecting, and it is
+        what makes "is the agent alive" answerable at all.
+
+        Still gated on `self._running`: a worker that has been stopped is not
+        checking in for work, and reporting it as present would be a different
+        lie.
+        """
         while True:
             if self._running:
                 with self._held_lock:
                     ids = list(self._held)
-                if ids:
-                    self.client.heartbeat(ids)
+                self.client.heartbeat(ids)
             time.sleep(HEARTBEAT_INTERVAL)
 
     def _queue_poll_loop(self):

backend/app/api/workers.py

@@ -23,7 +23,13 @@ from datetime import UTC, datetime
 from quart import Blueprint, jsonify, request
 
 from ..extensions import get_session
-from ..services.worker_control import LaneUpdateRefused, lane_view, set_lane
+from ..services.worker_control import (
+    LaneUpdateRefused,
+    lane_settings,
+    lane_view,
+    push_lane_cap,
+    store_lane_cap,
+)
 from ..services.worker_lanes import LANES_BY_NAME
 from ._responses import error_response as _bad
 
@@ -41,8 +47,14 @@ async def list_lanes():
     reply would show them the value from before their own change and read as
     the control having failed.
     """
+    # The session closes BEFORE the broker work. Holding a Postgres connection
+    # across a celery inspect is what made this page block the whole site —
+    # see `worker_control.LaneSettings`. This endpoint polls every 15s and the
+    # inspect budget is 11s, so each poll was pinning a connection for most of
+    # the interval.
     async with get_session() as session:
-        lanes = await lane_view(session)
+        settings = await lane_settings(session)
+    lanes = await lane_view(settings)
     return jsonify({
         "lanes": lanes,
         "fetched_at": datetime.now(UTC).isoformat(),
@@ -85,9 +97,13 @@ async def update_lane(name: str):
     if not isinstance(value, int) or isinstance(value, bool):
         return _bad("invalid_body", detail="slots_cap must be an integer")
 
+    # Store, close the session, THEN push. Same reason as the GET above, and
+    # more sharply here: a cap change could do three broker round trips, all
+    # of them previously with a connection held.
     async with get_session() as session:
         try:
-            result = await set_lane(session, lane, slots_cap=value)
+            was_cap = await store_lane_cap(session, lane, value)
         except LaneUpdateRefused as exc:
             return _bad("refused", detail=str(exc))
+    result = await push_lane_cap(lane, value, was_cap=was_cap)
     return jsonify(result)

backend/app/services/worker_control.py

@@ -334,14 +334,53 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]:
     return rows
 
 
-async def lane_view(session: AsyncSession) -> list[dict]:
+@dataclass
+class LaneSettings:
+    """What the DATABASE knows about the lanes — read and finished with before
+    anything touches the broker.
+
+    This exists because holding a Postgres connection across a celery round
+    trip is what made the System tab block the whole site (operator,
+    2026-09-23: *"something about changing the cap number is blocking to the
+    website"*).
+
+    `lane_view` used to take the session and keep it open through an inspect
+    whose budget is eleven seconds — and that page polls every fifteen. With a
+    lane not answering, every inspect ran to nearly its full budget, so each
+    poll pinned a connection for ten seconds. SQLAlchemy's default pool is
+    five connections plus ten overflow; a couple of browser tabs, the health
+    endpoint doing the same thing, and a cap change adding two more inspects
+    exhausts that, and every OTHER request then waits on a connection.
+
+    Nothing was slow in itself. The slowness was a scarce resource held across
+    it, which is why it surfaced as the whole site stalling rather than as one
+    slow page.
+    """
+
+    caps: dict[str, int]
+    oldest_by_queue: dict[str, datetime]
+
+
+async def lane_settings(session: AsyncSession) -> LaneSettings:
+    """Every DB read the lane view needs, in one short-lived session."""
+    rows = await _rows_by_name(session)
+    return LaneSettings(
+        caps={name: row.slots_cap for name, row in rows.items()},
+        oldest_by_queue=await _oldest_running_by_queue(session),
+    )
+
+
+async def lane_view(settings: LaneSettings) -> list[dict]:
     """Every lane: what is configured, what is live, what it may grow to.
 
-    One call rather than making the UI join three sources. `pending` is the
-    honest backlog — Redis depth PLUS reserved — because celery prefetches and
-    LLEN alone reads 0 while a worker holds tasks in memory.
+    Takes the settings rather than a session ON PURPOSE — see `LaneSettings`.
+    Everything below this line is broker work, and no database connection is
+    held while it happens.
+
+    `pending` is the honest backlog — Redis depth PLUS reserved — because
+    celery prefetches and LLEN alone reads 0 while a worker holds tasks in
+    memory.
     """
-    rows = await _rows_by_name(session)
     # A deadline, because this is a request path and `to_thread` on its own is
     # an await with no bound (rule 156). `inspect_lanes_sync` never raises and
     # every inner call has its own timeout, so the only way past the budget is
@@ -359,12 +398,12 @@ async def lane_view(session: AsyncSession) -> list[dict]:
         )
         live = {lane.name: LaneLiveState() for lane in LANES}
     depths = await asyncio.to_thread(_queue_depths_sync)
-    oldest = await _oldest_running_by_queue(session)
+    oldest = settings.oldest_by_queue
 
     now = datetime.now(UTC)
     out = []
     for lane in LANES:
-        row = rows[lane.name]
+        cap = settings.caps[lane.name]
         state = live[lane.name]
         # None for a queue the broker did not answer for, which must not be
         # silently summed as zero — an unknown depth is not an empty one.
@@ -376,11 +415,11 @@ async def lane_view(session: AsyncSession) -> list[dict]:
             "name": lane.name,
             "display_name": lane.display_name,
             "queues": list(lane.queues),
-            "slots_cap": row.slots_cap,
+            "slots_cap": cap,
             "ceiling": derived_ceiling(lane),
             # DERIVED, never stored. A cap of zero means no consumers, so
             # "off" and "may use no workers" cannot disagree.
-            "enabled": row.slots_cap > 0,
+            "enabled": cap > 0,
             "memory_bound": lane.memory_bound,
             "optional": lane.optional,
             # What raising this lane's cap will download, so the UI can say
@@ -485,30 +524,14 @@ class LaneUpdateRefused(ValueError):
     the UI shows — a greyed control with no explanation reads as a bug."""
 
 
-async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict:
-    """Store the operator's cap for `lane`, then make the live lane obey it.
+async def store_lane_cap(
+    session: AsyncSession, lane: Lane, slots_cap: int,
+) -> int:
+    """Validate and store the cap. Returns the PREVIOUS cap. DB only.
 
-    ONE value, since 2026-09-23. It used to take `slots`, `slots_cap`,
-    `enabled` and `autoscale`, which was four ways of saying two things — and
-    two of them were the caller's job to keep in agreement with each other.
-
-    ## What happens live, and what does not
-
-    A cap CHANGE is pushed immediately in one direction only: lowering it
-    shrinks the pool now, because a cap the operator just lowered should not
-    be exceeded for up to a minute. Raising it does NOT grow the pool here —
-    a cap is permission, not a request, and growing on permission would put
-    slots on a lane with nothing to do. The sizing pass adds them on its next
-    tick if there is work, which is the whole point of it being always on.
-
-    Consumers follow the cap in both directions and immediately: zero means
-    off, and off must take effect when it is asked for.
-
-    A failed PUSH is not a failed setting. The value is saved either way and
-    the sizing pass carries it within a minute; the result says `applied:
-    false` with a reason so the UI can say "saved, not yet live" rather than
-    "that didn't work" (lesson #4202 — a live change that does not survive,
-    with nothing saying so).
+    Split from the live push for the reason `LaneSettings` gives at length: a
+    Postgres connection must not be held across a celery round trip. Everything
+    here is fast and finished with before `push_lane_cap` starts.
     """
     rows = await _rows_by_name(session)
     row = rows[lane.name]
@@ -522,11 +545,40 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
             f"({ceiling} for {lane.display_name})"
         )
 
-    was_on = row.slots_cap > 0
+    was_cap = row.slots_cap
     row.slots_cap = slots_cap
     await session.commit()
+    # The previous value, because the push needs the DIRECTION: lowering a cap
+    # has to reach the running lane now, and raising one has nothing to say.
+    return was_cap
 
-    now_on = slots_cap > 0
+
+async def push_lane_cap(lane: Lane, slots_cap: int, *, was_cap: int) -> dict:
+    """Make the running lane obey a cap that is already stored. NO database.
+
+    ## What is pushed, and what is not
+
+    Consumers follow the cap immediately in BOTH directions: zero means off,
+    and off must take effect when it is asked for rather than up to a minute
+    later.
+
+    The pool is only ever pushed DOWNWARD. Raising a cap is permission, not a
+    request — growing on permission would put workers on a lane with nothing
+    to do — so the sizing pass spends it on its next tick if there is work.
+    That also makes the common case (raising a cap) free: no broker round trip
+    AT ALL, which is the difference between a control that answers instantly
+    and one that takes ten seconds. Keyed on the previous cap rather than on
+    "is it on" — the first cut only knew on/off, so it inspected on every
+    raise to find out whether the pool needed lowering, and the control it was
+    meant to make instant still waited out an inspect.
+
+    A failed push is not a failed setting. The value is already stored and the
+    sizing pass carries it within a minute; `applied: false` with a reason
+    lets the UI say "saved, not yet live" rather than "that didn't work"
+    (lesson #4202 — a live change that does not survive, with nothing saying
+    so).
+    """
+    was_on, now_on = was_cap > 0, slots_cap > 0
     applied, error = True, None
     if now_on != was_on:
         applied, error = await asyncio.to_thread(set_lane_enabled_sync, lane, now_on)
@@ -536,9 +588,10 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
         applied, error = await asyncio.to_thread(
             set_lane_slots_sync, lane, MIN_POOL_SLOTS,
         )
-    elif applied and now_on:
-        # Only DOWNWARD. See the docstring: raising a cap is permission, and
-        # the sizing pass decides whether there is work to spend it on.
+    elif applied and now_on and slots_cap < was_cap:
+        # LOWERED on a running lane. Only this direction needs a message, and
+        # only when the pool is actually above the new cap — so it reads the
+        # live pool rather than resizing blind. A raise never reaches here.
         live = await asyncio.to_thread(inspect_lanes_sync)
         current = live[lane.name].pool
         if current is not None and current > slots_cap:
@@ -561,8 +614,8 @@ async def set_lane(session: AsyncSession, lane: Lane, *, slots_cap: int) -> dict
 
     return {
         "name": lane.name,
-        "slots_cap": row.slots_cap,
-        "ceiling": ceiling,
+        "slots_cap": slots_cap,
+        "ceiling": derived_ceiling(lane),
         "enabled": now_on,
         "applied": applied,
         "apply_error": error,

tests/test_api_gpu.py

@@ -314,3 +314,67 @@ async def test_cpu_embed_never_blocks_gpu_crop_backfills(db):
         select(GpuJob.task).where(GpuJob.image_record_id == img.id)
     )).scalars().all())
     assert tasks == {"ccip", "siglip"}
+
+
+# --- an idle agent still checks in -------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_a_heartbeat_with_no_jobs_still_records_the_check_in(client, db):
+    """Operator, 2026-09-23: *"I'm running the gpu agent on my device and it
+    currently reads as 'offline' but it's running and has checked in
+    recently."*
+
+    It had — twelve minutes ago. The roster takes its agent check-in from the
+    `lease` and `heartbeat` calls, and calls an agent stopped after 300s of
+    silence. The agent's heartbeat loop was gated on holding leases, so an
+    IDLE agent sent none; the only check-in left was the lease poll, which
+    sleep mode backs off to a 900s ceiling. 900 against 300 — an idle agent
+    was structurally guaranteed to read as stopped.
+
+    So the empty heartbeat has to be a real check-in on the server side, not
+    merely tolerated. Asserted on `last_seen_at` moving, because "it returned
+    200" would pass against an endpoint that recorded nothing.
+    """
+    from backend.app.models import ServiceSeen
+
+    token = (await (await client.post("/api/gpu/token/rotate")).get_json())["token"]
+    hdr = {"Authorization": f"Bearer {token}"}
+
+    resp = await client.post(
+        "/api/gpu/jobs/heartbeat",
+        json={"agent_id": "desktop-agent", "job_ids": []}, headers=hdr,
+    )
+
+    assert resp.status_code == 200
+    assert (await resp.get_json())["extended"] == 0, "it extends no lease"
+
+    row = (await db.execute(
+        select(ServiceSeen).where(ServiceSeen.key == "agent:desktop-agent")
+    )).scalar_one()
+    assert row.kind == "agent"
+    assert row.display_name == "GPU agent (desktop-agent)"
+    assert row.last_seen_at is not None
+
+
+def test_the_agent_heartbeats_whether_or_not_it_holds_a_lease():
+    """The other half, in the agent itself — the half that was actually wrong.
+
+    Read from the source rather than by running the loop: it is a `while True`
+    with a sleep, so exercising it means threads and timing, and the property
+    is simply that the call is not behind a `if ids:`.
+    """
+    from pathlib import Path
+
+    src = (
+        Path(__file__).resolve().parents[1] / "agent" / "fc_agent" / "worker.py"
+    ).read_text()
+    loop = src[src.index("def _heartbeat_loop"):]
+    loop = loop[:loop.index("\n    def ")]
+
+    assert "self.client.heartbeat(ids)" in loop
+    assert "if ids:" not in loop, (
+        "the heartbeat is gated on holding leases again; an idle agent then "
+        "reads as stopped after 300s while sleep mode backs its lease poll "
+        "off to 900s"
+    )

tests/test_api_workers.py

@@ -255,3 +255,81 @@ async def _refreshed(db, name: str, expected: int) -> None:
     row = await _lane_row(db, name)
     await db.refresh(row)
     assert row.slots_cap == expected, "a refused write must store nothing"
+
+
+# --- no database connection is held across a broker round trip ---------------
+
+
+@pytest.mark.asyncio
+async def test_the_lane_read_holds_no_session_while_it_inspects(monkeypatch):
+    """Operator, 2026-09-23: *"something about changing the cap number is
+    blocking to the website... it shouldn't be"*.
+
+    Nothing here was slow in itself. A Postgres connection was held across a
+    celery inspect whose budget is eleven seconds, on a page that polls every
+    fifteen — so with a lane not answering, each poll pinned a connection for
+    most of the interval. SQLAlchemy's default pool is five plus ten overflow;
+    two browser tabs, the health endpoint doing the same, and a cap change
+    adding more inspects exhausts it, and every OTHER request then waits on a
+    connection. It surfaced as the whole site stalling rather than as one slow
+    page, which is why it took a screenshot to find.
+
+    Asserted STRUCTURALLY rather than by timing: `lane_view` must not accept a
+    session at all. A timing test would be flaky, and a mock-call-order test
+    would pass against a version that took the session and merely used it
+    early — the property that matters is that it CANNOT.
+    """
+    import inspect as _inspect
+
+    from backend.app.services.worker_control import lane_settings, lane_view
+
+    assert "session" not in _inspect.signature(lane_view).parameters, (
+        "lane_view takes a session again; the broker work must run with none held"
+    )
+    # And the DB half still exists, so the split did not simply lose the reads.
+    assert "session" in _inspect.signature(lane_settings).parameters
+
+
+@pytest.mark.asyncio
+async def test_the_cap_write_holds_no_session_while_it_pushes():
+    """The same property on the write path, where it was worse: a cap change
+    could make three broker round trips, each with a connection held."""
+    import inspect as _inspect
+
+    from backend.app.services.worker_control import push_lane_cap, store_lane_cap
+
+    assert "session" in _inspect.signature(store_lane_cap).parameters
+    assert "session" not in _inspect.signature(push_lane_cap).parameters, (
+        "push_lane_cap takes a session again; the push must run with none held"
+    )
+
+
+@pytest.mark.asyncio
+async def test_raising_a_cap_costs_no_broker_round_trip_at_all(
+    client, db, no_live_workers, monkeypatch,
+):
+    """The common case must be instant. Raising a cap is permission, not a
+    request — the sizing pass spends it — so there is nothing to tell the
+    broker, and the operator's `+` should answer immediately rather than
+    waiting out an inspect."""
+    from backend.app.services import worker_control as wc
+
+    calls = []
+    monkeypatch.setattr(
+        wc, "inspect_lanes_sync", lambda: calls.append("inspect") or {},
+    )
+    monkeypatch.setattr(
+        wc, "set_lane_slots_sync",
+        lambda *a, **k: calls.append("resize") or (True, None),
+    )
+    monkeypatch.setattr(
+        wc, "set_lane_enabled_sync",
+        lambda *a, **k: calls.append("consumers") or (True, None),
+    )
+
+    await client.post("/api/system/workers/worker", json={"slots_cap": 1})
+    calls.clear()
+    resp = await client.post("/api/system/workers/worker", json={"slots_cap": 6})
+
+    assert resp.status_code == 200
+    assert calls == [], f"raising a cap talked to the broker: {calls}"