From 55ad3d0b28540b56d50045253efab96aa27f2a9b Mon Sep 17 00:00:00 2001 From: Ettore Di Giacinto Date: Tue, 3 Nov 2026 08:22:43 +0000 Subject: [PATCH] fix(studio): cap the messages that wait for the next turn The queue of messages that wait while a turn ends or the harness swaps had no limit. Cap it at 32 messages and 256 KiB, and refuse a message over the cap with ErrTurnRunning before it reaches the transcript. Assisted-by: Claude Code:claude-sonnet-5-5 Signed-off-by: Ettore Di Giacinto --- core/services/studio/session/chat.go | 35 ++++++++++++++++++--- core/services/studio/session/export_test.go | 6 ++++ core/services/studio/session/queue_test.go | 17 ++++++++++ 3 files changed, 54 insertions(+), 4 deletions(-) diff --git a/core/services/studio/session/chat.go b/core/services/studio/session/chat.go index 163dc3ddae..26bc9f3277 100644 --- a/core/services/studio/session/chat.go +++ b/core/services/studio/session/chat.go @@ -689,6 +689,12 @@ func (c *Chat) send(ctx context.Context, msg Message, logged bool) error { c.mu.Unlock() return fmt.Errorf("%w: the turn is stopping", ErrTurnRunning) } + if c.pendingFullLocked(len(msg.Text) + len(msg.Note)) { + // Refused before the message reaches the transcript, so a retry + // does not log it twice. + c.mu.Unlock() + return fmt.Errorf("%w: too many messages are waiting for the turn", ErrTurnRunning) + } hs, turn, swapping := c.hs, c.turn, c.swapping if hs == nil || swapping { // The harness is being swapped: queue the message for the next @@ -752,18 +758,39 @@ func (c *Chat) send(ctx context.Context, msg Message, logged bool) error { return nil } +// Limits of the queue of messages that wait for the next turn. The queue only +// fills while a turn ends or the harness swaps, so a full one means a client +// that sends faster than the chat can take. +const ( + maxPendingMessages = 32 + maxPendingBytes = 256 << 10 +) + +// pendingFullLocked says whether a message of n more bytes does not fit in the +// queue. The caller holds c.mu. +func (c *Chat) pendingFullLocked(n int) bool { + if len(c.pending) >= maxPendingMessages { + return true + } + total := n + for _, p := range c.pending { + total += len(p) + } + return total > maxPendingBytes +} + // queueLogged queues text, which is in the transcript already, for the next // turn of the running loop, or starts it when the turn has ended meanwhile. -// When the turn was stopped meanwhile, the message waits for the turn to end -// and starts the next one: it is in the transcript, so it must not be refused -// any more. +// When the turn was stopped, or the queue filled up since Send checked it, the +// message waits for the turn to end and starts the next one: it was accepted +// into the transcript, so it must not be refused any more. func (c *Chat) queueLogged(ctx context.Context, msg Message, text string) error { c.mu.Lock() if !c.running { c.mu.Unlock() return c.send(ctx, msg, true) } - if c.stopped { + if c.stopped || c.pendingFullLocked(len(text)) { done := c.done c.mu.Unlock() select { diff --git a/core/services/studio/session/export_test.go b/core/services/studio/session/export_test.go index 1abd660442..acd1cafb8d 100644 --- a/core/services/studio/session/export_test.go +++ b/core/services/studio/session/export_test.go @@ -7,3 +7,9 @@ func (c *Chat) HarnessStateLen() int { defer c.mu.Unlock() return len(c.rec.HarnessState) } + +// The limits of the queue of waiting messages, for the specs. +const ( + MaxPendingMessages = maxPendingMessages + MaxPendingBytes = maxPendingBytes +) diff --git a/core/services/studio/session/queue_test.go b/core/services/studio/session/queue_test.go index 36dd6df29b..c9a98fa2e8 100644 --- a/core/services/studio/session/queue_test.go +++ b/core/services/studio/session/queue_test.go @@ -2,6 +2,8 @@ package session_test import ( "context" + "fmt" + "strings" "sync" "sync/atomic" "time" @@ -80,4 +82,19 @@ var _ = Describe("messages that wait for the next turn", func() { Expect(userTexts()).To(ContainElement("second"), "the message is not in the transcript") Eventually(func() []string { return h.Sent() }).Should(HaveLen(2), "the message never reached a turn") }) + + It("refuses a message when too many wait, before it reaches the transcript", func() { + var refused error + for i := 0; i < 2*session.MaxPendingMessages && refused == nil; i++ { + refused = chat.Send(context.Background(), session.Message{Text: fmt.Sprintf("m%d", i)}) + } + Expect(refused).To(MatchError(session.ErrTurnRunning)) + Expect(userTexts()).To(HaveLen(1+session.MaxPendingMessages), "the refused message was logged") + }) + + It("refuses a message when the waiting ones are too big", func() { + big := strings.Repeat("x", session.MaxPendingBytes/2+1) + Expect(chat.Send(context.Background(), session.Message{Text: big})).To(Succeed()) + Expect(chat.Send(context.Background(), session.Message{Text: big})).To(MatchError(session.ErrTurnRunning)) + }) })