Merge pull request #2518 from simonferquel-clanker/fix/steer-queue-newline-separator

runtime: append newline to non-last steer messages on multi-drain

Author: simonferquel

pkg/runtime/loop.go

@@ -47,6 +47,65 @@ func (r *LocalRuntime) appendSteerAndEmit(sess *session.Session, sm QueuedMessag
 	events <- UserMessage(sm.Content, sess.ID, sm.MultiContent, len(sess.Messages)-1)
 }
 
+// drainAndEmitSteered drains all messages from the steer queue and injects
+// them into the session as individual user messages. When multiple messages
+// are drained, a "\n" is appended to the content of every non-last message.
+// Some chat templates concatenate consecutive user messages without a
+// separator before tokenisation, which would cause trailing/leading word
+// fragments from adjacent messages to be glued together. The "\n" prevents
+// this without merging the messages into one.
+//
+// It also snapshots the message count before any messages are added and
+// returns it alongside the drained flag so the caller can pass it to
+// compactIfNeeded without a separate len(sess.GetAllMessages()) call.
+//
+// NOTE: the appended \n is persisted in the session message and included in
+// UserMessageEvent. This is a deliberate trade-off: because the runtime passes
+// chat.Message slices directly to the provider, this is the only injection
+// point that doesn't require restructuring. TUI consumers may see a trailing
+// newline on non-last steered messages in multi-drain batches.
+//
+// Returns (true, messageCountBefore) if any messages were drained and emitted;
+// (false, 0) otherwise.
+func (r *LocalRuntime) drainAndEmitSteered(ctx context.Context, sess *session.Session, events chan<- Event) (bool, int) {
+	steered := r.steerQueue.Drain(ctx)
+	if len(steered) == 0 {
+		return false, 0
+	}
+	messageCountBefore := len(sess.GetAllMessages())
+	for i, sm := range steered {
+		if i < len(steered)-1 {
+			sm = appendNewlineToQueuedMessage(sm)
+		}
+		r.appendSteerAndEmit(sess, sm, events)
+	}
+	return true, messageCountBefore
+}
+
+// appendNewlineToQueuedMessage returns sm with "\n" appended to its text
+// content; never mutates the caller's slice contents.
+// For plain-text messages Content is extended. For multi-content messages
+// only the last part is considered: if it is a text part, "\n" is appended
+// to it in a shallow copy of the slice. If the last part is not text type
+// (e.g. image), sm is returned unchanged — non-text parts carry their own
+// provider envelope that acts as a separator.
+func appendNewlineToQueuedMessage(sm QueuedMessage) QueuedMessage {
+	if len(sm.MultiContent) == 0 {
+		sm.Content += "\n"
+		return sm
+	}
+	// Only act if the last part is a text part.
+	last := len(sm.MultiContent) - 1
+	if sm.MultiContent[last].Type != chat.MessagePartTypeText {
+		return sm
+	}
+	// Shallow-copy the slice so we don't mutate the original.
+	parts := append([]chat.MessagePart(nil), sm.MultiContent...)
+	parts[last].Text += "\n"
+	sm.MultiContent = parts
+	return sm
+}
+
 // finalizeEventChannel performs cleanup at the end of a RunStream goroutine:
 // restores the previous elicitation channel, emits the StreamStopped event,
 // fires hooks, and closes the events channel.
@@ -302,11 +361,7 @@ func (r *LocalRuntime) RunStream(ctx context.Context, sess *session.Session) <-c
 
 			// Drain steer messages queued while idle or before the first model call
 			// (covers idle-window and first-turn-miss races).
