From ec9e4136c11fc9866d6e3cf6a6f9b3a57c510562 Mon Sep 17 00:00:00 2001
From: chapi456 <71797135+chapi456@users.noreply.github.com>
Date: Sat, 31 Oct 2026 13:39:17 +0100
Subject: [PATCH 1/2] feat(broker): opt-in round-robin turns (max turn,
 rotation, preemption)

Add a [turns] table, off by default:
- max_turn / max_turn_by: a holder past its maximum turn stops outranking a
  waiting lower priority (holds and pinned leases excepted).
- cycle: waiters of one priority are served in a configured order, starting
  after the last engine granted.
- preempt / preempt_gap: a strictly higher priority may drain a holder inside
  its min_turn, at most once per gap (default 45s).

Refs #275

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
---
 docs/leasing.md                |  80 ++++++++++++++
 internal/broker/broker.go      |  31 +++++-
 internal/broker/turns.go       | 114 ++++++++++++++++++++
 internal/broker/turns_test.go  | 188 +++++++++++++++++++++++++++++++++
 internal/cli/config.go         |  25 +++++
 internal/cli/daemon.go         |  13 +++
 internal/config/config.go      |  49 +++++++++
 internal/config/config_test.go |  26 +++++
 mkdocs.yml                     |   6 +-
 9 files changed, 529 insertions(+), 3 deletions(-)
 create mode 100644 internal/broker/turns.go
 create mode 100644 internal/broker/turns_test.go

