From b9c729457b55458542e7779bfd21afdb8c163692 Mon Sep 17 00:00:00 2001 From: Kyle Carberry Date: Fri, 6 Mar 2026 15:15:40 -0800 Subject: [PATCH] fix(chatd): queue interrupt messages to preserve conversation order (#22736) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem When `message_agent` is called with `interrupt=true`, two independent code paths race to persist messages: 1. `SendMessage` inserts the **user message** into `chat_messages` at time T1 2. `persistInterruptedStep` saves the partial **assistant response** at time T2 (T2 > T1) Since `chat_messages` are ordered by `(created_at, id)`, the assistant message ends up **after** the user message that triggered the interrupt. On reload, this produces a broken conversation where the interrupted response appears below the new user message — and Anthropic rejects the trailing assistant message as unsupported prefill. The root cause is that **two independent writers can't guarantee ordering**. Any solution involving timestamp manipulation or signal-then-wait coordination leaves race windows. ## Fix Route interrupt behavior through the existing queued message mechanism: 1. `SendMessage` with `BusyBehaviorInterrupt` now inserts into `chat_queued_messages` (not `chat_messages`) when the chat is busy 2. After queuing, `setChatWaiting` signals the running loop to stop 3. The deferred cleanup in `processChat` persists the partial assistant response first, then auto-promotes the queued user message This eliminates the race entirely: the assistant partial response and user message are written by the same serialized cleanup flow, so ordering is guaranteed by the DB's auto-incrementing `id` sequence. No timestamp hacks, no reordering at send time. Supersedes #22728 — fixes the root cause instead of reordering at prompt construction time. --- coderd/chatd/chatd.go | 40 ++++++++++++++++++++++++++++++++++---- coderd/chatd/chatd_test.go | 23 +++++++++++++--------- 2 files changed, 50 insertions(+), 13 deletions(-) diff --git a/coderd/chatd/chatd.go b/coderd/chatd/chatd.go index 2cad01f3e3..faacc91b17 100644 --- a/coderd/chatd/chatd.go +++ b/coderd/chatd/chatd.go @@ -182,8 +182,10 @@ type SendMessageBusyBehavior string const ( // SendMessageBusyBehaviorQueue queues user messages while the chat is busy. SendMessageBusyBehaviorQueue SendMessageBusyBehavior = "queue" - // SendMessageBusyBehaviorInterrupt inserts the message immediately and - // transitions the chat to pending, which interrupts the active run. + // SendMessageBusyBehaviorInterrupt queues the message and + // interrupts the active run. The queued message is + // auto-promoted after the interrupted assistant response is + // persisted, ensuring correct message ordering. SendMessageBusyBehaviorInterrupt SendMessageBusyBehavior = "interrupt" ) @@ -376,8 +378,15 @@ func (p *Server) SendMessage( modelConfigID = *opts.ModelConfigID } - if busyBehavior == SendMessageBusyBehaviorQueue && - shouldQueueUserMessage(lockedChat.Status) { + // Both queue and interrupt behaviors queue messages + // when the chat is busy. Interrupt additionally + // signals the running loop to stop so the queued + // message is promoted sooner. Crucially, this + // guarantees the interrupted assistant response is + // persisted (with a lower id/created_at) before the + // user message is promoted into chat_messages, + // preserving correct conversation order. + if shouldQueueUserMessage(lockedChat.Status) { existingQueued, err := tx.GetChatQueuedMessages(ctx, opts.ChatID) if err != nil { return xerrors.Errorf("get queued messages: %w", err) @@ -434,6 +443,29 @@ func (p *Server) SendMessage( p.publishChatStreamNotify(opts.ChatID, coderdpubsub.ChatStreamNotifyMessage{ QueueUpdate: true, }) + + // For interrupt behavior, signal the running loop to + // stop. setChatWaiting publishes a status notification + // that the worker's control subscriber detects, causing + // it to cancel with ErrInterrupted. The deferred cleanup + // in processChat then auto-promotes the queued message + // after persisting the partial assistant response. + if busyBehavior == SendMessageBusyBehaviorInterrupt { + updatedChat, err := p.setChatWaiting(ctx, opts.ChatID) + if err != nil { + // The message is already queued so the chat is + // not in a broken state — the user can still + // wait for the current run to finish. Log the + // error but don't fail the request. + p.logger.Error(ctx, "failed to interrupt chat for queued message", + slog.F("chat_id", opts.ChatID), + slog.Error(err), + ) + } else { + result.Chat = updatedChat + } + } + return result, nil } diff --git a/coderd/chatd/chatd_test.go b/coderd/chatd/chatd_test.go index a04b2b04b2..7fac515ecd 100644 --- a/coderd/chatd/chatd_test.go +++ b/coderd/chatd/chatd_test.go @@ -367,7 +367,7 @@ func TestSendMessageQueueBehaviorQueuesWhenBusy(t *testing.T) { require.Len(t, messages, 1) } -func TestSendMessageInterruptBehaviorSendsImmediatelyWhenBusy(t *testing.T) { +func TestSendMessageInterruptBehaviorQueuesAndInterruptsWhenBusy(t *testing.T) { t.Parallel() db, ps := dbtestutil.NewDB(t) @@ -399,26 +399,31 @@ func TestSendMessageInterruptBehaviorSendsImmediatelyWhenBusy(t *testing.T) { BusyBehavior: chatd.SendMessageBusyBehaviorInterrupt, }) require.NoError(t, err) - require.False(t, result.Queued) - require.Equal(t, database.ChatStatusPending, result.Chat.Status) - require.False(t, result.Chat.WorkerID.Valid) + + // The message should be queued, not inserted directly. + require.True(t, result.Queued) + require.NotNil(t, result.QueuedMessage) + + // The chat should transition to waiting (interrupt signal), + // not pending. + require.Equal(t, database.ChatStatusWaiting, result.Chat.Status) fromDB, err := db.GetChatByID(ctx, chat.ID) require.NoError(t, err) - require.Equal(t, database.ChatStatusPending, fromDB.Status) - require.False(t, fromDB.WorkerID.Valid) + require.Equal(t, database.ChatStatusWaiting, fromDB.Status) + // The message should be in the queue, not in chat_messages. queued, err := db.GetChatQueuedMessages(ctx, chat.ID) require.NoError(t, err) - require.Len(t, queued, 0) + require.Len(t, queued, 1) + // Only the initial user message should be in chat_messages. messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ ChatID: chat.ID, AfterID: 0, }) require.NoError(t, err) - require.Len(t, messages, 2) - require.Equal(t, messages[len(messages)-1].ID, result.Message.ID) + require.Len(t, messages, 1) } func TestEditMessageUpdatesAndTruncatesAndClearsQueue(t *testing.T) {