fix(workbench): coalesce runtime read queries (#1977)

internal/workbenchruntime/service.go

@@ -12,6 +12,7 @@ import (
 	"os/signal"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"
 
@@ -83,10 +84,18 @@ type Config struct {
 }
 
 type Server struct {
-	config Config
-	db     *sql.DB
-	cache  *derivedCache
-	mux    *http.ServeMux
+	config         Config
+	db             *sql.DB
+	cache          *derivedCache
+	mux            *http.ServeMux
+	singleflightMu sync.Mutex
+	singleflights  map[string]*singleflightMapCall
+}
+
+type singleflightMapCall struct {
+	done  chan struct{}
+	value map[string]any
+	err   error
 }
 
 var workbenchRuntimeReadIndexes = []string{
@@ -295,6 +304,15 @@ func (s *Server) handleFacts(w http.ResponseWriter, r *http.Request) {
 }
 
 func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[string]any, error) {
+	if key := factsSingleflightKey(query); key != "" {
+		return s.singleflightMap(ctx, "facts:"+key, func(callCtx context.Context) (map[string]any, error) {
+			return s.queryFactsResponseUncoalesced(callCtx, query)
+		})
+	}
+	return s.queryFactsResponseUncoalesced(ctx, query)
+}
+
+func (s *Server) queryFactsResponseUncoalesced(ctx context.Context, query factQuery) (map[string]any, error) {
 	plan := s.factsCachePlan(ctx, query)
 	if plan.Key != "" {
 		var entry factsCacheEntry
@@ -854,6 +872,15 @@ func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) {
 
 func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[string]any, error) {
 	limit, offset := sessionListPage(query)
+	if key := sessionsListSingleflightKey(query, limit, offset); key != "" {
+		return s.singleflightMap(ctx, "sessions:"+key, func(callCtx context.Context) (map[string]any, error) {
+			return s.listSessionsUncoalesced(callCtx, query, limit, offset)
+		})
+	}
+	return s.listSessionsUncoalesced(ctx, query, limit, offset)
+}
+
+func (s *Server) listSessionsUncoalesced(ctx context.Context, query sessionListQuery, limit int, offset int) (map[string]any, error) {
 	cacheKey, cacheMeta := s.sessionsSummaryCacheKey(query, limit, offset)
 	if cacheKey != "" {
 		var entry sessionsCacheEntry
@@ -939,6 +966,78 @@ func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[
 	return payload, nil
 }
 
+func (s *Server) singleflightMap(ctx context.Context, key string, fn func(context.Context) (map[string]any, error)) (map[string]any, error) {
+	if s == nil || strings.TrimSpace(key) == "" {
+		return fn(ctx)
+	}
+	s.singleflightMu.Lock()
+	if s.singleflights == nil {
+		s.singleflights = map[string]*singleflightMapCall{}
+	}
+	if existing := s.singleflights[key]; existing != nil {
+		s.singleflightMu.Unlock()
+		select {
+		case <-existing.done:
+			if existing.err != nil {
+				return nil, existing.err
+			}
+			return cloneJSONMap(existing.value), nil
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		}
+	}
+	call := &singleflightMapCall{done: make(chan struct{})}
+	s.singleflights[key] = call
+	s.singleflightMu.Unlock()
+
+	value, err := fn(ctx)
+	stored := cloneJSONMap(value)
+
+	s.singleflightMu.Lock()
+	call.value = stored
+	call.err = err
+	delete(s.singleflights, key)
+	close(call.done)
+	s.singleflightMu.Unlock()
+
+	return value, err
+}
+
+func factsSingleflightKey(query factQuery) string {
+	class := factsCacheClass(query)
+	if class == "" {
+		return ""
+	}
+	families := strings.Join(requestedFactFamilies(query), ",")
+	limit := boundedLimit(query.Limit, 0, 1000)
+	switch class {
+	case sessionMessagePageCacheClass:
+		sessionID := strings.TrimSpace(query.SessionID)
+		if sessionID == "" {
+			return ""
+		}
+		return strings.Join([]string{class, sessionID, families, strconv.Itoa(limit)}, "|")
+	case terminalTurnCacheClass, terminalTracePageCacheClass:
+		traceID := strings.TrimSpace(query.TraceID)
+		if traceID == "" {
+			return ""
+		}
+		return strings.Join([]string{class, traceID, families, strconv.FormatInt(query.AfterProjectedSeq, 10), strconv.Itoa(limit)}, "|")
+	default:
+		return ""
+	}
+}
+
+func sessionsListSingleflightKey(query sessionListQuery, limit int, offset int) string {
+	actorID := strings.TrimSpace(query.Actor.ID)
+	actorRole := strings.TrimSpace(query.Actor.Role)
+	includeSessionID := strings.TrimSpace(query.IncludeSessionID)
+	if actorID == "" && actorRole == "" && includeSessionID == "" {
+		return ""
+	}
+	return strings.Join([]string{sessionsSummaryCacheClass, actorID, actorRole, includeSessionID, strings.TrimSpace(query.Cursor), strconv.Itoa(limit), strconv.Itoa(offset)}, "|")
+}
+
 func (s *Server) listSessionsFromDB(ctx context.Context, query sessionListQuery, limit int, offset int) (map[string]any, error) {
 	ownerUserID := strings.TrimSpace(query.Actor.ID)
 	pageFacts, err := s.queryFacts(ctx, factQuery{OwnerUserID: ownerUserID, Limit: limit + offset + 1, Families: []string{"sessions"}, SessionsOrder: "updated_desc", SessionProjection: "summary"})