-			if steered := r.steerQueue.Drain(ctx); len(steered) > 0 {
-				messageCountBeforeSteer := len(sess.GetAllMessages())
-				for _, sm := range steered {
-					r.appendSteerAndEmit(sess, sm, events)
-				}
+			if drained, messageCountBeforeSteer := r.drainAndEmitSteered(ctx, sess, events); drained {
 				r.compactIfNeeded(ctx, sess, a, m, contextLimit, messageCountBeforeSteer, events)
 			}
 
@@ -435,11 +490,7 @@ func (r *LocalRuntime) RunStream(ctx context.Context, sess *session.Session) <-c
 			toolModelOverride = resolveToolCallModelOverride(res.Calls, agentTools)
 
 			// Drain steer messages that arrived during tool calls.
-			if steered := r.steerQueue.Drain(ctx); len(steered) > 0 {
-				for _, sm := range steered {
-					r.appendSteerAndEmit(sess, sm, events)
-				}
-
+			if drained, _ := r.drainAndEmitSteered(ctx, sess, events); drained {
 				r.compactIfNeeded(ctx, sess, a, m, contextLimit, messageCountBeforeTools, events)
 				continue
 			}
@@ -449,10 +500,7 @@ func (r *LocalRuntime) RunStream(ctx context.Context, sess *session.Session) <-c
 				r.executeStopHooks(ctx, sess, a, res.Content, events)
 
 				// Re-check steer queue: closes the race between the mid-loop drain and this stop.
-				if steered := r.steerQueue.Drain(ctx); len(steered) > 0 {
-					for _, sm := range steered {
-						r.appendSteerAndEmit(sess, sm, events)
-					}
+				if drained, _ := r.drainAndEmitSteered(ctx, sess, events); drained {
 					r.compactIfNeeded(ctx, sess, a, m, contextLimit, messageCountBeforeTools, events)
 					continue
 				}

pkg/runtime/runtime_test.go

@@ -2588,3 +2588,185 @@ func TestSteer_EndOfIterationRaceIsConsumedInCurrentRunStream(t *testing.T) {
 	assert.NotContains(t, steerSessionMsg.Message.Content, "<system-reminder>",
 		"end-of-iteration steer must NOT use the system-reminder envelope")
 }
+
+func TestAppendNewlineToQueuedMessage(t *testing.T) {
+	t.Parallel()
+
+	t.Run("plain-text message gets newline appended to Content", func(t *testing.T) {
+		sm := QueuedMessage{Content: "hello"}
+		got := appendNewlineToQueuedMessage(sm)
+		assert.Equal(t, "hello\n", got.Content)
+		assert.Nil(t, got.MultiContent)
+	})
+
+	t.Run("multi-content message with last part text gets newline on that part", func(t *testing.T) {
+		sm := QueuedMessage{
+			MultiContent: []chat.MessagePart{
+				{Type: chat.MessagePartTypeImageURL, ImageURL: &chat.MessageImageURL{URL: "https://example.com/img.png"}},
+				{Type: chat.MessagePartTypeText, Text: "and this"},
+			},
+		}
+		got := appendNewlineToQueuedMessage(sm)
+		// Last part is text — \n appended to it.
+		assert.Equal(t, "and this\n", got.MultiContent[1].Text)
+		// Image part unchanged.
+		assert.Equal(t, chat.MessagePartTypeImageURL, got.MultiContent[0].Type)
+	})
+
+	t.Run("multi-content message with last part non-text is returned unchanged", func(t *testing.T) {
+		sm := QueuedMessage{
+			MultiContent: []chat.MessagePart{
+				{Type: chat.MessagePartTypeText, Text: "look at this"},
+				{Type: chat.MessagePartTypeImageURL, ImageURL: &chat.MessageImageURL{URL: "https://example.com/img.png"}},
+			},
+		}
+		got := appendNewlineToQueuedMessage(sm)
+		// Last part is image — non-text parts have their own envelope separator;
+		// return unchanged.
+		assert.Equal(t, "look at this", got.MultiContent[0].Text)
+		assert.Equal(t, chat.MessagePartTypeImageURL, got.MultiContent[1].Type)
+	})
+
+	t.Run("multi-content message with no text part is returned unchanged", func(t *testing.T) {
+		sm := QueuedMessage{
+			MultiContent: []chat.MessagePart{
+				{Type: chat.MessagePartTypeImageURL, ImageURL: &chat.MessageImageURL{URL: "https://example.com/img.png"}},
+			},
+		}
+		got := appendNewlineToQueuedMessage(sm)
+		// Image-only messages have no text part to append \n to; they are immune to
+		// the run-on tokenisation problem because non-text parts carry their own
+		// envelope that acts as a separator. Return unchanged.
+		require.Len(t, got.MultiContent, 1)
+		assert.Equal(t, chat.MessagePartTypeImageURL, got.MultiContent[0].Type)
+	})
+
+	t.Run("original QueuedMessage is not mutated", func(t *testing.T) {
+		parts := []chat.MessagePart{
+			{Type: chat.MessagePartTypeText, Text: "original"},
+		}
+		sm := QueuedMessage{MultiContent: parts}
+		_ = appendNewlineToQueuedMessage(sm)
+		assert.Equal(t, "original", parts[0].Text, "original slice must not be mutated")
+	})
+
+	t.Run("plain-text original not mutated", func(t *testing.T) {
+		sm := QueuedMessage{Content: "x"}
+		_ = appendNewlineToQueuedMessage(sm)
+		assert.Equal(t, "x", sm.Content)
+	})
+}
+
+// TestDrainAndEmitSteered_MultipleMessages verifies that when multiple messages
+// are drained from the steer queue, each is emitted as a separate session
+// message and non-last messages have "\n" appended to their content, preventing
+// the LLM from tokenising adjacent words across message boundaries as a run-on
+// string.
+func TestDrainAndEmitSteered_MultipleMessages(t *testing.T) {
+	t.Parallel()
+
+	// Use a stream that never gets called — we only exercise drainAndEmitSteered directly.
+	prov := &mockProvider{id: "test/mock-model", stream: &mockStream{}}
+	root := agent.New("root", "You are a test agent", agent.WithModel(prov))
+	tm := team.New(team.WithAgents(root))
+
+	rt, err := NewLocalRuntime(tm, WithSessionCompaction(false), WithModelStore(mockModelStore{}))
+	require.NoError(t, err)
+
+	// Enqueue three plain-text steer messages before draining.
+	require.NoError(t, rt.Steer(QueuedMessage{Content: "first"}))
+	require.NoError(t, rt.Steer(QueuedMessage{Content: "second"}))
+	require.NoError(t, rt.Steer(QueuedMessage{Content: "third"}))
+
+	sess := session.New()
+	events := make(chan Event, 16)
+
+	drained, _ := rt.drainAndEmitSteered(t.Context(), sess, events)
+	close(events)
+
+	assert.True(t, drained, "should report messages were drained")
+
+	// Three separate session messages must have been added.
+	var userMsgs []string
+	for _, item := range sess.Messages {
+		if item.IsMessage() && item.Message.Message.Role == chat.MessageRoleUser {
+			userMsgs = append(userMsgs, item.Message.Message.Content)
+		}
+	}
+	require.Len(t, userMsgs, 3, "expected 3 independent user messages")
+
+	// Non-last messages must have "\n" appended; the last must not.
+	assert.Equal(t, "first\n", userMsgs[0])
+	assert.Equal(t, "second\n", userMsgs[1])
+	assert.Equal(t, "third", userMsgs[2])
+
+	// The UserMessageEvent contents must mirror the session messages.
+	var eventMsgs []string
+	for ev := range events {
+		if ue, ok := ev.(*UserMessageEvent); ok {
+			eventMsgs = append(eventMsgs, ue.Message)
+		}
+	}
+	require.Len(t, eventMsgs, 3)
+	assert.Equal(t, "first\n", eventMsgs[0])
+	assert.Equal(t, "second\n", eventMsgs[1])
+	assert.Equal(t, "third", eventMsgs[2])
+}
+
+// TestDrainAndEmitSteered_MultiContent verifies that the "\n" separator is
+// correctly appended to multi-content messages: specifically to the last text
+// part rather than the Content field.
+func TestDrainAndEmitSteered_MultiContent(t *testing.T) {
+	t.Parallel()
+
+	prov := &mockProvider{id: "test/mock-model", stream: &mockStream{}}
+	root := agent.New("root", "You are a test agent", agent.WithModel(prov))
+	tm := team.New(team.WithAgents(root))
+
+	rt, err := NewLocalRuntime(tm, WithSessionCompaction(false), WithModelStore(mockModelStore{}))
+	require.NoError(t, err)
+
+	// Two multi-content messages.
+	require.NoError(t, rt.Steer(QueuedMessage{
+		Content: "first",
+		MultiContent: []chat.MessagePart{
+			{Type: chat.MessagePartTypeText, Text: "first"},
+			{Type: chat.MessagePartTypeImageURL, ImageURL: &chat.MessageImageURL{URL: "https://example.com/a.png"}},
+			{Type: chat.MessagePartTypeText, Text: "first-text-after-img"},
+		},
+	}))
+	require.NoError(t, rt.Steer(QueuedMessage{
+		Content: "second",
+		MultiContent: []chat.MessagePart{
+			{Type: chat.MessagePartTypeText, Text: "second"},
+		},
+	}))
+
+	sess := session.New()
+	events := make(chan Event, 16)
+
+	drained, _ := rt.drainAndEmitSteered(t.Context(), sess, events)
+	close(events)
+
+	assert.True(t, drained)
+
+	// Two session messages.
+	var items []session.Item
+	for _, item := range sess.Messages {
+		if item.IsMessage() && item.Message.Message.Role == chat.MessageRoleUser {
+			items = append(items, item)
+		}
+	}
+	require.Len(t, items, 2)
+
+	// First message: last text part must have "\n" appended.
+	firstParts := items[0].Message.Message.MultiContent
+	require.Len(t, firstParts, 3)
+	assert.Equal(t, "first-text-after-img\n", firstParts[2].Text, "last text part of non-last message should have \\n")
+	assert.Equal(t, "first", firstParts[0].Text, "other text parts must be unchanged")
+
+	// Second (last) message: no modification.
+	secondParts := items[1].Message.Message.MultiContent
+	require.Len(t, secondParts, 1)
+	assert.Equal(t, "second", secondParts[0].Text, "last message must not be modified")
+}