feat(broker): opt-in round-robin turns (max turn, rotation, preemption) and gateway reply timeout

CHANGELOG.md

@@ -6,6 +6,15 @@ has a matching GitHub release whose body comes from its section below.
 
 ## [Unreleased]
 
+### Added
+
+- Round-robin turns, off by default (`[turns]` in `berth.toml`, [docs/leasing.md](docs/leasing.md)):
+  `max_turn` (and `max_turn_by` per engine) lets a waiting lower priority drain a holder that has had the card
+  that long, `cycle` serves waiters of one priority in a set order, and `preempt` with `preempt_gap` lets a
+  higher priority cut a `min_turn` short at most once per gap ([#275](https://github.com/T-Crypt/berth/issues/275)).
+- `[gateway] reply_timeout`: an engine that takes a request and sends no headers in that time gets it once more,
+  then the client gets a 504 with the code `no_reply` instead of waiting forever. Off by default.
+
 ## [v0.4.0] - 2026-10-29
 
 One berthd now governs every card on a machine. `berth run --gpu gpu1` puts an engine on one card,

docs/gateway.md

@@ -398,6 +398,22 @@ loopback, accepts loopback Host names plus the gateway's `--allow-host` names (f
 front), and takes a start only as a POST from its own page, so another site cannot make a visitor's
 browser drain the card.
 
+## An engine that takes a request and says nothing
+
+By default a chat request waits for the engine as long as the client does. A wedged engine then leaves the
+client hanging. Set a reply timeout to bound that:
+
+```toml
+[gateway]
+reply_timeout = "150s"   # 0 or unset = off, wait as long as the client does
+```
+
+If a live engine sends no response headers within `reply_timeout`, the gateway sends the same request once
+more (starting the engine again first if it went away). A second silence answers `504` with the error code
+`no_reply` naming the engine; a streaming request gets the same error as one SSE event. A request the engine
+did reply to is never replayed. Choose a value above the longest prompt processing time you expect before the
+first token, since the timer covers it.
+
 ## Drain a server at its request boundary
 
 An engine with `/slots` but no unload URL (Strata, a bare llama-server) would otherwise get SIGTERM mid-request when

docs/leasing.md

@@ -87,6 +87,86 @@
     or job boundary, and a minimum turn never cuts a request off. After a daemon restart, a
     re-attach starts a fresh turn.
 
+## Round-robin turns
+
+Two engines that share a card, say an LLM server and an image generator, each want it for a while and
+neither should starve the other. Berth already hands the card over at the request boundary and keeps a
+`min_turn`. The `[turns]` table adds three opt-in rules on top; with none of them set, nothing above changes.
+None of them cuts a request off: a drain still starts at the slot or job boundary, a pinned lease or a hold is
+never drained by them, and the driver must still show the memory free before the next grant.
+
+```toml
+[turns]
+max_turn    = "15m"                    # the longest a holder keeps the card against a lower priority
+cycle       = ["llm", "image"]         # who is served next among equal priorities
+preempt     = true                     # a higher priority may cut a min_turn short...
+preempt_gap = "45s"                    # ...at most once per this long (default 45s)
+
+[turns.max_turn_by]                    # per engine (its name, or --group), over max_turn
+image = "30m"
+```
+
+- **Maximum turn.** Priorities keep a lower-priority waiter out for as long as the holder runs. With
+  `max_turn`, a holder that has had the card that long stops outranking a waiting lower priority: the waiter
+  drains it at the next boundary. `0` (the default) is no limit.
+- **Rotation.** Waiters of one priority are normally served by arrival. With `cycle`, the engine after the
+  last one granted goes first, so `llm`, `image`, `llm`, `image` alternate even when `llm` asks more often.
+  Engines not listed go after the listed ones, by arrival.
+- **Preemption.** A `min_turn` keeps every waiter out, however urgent. With `preempt = true` a waiter of a
+  strictly higher priority (a person at the keyboard, say) may drain a holder inside its `min_turn`. Each such
+  cut makes the engine it empties reload from cold, so Berth allows one per `preempt_gap`; a second waiter
+  inside the gap is told when the next preemption is allowed. A waiter of equal or lower priority still waits
+  out the turn.
+
+```mermaid
+stateDiagram-v2
+    [*] --> Idle
+    Idle --> Turn: grant (a waiter reaches the head of the queue)
+    state Turn {
+        [*] --> Protected
+        Protected --> Open: min_turn ends
+        Open --> Expired: max_turn ends
+        note right of Protected
+            equal, lower priority: wait
+            higher priority: preempts if allowed,
+            once per preempt_gap
+        end note
+        note right of Open
+            equal priority drains the holder,
+            lower priority waits
+        end note
+        note right of Expired
+            any waiting lower priority
+            may drain the holder
+        end note
+    }
+    Turn --> Draining: a waiter may take the card
+    Draining --> Idle: the driver shows the memory free
+    Idle --> Turn: next waiter (the cycle picks among equal priorities)
+```
+
+The timeline of one card with two engines, `min_turn = 45s` and `preempt_gap = 45s`:
+
+```mermaid
+sequenceDiagram
+    participant A as LLM (interactive)
+    participant B as image generator (batch)
+    participant C as person (foreground)
+    A->>A: granted, turn starts
+    B-->>A: waits (lower priority, held)
+    Note over A: max_turn ends
+    A->>B: drained at the next boundary
+    B->>B: granted, min_turn 45 s protects it
+    C-->>B: asks for the card at 20 s
+    Note over B: preempt = true: the turn is cut short
+    B->>C: drained
+    C->>C: granted
+    Note over C: another foreground request within 45 s of this preemption waits for the gap to pass
+```
+
+`berth config show` prints the settings that are on. A wait reason names what the waiter is waiting for:
+"has its minimum turn, 3m12s left", or "a turn was preempted 10s ago; the next preemption is allowed in 35s".
+
 ## Games, apps and holds
 - **Claims cover work Berth does not start.** berthd scans the process table every 2 s.
   A *hold* claim keeps the card while its process runs: engines are drained and wait, with a

internal/broker/broker.go

@@ -126,6 +126,9 @@ type Config struct {
 	// JuggleWindow: an equal-priority drain of a holder granted less than
 	// this ago is a juggle, decided by rank. 0 = DefaultJuggleWindow; < 0 = off.
 	JuggleWindow time.Duration
+	// Turns is the round-robin policy (maximum turn, rotation, preemption);
+	// the zero value is off.
+	Turns Turns
 	// UsageFile keeps the card-hours per engine across restarts; "" = memory only.
 	UsageFile string
 	// FootprintFile keeps the VRAM footprints measured per engine across
@@ -479,6 +482,10 @@ type Broker struct {
 	saveFailed bool
 
 	verdicts map[string]verdict // loser engine key -> the juggle it lost
+	// lastKey is the engine key of the last grant, and preemptedAt the last
+	// time a waiter cut a minimum turn short (Turns).
+	lastKey     string
+	preemptedAt time.Time
 	// tracked is the claim watcher's last scan: a process (and every child)
 	// a track claim names. Its card memory is the claim's, never a stowaway.
 	tracked    map[int]string
@@ -1117,6 +1124,9 @@ func (b *Broker) schedule(now time.Time) {
 			continue
 		}
 		for _, v := range victims {
+			if w.prio != Hold && v.inTurn(now) {
+				b.notePreemption(now)
+			}
 			v.state = draining
 			// Its memory goes with the drain: an open lend is dropped, and
 			// never asked back, since the borrower still holds it (rule 2).
@@ -1142,6 +1152,7 @@ func (b *Broker) grant(w *lease, now time.Time) {
 	w.state = active
 	w.granted = now
 	w.grantedN = b.order()
+	b.lastKey = w.key()
 	w.releaseBaseline = b.snap.Baseline()
 	w.releaseGTTBaseline = b.snap.GTTBaseline()
 	w.contested = b.squatBytes()
@@ -1186,6 +1197,11 @@ func (b *Broker) sortQueue() {
 		if b.queue[i].prio != b.queue[j].prio {
 			return b.queue[i].prio > b.queue[j].prio
 		}
+		if len(b.cfg.Turns.Cycle) > 0 {
+			if ri, rj := b.cycleRank(b.queue[i].key()), b.cycleRank(b.queue[j].key()); ri != rj {
+				return ri < rj
+			}
+		}
 		return b.queue[i].arrivedN < b.queue[j].arrivedN
 	})
 }
@@ -1400,14 +1416,25 @@ func (b *Broker) evaluate(w *lease, now time.Time) (bool, string, []*lease) {
 			// for it to become parked (or fall back to a drain); evicting
 			// it mid-park would throw the park away.
 			pending = append(pending, h)
-		case h.pinned || h.prio > w.prio:
+		case h.pinned || (h.prio > w.prio && !b.turnExpired(h, now)):
+			// (a holder past its maximum turn no longer blocks a lower priority)
 			blockers = append(blockers, h)
 		case w.prio != Hold && h.inTurn(now):
 			// A minimum turn: the holder keeps the card until its turn
 			// ends, whatever the waiter's priority; only a hold (a game,
-			// a call, the operator's own berth hold) takes it out before.
+			// a call, the operator's own berth hold) takes it out before,
+			// or a higher-priority waiter when preemption is on and its
+			// gap has passed.
+			ok, why := b.preempts(h, w, now)
+			if ok {
+				drainable = append(drainable, h)
+				continue
+			}
 			kept = append(kept, h)
 			keptWhy = h.turnWhy(now)
+			if why != "" {
+				keptWhy += "; " + why
+			}
 			continue
 		case h.prio == w.prio:
 			if keep, why := b.keeps(h, w, now); keep {

internal/broker/turns.go

@@ -0,0 +1,114 @@
+// Copyright 2026 Trevin Tindall (T-Crypt)
+// SPDX-License-Identifier: Apache-2.0
+
+package broker
+
+// Round-robin turns. Everything here is off until the [turns] table sets it,
+// and with it off the broker decides exactly as before.
+//
+// A turn is the time one lease holds the card. Three rules shape it, none of
+// which ever cuts a request off (a drain still starts at the slot or job
+// boundary):
+//
+//   - A maximum turn. A holder past it stops being protected by its priority:
+//     a waiter of any lower priority may drain it, so a busy high-priority
+//     engine cannot keep the card from a waiting one indefinitely. A pinned
+//     lease and a hold are never drained.
+//   - Rotation. Waiters of one priority are served in the order of a
+//     configured cycle, starting with the engine after the last one granted,
+//     instead of strictly by arrival.
+//   - Preemption. A waiter of a strictly higher priority may drain a holder
+//     inside its minimum turn, at most once per preemption gap, since each
+//     such drain costs the engine it empties a cold reload.
+
+import (
+	"fmt"
+	"slices"
+	"time"
+)
+
+// DefaultPreemptGap is the gap between two preemptions when Preempt is on
+// and no gap is set.
+const DefaultPreemptGap = 45 * time.Second
+
+// Turns is the round-robin policy. The zero value is off.
+type Turns struct {
+	// MaxTurn is the longest a holder keeps the card against a waiter of
+	// lower priority; 0 = no limit. MaxTurnBy overrides it per engine key
+	// (a lease's group, else its engine name).
+	MaxTurn   time.Duration
+	MaxTurnBy map[string]time.Duration
+	// Cycle orders the waiters of one priority by engine key: the one after
+	// the last granted goes first. Keys not listed go after the listed ones,
+	// in arrival order. Empty = arrival order.
+	Cycle []string
+	// Preempt lets a waiter of a higher priority drain a holder inside its
+	// minimum turn, once per PreemptGap (0 = DefaultPreemptGap).
+	Preempt    bool
+	PreemptGap time.Duration
+}
+
+func (t Turns) maxTurn(key string) time.Duration {
+	if d, ok := t.MaxTurnBy[key]; ok {
+		return d
+	}
+	return t.MaxTurn
+}
+
+func (t Turns) preemptGap() time.Duration {
+	if t.PreemptGap > 0 {
+		return t.PreemptGap
+	}
+	return DefaultPreemptGap
+}
+
+// turnExpired is whether h has held the card past its maximum turn. A hold
+// has no maximum turn.
+func (b *Broker) turnExpired(h *lease, now time.Time) bool {
+	if h.prio == Hold || h.pinned || h.granted.IsZero() {
+		return false
+	}
+	max := b.cfg.Turns.maxTurn(h.key())
+	return max > 0 && now.Sub(h.granted) >= max
+}
+
+// preemptWait is how long until a preemption is allowed again; 0 = now.
+func (b *Broker) preemptWait(now time.Time) time.Duration {
+	if b.preemptedAt.IsZero() {
+		return 0
+	}
+	return max(0, b.cfg.Turns.preemptGap()-now.Sub(b.preemptedAt))
+}
+
+// preempts says whether waiter w may drain h inside h's minimum turn, and if
+// not, why (empty when preemption does not apply to this pair at all).
+func (b *Broker) preempts(h, w *lease, now time.Time) (bool, string) {
+	if !b.cfg.Turns.Preempt || w.prio <= h.prio {
+		return false, ""
+	}
+	if left := b.preemptWait(now); left > 0 {
+		return false, fmt.Sprintf("a turn was preempted %s ago; the next preemption is allowed in %s",
+			now.Sub(b.preemptedAt).Round(time.Second), left.Round(time.Second))
+	}
+	return true, ""
+}
+
+// notePreemption records a drain that cut a minimum turn short.
+func (b *Broker) notePreemption(now time.Time) {
+	b.preemptedAt = now
+}
+
+// cycleRank is where an engine key stands in the rotation: 0 is next after
+// the last key granted. Keys outside the cycle rank after every listed one.
+func (b *Broker) cycleRank(key string) int {
+	c := b.cfg.Turns.Cycle
+	i := slices.Index(c, key)
+	if i < 0 {
+		return len(c)
+	}
+	last := slices.Index(c, b.lastKey)
+	if last < 0 {
+		return i
+	}
+	return (i - last - 1 + len(c)) % len(c)
+}