gateway: a streamed request waiting for its key's slot (Concurrency) or for room in its minute (Requests per minute) has its agent sent the stream's 200 and SSE keepalives after 15s, as a held try's stream is, instead of nothing until it is sent (coeo91 on Discord: WorkBuddy, Trae and Qoder said it timed out)

serve's try took its lane slot and waited for the minute (04704111)
before holdWriter.watch started, so for up to 2 minutes (or QueueWait,
unbounded by default) the agent got no status, no headers, no byte.
keepQueued runs beside lanes.take and rpms.wait: past keepHeldAfter it
sends the stream's 200 and ": keepalive" comments every keepaliveEvery
(keepAlive, the mechanism held tries already use; none for Gemini, #934).
Once they are out the try is held, so the vendor's error, or the queue
turning it away, reaches the agent as its protocol's stream error
(failTo: Chat's error chunk, Responses' response.failed, Messages'
event: error), and the next candidate still answers in the same stream.
A wait shorter than keepHeldAfter sends nothing: its 429 keeps its
Retry-After. Non-streamed requests are unchanged.

What the agents do with it (read from their app bundles, not tried live):
- WorkBuddy 5.5.6 / WorkBuddy AI 5.6.2 (codebuddy-headless.js): no
  header timeout, first-token timer 20 min, so a 2-minute queue alone
  doesn't time it out there.
- Qoder 0.4.3: 60s to headers, then 60s more after them; its SSE parser
  drops comment lines, so keepalives don't extend that. Headers at 15s
  give it 75s instead of 60s; longer waits need firstPayloadTimeoutMs.
- Trae: its agent is downloaded at runtime and wasn't read.

Verified (Linux, golang:1.26.8 in Docker):
- queue_keepalive_test.go through lanGuard(New().Handler()), with an
  agent whose connection gives up after 400ms of silence: a request
  queued 1.2s for the minute (chat, responses, messages) and one queued
  for a Concurrency slot get 200, keepalives, then the reply; a vendor
  500 after keepalives is each protocol's stream error; a QueueWait
  waited out is the stream's error naming the limit; a short wait is
  still a plain 429. With keepQueued taken out, every one but the 429
  test fails with "i/o timeout". TestQueuedUnheldTryHeldOnceKeptAlive
  fails without the hold (": keepalive\n{\"error\":...}" in the stream).
- go test -tags nogui ./internal/gateway: ok (122s).
- go test -race on internal/gateway the way CI runs it (go test -race -c,
  build/race-shards.sh over 14 processes, -test.timeout=20m): 1258 tests,
  every shard ok. A single serial -race process passes 10m's default
  timeout, which is why CI shards it (#923).
- go vet -tags nogui ./internal/gateway for linux, windows and darwin;
  gofmt -l clean. After rebasing on 5e17a183: TestQueued, TestBurst,
  RPM, TestQuiet and TestSilentHeld under -race, and linux vet, pass.

docs/subsystems/gateway-routing.md

@@ -44,7 +44,7 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md).
 - A fallback happens only while none of the reply has been sent. An agent never gets half a reply from one upstream and the rest from another.
 - A request whose key is a Claude sign-in (`claudeSignIn`: `Authorization: Bearer sk-ant-oat…`, Claude Code signed in to claude.ai with magpie as its base URL) is never told a 401, nor a 403 saying "OAuth token has been revoked". magpie never passes that sign-in on, so such an answer is a provider refusing magpie's credential, or one of magpie's own Claude accounts lapsing; told it, Claude Code renews its own sign-in before each of its retries (ten for one request) and then asks to `/login`. Any error whose message mentions x-api-key, whatever its status, Claude Code tells as "Not logged in · Please run /login" too. It is told as a 502 in magpie's words, an `api_error` saying "<provider> refused magpie's credential (HTTP 401): …" with x-api-key said as "API key", and any other error such a request is told keeps its status with x-api-key said as "API key", as does a stream's error after its 200 (`keepsSignIn`, `claude_signin.go`). Recent calls and the usage log keep the provider's status and message. Any other 403, and a request with magpie's key or an API key, is answered as before. The sign-in's beta (`oauth-…` in `anthropic-beta`) isn't passed on either (`askedBetas`), so a provider gets the request as Claude Code with magpie's key sends it; a provider's own `header.anthropic-beta` still goes.
 - A 2xx that can't be an API's answer is the 502 it stands for (`notAnAPIReply`, #1012): a web page (`text/html`, or a body that begins as one: a sign-in page, Cloudflare's challenge), a body with nothing in it, or one sent as JSON that doesn't begin as JSON. Every upstream request goes through it (`forwardOnce`, and the built-in clients' own: the ChatGPT backend's `/responses`, Kiro, Zed, Qoder, Devin, Cursor, Command Code), so it fails over, rests and is logged as a 502 does, its reason naming the page's title. Only the first bytes are peeked, which a stream waits for anyway, and a compressed body is left as it came.
-- A streamed request is never left silent while its vendor says nothing. `holdWriter.watch` runs beside each try: past `keepHeldAfter` (15s) it sends a held try's agent the stream's 200 and SSE comments (`keepAlive`), none of the try, so another candidate may still answer in the same stream; once the reply passes, it puts a comment between two of its lines after `keepaliveEvery` of quiet. It stops after `keepaliveLongest` (5 minutes) with nothing from the vendor. A proxy in front such as Cloudflare (125s) otherwise ends the request (#947). Gemini streams get no comments (#934). A vendor that then ignores `stream:true` and answers 200 with its reply whole has it sent as the agent's stream (`wholeAsStream`: Chat, Responses, Anthropic), not as the stream's error, while a stream under a type that says neither JSON nor a stream goes through as it comes (`sseStart` reads its first bytes); a 200 that is no reply of the protocol's fails as a translated reply that didn't stream does (502, "… did not stream"), for the next candidate to answer, and the last one's reaches the agent as the stream's error.
+- A streamed request is never left silent while its vendor says nothing. `holdWriter.watch` runs beside each try: past `keepHeldAfter` (15s) it sends a held try's agent the stream's 200 and SSE comments (`keepAlive`), none of the try, so another candidate may still answer in the same stream; once the reply passes, it puts a comment between two of its lines after `keepaliveEvery` of quiet. It stops after `keepaliveLongest` (5 minutes) with nothing from the vendor. A try waiting for its key's or account's slot or for room in its minute (`lanes.take`, `rpms.wait`) is kept alive the same way (`keepQueued`), for as long as it waits; once the 200 is out, the try is held, and its failure or its queue turning it away reaches the agent as the stream's error (`failTo`), not a status (coeo91 on Discord). A wait shorter than `keepHeldAfter` sends nothing, so its 429 keeps its `Retry-After`. A proxy in front such as Cloudflare (125s) otherwise ends the request (#947). Gemini streams get no comments (#934). A vendor that then ignores `stream:true` and answers 200 with its reply whole has it sent as the agent's stream (`wholeAsStream`: Chat, Responses, Anthropic), not as the stream's error, while a stream under a type that says neither JSON nor a stream goes through as it comes (`sseStart` reads its first bytes); a 200 that is no reply of the protocol's fails as a translated reply that didn't stream does (502, "… did not stream"), for the next candidate to answer, and the last one's reaches the agent as the stream's error.
 - Claude's, GPT's and Gemini's reasoning is held (`refusesAfterThinking`, up to `holdThinking`, 4 minutes) so that a refusal after it goes to the next candidate unseen (#248). It is held only while one could: on the last candidate a refusal isn't asked again, so Claude's and GPT's reasoning streams as it comes there, and a refusal after it reaches the agent as the vendor's own. Held there, a lone relay's Claude or GPT showed Claude Code its thinking all at once with the text, and Codex none until then (iTianbao on X). Gemini's stays held on the last candidate too, for a reply that only reasoned to be asked again (#667). Any other model's reasoning is never held.
 - The order of the steps a request's accounts go through, how much the accounts' own order (*Make first*) counts under each routing, and how often an allowance is read are described for users in [How magpie picks an account](../reference.md#how-magpie-picks-an-account). Smart compares renewals window by window, biggest first, each truncated to the hour (`weighRouted`), so the order decides only when every window falls in the same hour. Affinity keeps a conversation on an account at 90–97%: only `SpentShareOf` (98%, 100% In order) stops it (`affine`), while the 90% tier (`lowShare`) orders only conversations nobody has answered.
 - A resting candidate is tried last, never dropped (`restLast`). A lone candidate is tried even while it rests, because there is no other.
@@ -65,7 +65,7 @@ served by a plugin are in [Provider and plugin ownership](provider-plugins.md).
 ## Verification
 
 ```sh
-go test -tags nogui ./internal/gateway -run 'TestQuiet|TestSilentHeld|TestRotateSpreadsSessions|TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestModelNotInPlanRestsTheModelNotTheKey|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace|TestImageTool|TestCodexNoCredits|TestClaudeGoesOnMessagesWhereTheRelayHasThem|TestAutoRelaysOnTheClientsOwnAPI|TestSearchOfferedStaysOnTheClientsAPI|TestCodexSearchAnsweredOnResponses|TestAgentEffortOnCursorLocal|TestAzureCompactionWithoutMagpiesReasoning|TestCodexCompact|TestProtection|TestNotAnAPIReply|TestAPIReplyUntouched|TestCodexBackendWebPage|TestSub2APIKeyLimitIsQuota|TestKeysWeighedByTheirWindows|TestGroupWeighsKeyWindows|TestKeyOutOfItsWindowRestsUntilReset|TestKeyPoolErrorDoesNotRestForWindow|TestKeyBackOnceItsWindowIsNotFull|TestRestingKeyReadAgainInAnOrderedGroup|TestClaudeAccountBackOnceItsWindowRenews|TestThinkingStreams|TestHeldThinkingKeepsTheAgentAlive|TestRefusalAfterThinkingFailsOver|TestGeminiEmptyRepliesAskedAgain|TestLocalUnredacted|TestCORS|TestForeignPagesRefused|TestGeminiThinkingOffAtItsLowest|TestGeminiThoughtsMarkedInExtraContent|TestClaudeSignInNotToldItsSignInFailed|TestSignInWriterReleasesA403|TestSignInBetaStaysWithMagpie|TestStructuredOutput|TestResponsesTextKeepsVerbosityBesideFormat|TestRefusedFormatToldInWords|RPM'
+go test -tags nogui ./internal/gateway -run 'TestQuiet|TestSilentHeld|TestRotateSpreadsSessions|TestRouting|TestSmartRouting|TestFallback|TestNoFallbackForOtherErrors|TestLastFallbackErrorReachesTheAgent|TestRateLimitedSinksToTheBack|TestGroupRateLimitedSinks|TestAffinity|TestSeveralKeysOnTakeOverFromEachOther|TestModelNotInPlanRestsTheModelNotTheKey|TestSubscriptionAccountsTakeOver|TestGroup|TestTrace|TestImageTool|TestCodexNoCredits|TestClaudeGoesOnMessagesWhereTheRelayHasThem|TestAutoRelaysOnTheClientsOwnAPI|TestSearchOfferedStaysOnTheClientsAPI|TestCodexSearchAnsweredOnResponses|TestAgentEffortOnCursorLocal|TestAzureCompactionWithoutMagpiesReasoning|TestCodexCompact|TestProtection|TestNotAnAPIReply|TestAPIReplyUntouched|TestCodexBackendWebPage|TestSub2APIKeyLimitIsQuota|TestKeysWeighedByTheirWindows|TestGroupWeighsKeyWindows|TestKeyOutOfItsWindowRestsUntilReset|TestKeyPoolErrorDoesNotRestForWindow|TestKeyBackOnceItsWindowIsNotFull|TestRestingKeyReadAgainInAnOrderedGroup|TestClaudeAccountBackOnceItsWindowRenews|TestThinkingStreams|TestHeldThinkingKeepsTheAgentAlive|TestRefusalAfterThinkingFailsOver|TestGeminiEmptyRepliesAskedAgain|TestLocalUnredacted|TestCORS|TestForeignPagesRefused|TestGeminiThinkingOffAtItsLowest|TestGeminiThoughtsMarkedInExtraContent|TestClaudeSignInNotToldItsSignInFailed|TestSignInWriterReleasesA403|TestSignInBetaStaysWithMagpie|TestStructuredOutput|TestResponsesTextKeepsVerbosityBesideFormat|TestRefusedFormatToldInWords|RPM|TestQueued'
 go test -tags nogui ./internal/provider -run 'TestGroup|TestCodexCreditsSwitch|TestRenewedAccountForgetsItsAllowance|TestNative|TestKeyAllowance|TestPlanKeyAllowance|TestClaudeAccountToldRenewed|TestPluginAccountToldRenewed|TestSkipsRedaction|TestSetRPM'
 ```
 

internal/gateway/fallback.go

@@ -1553,6 +1553,52 @@ func (h *holdWriter) release() {
 // watchEvery is how often watch looks at a try.
 var watchEvery = time.Second
 
+// keepQueued keeps the agent of a stream alive while its try waits for a
+// slot of its key's or account's (MaxConcurrency) or for room in its
+// minute (MaxRPM), before anything is sent to the vendor: past
+// keepHeldAfter it is sent the stream's 200 and SSE comments
+// (keepAlive), every keepaliveEvery for as long as the wait lasts, as a
+// try's held stream is. A wait of up to 2 minutes for the minute, or
+// QueueWait's for a slot, sent the agent nothing at all, which an agent's
+// or a proxy's timeout for its headers ended first (coeo91 on Discord:
+// WorkBuddy, Trae and Qoder said the request timed out). Once they are
+// sent the try is held (hold), so that a failure, or the queue turning it
+// away, reaches the agent as the stream's error (failTo), and another
+// candidate may still answer in the same stream. A wait shorter than
+// keepHeldAfter sends nothing, its 429 still a status with its
+// Retry-After. The returned func ends it, and returns once it has.
+func (h *holdWriter) keepQueued() func() {
+	if !h.streams || h.alive == nil || h.alive.proto == provider.Gemini {
+		return func() {}
+	}
+	done, over := make(chan struct{}), make(chan struct{})
+	go func() {
+		defer close(over)
+		tick := time.NewTicker(watchEvery)
+		defer tick.Stop()
+		began := time.Now()
+		for {
+			select {
+			case <-done:
+				return
+			case <-tick.C:
+				h.mu.Lock()
+				if (time.Since(began) >= keepHeldAfter || h.alive.sent) && (h.ctx == nil || h.ctx.Err() == nil) {
+					h.keepAlive()
+					if h.alive.sent {
+						h.hold = true
+					}
+				}
+				h.mu.Unlock()
+			}
+		}
+	}()
+	return func() {
+		close(done)
+		<-over
+	}
+}
+
 // watch lets the try go — stop, with slow said — once firstWait passes
 // with no first content and nothing of it sent to the agent: the stream
 // held, or no reply begun at all. And it keeps the agent of a stream alive

internal/gateway/gateway.go

@@ -1909,6 +1909,8 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro
 			// its queue full, or waited out (QueueLimit, QueueWait), it is
 			// turned away below
 			waited := time.Now()
+			// a streaming agent waiting is kept alive meanwhile
+			queuedAlive := hw.keepQueued()
 			release, err := s.lanes.take(ctx, c.who(), c.p.LaneLimit(), c.p.QueueLimit, time.Duration(c.p.QueueWait)*time.Second)
 			if err == nil && c.p.RPMLimit() > 0 {
 				// with a MaxRPM, it then waits for room in the minute, and
@@ -1920,6 +1922,7 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, from provider.Pro
 					ctx = s.paidFor(ctx, c.p, c.who())
 				}
 			}
+			queuedAlive()
 			queued = time.Since(waited).Milliseconds()
 			var tooMany *errRPM
 			if errors.Is(err, errQueueFull) || errors.Is(err, errQueueWait) || errors.As(err, &tooMany) {

internal/gateway/queue_keepalive_test.go

@@ -0,0 +1,275 @@
+package gateway
+
+import (
+	"bufio"
+	"context"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/yetone/magpie/internal/provider"
+)
+
+// streamVendor streams a short reply on chat completions, Responses and
+// Messages; fail is how many requests it answers 500 "overloaded" first,
+// after the first it lets through.
+type streamVendor struct {
+	mu    sync.Mutex
+	seen  int
+	fail  int
+	block chan struct{} // the first request waits on it, if set
+}
+
+func (v *streamVendor) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	io.Copy(io.Discard, r.Body)
+	v.mu.Lock()
+	v.seen++
+	n := v.seen
+	failing := n > 1 && v.fail > 0
+	if failing {
+		v.fail--
+	}
+	v.mu.Unlock()
+	if n == 1 && v.block != nil {
+		select {
+		case <-v.block:
+		case <-r.Context().Done():
+			return
+		}
+	}
+	if failing {
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(500)
+		io.WriteString(w, `{"error":{"type":"api_error","message":"the vendor is overloaded"}}`)
+		return
+	}
+	w.Header().Set("Content-Type", "text/event-stream")
+	switch {
+	case strings.HasSuffix(r.URL.Path, "/messages"):
+		io.WriteString(w, "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"model\":\"m\",\"content\":[],\"usage\":{\"input_tokens\":3,\"output_tokens\":0}}}\n\n"+
+			"event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n"+
+			"event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"queued ok\"}}\n\n"+
+			"event: content_block_stop\ndata: {\"type\":\"content_block_stop\",\"index\":0}\n\n"+
+			"event: message_delta\ndata: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":2}}\n\n"+
+			"event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")
+	case strings.HasSuffix(r.URL.Path, "/responses"):
+		io.WriteString(w, quietCreated+quietDelta("queued ok")+quietDone)
+	default:
+		io.WriteString(w, "data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{\"role\":\"assistant\",\"content\":\"queued ok\"}}]}\n\n"+
+			"data: {\"id\":\"c1\",\"object\":\"chat.completion.chunk\",\"model\":\"m\",\"choices\":[{\"index\":0,\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":2,\"total_tokens\":5}}\n\n"+
+			"data: [DONE]\n\n")
+	}
+}
+
+// idleConn is a connection whose reads time out after idle with nothing
+// read, as an agent's HTTP client with an idle timeout (WorkBuddy, Trae,
+// Qoder): any byte, a comment's too, starts it again.
+type idleConn struct {
+	net.Conn
+	idle time.Duration
+}
+
+func (c idleConn) Read(b []byte) (int, error) {
+	c.SetReadDeadline(time.Now().Add(c.idle))
+	return c.Conn.Read(b)
+}
+
+// idleClient is an HTTP client that gives up once it hears nothing for
+// idle, headers or body.
+func idleClient(idle time.Duration) *http.Client {
+	d := &net.Dialer{}
+	return &http.Client{Transport: &http.Transport{
+		DisableKeepAlives: true,
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			c, err := d.DialContext(ctx, network, addr)
+			if err != nil {
+				return nil, err
+			}
+			return idleConn{c, idle}, nil
+		},
+	}}
+}
+
+// queueGateway is the real server (lanGuard in front) with a provider "q"
+// of one key, set up by p, and every keepalive within milliseconds.
+func queueGateway(t *testing.T, v *streamVendor, set func(*provider.Provider)) string {
+	t.Helper()
+	fresh(t)
+	keepFast(t, keepaliveGap, 40*time.Millisecond, keepaliveLongest)
+	k, w := keepHeldAfter, watchEvery
+	keepHeldAfter, watchEvery = 80*time.Millisecond, 10*time.Millisecond
+	t.Cleanup(func() { keepHeldAfter, watchEvery = k, w })
+	up := httptest.NewServer(v)
+	t.Cleanup(up.Close)
+	p := provider.Provider{ID: "q", Name: "Queued", Key: "k", Models: []string{"m"}, Chat: up.URL + "/v1", Responses: up.URL + "/v1", Anthropic: up.URL}
+	set(&p)
+	if err := provider.Save(p); err != nil {
+		t.Fatal(err)
+	}
+	gw := httptest.NewServer(lanGuard(New().Handler()))
+	t.Cleanup(gw.Close)
+	return gw.URL
+}
+
+var queueAsks = []struct{ name, path, body string }{
+	{"chat", "/v1/chat/completions", `{"model":"q/m","stream":true,"messages":[{"role":"user","content":"hi"}]}`},
+	{"responses", "/v1/responses", `{"model":"q/m","stream":true,"input":"hi"}`},
+	{"messages", "/v1/messages", `{"model":"q/m","stream":true,"max_tokens":50,"messages":[{"role":"user","content":"hi"}]}`},
+}
+
+// askIdle sends body to path with an agent that gives up after idle of
+// silence, and returns the status, the stream read, and the read's error.
+func askIdle(gw, path, body string, idle time.Duration) (int, string, error) {
+	req, _ := http.NewRequest("POST", gw+path, strings.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	res, err := idleClient(idle).Do(req)
+	if err != nil {
+		return 0, "", err
+	}
+	defer res.Body.Close()
+	var b strings.Builder
+	_, err = io.Copy(&b, bufio.NewReader(res.Body))
+	return res.StatusCode, b.String(), err
+}
+
+// coeo91 on Discord: with a Requests per minute limit, a request held for
+// room in the minute had its agent hear nothing at all — WorkBuddy, Trae
+// and Qoder said it timed out. Through the real server, a streaming
+// request that waits 1.2s for the minute with an agent that gives up
+// after 400ms of silence is sent the stream's 200 and keepalives
+// meanwhile, then the vendor's reply, in each protocol's stream.
+func TestQueuedStreamKeptAliveForTheMinute(t *testing.T) {
+	for _, a := range queueAsks {
+		t.Run(a.name, func(t *testing.T) {
+			shortMinute(t, 1200*time.Millisecond, time.Minute)
+			gw := queueGateway(t, &streamVendor{}, func(p *provider.Provider) { p.MaxRPM = 1 })
+			if code, body, err := askIdle(gw, a.path, a.body, 400*time.Millisecond); err != nil || code != 200 {
+				t.Fatalf("first: %d %v %s", code, err, body)
+			}
+			began := time.Now()
+			code, body, err := askIdle(gw, a.path, a.body, 400*time.Millisecond)
+			if err != nil {
+				t.Fatalf("the agent gave up after %s of a queued request: %v (read %q)", time.Since(began).Round(time.Millisecond), err, body)
+			}
+			if code != 200 || !strings.Contains(body, ": keepalive") || !strings.Contains(body, "queued ok") {
+				t.Fatalf("queued = %d %q, want 200, keepalives, then the reply", code, body)
+			}
+			if d := time.Since(began); d < time.Second {
+				t.Fatalf("answered after %s, inside the minute", d)
+			}
+		})
+	}
+}
+
+// A slot of MaxConcurrency waited for is kept alive the same way.
+func TestQueuedStreamKeptAliveForASlot(t *testing.T) {
+	one := 1
+	v := &streamVendor{block: make(chan struct{})}
+	gw := queueGateway(t, v, func(p *provider.Provider) { p.MaxConcurrency = &one })
+	first := make(chan error, 1)
+	go func() {
+		_, _, err := askIdle(gw, queueAsks[0].path, queueAsks[0].body, 5*time.Second)
+		first <- err
+	}()
+	time.AfterFunc(time.Second, func() { close(v.block) })
+	time.Sleep(100 * time.Millisecond) // the first has the slot
+	code, body, err := askIdle(gw, queueAsks[0].path, queueAsks[0].body, 400*time.Millisecond)
+	if err != nil || code != 200 || !strings.Contains(body, ": keepalive") || !strings.Contains(body, "queued ok") {
+		t.Fatalf("queued for a slot = %d %v %q, want 200, keepalives, then the reply", code, err, body)
+	}
+	if err := <-first; err != nil {
+		t.Fatal(err)
+	}
+}
+
+// Once a queued request's agent has the stream's 200, the vendor failing
+// after it reaches the agent as its stream's error event, in its
+// protocol's own shape, not as a status the agent can no longer be told,
+// nor a JSON error body inside the stream.
+func TestQueuedStreamFailureIsTheStreamsError(t *testing.T) {
+	wants := map[string][]string{
+		"chat":      {`"error"`, "the vendor is overloaded"},
+		"responses": {"event: response.failed", `"status":"failed"`, "the vendor is overloaded"},
+		"messages":  {"event: error", `"type":"error"`, "the vendor is overloaded"},
+	}
+	for _, a := range queueAsks {
+		t.Run(a.name, func(t *testing.T) {
+			shortMinute(t, 1200*time.Millisecond, time.Minute)
+			gw := queueGateway(t, &streamVendor{fail: 10}, func(p *provider.Provider) { p.MaxRPM = 1 })
+			if code, body, err := askIdle(gw, a.path, a.body, 400*time.Millisecond); err != nil || code != 200 {
+				t.Fatalf("first: %d %v %s", code, err, body)
+			}
+			code, body, err := askIdle(gw, a.path, a.body, 2*time.Second)
+			if err != nil || code != 200 || !strings.Contains(body, ": keepalive") {
+				t.Fatalf("queued = %d %v %q, want the stream's 200 and keepalives", code, err, body)
+			}
+			for _, w := range wants[a.name] {
+				if !strings.Contains(body, w) {
+					t.Fatalf("stream %q doesn't say %s", body, w)
+				}
+			}
+			if strings.Contains(body, `{"error":{"type":"api_error","message":"the vendor is overloaded"}}`+"\n") || strings.HasPrefix(strings.TrimLeft(strings.ReplaceAll(body, ": keepalive\n", ""), "\n"), "{") {
+				t.Fatalf("the vendor's JSON error went into the stream raw: %q", body)
+			}
+		})
+	}
+}
+
+// A request that waits out its queue (QueueWait) after keepalives were
+// sent is told so as its stream's error, saying the limit.
+func TestQueuedStreamWaitedOutIsTheStreamsError(t *testing.T) {
+	one := 1
+	v := &streamVendor{block: make(chan struct{})}
+	defer close(v.block)
+	gw := queueGateway(t, v, func(p *provider.Provider) { p.MaxConcurrency, p.QueueWait = &one, 1 })
+	go askIdle(gw, queueAsks[2].path, queueAsks[2].body, 5*time.Second)
+	time.Sleep(100 * time.Millisecond)
+	code, body, err := askIdle(gw, queueAsks[2].path, queueAsks[2].body, 400*time.Millisecond)
+	if err != nil || code != 200 || !strings.Contains(body, ": keepalive") || !strings.Contains(body, "event: error") || !strings.Contains(body, "requests at once is its limit") {
+		t.Fatalf("waited out = %d %v %q, want keepalives, then the limit as the stream's error", code, err, body)
+	}
+}
+
+// A wait shorter than keepHeldAfter sends nothing ahead: turned away, it
+// is still a 429 with its Retry-After.
+func TestQueuedBrieflyStill429(t *testing.T) {
+	shortMinute(t, 1500*time.Millisecond, 50*time.Millisecond)
+	gw := queueGateway(t, &streamVendor{}, func(p *provider.Provider) { p.MaxRPM = 1 })
+	if code, _, err := askIdle(gw, queueAsks[0].path, queueAsks[0].body, time.Second); err != nil || code != 200 {
+		t.Fatalf("first: %d %v", code, err)
+	}
+	code, body, err := askIdle(gw, queueAsks[0].path, queueAsks[0].body, time.Second)
+	if err != nil || code != 429 || strings.Contains(body, "keepalive") {
+		t.Fatalf("second = %d %v %q, want a plain 429", code, err, body)
+	}
+}
+
+// A try that wasn't held — the last one, its retries spent — and whose
+// agent was kept alive while it waited is held from then on: the vendor's
+// error is the stream's to tell (failTo), not a JSON body written into
+// the stream the agent already has.
+func TestQueuedUnheldTryHeldOnceKeptAlive(t *testing.T) {
+	k, w := keepHeldAfter, watchEvery
+	keepHeldAfter, watchEvery = 20*time.Millisecond, 5*time.Millisecond
+	t.Cleanup(func() { keepHeldAfter, watchEvery = k, w })
+	rec := httptest.NewRecorder()
+	hw := newHoldWriter(rec, false)
+	hw.alive, hw.streams = &keptAlive{proto: provider.Chat}, true
+	stop := hw.keepQueued()
+	time.Sleep(80 * time.Millisecond)
+	stop()
+	if !hw.alive.sent {
+		t.Fatal("the agent wasn't kept alive while its try waited")
+	}
+	hw.Header().Set("Content-Type", "application/json")
+	hw.WriteHeader(400)
+	hw.Write([]byte(`{"error":{"message":"bad request"}}`))
+	if hw.passing || strings.Contains(rec.Body.String(), "bad request") {
+		t.Fatalf("the vendor's error went into the kept-alive stream: passing %v, %q", hw.passing, rec.Body.String())
+	}
+}