diff --git a/internal/workbenchruntime/service.go b/internal/workbenchruntime/service.go index 6db67401..9b6ad3ea 100644 --- a/internal/workbenchruntime/service.go +++ b/internal/workbenchruntime/service.go @@ -1349,17 +1349,15 @@ func (s *Server) querySummaryFacts(ctx context.Context, sessions []map[string]an sessionIDs := uniqueStrings(mapSessions(sessions, sessionID)) facts := emptyFacts() if len(sessionIDs) > 0 { - messageSummaries, err := s.queryMessageSummaries(ctx, sessionIDs) + messageSummaries, messages, turns, err := s.querySessionSummaryFacts(ctx, sessionIDs) if err != nil { return nil, err } - facts = mergeFacts(facts, map[string][]any{"messageSummaries": messageSummaries}) - - bySession, err := s.queryFacts(ctx, factQuery{SessionIDs: sessionIDs, Families: []string{"messages", "turns"}, Limit: summarySessionFactLimit(len(sessionIDs)), MessagesOrder: "updated_desc", TurnsOrder: "updated_desc"}) - if err != nil { - return nil, err - } - facts = mergeFacts(facts, bySession) + facts = mergeFacts(facts, map[string][]any{ + "messageSummaries": messageSummaries, + "messages": messages, + "turns": turns, + }) } traceIDs := summaryTraceIDs(sessions, facts) if len(traceIDs) > 0 { @@ -1372,18 +1370,35 @@ func (s *Server) querySummaryFacts(ctx context.Context, sessions []map[string]an return facts, nil } -func summarySessionFactLimit(sessionCount int) int { - if sessionCount <= 0 { - return 0 +func (s *Server) querySessionSummaryFacts(ctx context.Context, sessionIDs []string) ([]any, []any, []any, error) { + const recentFactsPerSession = 8 + var messageSummaries, messages, turns []any + var messageSummaryErr, messagesErr, turnsErr error + var wait sync.WaitGroup + wait.Add(3) + go func() { + defer wait.Done() + messageSummaries, messageSummaryErr = s.queryMessageSummaries(ctx, sessionIDs) + }() + go func() { + defer wait.Done() + messages, messagesErr = s.queryRecentSessionFacts(ctx, "workbench_messages", "message_json", "message_id", sessionIDs, recentFactsPerSession) + }() + go func() { + defer wait.Done() + turns, turnsErr = s.queryRecentSessionFacts(ctx, "workbench_turns", "turn_json", "turn_id", sessionIDs, recentFactsPerSession) + }() + wait.Wait() + if messageSummaryErr != nil { + return nil, nil, nil, messageSummaryErr } - limit := sessionCount * 8 - if limit < sessionCount { - return sessionCount + if messagesErr != nil { + return nil, nil, nil, messagesErr } - if limit > 1000 { - return 1000 + if turnsErr != nil { + return nil, nil, nil, turnsErr } - return limit + return messageSummaries, messages, turns, nil } func summaryTraceIDs(sessions []map[string]any, facts map[string][]any) []string { @@ -1741,6 +1756,75 @@ LEFT JOIN LATERAL ( return result, nil } +func (s *Server) queryRecentSessionFacts(ctx context.Context, table string, jsonColumn string, idColumn string, sessionIDs []string, perSessionLimit int) ([]any, error) { + cleaned := uniqueStrings(sessionIDs) + if len(cleaned) == 0 || perSessionLimit <= 0 { + return []any{}, nil + } + if !validRecentSessionFactSource(table, jsonColumn, idColumn) { + return nil, wrapQueryStage("workbench_runtime.db.session_recent_facts", fmt.Errorf("unsupported recent session fact source: %s", table)) + } + args := make([]any, 0, len(cleaned)+1) + valueRows := make([]string, 0, len(cleaned)) + for _, sid := range cleaned { + args = append(args, sid) + valueRows = append(valueRows, fmt.Sprintf("($%d::text)", len(args))) + } + args = append(args, perSessionLimit) + limitPlaceholder := fmt.Sprintf("$%d", len(args)) + sqlText := fmt.Sprintf(`SELECT recent.%s +FROM (VALUES %s) AS requested(session_id) +CROSS JOIN LATERAL ( + SELECT %s, updated_at, %s + FROM %s + WHERE session_id = requested.session_id + ORDER BY updated_at DESC, %s DESC + LIMIT %s +) AS recent +ORDER BY recent.updated_at DESC, recent.%s DESC`, jsonColumn, strings.Join(valueRows, ", "), jsonColumn, idColumn, table, idColumn, limitPlaceholder, idColumn) + stage := "workbench_runtime.db." + factFamilyForTable(table) + "_recent_by_session" + result := []any{} + err := s.withOtelInternalSpan(ctx, stage, s.dbQueryAttrs(table, len(args), map[string]any{ + "db.index.expected": recentSessionFactExpectedIndex(table), + "db.query.shape": "values_lateral_per_session_limit", + "db.query.session_count": int64(len(cleaned)), + "db.query.per_session_max": int64(perSessionLimit), + }), func(spanCtx context.Context) error { + rows, err := s.db.QueryContext(spanCtx, sqlText, args...) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var raw sql.NullString + if err := rows.Scan(&raw); err != nil { + return err + } + var item any + if raw.Valid && strings.TrimSpace(raw.String) != "" && json.Unmarshal([]byte(raw.String), &item) == nil && item != nil { + result = append(result, item) + } + } + return rows.Err() + }) + if err != nil { + return nil, wrapQueryStage(stage, err) + } + return result, nil +} + +func validRecentSessionFactSource(table string, jsonColumn string, idColumn string) bool { + return table == "workbench_messages" && jsonColumn == "message_json" && idColumn == "message_id" || + table == "workbench_turns" && jsonColumn == "turn_json" && idColumn == "turn_id" +} + +func recentSessionFactExpectedIndex(table string) string { + if table == "workbench_messages" { + return "idx_workbench_messages_session_updated" + } + return "idx_workbench_turns_session_updated" +} + func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) { clauses := []string{} args := []any{} diff --git a/web/hwlab-cloud-web/src/components/workbench/SessionRail.vue b/web/hwlab-cloud-web/src/components/workbench/SessionRail.vue index fcfb932a..baed6887 100644 --- a/web/hwlab-cloud-web/src/components/workbench/SessionRail.vue +++ b/web/hwlab-cloud-web/src/components/workbench/SessionRail.vue @@ -98,32 +98,30 @@ function formatSessionUpdatedTime(value: string | null | undefined): string {

显式会话

- +
- - +

已显示全部 sessions

+

{{ workbench.sessionListLoadMoreError }}

+
- - + +