diff --git a/docs/leasing.md b/docs/leasing.md
index 7a96734..bf5f38c 100644
--- a/docs/leasing.md
+++ b/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
diff --git a/internal/broker/broker.go b/internal/broker/broker.go
index 4732cb9..6a70e1e 100644
--- a/internal/broker/broker.go
+++ b/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 {
diff --git a/internal/broker/turns.go b/internal/broker/turns.go
new file mode 100644
index 0000000..6567873
--- /dev/null
+++ b/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)
+}
diff --git a/internal/broker/turns_test.go b/internal/broker/turns_test.go
new file mode 100644
index 0000000..6dd14ce
--- /dev/null
+++ b/internal/broker/turns_test.go
@@ -0,0 +1,188 @@
+// Copyright 2026 Trevin Tindall (T-Crypt)
+// SPDX-License-Identifier: Apache-2.0
+
+package broker
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/T-Crypt/berth/internal/procs"
+)
+
+// startTurnsBroker is startTurnBroker with a round-robin policy.
+func startTurnsBroker(t *testing.T, p Turns) (*Broker, *minClock) {
+	t.Helper()
+	b, clk := startTurnBroker(t, procs.Map{})
+	b.cfg.Turns = p
+	return b, clk
+}
+
+// With no [turns] policy a higher-priority holder keeps the card against a
+// lower-priority waiter however long it holds it: the behaviour before turns.
+func TestNoPolicyKeepsAHigherPriorityHolderForever(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{})
+	must(t, acquire(b, Request{Engine: "llm", Priority: Foreground}))
+	clk.set(turnBase.Add(24 * time.Hour))
+
+	w := acquireW(b, Request{Engine: "image", Priority: Batch})
+	pending(t, w.ch)
+	select {
+	case why := <-holderDrains(b, "llm"):
+		t.Fatalf("drained with no policy: %s", why)
+	default:
+	}
+	if r := firstReason(t, w); !strings.Contains(r, "held by") {
+		t.Errorf("reason = %q, want a holder named", r)
+	}
+}
+
+// Past its maximum turn a holder no longer outranks a waiting lower priority.
+func TestMaxTurnLetsALowerPriorityWaiterDrainTheHolder(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{MaxTurn: 10 * time.Minute})
+	holder := must(t, acquire(b, Request{Engine: "llm", Priority: Foreground}))
+
+	w := acquireW(b, Request{Engine: "image", Priority: Batch})
+	pending(t, w.ch)
+	clk.set(turnBase.Add(9 * time.Minute))
+	b.Tick(t.Context())
+	select {
+	case why := <-holderDrains(b, "llm"):
+		t.Fatalf("drained before the maximum turn: %s", why)
+	default:
+	}
+
+	clk.set(turnBase.Add(10 * time.Minute))
+	b.Tick(t.Context())
+	if why := drained(t, holder); !strings.Contains(why, "image") {
+		t.Fatalf("drain reason = %q, want the waiter named", why)
+	}
+}
+
+// The per-engine maximum turn overrides the default, and a hold has none.
+func TestMaxTurnByEngineAndHoldsAreExempt(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{MaxTurn: time.Hour, MaxTurnBy: map[string]time.Duration{"llm": time.Minute}})
+	holder := must(t, acquire(b, Request{Engine: "llm", Priority: Foreground}))
+	clk.set(turnBase.Add(2 * time.Minute))
+	w := acquireW(b, Request{Engine: "image", Priority: Batch})
+	pending(t, w.ch)
+	if why := drained(t, holder); !strings.Contains(why, "image") {
+		t.Fatalf("drain reason = %q", why)
+	}
+
+	b2, clk2 := startTurnsBroker(t, Turns{MaxTurn: time.Minute})
+	must(t, acquire(b2, Request{Engine: "game", Priority: Hold, Pinned: true}))
+	clk2.set(turnBase.Add(time.Hour))
+	w2 := acquireW(b2, Request{Engine: "image", Priority: Batch})
+	pending(t, w2.ch)
+	select {
+	case why := <-holderDrains(b2, "game"):
+		t.Fatalf("a hold was drained: %s", why)
+	default:
+	}
+}
+
+// Of two waiters at one priority, the cycle serves the engine after the last
+// one granted first, not the earlier arrival.
+func TestCycleServesTheNextEngineFirst(t *testing.T) {
+	b, _ := startTurnsBroker(t, Turns{Cycle: []string{"a", "b", "c"}})
+	first := must(t, acquire(b, Request{Engine: "a", Priority: Interactive}))
+
+	wc := acquireW(b, Request{Engine: "c", Priority: Interactive})
+	pending(t, wc.ch)
+	// b arrives after c, but is next in the cycle after a.
+	wb := acquireW(b, Request{Engine: "b", Priority: Interactive})
+	pending(t, wb.ch)
+	first.Release()
+	b.Tick(t.Context())
+	must(t, wb.ch)
+	pending(t, wc.ch)
+}
+
+// Without a cycle the same waiters are served by arrival.
+func TestNoCycleServesByArrival(t *testing.T) {
+	b, _ := startTurnsBroker(t, Turns{})
+	first := must(t, acquire(b, Request{Engine: "a", Priority: Interactive}))
+	wc := acquireW(b, Request{Engine: "c", Priority: Interactive})
+	pending(t, wc.ch)
+	wb := acquireW(b, Request{Engine: "b", Priority: Interactive})
+	pending(t, wb.ch)
+	first.Release()
+	b.Tick(t.Context())
+	must(t, wc.ch)
+	pending(t, wb.ch)
+}
+
+// With preemption off, a higher priority waits out a minimum turn.
+func TestPreemptionOffWaitsOutTheTurn(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{})
+	must(t, acquire(b, Request{Engine: "image", Priority: Interactive, MinTurn: 5 * time.Minute}))
+	clk.set(turnBase.Add(10 * time.Second))
+	w := acquireW(b, Request{Engine: "llm", Priority: Foreground})
+	pending(t, w.ch)
+	select {
+	case why := <-holderDrains(b, "image"):
+		t.Fatalf("drained inside the minimum turn: %s", why)
+	default:
+	}
+}
+
+// With preemption on, a higher priority cuts a minimum turn; an equal one
+// still waits it out.
+func TestPreemptionCutsAMinimumTurnForAHigherPriorityOnly(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{Preempt: true})
+	holder := must(t, acquire(b, Request{Engine: "image", Priority: Interactive, MinTurn: 5 * time.Minute}))
+	clk.set(turnBase.Add(10 * time.Second))
+
+	eq := acquireW(b, Request{Engine: "other", Priority: Interactive})
+	pending(t, eq.ch)
+	select {
+	case why := <-holderDrains(b, "image"):
+		t.Fatalf("an equal priority preempted: %s", why)
+	default:
+	}
+	if r := firstReason(t, eq); !strings.Contains(r, "minimum turn") {
+		t.Errorf("reason = %q", r)
+	}
+
+	hi := acquireW(b, Request{Engine: "llm", Priority: Foreground})
+	pending(t, hi.ch)
+	if why := drained(t, holder); !strings.Contains(why, "llm") {
+		t.Fatalf("drain reason = %q, want the preempting waiter named", why)
+	}
+}
+
+// A second preemption inside the gap is refused with the time left, and is
+// allowed again once the gap has passed.
+func TestPreemptionIsLimitedToOncePerGap(t *testing.T) {
+	b, clk := startTurnsBroker(t, Turns{Preempt: true, PreemptGap: time.Minute})
+	h1 := must(t, acquire(b, Request{Engine: "image", Priority: Batch, MinTurn: 10 * time.Minute}))
+	clk.set(turnBase.Add(5 * time.Second))
+	w1 := acquireW(b, Request{Engine: "llm", Priority: Interactive, MinTurn: 10 * time.Minute})
+	pending(t, w1.ch)
+	drained(t, h1) // preemption number one
+	h1.Release()
+	b.Tick(t.Context())
+	must(t, w1.ch)
+
+	clk.set(turnBase.Add(15 * time.Second))
+	w2 := acquireW(b, Request{Engine: "urgent", Priority: Foreground})
+	pending(t, w2.ch)
+	select {
+	case why := <-holderDrains(b, "llm"):
+		t.Fatalf("a second preemption inside the gap: %s", why)
+	default:
+	}
+	if r := firstReason(t, w2); !strings.Contains(r, "next preemption is allowed in 50s") {
+		t.Errorf("reason = %q, want the time left", r)
+	}
+
+	clk.set(turnBase.Add(66 * time.Second))
+	b.Tick(t.Context())
+	select {
+	case <-holderDrains(b, "llm"):
+	case <-time.After(2 * time.Second):
+		t.Fatal("no preemption after the gap")
+	}
+}
diff --git a/internal/cli/config.go b/internal/cli/config.go
index 5cba9ac..163ee7c 100644
--- a/internal/cli/config.go
+++ b/internal/cli/config.go
@@ -9,6 +9,7 @@ import (
 	"os"
 	"strings"
 
+	"github.com/T-Crypt/berth/internal/broker"
 	"github.com/T-Crypt/berth/internal/config"
 	"github.com/T-Crypt/berth/internal/proto"
 	"github.com/T-Crypt/berth/internal/suite"
@@ -106,6 +107,7 @@ func configBullets(c *Context, cfg config.Config) {
 	harborGatewayBlock(c, cfg)
 	if len(cfg.Priority.Order) > 0 || c.Verbose {
 		priorityBullet(c, cfg)
+		turnsBullet(c, cfg)
 	}
 	if cfg.AlertsSet() || c.Verbose {
 		alertsBullet(c, cfg)
@@ -207,6 +209,29 @@ func priorityBullet(c *Context, cfg config.Config) {
 	c.Bullet("Priority: %s, juggle window %s", rank, window)
 }
 
+// turnsBullet names the round-robin settings that are on; with none set it
+// says nothing, since the broker then decides as it always did.
+func turnsBullet(c *Context, cfg config.Config) {
+	t := cfg.Turns
+	var parts []string
+	if t.MaxTurn.Duration > 0 || len(t.MaxTurnBy) > 0 {
+		parts = append(parts, "max turn "+t.MaxTurn.String())
+	}
+	if len(t.Cycle) > 0 {
+		parts = append(parts, "cycle ["+strings.Join(t.Cycle, " ")+"]")
+	}
+	if t.Preempt {
+		gap := t.PreemptGap.String()
+		if t.PreemptGap.Duration == 0 {
+			gap = broker.DefaultPreemptGap.String()
+		}
+		parts = append(parts, "preemption every "+gap+" at most")
+	}
+	if len(parts) > 0 {
+		c.Bullet("Turns: %s", strings.Join(parts, ", "))
+	}
+}
+
 func alertsBullet(c *Context, cfg config.Config) {
 	var parts []string
 	for _, k := range []struct {
diff --git a/internal/cli/daemon.go b/internal/cli/daemon.go
index 1ef48e2..0f3c723 100644
--- a/internal/cli/daemon.go
+++ b/internal/cli/daemon.go
@@ -165,6 +165,7 @@ func runDaemon(c *Context) int {
 			FootprintFile:  cardFile(*footprintFile, i, name),
 			Order:          cfg.Priority.Order,
 			JuggleWindow:   cfg.Priority.JuggleWindow.Duration,
+			Turns:          turnPolicy(cfg.Turns),
 			Busy:           watcher.Busy,
 			Comfy:          comfy,
 			ReattachWindow: *reattachWin,
@@ -452,6 +453,18 @@ func cardFile(path string, i int, name string) string {
 
 // parkFloorBytes is the RAM a park must leave the machine: the [park]
 // floor_gib plus the headroom floor, in bytes.
+// turnPolicy is the broker's round-robin policy from the [turns] table.
+func turnPolicy(t config.TurnsConfig) broker.Turns {
+	p := broker.Turns{MaxTurn: t.MaxTurn.Duration, Cycle: t.Cycle, Preempt: t.Preempt, PreemptGap: t.PreemptGap.Duration}
+	if len(t.MaxTurnBy) > 0 {
+		p.MaxTurnBy = make(map[string]time.Duration, len(t.MaxTurnBy))
+		for k, d := range t.MaxTurnBy {
+			p.MaxTurnBy[k] = d.Duration
+		}
+	}
+	return p
+}
+
 func parkFloorBytes(cfg config.Config, headroomMib uint64) uint64 {
 	floor := cfg.ParkFloorGiB() + float64(headroomMib)/1024
 	return uint64(floor * float64(gpu.GiB))
diff --git a/internal/config/config.go b/internal/config/config.go
index 6a4c690..0119e4d 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -202,6 +202,11 @@ type Config struct {
 		// --listen. Loopback only until the API has authentication.
 		// Empty = the built-in default.
 		Listen string `toml:"listen" json:"listen,omitempty" yaml:"listen,omitempty"`
+		// ReplyTimeout: an engine that takes a request and sends no response
+		// headers for this long gets it once more (started again if it went
+		// away), then the client gets a 504 naming the engine. 0 = off, the
+		// default: the request waits for the engine as long as the client does.
+		ReplyTimeout Duration `toml:"reply_timeout,omitzero" json:"reply_timeout,omitempty" yaml:"reply_timeout,omitempty"`
 	} `toml:"gateway,omitempty" json:"gateway" yaml:"gateway"`
 	Notify struct {
 		// Enabled turns berth notify's desktop notifications on or off;
@@ -225,6 +230,7 @@ type Config struct {
 	Engine    []Engine        `toml:"engine,omitempty" json:"engine,omitempty" yaml:"engine,omitempty"`
 	Alerts    AlertsConfig    `toml:"alerts" json:"alerts" yaml:"alerts"`
 	Priority  PriorityConfig  `toml:"priority,omitempty" json:"priority" yaml:"priority"`
+	Turns     TurnsConfig     `toml:"turns,omitempty" json:"turns" yaml:"turns"`
 	Claims    ClaimsConfig    `toml:"claims,omitempty" json:"claims" yaml:"claims"`
 	Squatters SquattersConfig `toml:"squatters,omitempty" json:"squatters" yaml:"squatters"`
 	Claim     []ClaimRule     `toml:"claim,omitempty" json:"claim,omitempty" yaml:"claim,omitempty"`
@@ -444,6 +450,25 @@ type PriorityConfig struct {
 	JuggleWindow Duration `toml:"juggle_window" json:"juggle_window" yaml:"juggle_window"`
 }
 
+// TurnsConfig is the round-robin policy, [turns] in berth.toml. Every
+// setting is off by default; with none set the broker decides as it always
+// did (priority, then arrival, with each engine's own min_turn).
+type TurnsConfig struct {
+	// MaxTurn is the longest a holder keeps the card against a waiter of
+	// lower priority: past it, the holder is drained for the waiter (at a
+	// job boundary; a request is never cut off). 0 = no limit. MaxTurnBy
+	// sets it per engine, keyed by name (or --group).
+	MaxTurn   Duration            `toml:"max_turn,omitzero" json:"max_turn,omitempty" yaml:"max_turn,omitempty"`
+	MaxTurnBy map[string]Duration `toml:"max_turn_by,omitempty" json:"max_turn_by,omitempty" yaml:"max_turn_by,omitempty"`
+	// Cycle serves waiters of one priority in this order of engine keys,
+	// starting with the one after the last granted. Empty = by arrival.
+	Cycle []string `toml:"cycle,omitempty" json:"cycle,omitempty" yaml:"cycle,omitempty"`
+	// Preempt lets a waiter of a higher priority drain a holder inside its
+	// min_turn, at most once per PreemptGap (default 45s). Off by default.
+	Preempt    bool     `toml:"preempt,omitzero" json:"preempt,omitempty" yaml:"preempt,omitempty"`
+	PreemptGap Duration `toml:"preempt_gap,omitzero" json:"preempt_gap,omitempty" yaml:"preempt_gap,omitempty"`
+}
+
 // AlertsConfig is the threshold table the node evaluates on its sample cycle.
 // A key with no entry is not watched. probe and stuck are built in; they
 // cannot be configured off.
@@ -609,6 +634,30 @@ func (c *Config) Validate() error {
 		}
 		ordered[o] = true
 	}
+	if c.Gateway.ReplyTimeout.Duration < 0 {
+		errs = append(errs, fmt.Errorf("gateway.reply_timeout %s must be >= 0", c.Gateway.ReplyTimeout))
+	}
+	if c.Turns.MaxTurn.Duration < 0 {
+		errs = append(errs, fmt.Errorf("turns.max_turn %s must be >= 0", c.Turns.MaxTurn))
+	}
+	if c.Turns.PreemptGap.Duration < 0 {
+		errs = append(errs, fmt.Errorf("turns.preempt_gap %s must be >= 0", c.Turns.PreemptGap))
+	}
+	for k, d := range c.Turns.MaxTurnBy {
+		if d.Duration < 0 {
+			errs = append(errs, fmt.Errorf("turns.max_turn_by.%s: %s must be >= 0", k, d))
+		}
+	}
+	cycled := map[string]bool{}
+	for i, k := range c.Turns.Cycle {
+		switch {
+		case strings.TrimSpace(k) == "":
+			errs = append(errs, fmt.Errorf("turns.cycle[%d] is empty", i))
+		case cycled[k]:
+			errs = append(errs, fmt.Errorf("turns.cycle lists %q twice", k))
+		}
+		cycled[k] = true
+	}
 	seen := map[string]bool{}
 	served := map[string]string{} // model -> the engine that serves it
 	for i, e := range c.Engine {
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index b6e4cb9..2b9ec3f 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -997,3 +997,29 @@ func TestSquattersAllow(t *testing.T) {
 		t.Errorf("allow = %q, want chrome,WindowsTerminal", got)
 	}
 }
+
+// Every [turns] and reply_timeout setting is off until the file sets it.
+func TestTurnsOffByDefaultAndValidated(t *testing.T) {
+	c, err := Load(write(t, ""))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.Turns.MaxTurn.Duration != 0 || c.Turns.Preempt || len(c.Turns.Cycle) != 0 || c.Gateway.ReplyTimeout.Duration != 0 {
+		t.Fatalf("defaults are not off: %+v %v", c.Turns, c.Gateway.ReplyTimeout)
+	}
+	c, err = Load(write(t, "[turns]\nmax_turn = \"15m\"\ncycle = [\"llm\", \"image\"]\npreempt = true\npreempt_gap = \"45s\"\n[turns.max_turn_by]\nimage = \"30m\"\n[gateway]\nreply_timeout = \"150s\"\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.Turns.MaxTurn.Minutes() != 15 || c.Turns.MaxTurnBy["image"].Minutes() != 30 || !c.Turns.Preempt ||
+		c.Turns.PreemptGap.Seconds() != 45 || len(c.Turns.Cycle) != 2 || c.Gateway.ReplyTimeout.Seconds() != 150 {
+		t.Fatalf("turns = %+v reply = %v", c.Turns, c.Gateway.ReplyTimeout)
+	}
+	_, err = Load(write(t, "[turns]\nmax_turn = \"-1m\"\npreempt_gap = \"-1s\"\ncycle = [\"a\", \"\", \"a\"]\n[turns.max_turn_by]\nx = \"-2m\"\n[gateway]\nreply_timeout = \"-1s\"\n"))
+	for _, want := range []string{"turns.max_turn -1m0s must be >= 0", "turns.preempt_gap", "turns.cycle[1] is empty",
+		`turns.cycle lists "a" twice`, "turns.max_turn_by.x", "gateway.reply_timeout"} {
+		if err == nil || !strings.Contains(err.Error(), want) {
+			t.Errorf("error lacks %q: %v", want, err)
+		}
+	}
+}
diff --git a/mkdocs.yml b/mkdocs.yml
index 47a44ed..c537f2a 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -45,7 +45,11 @@ markdown_extensions:
   - md_in_html
   - toc:
       permalink: true
-  - pymdownx.superfences
+  - pymdownx.superfences:
+      custom_fences:
+        - name: mermaid
+          class: mermaid
+          format: !!python/name:pymdownx.superfences.fence_code_format
 
 validation:
   links:

From 9f52cdeebd769d62a35637fef00fbd839e8be4d8 Mon Sep 17 00:00:00 2001
From: chapi456 <71797135+chapi456@users.noreply.github.com>
Date: Sat, 31 Oct 2026 13:39:24 +0100
Subject: [PATCH 2/2] feat(gateway): reply_timeout replays a silent engine's
 request once, then 504

With [gateway] reply_timeout set, an engine that takes a chat request and
sends no response headers in that time gets it once more (started again if it
went away); a second silence answers 504 with the code no_reply instead of
leaving the client hanging. Off by default. The changelog covers both this
and the turns commit.

Refs #275

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md                          |  9 +++
 docs/gateway.md                       | 16 +++++
 internal/cli/gateway.go               |  2 +-
 internal/gateway/gateway.go           | 73 ++++++++++++++++++++++-
 internal/gateway/replytimeout_test.go | 86 +++++++++++++++++++++++++++
 5 files changed, 183 insertions(+), 3 deletions(-)
 create mode 100644 internal/gateway/replytimeout_test.go

diff --git a/CHANGELOG.md b/CHANGELOG.md
index c1d73a6..c155f22 100644
--- a/CHANGELOG.md
+++ b/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,
diff --git a/docs/gateway.md b/docs/gateway.md
index 7b5648f..4ba7407 100644
--- a/docs/gateway.md
+++ b/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
diff --git a/internal/cli/gateway.go b/internal/cli/gateway.go
index 5654e6c..9bb1a48 100644
--- a/internal/cli/gateway.go
+++ b/internal/cli/gateway.go
@@ -82,7 +82,7 @@ func runGateway(c *Context) int {
 	if err != nil {
 		return c.fail(Systemf("api key: %v", err))
 	}
-	g := &gateway.Gateway{Socket: *sock, Engines: cfg.Engine, Ready: *ready, Key: key, Log: log, HeadroomFloor: *headroomMiB << 20,
+	g := &gateway.Gateway{Socket: *sock, Engines: cfg.Engine, Ready: *ready, ReplyTimeout: cfg.Gateway.ReplyTimeout.Duration, Key: key, Log: log, HeadroomFloor: *headroomMiB << 20,
 		AllowHosts: allow, StateFile: defaultStateFile("gateway-engines.json"),
 		CostFile: defaultStateFile("swapcost.json"), DeferredFile: defaultStateFile("comfyui-queue.json"), ConfigPath: *cfgPath}
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go
index ec96d20..0c0a2e9 100644
--- a/internal/gateway/gateway.go
+++ b/internal/gateway/gateway.go
@@ -50,6 +50,10 @@ type Gateway struct {
 	Engines []config.Engine // configured engines; those with Start can be started on demand
 	Ready   time.Duration   // how long a start may wait for a berth and then to answer; 0 = 10 min
 	Key     string          // the node's API key; required on every request
+	// ReplyTimeout is how long a live engine may take to start replying to a
+	// chat request before the gateway sends it once more, then answers 504;
+	// 0 = wait for the engine as long as the client does.
+	ReplyTimeout time.Duration
 	// AllowHosts are extra Host names the loopback gateway answers to: the name a
 	// same-machine reverse proxy (tailscale serve) forwards. The listener stays on
 	// loopback and the key and Origin checks still apply.
@@ -597,16 +601,81 @@ func (g *Gateway) forward(w http.ResponseWriter, r *http.Request) {
 		apiError(w, http.StatusInternalServerError, "engine url: "+err.Error())
 		return
 	}
+	rw := w
+	if preamble {
+		rw = bodyOnly{w} // the SSE preamble owns the headers
+	}
+	if g.ReplyTimeout <= 0 {
+		g.proxyTo(rw, r, u, body, nil)
+		return
+	}
+	// A reply timeout is set: an engine that takes the request and says
+	// nothing is asked once more (started again if it went away), then
+	// answered 504, instead of leaving the client hanging.
+	var silent bool
+	g.proxyTo(rw, r, u, body, &silent)
+	if !silent || r.Context().Err() != nil {
+		return
+	}
+	g.log().Warn("engine did not reply, replaying the request once", "engine", engineName(t.engine), "request", req.Model, "after", g.ReplyTimeout)
+	if t.engine != nil {
+		if err := g.ensure(r.Context(), t.engine, false); err != nil {
+			g.replayFailed(w, preamble, engineName(t.engine), req.Model, err)
+			return
+		}
+	}
+	silent = false
+	g.proxyTo(rw, r, u, body, &silent)
+	if silent && r.Context().Err() == nil {
+		g.replayFailed(w, preamble, engineName(t.engine), req.Model,
+			fmt.Errorf("the engine did not reply in %s, twice", g.ReplyTimeout))
+	}
+}
+
+// proxyTo sends the buffered request to the engine at u. With silent set,
+// an engine that sends no response headers within ReplyTimeout is reported
+// there and nothing is written, so the caller may send the request again.
+func (g *Gateway) proxyTo(w http.ResponseWriter, r *http.Request, u *url.URL, body []byte, silent *bool) {
 	p := httputil.NewSingleHostReverseProxy(u)
 	p.FlushInterval = -1 // stream tokens as the engine sends them
+	if silent != nil {
+		tr := http.DefaultTransport.(*http.Transport).Clone()
+		tr.ResponseHeaderTimeout = g.ReplyTimeout
+		p.Transport = tr
+		p.ErrorHandler = func(rw http.ResponseWriter, _ *http.Request, err error) {
+			var ne net.Error
+			if errors.As(err, &ne) && ne.Timeout() {
+				*silent = true
+				return
+			}
+			g.log().Warn("gateway proxy", "err", err)
+			rw.WriteHeader(http.StatusBadGateway)
+		}
+	}
 	r.Body = io.NopCloser(bytes.NewReader(body))
 	r.ContentLength = int64(len(body))
 	r.Host = u.Host
+	p.ServeHTTP(w, r)
+}
+
+func engineName(e *config.Engine) string {
+	if e == nil {
+		return "engine"
+	}
+	return e.Name
+}
+
+// replayFailed answers a request the engine never replied to, a 504 with
+// the reason; a streaming request gets the same error as an SSE event.
+func (g *Gateway) replayFailed(w http.ResponseWriter, preamble bool, engine, what string, err error) {
+	g.log().Warn("gateway request not answered", "engine", engine, "request", what, "err", err)
 	if preamble {
-		p.ServeHTTP(bodyOnly{w}, r) // the SSE preamble owns the headers
+		b, _ := json.Marshal(berthError("no_reply", err.Error()))
+		fmt.Fprintf(w, "data: %s\ndata: [DONE]\n\n", b)
+		_ = http.NewResponseController(w).Flush()
 		return
 	}
-	p.ServeHTTP(w, r)
+	apiErrorCode(w, http.StatusGatewayTimeout, "no_reply", err.Error())
 }
 
 // keepAliveEvery is how often a streaming request that waits for its
diff --git a/internal/gateway/replytimeout_test.go b/internal/gateway/replytimeout_test.go
new file mode 100644
index 0000000..45bcc7b
--- /dev/null
+++ b/internal/gateway/replytimeout_test.go
@@ -0,0 +1,86 @@
+// Copyright 2026 Trevin Tindall (T-Crypt)
+// SPDX-License-Identifier: Apache-2.0
+
+package gateway
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// An engine that takes the request and sends no headers within ReplyTimeout
+// is reported silent and nothing reaches the client, so the caller can send
+// the buffered request again; an engine that replies in time is proxied.
+func TestProxyToReportsASilentEngine(t *testing.T) {
+	var calls atomic.Int32
+	release := make(chan struct{})
+	engine := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if calls.Add(1) == 1 {
+			select { // the first request is swallowed
+			case <-release:
+			case <-r.Context().Done():
+			}
+			return
+		}
+		_, _ = io.WriteString(w, "pong")
+	}))
+	defer engine.Close()
+	defer close(release)
+	u, _ := url.Parse(engine.URL)
+	g := &Gateway{ReplyTimeout: 150 * time.Millisecond}
+	body := []byte(`{"model":"m"}`)
+
+	var silent bool
+	rec := httptest.NewRecorder()
+	g.proxyTo(rec, httptest.NewRequest("POST", "/v1/chat/completions", strings.NewReader(string(body))), u, body, &silent)
+	if !silent {
+		t.Fatalf("a silent engine was not reported (code %d, body %q)", rec.Code, rec.Body.String())
+	}
+	if rec.Body.Len() != 0 {
+		t.Errorf("the client was written to: %q", rec.Body.String())
+	}
+
+	silent = false
+	rec = httptest.NewRecorder()
+	g.proxyTo(rec, httptest.NewRequest("POST", "/v1/chat/completions", strings.NewReader(string(body))), u, body, &silent)
+	if silent || rec.Body.String() != "pong" {
+		t.Fatalf("replay: silent=%v body=%q, want the engine's reply", silent, rec.Body.String())
+	}
+}
+
+// With no ReplyTimeout, proxyTo is the plain reverse proxy it was.
+func TestProxyToWithoutATimeoutIsThePlainProxy(t *testing.T) {
+	engine := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		_, _ = w.Write(b)
+	}))
+	defer engine.Close()
+	u, _ := url.Parse(engine.URL)
+	rec := httptest.NewRecorder()
+	(&Gateway{}).proxyTo(rec, httptest.NewRequest("POST", "/", nil), u, []byte("echo"), nil)
+	if rec.Body.String() != "echo" {
+		t.Fatalf("body = %q", rec.Body.String())
+	}
+}
+
+// The second silence is a 504 with a reason code, as JSON; a streaming
+// request gets the error as one SSE event.
+func TestReplayFailedAnswers504(t *testing.T) {
+	g := &Gateway{}
+	rec := httptest.NewRecorder()
+	g.replayFailed(rec, false, "llm", "m", io.ErrUnexpectedEOF)
+	if rec.Code != http.StatusGatewayTimeout || !strings.Contains(rec.Body.String(), `"code":"no_reply"`) {
+		t.Fatalf("code %d body %q", rec.Code, rec.Body.String())
+	}
+	rec = httptest.NewRecorder()
+	g.replayFailed(rec, true, "llm", "m", io.ErrUnexpectedEOF)
+	if !strings.Contains(rec.Body.String(), "no_reply") || !strings.Contains(rec.Body.String(), "[DONE]") {
+		t.Fatalf("sse body %q", rec.Body.String())
+	}
+}