fix(#2053): filter sub-agent streaming events from parent session persistence

During task transfers, sub-agent streaming events (AgentChoiceEvent,
AgentChoiceReasoningEvent, MessageAddedEvent) were forwarded through
the parent session's event channel. The PersistentRuntime's
handleEvent checked sess.IsSubSession(), but sess was always the
parent session, so the guard never triggered. This caused sub-agent
assistant messages to be persisted directly into the parent session's
message history, corrupting it.

On session restore, the parent session contained interleaved sub-agent
messages with tool_use blocks that had no corresponding tool_result
messages in the parent context, causing Anthropic API errors:
"unexpected tool_use_id found in tool_result blocks".

Add SessionID field to AgentChoiceEvent and AgentChoiceReasoningEvent,
and filter all streaming/message events by comparing the event's
SessionID against the parent session's ID. Events from sub-sessions
are now silently skipped during persistence (they are persisted
separately via SubSessionCompletedEvent).

Assisted-By: docker-agent

Author: aheritier

pkg/runtime/event.go

@@ -138,29 +138,33 @@ func StreamStarted(sessionID, agentName string) Event {
 }
 
 type AgentChoiceEvent struct {
-	Type    string `json:"type"`
-	Content string `json:"content"`
+	Type      string `json:"type"`
+	Content   string `json:"content"`
+	SessionID string `json:"session_id,omitempty"`
 	AgentContext
 }
 
-func AgentChoice(agentName, content string) Event {
+func AgentChoice(agentName, sessionID, content string) Event {
 	return &AgentChoiceEvent{
 		Type:         "agent_choice",
 		Content:      content,
+		SessionID:    sessionID,
 		AgentContext: newAgentContext(agentName),
 	}
 }
 
 type AgentChoiceReasoningEvent struct {
-	Type    string `json:"type"`
-	Content string `json:"content"`
+	Type      string `json:"type"`
+	Content   string `json:"content"`
+	SessionID string `json:"session_id,omitempty"`
 	AgentContext
 }
 
-func AgentChoiceReasoning(agentName, content string) Event {
+func AgentChoiceReasoning(agentName, sessionID, content string) Event {
 	return &AgentChoiceReasoningEvent{
 		Type:         "agent_choice_reasoning",
 		Content:      content,
+		SessionID:    sessionID,
 		AgentContext: newAgentContext(agentName),
 	}
 }

pkg/runtime/persistent_runtime.go

@@ -73,13 +73,19 @@ func (r *PersistentRuntime) handleEvent(ctx context.Context, sess *session.Sessi
 
 	switch e := event.(type) {
 	case *AgentChoiceEvent:
+		if e.SessionID != sess.ID {
+			return
+		}
 		// Accumulate streaming content
 		streaming.content.WriteString(e.Content)
 		streaming.agentName = e.AgentName
 
 		r.persistStreamingContent(ctx, sess.ID, streaming)
 
 	case *AgentChoiceReasoningEvent:
+		if e.SessionID != sess.ID {
+			return
+		}
 		// Accumulate streaming reasoning content
 		streaming.reasoningContent.WriteString(e.Content)
 		streaming.agentName = e.AgentName
@@ -98,6 +104,9 @@ func (r *PersistentRuntime) handleEvent(ctx context.Context, sess *session.Sessi
 		}
 
 	case *MessageAddedEvent:
+		if e.SessionID != sess.ID {
+			return
+		}
 		// Finalize the streaming message with complete metadata
 		if streaming.messageID != 0 {
 			// Update the existing streaming message with final content

pkg/runtime/runtime_test.go

@@ -282,7 +282,7 @@ func TestSimple(t *testing.T) {
 		UserMessage("Hi", sess.ID, nil, 0),
 		StreamStarted(sess.ID, "root"),
 		ToolsetInfo(0, false, "root"),
-		AgentChoice("root", "Hello"),
+		AgentChoice("root", sess.ID, "Hello"),
 		MessageAdded(sess.ID, msgAdded.Message, "root"),
 		NewTokenUsageEvent(sess.ID, "root", &Usage{InputTokens: 3, OutputTokens: 2, ContextLength: 5, LastMessage: &MessageUsage{
 			Usage: chat.Usage{InputTokens: 3, OutputTokens: 2},
@@ -321,11 +321,11 @@ func TestMultipleContentChunks(t *testing.T) {
 		UserMessage("Please greet me", sess.ID, nil, 0),
 		StreamStarted(sess.ID, "root"),
 		ToolsetInfo(0, false, "root"),
-		AgentChoice("root", "Hello "),
-		AgentChoice("root", "there, "),
-		AgentChoice("root", "how "),
-		AgentChoice("root", "are "),
-		AgentChoice("root", "you?"),
+		AgentChoice("root", sess.ID, "Hello "),
+		AgentChoice("root", sess.ID, "there, "),
+		AgentChoice("root", sess.ID, "how "),
+		AgentChoice("root", sess.ID, "are "),
+		AgentChoice("root", sess.ID, "you?"),
 		MessageAdded(sess.ID, msgAdded.Message, "root"),
 		NewTokenUsageEvent(sess.ID, "root", &Usage{InputTokens: 8, OutputTokens: 12, ContextLength: 20, LastMessage: &MessageUsage{
 			Usage: chat.Usage{InputTokens: 8, OutputTokens: 12},
@@ -362,9 +362,9 @@ func TestWithReasoning(t *testing.T) {
 		UserMessage("Hi", sess.ID, nil, 0),
 		StreamStarted(sess.ID, "root"),
 		ToolsetInfo(0, false, "root"),
-		AgentChoiceReasoning("root", "Let me think about this..."),
-		AgentChoiceReasoning("root", " I should respond politely."),
-		AgentChoice("root", "Hello, how can I help you?"),
+		AgentChoiceReasoning("root", sess.ID, "Let me think about this..."),
+		AgentChoiceReasoning("root", sess.ID, " I should respond politely."),
+		AgentChoice("root", sess.ID, "Hello, how can I help you?"),
 		MessageAdded(sess.ID, msgAdded.Message, "root"),
 		NewTokenUsageEvent(sess.ID, "root", &Usage{InputTokens: 10, OutputTokens: 15, ContextLength: 25, LastMessage: &MessageUsage{
 			Usage: chat.Usage{InputTokens: 10, OutputTokens: 15},
@@ -402,10 +402,10 @@ func TestMixedContentAndReasoning(t *testing.T) {
 		UserMessage("Hi there", sess.ID, nil, 0),
 		StreamStarted(sess.ID, "root"),
 		ToolsetInfo(0, false, "root"),
-		AgentChoiceReasoning("root", "The user wants a greeting"),
-		AgentChoice("root", "Hello!"),
-		AgentChoiceReasoning("root", " I should be friendly"),
-		AgentChoice("root", " How can I help you today?"),
+		AgentChoiceReasoning("root", sess.ID, "The user wants a greeting"),
+		AgentChoice("root", sess.ID, "Hello!"),
+		AgentChoiceReasoning("root", sess.ID, " I should be friendly"),
+		AgentChoice("root", sess.ID, " How can I help you today?"),
 		MessageAdded(sess.ID, msgAdded.Message, "root"),
 		NewTokenUsageEvent(sess.ID, "root", &Usage{InputTokens: 15, OutputTokens: 20, ContextLength: 35, LastMessage: &MessageUsage{
 			Usage: chat.Usage{InputTokens: 15, OutputTokens: 20},

pkg/runtime/streaming.go

@@ -164,7 +164,7 @@ func (r *LocalRuntime) handleStream(ctx context.Context, stream chat.MessageStre
 		}
 
 		if choice.Delta.ReasoningContent != "" {
-			events <- AgentChoiceReasoning(a.Name(), choice.Delta.ReasoningContent)
+			events <- AgentChoiceReasoning(a.Name(), sess.ID, choice.Delta.ReasoningContent)
 			fullReasoningContent.WriteString(choice.Delta.ReasoningContent)
 		}
 
@@ -174,7 +174,7 @@ func (r *LocalRuntime) handleStream(ctx context.Context, stream chat.MessageStre
 		}
 
 		if choice.Delta.Content != "" {
-			events <- AgentChoice(a.Name(), choice.Delta.Content)
+			events <- AgentChoice(a.Name(), sess.ID, choice.Delta.Content)
 			fullContent.WriteString(choice.Delta.Content)
 		}
 	}