feat: a saturated lane can grow itself, within the cap the operator set (4297)

Milestone 422 step 7 — the one sweep in this milestone that decides rather
than obeys, so it is off until a lane is opted in, bounded by the operator's
cap, floored at the operator's value, and it reports every decision including
the ones where it did nothing.

Growth needs BOTH halves: all slots busy AND a backlog. Depth alone means
celery is about to pick those up and growing would add idle children (#1253
is that bug in the GPU agent); saturation alone means the lane is busy with
exactly as much work as exists. The backlog is depth PLUS reserved, because
celery prefetches and LLEN reads 0 while a worker holds thirty tasks in
memory — the case an LLEN-only autoscaler misses entirely, and the reason
step 2 plumbed `reserved` through.

The two sweeps had to be taught not to fight. The reconcile drives every
lane to its stored slots every five minutes, which would have reverted each
grow on the next tick: grow, revert, grow, revert, forever. For an
autoscaling lane the stored value is now a FLOOR — restored when a lane
falls below it, never taken back above it.

The operator's "a task that runs for x concurrent time" idea stays a UI
warning rather than a trigger: a long task does not finish sooner because
the lane gained a slot, so scaling on it would spend memory to change
nothing. Read from `task_run` on our own wall clock, not celery's
`time_start`, which is the WORKER's monotonic clock and would produce a
duration that is meaningless in the direction that matters — plausible.

Caught while reading it back: the first version read the stored slots as the
CURRENT pool. The autoscaler never writes that row, so every tick would have
proposed floor+1 — resizing nothing, reporting `grew` anyway (a replica
already past the target is issued no message and reports success), and
capping the lane one slot above its floor forever while claiming otherwise.
It now reads the live pool and keeps the stored value purely as the floor,
and the tests fix the two to different numbers so an equal-fixture pass
cannot hide it again.

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

backend/app/services/worker_control.py

@@ -50,11 +50,12 @@ from __future__ import annotations
 import asyncio
 import logging
 from dataclasses import dataclass, field
+from datetime import UTC, datetime
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from ..models import WorkerLane
+from ..models import TaskRun, WorkerLane
 from .worker_lanes import LANES, LANES_BY_QUEUE_KEY, Lane, derived_ceiling
 
 log = logging.getLogger(__name__)
