Separate main chat from subagent group stream

pkg/agent/runtime_admin.go

@@ -393,6 +393,22 @@ func (al *AgentLoop) HandleSubagentRuntime(ctx context.Context, action string, a
 			"thread": thread,
 			"items":  stream,
 		}, nil
+	case "stream_all":
+		tasks := sm.ListTasks()
+		sort.Slice(tasks, func(i, j int) bool {
+			left := maxInt64(tasks[i].Updated, tasks[i].Created)
+			right := maxInt64(tasks[j].Updated, tasks[j].Created)
+			if left != right {
+				return left > right
+			}
+			return tasks[i].ID > tasks[j].ID
+		})
+		taskLimit := runtimeIntArg(args, "task_limit", 16)
+		if taskLimit > 0 && len(tasks) > taskLimit {
+			tasks = tasks[:taskLimit]
+		}
+		items := mergeAllSubagentStreams(sm, tasks, runtimeIntArg(args, "limit", 200))
+		return map[string]interface{}{"found": true, "items": items}, nil
 	case "inbox":
 		agentID := runtimeStringArg(args, "agent_id")
 		if agentID == "" {
@@ -460,6 +476,99 @@ func mergeSubagentStream(events []tools.SubagentRunEvent, messages []tools.Agent
 	return items
 }
 
+func mergeAllSubagentStreams(sm *tools.SubagentManager, tasks []*tools.SubagentTask, limit int) []map[string]interface{} {
+	if sm == nil || len(tasks) == 0 {
+		return nil
+	}
+	items := make([]map[string]interface{}, 0)
+	seenEvents := map[string]struct{}{}
+	seenMessages := map[string]struct{}{}
+	for _, task := range tasks {
+		if task == nil {
+			continue
+		}
+		if events, err := sm.Events(task.ID, limit); err == nil {
+			for _, evt := range events {
+				key := fmt.Sprintf("%s:%s:%d:%s", evt.RunID, evt.Type, evt.At, evt.Message)
+				if _, ok := seenEvents[key]; ok {
+					continue
+				}
+				seenEvents[key] = struct{}{}
+				items = append(items, map[string]interface{}{
+					"kind":        "event",
+					"at":          evt.At,
+					"task_id":     task.ID,
+					"label":       task.Label,
+					"run_id":      evt.RunID,
+					"agent_id":    firstNonEmptyString(evt.AgentID, task.AgentID),
+					"event_type":  evt.Type,
+					"status":      evt.Status,
+					"message":     evt.Message,
+					"retry_count": evt.RetryCount,
+				})
+			}
+		}
+		if strings.TrimSpace(task.ThreadID) == "" {
+			continue
+		}
+		if messages, err := sm.ThreadMessages(task.ThreadID, limit); err == nil {
+			for _, msg := range messages {
+				if _, ok := seenMessages[msg.MessageID]; ok {
+					continue
+				}
+				seenMessages[msg.MessageID] = struct{}{}
+				items = append(items, map[string]interface{}{
+					"kind":           "message",
+					"at":             msg.CreatedAt,
+					"task_id":        task.ID,
+					"label":          task.Label,
+					"message_id":     msg.MessageID,
+					"thread_id":      msg.ThreadID,
+					"from_agent":     msg.FromAgent,
+					"to_agent":       msg.ToAgent,
+					"reply_to":       msg.ReplyTo,
+					"correlation_id": msg.CorrelationID,
+					"message_type":   msg.Type,
+					"content":        msg.Content,
+					"status":         msg.Status,
+					"requires_reply": msg.RequiresReply,
+				})
+			}
+		}
+	}
+	sort.Slice(items, func(i, j int) bool {
+		left, _ := items[i]["at"].(int64)
+		right, _ := items[j]["at"].(int64)
+		if left != right {
+			return left < right
+		}
+		return fmt.Sprintf("%v", items[i]["task_id"]) < fmt.Sprintf("%v", items[j]["task_id"])
+	})
+	if limit > 0 && len(items) > limit {
+		items = items[len(items)-limit:]
+	}
+	return items
+}
+
+func maxInt64(values ...int64) int64 {
+	var out int64
+	for _, v := range values {
+		if v > out {
+			out = v
+		}
+	}
+	return out
+}
+
+func firstNonEmptyString(values ...string) string {
+	for _, v := range values {
+		if strings.TrimSpace(v) != "" {
+			return strings.TrimSpace(v)
+		}
+	}
+	return ""
+}
+
 func cloneSubagentTask(in *tools.SubagentTask) *tools.SubagentTask {
 	if in == nil {
 		return nil

pkg/agent/runtime_admin_test.go

@@ -389,3 +389,48 @@ func TestHandleSubagentRuntimeStream(t *testing.T) {
 		t.Fatalf("expected merged event and message items, got %#v", items)
 	}
 }
+
+func TestHandleSubagentRuntimeStreamAll(t *testing.T) {
+	workspace := t.TempDir()
+	manager := tools.NewSubagentManager(nil, workspace, nil)
+	manager.SetRunFunc(func(ctx context.Context, task *tools.SubagentTask) (string, error) {
+		return "stream-all-result", nil
+	})
+	loop := &AgentLoop{
+		workspace:       workspace,
+		subagentManager: manager,
+		subagentRouter:  tools.NewSubagentRouter(manager),
+	}
+
+	if _, err := loop.HandleSubagentRuntime(context.Background(), "spawn", map[string]interface{}{
+		"task":     "prepare grouped stream task",
+		"agent_id": "coder",
+		"channel":  "webui",
+		"chat_id":  "webui",
+	}); err != nil {
+		t.Fatalf("spawn failed: %v", err)
+	}
+	for i := 0; i < 50; i++ {
+		tasks := manager.ListTasks()
+		if len(tasks) > 0 && tasks[0].Status == "completed" {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	out, err := loop.HandleSubagentRuntime(context.Background(), "stream_all", map[string]interface{}{
+		"limit":      100,
+		"task_limit": 10,
+	})
+	if err != nil {
+		t.Fatalf("stream_all failed: %v", err)
+	}
+	payload, ok := out.(map[string]interface{})
+	if !ok || payload["found"] != true {
+		t.Fatalf("unexpected stream_all payload: %#v", out)
+	}
+	items, ok := payload["items"].([]map[string]interface{})
+	if !ok || len(items) == 0 {
+		t.Fatalf("expected grouped stream items, got %#v", payload["items"])
+	}
+}