@@ -103,6 +104,16 @@ class LaneLiveState:
         no replica reported — unknown, never zero."""
         return max(self.pools.values()) if self.pools else None
 
+    @property
+    def capacity(self) -> int:
+        """Total slots across replicas — how many tasks this lane can run at
+        once. Distinct from `pool`, and the two must not be confused: `pool`
+        is the DIAL (one process's size, what grow/shrink move), `capacity` is
+        the CAPABILITY. Asking "is this lane saturated" compares `active`,
+        which is summed across replicas, against this — against `pool` it
+        would call two half-busy replicas of 4 saturated at 4 active."""
+        return sum(self.pools.values())
+
 
 def _lane_for_queues(queues: tuple[str, ...]) -> Lane | None:
     return LANES_BY_QUEUE_KEY.get(tuple(sorted(queues)))
@@ -260,6 +271,7 @@ async def _rows_by_name(session: AsyncSession) -> dict[str, WorkerLane]:
             slots=lane.default_slots,
             slots_cap=lane.default_slots_cap,
             enabled=lane.default_enabled,
+            autoscale=lane.default_autoscale,
         )
         session.add(row)
         rows[lane.name] = row
@@ -278,7 +290,9 @@ async def lane_view(session: AsyncSession) -> list[dict]:
     rows = await _rows_by_name(session)
     live = await asyncio.to_thread(inspect_lanes_sync)
     depths = await asyncio.to_thread(_queue_depths_sync)
+    oldest = await _oldest_running_by_queue(session)
 
+    now = datetime.now(UTC)
     out = []
     for lane in LANES:
         row = rows[lane.name]
@@ -297,6 +311,7 @@ async def lane_view(session: AsyncSession) -> list[dict]:
             "slots_cap": row.slots_cap,
             "ceiling": derived_ceiling(lane),
             "enabled": row.enabled,
+            "autoscale": row.autoscale,
             "memory_bound": lane.memory_bound,
             "optional": lane.optional,
             # What enabling this lane will download, so the UI can say WHICH
@@ -321,10 +336,55 @@ async def lane_view(session: AsyncSession) -> list[dict]:
             },
             "queue_depth": depth,
             "pending": None if depth is None else depth + state.reserved,
+            # How long the oldest still-running task on this lane has been
+            # going, in minutes. The operator asked for a trigger here — grow
+            # a lane whose tasks run past some duration — and it stayed a
+            # REPORT: a long task does not finish sooner because the lane
+            # gained a slot, so scaling on it would spend memory to change
+            # nothing. Shown so they can see a lane wedged on one slow job,
+            # which is the genuinely useful half of the idea.
+            "oldest_running_minutes": _minutes_since(
+                min(
+                    (oldest[q] for q in lane.queues if q in oldest),
+                    default=None,
+                ),
+                now,
+            ),
         })
     return out
 
 
+async def _oldest_running_by_queue(session: AsyncSession) -> dict[str, datetime]:
+    """When the longest-running unfinished task on each queue started.
+
+    Read from `task_run`, which is OUR OWN table on OUR OWN wall clock, and
+    deliberately not from celery's `inspect active()`. Those entries carry a
+    `time_start` taken from the WORKER's `time.monotonic()` — a clock with an
+    arbitrary origin per process. Subtracting it from this process's wall
+    clock produces a number that looks like a duration and is meaningless, and
+    it would be meaningless in the direction that matters: plausible.
+
+    `task_run` also already carries the per-queue staleness thresholds the
+    recovery sweep uses, so a row still `running` here is one the system
+    itself considers legitimately in flight rather than abandoned.
+    """
+    result = await session.execute(
+        select(TaskRun.queue, func.min(TaskRun.started_at))
+        .where(TaskRun.status == "running", TaskRun.finished_at.is_(None))
+        .group_by(TaskRun.queue)
+    )
+    return {queue: started for queue, started in result if started is not None}
+
+
+def _minutes_since(started: datetime | None, now: datetime) -> int | None:
+    """Whole minutes, or None when nothing is running. Never negative: a row
+    written by a container whose clock is a few seconds ahead must read as 0
+    rather than as a task that starts in the future."""
+    if started is None:
+        return None
+    return max(0, int((now - started).total_seconds() // 60))
+
+
 def _queue_depths_sync() -> dict[str, int | None]:
     """Redis LLEN per queue. None for one that did not answer — see lane_view.
 
@@ -362,6 +422,7 @@ async def set_lane(
     slots: int | None = None,
     slots_cap: int | None = None,
     enabled: bool | None = None,
+    autoscale: bool | None = None,
 ) -> dict:
     """Store the operator's choice, then push it to the running lane.
 
@@ -383,6 +444,7 @@ async def set_lane(
     new_cap = row.slots_cap if slots_cap is None else slots_cap
     new_slots = row.slots if slots is None else slots
     new_enabled = row.enabled if enabled is None else enabled
+    new_autoscale = row.autoscale if autoscale is None else autoscale
 
     ceiling = derived_ceiling(lane)
     if new_cap < 0 or new_slots < 0:
@@ -398,6 +460,7 @@ async def set_lane(
     row.slots_cap = new_cap
     row.slots = new_slots
     row.enabled = new_enabled
+    row.autoscale = new_autoscale
     await session.commit()
 
     applied, error = True, None
@@ -428,6 +491,7 @@ async def set_lane(
         "slots_cap": row.slots_cap,
         "ceiling": ceiling,
         "enabled": row.enabled,
+        "autoscale": row.autoscale,
         "applied": applied,
         "apply_error": error,
         # Tells the card to say a download has started rather than leaving the
@@ -458,11 +522,28 @@ def _enqueue_model_fetch() -> bool:
         return False
 
 
-def reconcile_lanes_sync(desired: dict[str, tuple[int, bool]]) -> dict:
+def reconcile_lanes_sync(
+    desired: dict[str, tuple[int, bool]],
+    autoscaling: frozenset[str] = frozenset(),
+) -> dict:
     """Drive every RUNNING lane to its stored slots and enabled flag.
 
     `desired` is lane name -> (slots, enabled), read from the database by the
-    caller. This function touches no database: the celery task that schedules
+    caller. `autoscaling` names the lanes the autoscaler is allowed to move.
+
+    ## For an autoscaling lane the stored value is a FLOOR, not a target
+
+    Step 7's autoscaler raises a saturated lane's live pool without changing
+    its row — the row holds what the OPERATOR set. If this pass treated that
+    row as an exact target it would shrink the lane back on the very next
+    tick, and the two sweeps would fight forever at five-minute intervals:
+    grow, revert, grow, revert. That is lesson #4183's failure arriving
+    between two enforcers rather than inside one.
+
+    So for those lanes the target becomes `max(stored, current)` — this pass
+    still restores a lane that came back from a restart below what the
+    operator set, and never takes back what the autoscaler added. Bringing it
+    down is the autoscaler's job, and it does so only to that same floor. This function touches no database: the celery task that schedules
     it owns the sync session, and keeping the DB out of here is what lets the
     same code be called from anywhere that already knows the target.
 
@@ -521,17 +602,177 @@ def reconcile_lanes_sync(desired: dict[str, tuple[int, bool]]) -> dict:
             changed.append(lane.name)
 
         current = state.pool
-        if current is not None and current == slots:
+        # The floor, for a lane the autoscaler manages. See the docstring.
+        target_slots = slots
+        if lane.name in autoscaling and current is not None:
+            target_slots = max(slots, current)
+        if current is not None and current == target_slots:
             continue
-        ok, err = set_lane_slots_sync(lane, slots, live=state)
+        ok, err = set_lane_slots_sync(lane, target_slots, live=state)
         if ok:
             if lane.name not in changed:
                 changed.append(lane.name)
             log.info(
                 "worker_control: %s reconciled %s -> %s slots",
-                lane.name, current, slots,
+                lane.name, current, target_slots,
             )
         else:
             failed[lane.name] = err or "could not resize"
 
     return {"changed": changed, "skipped": skipped, "failed": failed}
+
+
+# --- the autoscaler (step 7) --------------------------------------------------
+#
+# The only part of this milestone that acts without anyone asking. Everything
+# above does what an operator pressed; this decides. So it is off by default,
+# opted into per lane, bounded by the cap the operator set, and it reports what
+# it did rather than moving numbers silently.
+
+# ## Why these three numbers are not in Settings
+#
+# Rule 25 puts anything an operator might want to tune in the UI, and the
+# knobs that decide what this does ARE there: whether a lane autoscales at
+# all, its cap, and its floor — all DB-backed, all changeable without a
+# restart. What is left here is the POLICY's internals, and exposing them
+# would add four numbers per lane to a card whose whole value is being
+# readable at a glance, to tune a decision the operator has a better lever
+# for. If growth turns out to be too eager or too shy in practice, that is a
+# reason to change these values for everyone, not to ask each operator to
+# discover them.
+
+# All slots busy AND this many tasks waiting before a lane may grow.
+#
+# The AND is the design. Depth with free slots means nothing — celery is about
+# to pick those up, and growing the pool would add idle children. Saturation
+# with an empty queue means nothing either: the lane is busy with exactly as
+# much work as exists. Only both together say "there is more work than this
+# lane can reach".
+AUTOSCALE_BACKLOG_THRESHOLD = 10
+
+# Grow by one slot per tick, never to the cap in one jump. A lane that is
+# saturated because of one slow burst settles a slot or two above where it
+# started rather than at its ceiling, and the next tick re-measures rather
+# than committing to a guess made once.
+AUTOSCALE_STEP = 1
+
+# Hysteresis: shrink only when the backlog is well BELOW the grow threshold,
+# not merely under it. Equal thresholds flap — one task arriving and leaving
+# would grow and shrink the lane forever at the tick interval, which is
+# lesson #4183's churn arriving through a different door.
+AUTOSCALE_SHRINK_BELOW = 2
+
+
+@dataclass
+class AutoscaleDecision:
+    """What the autoscaler did to one lane, and why — in the operator's terms.
+
+    A reason string on every outcome including "nothing", because an
+    autoscaler that only speaks when it acts is one nobody can debug when it
+    does not.
+    """
+
+    lane: str
+    action: str  # "grew" | "shrank" | "held"
+    slots: int
+    reason: str
+
+
+def autoscale_lanes_sync(
+    lanes: dict[str, tuple[int, int, bool]],
+) -> list[AutoscaleDecision]:
+    """Decide and apply one round of autoscaling.
+
+    `lanes` is name -> (slots_cap, configured_slots, autoscale_on), read from
+    the database by the caller — this function touches no database, for the
+    same reason `reconcile_lanes_sync` does not.
+
+    ## What is current, and what is the floor
+
+    The value this moves is the LIVE pool, read from `inspect`. The stored
+    `configured_slots` is what the operator set and is only the FLOOR: growth
+    goes above it and a shrink returns to it, never below.
+
+    The two are deliberately not the same number, and reading the stored value
+    as "current" is the mistake that makes this function useless in a way no
+    unit test of a single tick would show. The autoscaler never writes the
+    row, so the stored value never moves; a tick that computed `stored + 1`
+    would propose the same target forever, cap the lane one slot above the
+    floor no matter the load, and — because resizing a replica already at the
+    target issues nothing and reports success — claim `grew` on every tick
+    while nothing changed. Lesson #4183's non-convergence, arriving with a
+    success message attached.
+
+    So: `current = state.pool`, `configured` is the floor, and both `grew` and
+    `shrank` mean the live pool actually moved.
+    """
+    live = inspect_lanes_sync()
+    depths = _queue_depths_sync()
+    out: list[AutoscaleDecision] = []
+
+    for lane in LANES:
+        target = lanes.get(lane.name)
+        if target is None:
+            continue
+        cap, configured, on = target
+        if not on:
+            continue
+
+        state = live[lane.name]
+        if not state.present or state.pool is None:
+            # Nothing answered. Not "idle" — unknown, and a decision drawn
+            # from an unswept read is exactly what snippet #3969 warns about.
+            # `configured` is reported because there is no live number to
+            # report; it is what the lane will come back at.
+            out.append(AutoscaleDecision(
+                lane.name, "held", configured, "lane is not answering",
+            ))
+            continue
+
+        current = state.pool
+        known = [depths.get(q) for q in lane.queues]
+        if all(d is None for d in known):
+            out.append(AutoscaleDecision(
+                lane.name, "held", current, "queue depth unavailable",
+            ))
+            continue
+        backlog = sum(d for d in known if d is not None) + state.reserved
+        # Against CAPACITY, not against the dial: `active` is summed across
+        # replicas, so comparing it to one replica's pool size would call two
+        # half-busy replicas of 4 saturated at 4 active and grow a lane that
+        # has idle slots.
+        saturated = state.active >= state.capacity > 0
+        busy = backlog >= AUTOSCALE_BACKLOG_THRESHOLD
+
+        if saturated and busy and current < cap:
+            new = min(cap, current + AUTOSCALE_STEP)
+            ok, err = set_lane_slots_sync(lane, new, live=state)
+            out.append(AutoscaleDecision(
+                lane.name, "grew" if ok else "held", new if ok else current,
+                f"{backlog} waiting and all {state.capacity} slots busy"
+                if ok else f"could not grow: {err}",
+            ))
+        elif saturated and busy:
+            # At the cap with work still waiting. Said out loud rather than
+            # held silently: this is the operator's own ceiling doing its job,
+            # and it is the moment they would want to know they set it.
+            out.append(AutoscaleDecision(
+                lane.name, "held", current,
+                f"{backlog} waiting but the cap is {cap}",
+            ))
+        elif current > configured and backlog <= AUTOSCALE_SHRINK_BELOW:
+            new = max(configured, current - AUTOSCALE_STEP)
+            ok, err = set_lane_slots_sync(lane, new, live=state)
+            out.append(AutoscaleDecision(
+                lane.name, "shrank" if ok else "held", new if ok else current,
+                f"backlog cleared, back toward {configured}"
+                if ok else f"could not shrink: {err}",
+            ))
+        else:
+            # The fixed point. A settled lane sends nothing and says so —
+            # the tick is one inspect and one LLEN sweep, no control messages.
+            out.append(AutoscaleDecision(
+                lane.name, "held", current,
+                f"{backlog} waiting, {state.active} busy",
+            ))
+    return out