pikasTech-HWLAB/internal/workbenchruntime/service.go

package workbenchruntime

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"syscall"
	"time"

	_ "github.com/jackc/pgx/v4/stdlib"
)

const serviceID = "hwlab-workbench-runtime"

const (
	sessionSummaryQueryMaxAttempts    = 2
	sessionSummaryQueryAttemptTimeout = 2 * time.Second
	sessionSummaryQueryRetryBackoff   = 150 * time.Millisecond
	sessionsSummaryCacheClass         = "sessions.summary"
	terminalTurnCacheClass            = "turn.terminal.snapshot"
	terminalTracePageCacheClass       = "trace.terminal.events"
)

type sessionsCacheEntry struct {
	CachedAt            string         `json:"cachedAt"`
	Payload             map[string]any `json:"payload"`
	PayloadBytes        int            `json:"payloadBytes"`
	TTLMillis           int64          `json:"ttlMs"`
	ProjectionRevision  string         `json:"projectionRevision,omitempty"`
	ProjectionSeqMax    int64          `json:"projectionSeqMax,omitempty"`
	SessionUpdatedAtMax string         `json:"sessionUpdatedAtMax,omitempty"`
	SessionCount        int            `json:"sessionCount,omitempty"`
	ValuesRedacted      bool           `json:"valuesRedacted"`
}

type factsCacheEntry struct {
	CachedAt       string         `json:"cachedAt"`
	Payload        map[string]any `json:"payload"`
	PayloadBytes   int            `json:"payloadBytes"`
	TTLMillis      int64          `json:"ttlMs"`
	ValuesRedacted bool           `json:"valuesRedacted"`
}

type factsCachePlan struct {
	Class            string
	Key              string
	Meta             map[string]any
	TraceID          string
	ProjectionSeq    int64
	ProjectionStatus string
	ProjectionHealth string
	Status           string
}

type sessionsSummaryRevision struct {
	ProjectionRevision  string
	ProjectionSeqMax    int64
	SessionUpdatedAtMax string
	SessionCount        int
	RevisionSource      string
}

type Config struct {
	Addr           string
	DatabaseURL    string
	QueryTimeout   time.Duration
	ReadinessDelay time.Duration
	PoolMax        int
	Cache          cacheConfig
}

type Server struct {
	config Config
	db     *sql.DB
	cache  *derivedCache
	mux    *http.ServeMux
}

var workbenchRuntimeReadIndexes = []string{
	"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_updated ON workbench_sessions(updated_at DESC, session_id)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_owner_updated ON workbench_sessions(owner_user_id, updated_at DESC, session_id)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_sessions_status_updated ON workbench_sessions(status, updated_at DESC, session_id)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_messages_session_updated ON workbench_messages(session_id, updated_at ASC, message_id)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_messages_session_role_updated ON workbench_messages(session_id, role, updated_at ASC, message_id)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_turns_trace_updated ON workbench_turns(trace_id, updated_at DESC)",
	"CREATE INDEX IF NOT EXISTS idx_workbench_projection_checkpoints_trace_updated ON workbench_projection_checkpoints(trace_id, updated_at DESC)",
}

type factQuery struct {
	Families           []string `json:"families"`
	FactFamilies       []string `json:"factFamilies"`
	Limit              int      `json:"limit"`
	AfterProjectedSeq  int64    `json:"afterProjectedSeq"`
	Order              string   `json:"order"`
	SessionsOrder      string   `json:"sessionsOrder"`
	MessagesOrder      string   `json:"messagesOrder"`
	PartsOrder         string   `json:"partsOrder"`
	TurnsOrder         string   `json:"turnsOrder"`
	CheckpointsOrder   string   `json:"checkpointsOrder"`
	SessionProjection  string   `json:"sessionProjection"`
	SessionID          string   `json:"sessionId"`
	SessionIDs         []string `json:"sessionIds"`
	TraceID            string   `json:"traceId"`
	TraceIDs           []string `json:"traceIds"`
	MessageID          string   `json:"messageId"`
	MessageIDs         []string `json:"messageIds"`
	TurnID             string   `json:"turnId"`
	TurnIDs            []string `json:"turnIds"`
	Status             string   `json:"status"`
	Statuses           []string `json:"statuses"`
	OwnerUserID        string   `json:"ownerUserId"`
	OwnerUserIDs       []string `json:"ownerUserIds"`
	ConversationID     string   `json:"conversationId"`
	ConversationIDs    []string `json:"conversationIds"`
	EventType          string   `json:"eventType"`
	EventTypes         []string `json:"eventTypes"`
	ProjectionStatus   string   `json:"projectionStatus"`
	ProjectionStatuses []string `json:"projectionStatuses"`
	ProjectionHealth   string   `json:"projectionHealth"`
	ProjectionHealths  []string `json:"projectionHealths"`
	RunID              string   `json:"runId"`
	RunIDs             []string `json:"runIds"`
	CommandID          string   `json:"commandId"`
	CommandIDs         []string `json:"commandIds"`
}

type traceEventsQuery struct {
	TraceID         string `json:"traceId"`
	SessionID       string `json:"sessionId"`
	WorkerSessionID string `json:"workerSessionId"`
	Level           string `json:"level"`
	Limit           int    `json:"limit"`
}

type outboxQuery struct {
	AfterSeq  int64  `json:"afterSeq"`
	Limit     int    `json:"limit"`
	TraceID   string `json:"traceId"`
	SessionID string `json:"sessionId"`
}

type sessionListQuery struct {
	Limit            int          `json:"limit"`
	Offset           int          `json:"offset"`
	Cursor           string       `json:"cursor"`
	IncludeSessionID string       `json:"includeSessionId"`
	Actor            sessionActor `json:"actor"`
}

type sessionActor struct {
	ID   string `json:"id"`
	Role string `json:"role"`
}

type fieldColumn struct {
	field  string
	values []string
	column string
}

func Run(ctx context.Context) error {
	config := configFromEnv()
	if strings.TrimSpace(config.DatabaseURL) == "" {
		return errors.New("HWLAB_WORKBENCH_RUNTIME_DB_URL or HWLAB_CLOUD_DB_URL is required")
	}
	db, err := sql.Open("pgx", config.DatabaseURL)
	if err != nil {
		return err
	}
	defer db.Close()
	if config.PoolMax > 0 {
		db.SetMaxOpenConns(config.PoolMax)
		db.SetMaxIdleConns(config.PoolMax)
	}
	if err := ensureWorkbenchRuntimeReadIndexes(ctx, db); err != nil {
		return err
	}
	server := NewServer(config, db)
	defer server.Close()
	httpServer := &http.Server{Addr: config.Addr, Handler: server.otelHTTPHandler(server.mux), ReadHeaderTimeout: 5 * time.Second}
	errCh := make(chan error, 1)
	go func() {
		log.Printf("%s listening addr=%s", serviceID, config.Addr)
		errCh <- httpServer.ListenAndServe()
	}()
	stopCh := make(chan os.Signal, 1)
	signal.Notify(stopCh, syscall.SIGINT, syscall.SIGTERM)
	select {
	case <-ctx.Done():
	case <-stopCh:
	case err := <-errCh:
		if err != nil && !errors.Is(err, http.ErrServerClosed) {
			return err
		}
	}
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	return httpServer.Shutdown(shutdownCtx)
}

func ensureWorkbenchRuntimeReadIndexes(ctx context.Context, db *sql.DB) error {
	ensureCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	for _, statement := range workbenchRuntimeReadIndexes {
		if _, err := db.ExecContext(ensureCtx, statement); err != nil {
			return fmt.Errorf("ensure workbench runtime read index: %w", err)
		}
	}
	return nil
}

func NewServer(config Config, db *sql.DB) *Server {
	server := &Server{config: config, db: db, cache: newDerivedCache(config.Cache), mux: http.NewServeMux()}
	server.routes()
	return server
}

func (s *Server) Close() error {
	if s == nil || s.cache == nil {
		return nil
	}
	return s.cache.Close()
}

func (s *Server) routes() {
	s.mux.HandleFunc("/health/live", s.handleLive)
	s.mux.HandleFunc("/health/ready", s.handleReady)
	s.mux.HandleFunc("/metrics", s.handleMetrics)
	s.mux.HandleFunc("/v1/workbench-runtime/facts", s.handleFacts)
	s.mux.HandleFunc("/v1/workbench-runtime/trace-events", s.handleTraceEvents)
	s.mux.HandleFunc("/v1/workbench-runtime/sessions", s.handleSessions)
	s.mux.HandleFunc("/v1/workbench-runtime/projection-outbox", s.handleProjectionOutbox)
}

func (s *Server) handleMetrics(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		methodNotAllowed(w, "GET")
		return
	}
	w.Header().Set("content-type", "text/plain; version=0.0.4; charset=utf-8")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write([]byte(s.cache.PrometheusText(serviceID)))
}

func (s *Server) handleLive(w http.ResponseWriter, r *http.Request) {
	cache := s.cache.Diagnostic(r.Context())
	if s.config.Cache.UnavailableLivenessFailure && cache["enabled"] == true && cache["available"] != true {
		writeJSON(w, http.StatusServiceUnavailable, map[string]any{"ok": false, "serviceId": serviceID, "status": "cache-unavailable", "cache": cache, "valuesRedacted": true})
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "live", "cache": cache, "valuesRedacted": true})
}

func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
	ctx, cancel := context.WithTimeout(r.Context(), s.config.ReadinessDelay)
	defer cancel()
	if err := s.db.PingContext(ctx); err != nil {
		writeAPIError(w, http.StatusServiceUnavailable, "workbench_runtime_db_unavailable", "workbench runtime database is unavailable")
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "ready", "cache": s.cache.Diagnostic(r.Context()), "valuesRedacted": true})
}

func (s *Server) handleFacts(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		methodNotAllowed(w, "POST")
		return
	}
	var query factQuery
	if !decodeJSON(w, r, &query) {
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
	defer cancel()
	payload, err := s.queryFactsResponse(ctx, query)
	if err != nil {
		writeQueryError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, payload)
}

func (s *Server) queryFactsResponse(ctx context.Context, query factQuery) (map[string]any, error) {
	plan := s.factsCachePlan(ctx, query)
	if plan.Key != "" {
		var entry factsCacheEntry
		hit, err := s.cache.Get(ctx, plan.Key, &entry)
		if err != nil {
			setFactsCacheStatus(plan.Meta, "unavailable")
			plan.Meta["errorKind"] = classifyCacheError(err)
		} else if hit && entry.Payload != nil {
			ageMs, stale := factsCacheEntryAgeMs(entry, time.Now().UTC())
			if !stale {
				payload := factsCacheResponsePayload(entry.Payload, s.persistenceSummary())
				setFactsCacheStatus(plan.Meta, "hit")
				plan.Meta["cacheAgeMs"] = ageMs
				plan.Meta["dbQueryAvoided"] = true
				plan.Meta["dbQueryDurationMs"] = int64(0)
				plan.Meta["payloadBytes"] = entry.PayloadBytes
				plan.Meta["ttlMs"] = entry.TTLMillis
				plan.Meta["freshnessSloMs"] = entry.TTLMillis
				attachFactsCacheMetadata(payload, plan.Meta)
				s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
				return payload, nil
			}
			setFactsCacheStatus(plan.Meta, "stale")
			plan.Meta["cacheAgeMs"] = ageMs
			plan.Meta["freshnessSloMs"] = entry.TTLMillis
		}
	}

	dbStart := time.Now()
	facts, err := s.queryFacts(ctx, query)
	dbDuration := time.Since(dbStart)
	if err != nil {
		return nil, err
	}
	payload := factsResponsePayload(facts, s.persistenceSummary())
	plan.Meta["dbQueryAvoided"] = false
	plan.Meta["dbQueryDurationMs"] = dbDuration.Milliseconds()
	if plan.Key == "" {
		attachFactsCacheMetadata(payload, plan.Meta)
		s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
		return payload, nil
	}

	ttl, unstable, ttlReason := factsCacheTTL(plan, s.config.Cache)
	plan.Meta["unstable"] = unstable
	plan.Meta["ttlMs"] = ttl.Milliseconds()
	plan.Meta["freshnessSloMs"] = ttl.Milliseconds()
	plan.Meta["ttlReason"] = ttlReason
	storedPayload := factsCacheStorablePayload(payload)
	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
	if payloadErr != nil {
		setFactsCacheStatus(plan.Meta, "skipped")
		plan.Meta["skipReason"] = classifyCacheError(payloadErr)
		attachFactsCacheMetadata(payload, plan.Meta)
		s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
		return payload, nil
	}
	plan.Meta["payloadBytes"] = len(payloadBytes)
	if ttl <= 0 {
		setFactsCacheStatus(plan.Meta, "skipped")
		plan.Meta["skipReason"] = ttlReason
		attachFactsCacheMetadata(payload, plan.Meta)
		s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
		return payload, nil
	}
	entry := factsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ValuesRedacted: true}
	if err := s.cache.Set(ctx, plan.Key, entry, ttl); err != nil {
		setFactsCacheStatus(plan.Meta, "set_failed")
		plan.Meta["errorKind"] = classifyCacheError(err)
		attachFactsCacheMetadata(payload, plan.Meta)
		s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
		return payload, nil
	}
	if plan.Meta["status"] != "stale" {
		setFactsCacheStatus(plan.Meta, "miss")
	}
	plan.Meta["stored"] = true
	attachFactsCacheMetadata(payload, plan.Meta)
	s.observeCacheDiagnostic(ctx, plan.Class, plan.Meta)
	return payload, nil
}

func (s *Server) observeCacheDiagnostic(ctx context.Context, class string, meta map[string]any) {
	if s == nil || meta == nil {
		return
	}
	cacheClass := first(text(meta["class"]), class, "unknown")
	status := first(text(meta["status"]), "unknown")
	if s.cache != nil {
		if status == "stale" {
			s.cache.RecordStale(cacheClass)
		}
		if boolValue(meta["dbQueryAvoided"]) {
			s.cache.RecordDBQueryAvoided(cacheClass)
		}
	}
	attrs := map[string]any{
		"cacheStatus":                   status,
		"cacheKeyClass":                 safeCacheToken(cacheClass),
		"workbench.cache.status":        status,
		"workbench.cache.key_class":     safeCacheToken(cacheClass),
		"workbench.cache.db_avoided":    boolValue(meta["dbQueryAvoided"]),
		"workbench.cache.values_source": "diagnostic",
		"valuesRedacted":                true,
	}
	if meta["cacheAgeMs"] != nil {
		age := int64FromAny(meta["cacheAgeMs"])
		attrs["cacheAgeMs"] = age
		attrs["workbench.cache.age_ms"] = age
	}
	if meta["projectionSeq"] != nil {
		seq := int64FromAny(meta["projectionSeq"])
		attrs["projectionSeq"] = seq
		attrs["workbench.projection.seq"] = seq
	}
	if meta["payloadBytes"] != nil {
		attrs["workbench.cache.payload_bytes"] = int64FromAny(meta["payloadBytes"])
	}
	if meta["ttlMs"] != nil {
		attrs["workbench.cache.ttl_ms"] = int64FromAny(meta["ttlMs"])
	}
	if meta["freshnessSloMs"] != nil {
		attrs["workbench.cache.freshness_slo_ms"] = int64FromAny(meta["freshnessSloMs"])
	}
	if revisionSource := text(meta["revisionSource"]); revisionSource != "" {
		attrs["workbench.cache.revision_source"] = revisionSource
	}
	if invalidationMode := text(meta["invalidationMode"]); invalidationMode != "" {
		attrs["workbench.cache.invalidation_mode"] = invalidationMode
	}
	_ = s.withOtelInternalSpan(ctx, "workbench-runtime.cache", attrs, func(context.Context) error { return nil })
}

func factsResponsePayload(facts map[string][]any, persistence map[string]any) map[string]any {
	count := 0
	for _, rows := range facts {
		count += len(rows)
	}
	return map[string]any{
		"contractVersion": "workbench-runtime-facts-v1",
		"facts":           facts,
		"count":           count,
		"persistence":     persistence,
		"valuesRedacted":  true,
	}
}

func (s *Server) factsCachePlan(ctx context.Context, query factQuery) factsCachePlan {
	class := factsCacheClass(query)
	meta := factsCacheMetadata(class, "disabled")
	plan := factsCachePlan{Class: class, Meta: meta, TraceID: strings.TrimSpace(query.TraceID)}
	if s == nil || s.cache == nil || !s.config.Cache.Enabled || class == "" {
		return plan
	}
	meta["enabled"] = true
	if plan.TraceID == "" {
		setFactsCacheStatus(meta, "skipped")
		meta["skipReason"] = "missing_trace"
		return plan
	}
	state, err := s.traceProjectionState(ctx, plan.TraceID)
	if err != nil {
		setFactsCacheStatus(meta, "skipped")
		meta["skipReason"] = "projection_revision_unavailable"
		return plan
	}
	plan.ProjectionSeq = state.ProjectedSeq
	plan.ProjectionStatus = state.ProjectionStatus
	plan.ProjectionHealth = state.ProjectionHealth
	plan.Status = state.Status
	meta["projectionSeq"] = state.ProjectedSeq
	meta["projectionRevision"] = projectionRevisionToken(state.ProjectedSeq)
	meta["projectionStatus"] = state.ProjectionStatus
	meta["projectionHealth"] = state.ProjectionHealth
	meta["traceStatus"] = state.Status
	meta["revisionSource"] = "workbench_projection_checkpoints"
	meta["invalidationMode"] = "projection_revision_key"
	if state.ProjectedSeq <= 0 {
		setFactsCacheStatus(meta, "skipped")
		meta["skipReason"] = "missing_projection_revision"
		return plan
	}
	cursor := fmt.Sprintf("after:%d|limit:%d|families:%s", query.AfterProjectedSeq, boundedLimit(query.Limit, 0, 1000), strings.Join(requestedFactFamilies(query), ","))
	authority := fmt.Sprintf("contract=workbench-runtime-facts-v1|class=%s|projectionStatus=%s|projectionHealth=%s", class, state.ProjectionStatus, state.ProjectionHealth)
	key, err := s.cache.BuildKey(cacheKeyParts{Class: class, ActorID: "workbench-runtime", TraceID: plan.TraceID, Cursor: cursor, ProjectionSeq: state.ProjectedSeq, Authority: authority})
	if err != nil {
		setFactsCacheStatus(meta, "skipped")
		meta["skipReason"] = classifyCacheError(err)
		return plan
	}
	plan.Key = key
	setFactsCacheStatus(meta, "miss")
	return plan
}

type traceProjectionState struct {
	ProjectedSeq     int64
	ProjectionStatus string
	ProjectionHealth string
	Status           string
}

func (s *Server) traceProjectionState(ctx context.Context, traceID string) (traceProjectionState, error) {
	facts, err := s.queryFacts(ctx, factQuery{TraceID: traceID, Families: []string{"checkpoints", "turns"}, Limit: 1, TurnsOrder: "updated_desc", CheckpointsOrder: "updated_desc"})
	if err != nil {
		return traceProjectionState{}, err
	}
	checkpoint := firstObject(facts["checkpoints"])
	turn := firstObject(facts["turns"])
	state := traceProjectionState{
		ProjectedSeq:     int64FromAny(seq(checkpoint)),
		ProjectionStatus: normalizeProjectionStatus(text(checkpoint["projectionStatus"])),
		ProjectionHealth: normalizeProjectionHealth(text(checkpoint["projectionHealth"]), normalizeProjectionStatus(text(checkpoint["projectionStatus"]))),
		Status:           normalizeStatus(first(text(checkpoint["status"]), text(turn["status"]))),
	}
	if state.ProjectionStatus == "" {
		state.ProjectionStatus = "unknown"
	}
	if state.ProjectionHealth == "" {
		state.ProjectionHealth = "unknown"
	}
	return state, nil
}

func factsCacheClass(query factQuery) string {
	if strings.TrimSpace(query.TraceID) == "" {
		return ""
	}
	families := requestedFactFamilies(query)
	if stringSetEqual(families, []string{"traceEvents"}) {
		return terminalTracePageCacheClass
	}
	if query.AfterProjectedSeq == 0 && stringSetEqual(families, []string{"sessions", "messages", "turns", "checkpoints"}) {
		return terminalTurnCacheClass
	}
	return ""
}

func requestedFactFamilies(query factQuery) []string {
	selected := query.Families
	if len(selected) == 0 {
		selected = query.FactFamilies
	}
	result := []string{}
	seen := map[string]bool{}
	for _, family := range selected {
		family = strings.TrimSpace(family)
		if family != "" && !seen[family] {
			result = append(result, family)
			seen[family] = true
		}
	}
	return result
}

func stringSetEqual(values []string, expected []string) bool {
	if len(values) != len(expected) {
		return false
	}
	set := map[string]bool{}
	for _, value := range values {
		set[value] = true
	}
	for _, value := range expected {
		if !set[value] {
			return false
		}
	}
	return true
}

func factsCacheTTL(plan factsCachePlan, config cacheConfig) (time.Duration, bool, string) {
	stableProjection := plan.ProjectionStatus == "caught-up" || plan.ProjectionStatus == "blocked" || plan.ProjectionStatus == "stalled"
	terminal := isTerminal(plan.Status) || stableProjection
	if !terminal {
		if config.RunningPageTTL <= 0 {
			return 0, true, "running_page_ttl_disabled"
		}
		return config.RunningPageTTL, true, "running_or_projecting_trace"
	}
	if plan.Class == terminalTracePageCacheClass {
		if config.TerminalTracePageTTL <= 0 {
			return 0, false, "terminal_trace_page_ttl_disabled"
		}
		return config.TerminalTracePageTTL, false, "terminal_trace_page"
	}
	if config.TerminalTurnTTL <= 0 {
		return 0, false, "terminal_turn_ttl_disabled"
	}
	return config.TerminalTurnTTL, false, "terminal_turn_snapshot"
}

func factsCacheMetadata(class string, status string) map[string]any {
	if class == "" {
		class = "unknown"
	}
	meta := map[string]any{
		"class":                class,
		"schemaVersion":        cacheKeySchemaVersion,
		"enabled":              false,
		"status":               status,
		"hit":                  false,
		"miss":                 false,
		"stale":                false,
		"unavailable":          false,
		"dbQueryAvoided":       false,
		"cacheAgeMs":           nil,
		"dbQueryDurationMs":    nil,
		"payloadBytes":         nil,
		"ttlMs":                nil,
		"freshnessSloMs":       nil,
		"projectionRevision":   nil,
		"invalidationMode":     nil,
		"revisionSource":       nil,
		"valuesRedacted":       true,
		"secretMaterialStored": false,
	}
	setFactsCacheStatus(meta, status)
	return meta
}

func setFactsCacheStatus(meta map[string]any, status string) {
	if meta == nil {
		return
	}
	meta["status"] = status
	meta["hit"] = status == "hit"
	meta["miss"] = status == "miss"
	meta["stale"] = status == "stale"
	meta["unavailable"] = status == "unavailable" || status == "set_failed"
}

func attachFactsCacheMetadata(payload map[string]any, meta map[string]any) {
	if payload == nil || meta == nil {
		return
	}
	payload["cache"] = meta
}

func factsCacheStorablePayload(payload map[string]any) map[string]any {
	clone := cloneJSONMap(payload)
	delete(clone, "cache")
	delete(clone, "persistence")
	delete(clone, "secretMaterialStored")
	return objectMap(sanitizeFactsCachePayload(clone))
}

func sanitizeFactsCachePayload(value any) any {
	switch item := value.(type) {
	case map[string]any:
		cleaned := map[string]any{}
		for key, child := range item {
			if cachePayloadKeyRejected(key) {
				continue
			}
			cleaned[key] = sanitizeFactsCachePayload(child)
		}
		return cleaned
	case []any:
		cleaned := make([]any, 0, len(item))
		for _, child := range item {
			cleaned = append(cleaned, sanitizeFactsCachePayload(child))
		}
		return cleaned
	case []map[string]any:
		cleaned := make([]any, 0, len(item))
		for _, child := range item {
			cleaned = append(cleaned, sanitizeFactsCachePayload(child))
		}
		return cleaned
	default:
		return value
	}
}

func factsCacheResponsePayload(payload map[string]any, persistence map[string]any) map[string]any {
	clone := cloneJSONMap(payload)
	clone["persistence"] = persistence
	clone["valuesRedacted"] = true
	return clone
}

func factsCacheEntryAgeMs(entry factsCacheEntry, now time.Time) (int64, bool) {
	cachedAt, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(entry.CachedAt))
	if err != nil {
		return 0, true
	}
	age := now.Sub(cachedAt)
	if age < 0 {
		age = 0
	}
	if entry.TTLMillis > 0 && age.Milliseconds() > entry.TTLMillis {
		return age.Milliseconds(), true
	}
	return age.Milliseconds(), false
}

func firstObject(rows []any) map[string]any {
	if len(rows) == 0 {
		return map[string]any{}
	}
	return objectMap(rows[0])
}

func int64FromAny(value any) int64 {
	switch typed := value.(type) {
	case int64:
		return typed
	case int:
		return int64(typed)
	case float64:
		return int64(typed)
	case json.Number:
		parsed, _ := typed.Int64()
		return parsed
	default:
		return 0
	}
}

func (s *Server) handleTraceEvents(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		methodNotAllowed(w, "POST")
		return
	}
	var query traceEventsQuery
	if !decodeJSON(w, r, &query) {
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
	defer cancel()
	events, err := s.queryAgentTraceEvents(ctx, query)
	if err != nil {
		writeQueryError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"contractVersion": "workbench-runtime-trace-events-v1",
		"events":          events,
		"count":           len(events),
		"persistence":     s.persistenceSummary(),
		"valuesRedacted":  true,
	})
}

func (s *Server) handleProjectionOutbox(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		methodNotAllowed(w, "POST")
		return
	}
	var query outboxQuery
	if !decodeJSON(w, r, &query) {
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
	defer cancel()
	rows, err := s.readProjectionOutbox(ctx, query)
	if err != nil {
		writeQueryError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"contractVersion": "workbench-runtime-projection-outbox-v1",
		"rows":            rows,
		"count":           len(rows),
		"persistence":     s.persistenceSummary(),
		"valuesRedacted":  true,
	})
}

func (s *Server) handleSessions(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		methodNotAllowed(w, "POST")
		return
	}
	var query sessionListQuery
	if !decodeJSON(w, r, &query) {
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
	defer cancel()
	payload, err := s.listSessions(ctx, query)
	if err != nil {
		writeQueryError(w, err)
		return
	}
	writeJSON(w, http.StatusOK, payload)
}

func (s *Server) listSessions(ctx context.Context, query sessionListQuery) (map[string]any, error) {
	limit, offset := sessionListPage(query)
	cacheKey, cacheMeta := s.sessionsSummaryCacheKey(query, limit, offset)
	if cacheKey != "" {
		var entry sessionsCacheEntry
		hit, err := s.cache.Get(ctx, cacheKey, &entry)
		if err != nil {
			setSessionsCacheStatus(cacheMeta, "unavailable")
			cacheMeta["errorKind"] = classifyCacheError(err)
		} else if hit && entry.Payload != nil {
			ageMs, stale := sessionsCacheEntryAgeMs(entry, time.Now().UTC())
			if !stale {
				payload := sessionsCacheResponsePayload(entry.Payload, s.persistenceSummary())
				setSessionsCacheStatus(cacheMeta, "hit")
				cacheMeta["cacheAgeMs"] = ageMs
				cacheMeta["dbQueryAvoided"] = true
				cacheMeta["dbQueryDurationMs"] = int64(0)
				cacheMeta["payloadBytes"] = entry.PayloadBytes
				cacheMeta["ttlMs"] = entry.TTLMillis
				cacheMeta["freshnessSloMs"] = entry.TTLMillis
				applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"})
				attachSessionsCacheMetadata(payload, cacheMeta)
				s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
				return payload, nil
			}
			setSessionsCacheStatus(cacheMeta, "stale")
			cacheMeta["cacheAgeMs"] = ageMs
			cacheMeta["freshnessSloMs"] = entry.TTLMillis
			applySessionsCacheRevisionMetadata(cacheMeta, sessionsSummaryRevision{ProjectionRevision: entry.ProjectionRevision, ProjectionSeqMax: entry.ProjectionSeqMax, SessionUpdatedAtMax: entry.SessionUpdatedAtMax, SessionCount: entry.SessionCount, RevisionSource: "cached_payload"})
		}
	}

	dbStart := time.Now()
	payload, err := s.listSessionsFromDB(ctx, query, limit, offset)
	dbDuration := time.Since(dbStart)
	if err != nil {
		return nil, err
	}
	cacheMeta["dbQueryAvoided"] = false
	cacheMeta["dbQueryDurationMs"] = dbDuration.Milliseconds()
	if cacheKey == "" {
		attachSessionsCacheMetadata(payload, cacheMeta)
		s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
		return payload, nil
	}

	ttl, unstable, ttlReason := sessionsSummaryCacheTTL(payload, s.config.Cache)
	cacheMeta["unstable"] = unstable
	cacheMeta["ttlMs"] = ttl.Milliseconds()
	cacheMeta["freshnessSloMs"] = ttl.Milliseconds()
	cacheMeta["ttlReason"] = ttlReason
	revision := sessionsSummaryRevisionForPayload(payload)
	applySessionsCacheRevisionMetadata(cacheMeta, revision)
	storedPayload := sessionsCacheStorablePayload(payload)
	payloadBytes, payloadErr := cachePayloadBytes(storedPayload, s.config.Cache.MaxPayloadBytes)
	if payloadErr != nil {
		setSessionsCacheStatus(cacheMeta, "skipped")
		cacheMeta["skipReason"] = classifyCacheError(payloadErr)
		attachSessionsCacheMetadata(payload, cacheMeta)
		s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
		return payload, nil
	}
	cacheMeta["payloadBytes"] = len(payloadBytes)
	if ttl <= 0 {
		setSessionsCacheStatus(cacheMeta, "skipped")
		cacheMeta["skipReason"] = ttlReason
		attachSessionsCacheMetadata(payload, cacheMeta)
		s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
		return payload, nil
	}
	entry := sessionsCacheEntry{CachedAt: time.Now().UTC().Format(time.RFC3339Nano), Payload: storedPayload, PayloadBytes: len(payloadBytes), TTLMillis: ttl.Milliseconds(), ProjectionRevision: revision.ProjectionRevision, ProjectionSeqMax: revision.ProjectionSeqMax, SessionUpdatedAtMax: revision.SessionUpdatedAtMax, SessionCount: revision.SessionCount, ValuesRedacted: true}
	if err := s.cache.Set(ctx, cacheKey, entry, ttl); err != nil {
		setSessionsCacheStatus(cacheMeta, "set_failed")
		cacheMeta["errorKind"] = classifyCacheError(err)
		attachSessionsCacheMetadata(payload, cacheMeta)
		s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
		return payload, nil
	}
	if cacheMeta["status"] != "stale" {
		setSessionsCacheStatus(cacheMeta, "miss")
	}
	cacheMeta["stored"] = true
	attachSessionsCacheMetadata(payload, cacheMeta)
	s.observeCacheDiagnostic(ctx, sessionsSummaryCacheClass, cacheMeta)
	return payload, nil
}

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"})
	if err != nil {
		return nil, err
	}
	natural := visibleSessions(pageFacts["sessions"], query.Actor)
	if offset > 0 {
		if offset >= len(natural) {
			natural = []map[string]any{}
		} else {
			natural = natural[offset:]
		}
	}
	if len(natural) > limit+1 {
		natural = natural[:limit+1]
	}
	includeID := strings.TrimSpace(query.IncludeSessionID)
	if includeID != "" && !sessionListContainsRoute(natural, includeID) {
		includeFacts, err := s.queryIncludedSession(ctx, includeID)
		if err != nil {
			return nil, err
		}
		included := visibleSessions(includeFacts["sessions"], query.Actor)
		if len(included) > 0 {
			pageFacts = mergeFacts(pageFacts, includeFacts)
			natural = append([]map[string]any{included[0]}, natural...)
		}
	}
	hasMore := len(natural) > limit
	pageSessions := natural
	if len(pageSessions) > limit {
		pageSessions = pageSessions[:limit]
	}
	summaryFacts, err := s.querySummaryFacts(ctx, pageSessions)
	if err != nil {
		return nil, err
	}
	facts := mergeFacts(pageFacts, summaryFacts)
	summaries := []any{}
	for _, session := range pageSessions {
		if summary := sessionSummary(session, facts); summary != nil {
			summaries = append(summaries, summary)
		}
	}
	return map[string]any{"ok": true, "status": "succeeded", "contractVersion": "workbench-sessions-v1", "sessions": summaries, "count": len(summaries), "cursor": cursorOrNil(offset), "hasMore": hasMore, "nextCursor": nextCursor(offset, limit, hasMore), "persistence": s.persistenceSummary(), "servedBy": serviceID, "valuesRedacted": true, "secretMaterialStored": false}, nil
}

func sessionListPage(query sessionListQuery) (int, int) {
	limit := boundedLimit(query.Limit, 20, 100)
	offset := query.Offset
	if offset <= 0 {
		offset = cursorOffset(query.Cursor)
	}
	if offset < 0 {
		offset = 0
	}
	return limit, offset
}

func (s *Server) sessionsSummaryCacheKey(query sessionListQuery, limit int, offset int) (string, map[string]any) {
	meta := sessionsCacheMetadata("disabled")
	if s == nil || s.cache == nil || !s.config.Cache.Enabled {
		return "", meta
	}
	meta["enabled"] = true
	actorID := strings.TrimSpace(query.Actor.ID)
	if actorID == "" {
		setSessionsCacheStatus(meta, "skipped")
		meta["skipReason"] = "missing_actor"
		return "", meta
	}
	includeID := strings.TrimSpace(query.IncludeSessionID)
	actorRole := strings.TrimSpace(query.Actor.Role)
	cursor := fmt.Sprintf("idx:%d|limit:%d|include:%s|role:%s", offset, limit, includeID, actorRole)
	authority := fmt.Sprintf("contract=workbench-sessions-v1|limit=%d|offset=%d|include=%t|role=%s", limit, offset, includeID != "", actorRole)
	key, err := s.cache.BuildKey(cacheKeyParts{Class: sessionsSummaryCacheClass, ActorID: actorID, SessionID: includeID, Cursor: cursor, ProjectionRevision: "workbench-sessions-v1", Authority: authority})
	if err != nil {
		setSessionsCacheStatus(meta, "skipped")
		meta["skipReason"] = classifyCacheError(err)
		return "", meta
	}
	setSessionsCacheStatus(meta, "miss")
	return key, meta
}

func sessionsCacheMetadata(status string) map[string]any {
	meta := map[string]any{
		"class":                sessionsSummaryCacheClass,
		"schemaVersion":        cacheKeySchemaVersion,
		"enabled":              false,
		"status":               status,
		"hit":                  false,
		"miss":                 false,
		"stale":                false,
		"unavailable":          false,
		"dbQueryAvoided":       false,
		"cacheAgeMs":           nil,
		"dbQueryDurationMs":    nil,
		"payloadBytes":         nil,
		"ttlMs":                nil,
		"freshnessSloMs":       nil,
		"projectionRevision":   nil,
		"projectionSeq":        nil,
		"sessionUpdatedAtMax":  nil,
		"sessionCount":         nil,
		"revisionSource":       nil,
		"invalidationMode":     "ttl_slo",
		"valuesRedacted":       true,
		"secretMaterialStored": false,
	}
	setSessionsCacheStatus(meta, status)
	return meta
}

func setSessionsCacheStatus(meta map[string]any, status string) {
	if meta == nil {
		return
	}
	meta["status"] = status
	meta["hit"] = status == "hit"
	meta["miss"] = status == "miss"
	meta["stale"] = status == "stale"
	meta["unavailable"] = status == "unavailable" || status == "set_failed"
}

func attachSessionsCacheMetadata(payload map[string]any, meta map[string]any) {
	if payload == nil || meta == nil {
		return
	}
	payload["cache"] = meta
}

func sessionsSummaryCacheTTL(payload map[string]any, config cacheConfig) (time.Duration, bool, string) {
	if sessionsSummaryPayloadUnstable(payload) {
		if config.RunningPageTTL <= 0 {
			return 0, true, "running_page_ttl_disabled"
		}
		return config.RunningPageTTL, true, "running_or_projecting_page"
	}
	if config.SessionsSummaryTTL <= 0 {
		return 0, false, "sessions_summary_ttl_disabled"
	}
	return config.SessionsSummaryTTL, false, "stable_page"
}

func sessionsSummaryRevisionForPayload(payload map[string]any) sessionsSummaryRevision {
	rows := sessionsPayloadRows(payload["sessions"])
	revision := sessionsSummaryRevision{RevisionSource: "response_payload"}
	maxUpdatedAt := ""
	var maxSeq int64
	for _, row := range rows {
		item := objectMap(row)
		if len(item) == 0 {
			continue
		}
		revision.SessionCount++
		if updatedAt := text(item["updatedAt"]); updatedAt > maxUpdatedAt {
			maxUpdatedAt = updatedAt
		}
		maxSeq = maxInt64(maxSeq, int64FromAny(item["projectedSeq"]))
		projection := objectMap(item["projection"])
		maxSeq = maxInt64(maxSeq, int64FromAny(projection["lastProjectedSeq"]))
		turnSummary := objectMap(item["turnSummary"])
		maxSeq = maxInt64(maxSeq, int64FromAny(turnSummary["eventCount"]))
	}
	revision.ProjectionSeqMax = maxSeq
	revision.SessionUpdatedAtMax = maxUpdatedAt
	raw := fmt.Sprintf("contract=workbench-sessions-v1|count=%d|maxProjectionSeq=%d|maxUpdatedAt=%s", revision.SessionCount, revision.ProjectionSeqMax, revision.SessionUpdatedAtMax)
	revision.ProjectionRevision = "digest:" + cacheHash(raw)
	return revision
}

func applySessionsCacheRevisionMetadata(meta map[string]any, revision sessionsSummaryRevision) {
	if meta == nil {
		return
	}
	if revision.ProjectionRevision != "" {
		meta["projectionRevision"] = revision.ProjectionRevision
	}
	meta["projectionSeq"] = revision.ProjectionSeqMax
	meta["sessionUpdatedAtMax"] = nilIfEmpty(revision.SessionUpdatedAtMax)
	meta["sessionCount"] = revision.SessionCount
	meta["revisionSource"] = first(revision.RevisionSource, "response_payload")
	meta["invalidationMode"] = "ttl_slo"
}

func projectionRevisionToken(projectedSeq int64) string {
	if projectedSeq <= 0 {
		return ""
	}
	return "seq:" + strconv.FormatInt(projectedSeq, 10)
}

func maxInt64(a int64, b int64) int64 {
	if b > a {
		return b
	}
	return a
}

func nilIfEmpty(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

func sessionsSummaryPayloadUnstable(payload map[string]any) bool {
	for _, row := range sessionsPayloadRows(payload["sessions"]) {
		item := objectMap(row)
		if len(item) == 0 {
			continue
		}
		if boolValue(item["running"]) || !boolValue(item["terminal"]) {
			return true
		}
		if status := normalizeProjectionStatus(text(item["projectionStatus"])); status != "" && status != "caught-up" {
			return true
		}
		turnSummary := objectMap(item["turnSummary"])
		if len(turnSummary) > 0 && (boolValue(turnSummary["running"]) || !boolValue(turnSummary["terminal"])) {
			return true
		}
	}
	return false
}

func sessionsPayloadRows(value any) []any {
	switch rows := value.(type) {
	case []any:
		return rows
	case []map[string]any:
		out := make([]any, 0, len(rows))
		for _, row := range rows {
			out = append(out, row)
		}
		return out
	default:
		return []any{}
	}
}

func sessionsCacheStorablePayload(payload map[string]any) map[string]any {
	clone := cloneJSONMap(payload)
	delete(clone, "cache")
	delete(clone, "persistence")
	delete(clone, "secretMaterialStored")
	return clone
}

func sessionsCacheResponsePayload(payload map[string]any, persistence map[string]any) map[string]any {
	clone := cloneJSONMap(payload)
	clone["persistence"] = persistence
	clone["secretMaterialStored"] = false
	clone["valuesRedacted"] = true
	clone["servedBy"] = serviceID
	return clone
}

func sessionsCacheEntryAgeMs(entry sessionsCacheEntry, now time.Time) (int64, bool) {
	cachedAt, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(entry.CachedAt))
	if err != nil {
		return 0, true
	}
	age := now.Sub(cachedAt)
	if age < 0 {
		age = 0
	}
	if entry.TTLMillis > 0 && age.Milliseconds() > entry.TTLMillis {
		return age.Milliseconds(), true
	}
	return age.Milliseconds(), false
}

func cloneJSONMap(value map[string]any) map[string]any {
	if value == nil {
		return map[string]any{}
	}
	payload, err := json.Marshal(value)
	if err != nil {
		return map[string]any{}
	}
	var clone map[string]any
	if err := json.Unmarshal(payload, &clone); err != nil || clone == nil {
		return map[string]any{}
	}
	return clone
}

func (s *Server) queryIncludedSession(ctx context.Context, routeID string) (map[string][]any, error) {
	if strings.HasPrefix(routeID, "ses_") {
		return s.queryFacts(ctx, factQuery{SessionID: routeID, Limit: 1, Families: []string{"sessions"}, SessionProjection: "summary"})
	}
	return s.queryFacts(ctx, factQuery{ConversationID: routeID, Limit: 1, Families: []string{"sessions"}, SessionProjection: "summary"})
}

func (s *Server) querySummaryFacts(ctx context.Context, sessions []map[string]any) (map[string][]any, error) {
	sessionIDs := uniqueStrings(mapSessions(sessions, sessionID))
	traceIDs := uniqueStrings(mapSessions(sessions, lastTraceID))
	facts := emptyFacts()
	if len(sessionIDs) > 0 {
		messageSummaries, err := s.queryMessageSummaries(ctx, sessionIDs)
		if err != nil {
			return nil, err
		}
		facts["messageSummaries"] = messageSummaries
	}
	if len(traceIDs) > 0 {
		byTrace, err := s.queryFacts(ctx, factQuery{TraceIDs: traceIDs, Families: []string{"turns", "checkpoints"}, Limit: len(traceIDs), TurnsOrder: "updated_desc", CheckpointsOrder: "updated_desc"})
		if err != nil {
			return nil, err
		}
		facts = mergeFacts(facts, byTrace)
	}
	return facts, nil
}

func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][]any, error) {
	families := factFamilySet(query)
	facts := emptyFacts()
	var err error
	if families["sessions"] {
		facts["sessions"], err = s.queryFactRows(ctx, "workbench_sessions", "session_json", query, []fieldColumn{
			{"sessionId", query.SessionIDs, "session_id"},
			{"traceId", query.TraceIDs, "last_trace_id"},
			{"ownerUserId", query.OwnerUserIDs, "owner_user_id"},
			{"conversationId", query.ConversationIDs, "conversation_id"},
			{"status", query.Statuses, "status"},
		})
		if err != nil {
			return nil, err
		}
	}
	if families["messages"] {
		facts["messages"], err = s.queryFactRows(ctx, "workbench_messages", "message_json", query, []fieldColumn{
			{"messageId", query.MessageIDs, "message_id"},
			{"sessionId", query.SessionIDs, "session_id"},
			{"traceId", query.TraceIDs, "trace_id"},
			{"turnId", query.TurnIDs, "turn_id"},
			{"status", query.Statuses, "status"},
		})
		if err != nil {
			return nil, err
		}
	}
	if families["parts"] {
		facts["parts"], err = s.queryFactRows(ctx, "workbench_parts", "part_json", query, []fieldColumn{
			{"messageId", query.MessageIDs, "message_id"},
			{"sessionId", query.SessionIDs, "session_id"},
			{"traceId", query.TraceIDs, "trace_id"},
			{"turnId", query.TurnIDs, "turn_id"},
			{"status", query.Statuses, "status"},
		})
		if err != nil {
			return nil, err
		}
	}
	if families["turns"] {
		facts["turns"], err = s.queryFactRows(ctx, "workbench_turns", "turn_json", query, []fieldColumn{
			{"turnId", query.TurnIDs, "turn_id"},
			{"sessionId", query.SessionIDs, "session_id"},
			{"traceId", query.TraceIDs, "trace_id"},
			{"messageId", query.MessageIDs, "message_id"},
			{"status", query.Statuses, "status"},
		})
		if err != nil {
			return nil, err
		}
	}
	if families["traceEvents"] {
		facts["traceEvents"], err = s.queryFactRows(ctx, "workbench_trace_events", "event_json", query, []fieldColumn{
			{"traceId", query.TraceIDs, "trace_id"},
			{"sessionId", query.SessionIDs, "session_id"},
			{"turnId", query.TurnIDs, "turn_id"},
			{"messageId", query.MessageIDs, "message_id"},
			{"eventType", query.EventTypes, "event_type"},
		})
		if err != nil {
			return nil, err
		}
	}
	if families["checkpoints"] {
		facts["checkpoints"], err = s.queryFactRows(ctx, "workbench_projection_checkpoints", "checkpoint_json", query, []fieldColumn{
			{"traceId", query.TraceIDs, "trace_id"},
			{"sessionId", query.SessionIDs, "session_id"},
			{"turnId", query.TurnIDs, "turn_id"},
			{"runId", query.RunIDs, "run_id"},
			{"commandId", query.CommandIDs, "command_id"},
			{"projectionStatus", query.ProjectionStatuses, "projection_status"},
			{"projectionHealth", query.ProjectionHealths, "projection_health"},
		})
		if err != nil {
			return nil, err
		}
	}
	return facts, nil
}

func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn string, query factQuery, fields []fieldColumn) ([]any, error) {
	clauses, args := factClauses(query, fields)
	limit := boundedLimit(query.Limit, 0, 1000)
	family := factFamilyForTable(table)
	stage := "workbench_runtime.db." + family
	if table == "workbench_trace_events" {
		if query.AfterProjectedSeq > 0 {
			args = append(args, query.AfterProjectedSeq)
			clauses = append(clauses, fmt.Sprintf("projected_seq > $%d", len(args)))
		}
		sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY projected_seq ASC", jsonColumn, table, whereSQL(clauses))
		if limit > 0 {
			args = append(args, limit)
			sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
		}
		return s.queryJSONColumn(ctx, stage, table, sqlText, args)
	}
	orderDirection := "ASC"
	if factOrder(query, family) == "updated_desc" {
		orderDirection = "DESC"
	}
	if table == "workbench_sessions" && query.SessionProjection == "summary" {
		stage = "workbench_runtime.db.sessions_summary"
		sqlText := fmt.Sprintf("SELECT %s FROM workbench_sessions%s ORDER BY updated_at %s", sessionSummarySelectClause(), whereSQL(clauses), orderDirection)
		if limit > 0 {
			args = append(args, limit)
			sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
		}
		return s.querySessionSummaries(ctx, stage, sqlText, args)
	}
	sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY updated_at %s", jsonColumn, table, whereSQL(clauses), orderDirection)
	if limit > 0 {
		args = append(args, limit)
		sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
	}
	return s.queryJSONColumn(ctx, stage, table, sqlText, args)
}

func (s *Server) queryJSONColumn(ctx context.Context, stage string, table string, sqlText string, args []any) ([]any, error) {
	result := []any{}
	err := s.withOtelInternalSpan(ctx, stage, s.dbQueryAttrs(table, len(args), nil), 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 (s *Server) dbQueryAttrs(table string, argCount int, extra map[string]any) map[string]any {
	stats := s.db.Stats()
	attrs := map[string]any{
		"db.system":                   "postgresql",
		"db.operation.name":           "SELECT",
		"db.sql.table":                table,
		"db.query.arg_count":          int64(argCount),
		"db.pool.max_open":            int64(s.config.PoolMax),
		"db.pool.open_connections":    int64(stats.OpenConnections),
		"db.pool.in_use":              int64(stats.InUse),
		"db.pool.idle":                int64(stats.Idle),
		"db.pool.wait_count":          int64(stats.WaitCount),
		"db.pool.wait_duration_ms":    stats.WaitDuration.Milliseconds(),
		"db.pool.max_idle_closed":     int64(stats.MaxIdleClosed),
		"db.pool.max_lifetime_closed": int64(stats.MaxLifetimeClosed),
	}
	for key, value := range extra {
		attrs[key] = value
	}
	return attrs
}

func (s *Server) querySessionSummaries(ctx context.Context, stage string, sqlText string, args []any) ([]any, error) {
	var lastErr error
	for attempt := 1; attempt <= sessionSummaryQueryMaxAttempts; attempt++ {
		attemptResult, err := s.querySessionSummariesAttempt(ctx, stage, sqlText, args, attempt, sessionSummaryQueryMaxAttempts)
		if err == nil {
			return attemptResult, nil
		}
		lastErr = err
		if attempt == sessionSummaryQueryMaxAttempts || !retryableSessionSummaryQueryError(err) || ctx.Err() != nil {
			break
		}
		backoff := sessionSummaryRetryBackoff(attempt)
		timer := time.NewTimer(backoff)
		select {
		case <-ctx.Done():
			timer.Stop()
			return nil, wrapQueryStage(stage, ctx.Err())
		case <-timer.C:
		}
	}
	return nil, wrapQueryStage(stage, lastErr)
}

func (s *Server) querySessionSummariesAttempt(ctx context.Context, stage string, sqlText string, args []any, attempt int, maxAttempts int) ([]any, error) {
	result := []any{}
	attemptCtx, cancel := context.WithTimeout(ctx, sessionSummaryQueryAttemptTimeout)
	defer cancel()
	attrs := s.dbQueryAttrs("workbench_sessions", len(args), map[string]any{
		"db.index.expected": "idx_workbench_sessions_owner_updated|idx_workbench_sessions_updated",
		"retryAttempt":      int64(attempt),
		"retryMax":          int64(maxAttempts),
	})
	if attempt > 1 {
		attrs["retryBackoffMs"] = sessionSummaryRetryBackoff(attempt - 1).Milliseconds()
	}
	err := s.withOtelInternalSpan(attemptCtx, stage, attrs, 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 sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID sql.NullString
			var projectedSeq sql.NullInt64
			var createdAt, updatedAt sql.NullString
			if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &createdAt, &updatedAt); err != nil {
				return err
			}
			if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
				continue
			}
			item := map[string]any{
				"id":             sessionID.String,
				"sessionId":      sessionID.String,
				"ownerUserId":    nullableString(ownerUserID),
				"projectId":      nullableString(projectID),
				"conversationId": nullableString(conversationID),
				"threadId":       nullableString(threadID),
				"status":         first(nullableStringValue(status), "unknown"),
				"lastTraceId":    nullableString(lastTraceID),
				"projectedSeq":   nullableInt(projectedSeq),
				"createdAt":      nullableTimeText(createdAt),
				"updatedAt":      first(nullableTimeText(updatedAt), nullableTimeText(createdAt)),
				"valuesPrinted":  false,
				"valuesRedacted": true,
			}
			result = append(result, item)
		}
		return rows.Err()
	})
	return result, err
}

func sessionSummaryRetryBackoff(attempt int) time.Duration {
	if attempt <= 1 {
		return sessionSummaryQueryRetryBackoff
	}
	return time.Duration(attempt) * sessionSummaryQueryRetryBackoff
}

func retryableSessionSummaryQueryError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
		return true
	}
	switch sqlErrorCode(err) {
	case "40001", "40P01", "55P03", "57014", "query_timeout":
		return true
	default:
		return false
	}
}

func (s *Server) queryMessageSummaries(ctx context.Context, sessionIDs []string) ([]any, error) {
	cleaned := uniqueStrings(sessionIDs)
	if len(cleaned) == 0 {
		return []any{}, nil
	}
	args := []any{}
	placeholders := []string{}
	for _, sid := range cleaned {
		args = append(args, sid)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(args)))
	}
	bySession := map[string]map[string]any{}
	countSQL := fmt.Sprintf("SELECT session_id, COUNT(*) AS message_count FROM workbench_messages WHERE session_id IN (%s) GROUP BY session_id", strings.Join(placeholders, ", "))
	countStage := "workbench_runtime.db.message_counts"
	err := s.withOtelInternalSpan(ctx, countStage, s.dbQueryAttrs("workbench_messages", len(args), map[string]any{"db.index.expected": "idx_workbench_messages_session_updated"}), func(spanCtx context.Context) error {
		rows, err := s.db.QueryContext(spanCtx, countSQL, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var sessionID sql.NullString
			var messageCount sql.NullInt64
			if err := rows.Scan(&sessionID, &messageCount); err != nil {
				return err
			}
			if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
				continue
			}
			item := map[string]any{"sessionId": sessionID.String, "messageCount": nullableInt(messageCount), "valuesRedacted": true}
			bySession[sessionID.String] = item
		}
		return rows.Err()
	})
	if err != nil {
		return nil, wrapQueryStage(countStage, err)
	}
	firstUserSQL := fmt.Sprintf("SELECT DISTINCT ON (session_id) session_id, message_json FROM workbench_messages WHERE session_id IN (%s) AND role = 'user' ORDER BY session_id ASC, updated_at ASC, message_id ASC", strings.Join(placeholders, ", "))
	firstUserStage := "workbench_runtime.db.first_user_messages"
	err = s.withOtelInternalSpan(ctx, firstUserStage, s.dbQueryAttrs("workbench_messages", len(args), map[string]any{"db.index.expected": "idx_workbench_messages_session_role_updated"}), func(spanCtx context.Context) error {
		rows, err := s.db.QueryContext(spanCtx, firstUserSQL, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var sessionID, firstUserMessageJSON sql.NullString
			if err := rows.Scan(&sessionID, &firstUserMessageJSON); err != nil {
				return err
			}
			if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
				continue
			}
			item := bySession[sessionID.String]
			if item == nil {
				item = map[string]any{"sessionId": sessionID.String, "messageCount": 0, "valuesRedacted": true}
				bySession[sessionID.String] = item
			}
			if firstUserMessageJSON.Valid && strings.TrimSpace(firstUserMessageJSON.String) != "" {
				var message map[string]any
				if json.Unmarshal([]byte(firstUserMessageJSON.String), &message) == nil && len(message) > 0 {
					item["firstUserMessagePreview"] = firstUserPreview([]map[string]any{message})
				}
			}
		}
		return rows.Err()
	})
	if err != nil {
		return nil, wrapQueryStage(firstUserStage, err)
	}
	result := []any{}
	for _, sid := range cleaned {
		if item := bySession[sid]; item != nil {
			result = append(result, item)
		}
	}
	return result, nil
}

func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) {
	clauses := []string{}
	args := []any{}
	addTextClause(&clauses, &args, "trace_id", query.TraceID)
	addTextClause(&clauses, &args, "agent_session_id", query.SessionID)
	addTextClause(&clauses, &args, "worker_session_id", query.WorkerSessionID)
	addTextClause(&clauses, &args, "level", query.Level)
	sqlText := fmt.Sprintf("SELECT event_json FROM agent_trace_events%s ORDER BY occurred_at ASC, id ASC", whereSQL(clauses))
	if limit := boundedLimit(query.Limit, 0, 1000); limit > 0 {
		args = append(args, limit)
		sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
	}
	return s.queryJSONColumn(ctx, "workbench_runtime.db.agent_trace_events", "agent_trace_events", sqlText, args)
}

func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([]any, error) {
	limit := boundedLimit(query.Limit, 100, 500)
	args := []any{query.AfterSeq}
	clauses := []string{"outbox_seq > $1"}
	addTextClause(&clauses, &args, "trace_id", query.TraceID)
	addTextClause(&clauses, &args, "session_id", query.SessionID)
	args = append(args, limit)
	sqlText := fmt.Sprintf("SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE %s ORDER BY outbox_seq ASC LIMIT $%d", strings.Join(clauses, " AND "), len(args))
	stage := "workbench_runtime.db.projection_outbox"
	result := []any{}
	err := s.withOtelInternalSpan(ctx, stage, map[string]any{
		"db.system":          "postgresql",
		"db.operation.name":  "SELECT",
		"db.sql.table":       "workbench_projection_outbox",
		"db.query.arg_count": int64(len(args)),
	}, 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 outboxSeq, projectedSeq, sourceSeq sql.NullInt64
			var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString
			var terminal, sealed sql.NullBool
			var raw sql.NullString
			var createdAt sql.NullTime
			if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil {
				return err
			}
			var payload any
			if raw.Valid && strings.TrimSpace(raw.String) != "" {
				_ = json.Unmarshal([]byte(raw.String), &payload)
			}
			result = append(result, map[string]any{
				"outboxSeq":     nullableInt(outboxSeq),
				"traceId":       nullableString(traceID),
				"sessionId":     nullableString(sessionID),
				"turnId":        nullableString(turnID),
				"messageId":     nullableString(messageID),
				"projectedSeq":  nullableInt(projectedSeq),
				"sourceSeq":     nullableInt(sourceSeq),
				"sourceEventId": nullableString(sourceEventID),
				"commitType":    nullableString(commitType),
				"terminal":      nullableBool(terminal),
				"sealed":        nullableBool(sealed),
				"payload":       payload,
				"createdAt":     nullableTime(createdAt),
				"valuesPrinted": false,
			})
		}
		return rows.Err()
	})
	if err != nil {
		return nil, wrapQueryStage(stage, err)
	}
	return result, nil
}

func factClauses(query factQuery, fields []fieldColumn) ([]string, []any) {
	clauses := []string{}
	args := []any{}
	for _, field := range fields {
		value := singularField(query, field.field)
		addTextClause(&clauses, &args, field.column, value)
		addTextListClause(&clauses, &args, field.column, field.values)
	}
	return clauses, args
}

func addTextClause(clauses *[]string, args *[]any, column string, value string) {
	text := strings.TrimSpace(value)
	if text == "" {
		return
	}
	*args = append(*args, text)
	*clauses = append(*clauses, fmt.Sprintf("%s = $%d", column, len(*args)))
}

func addTextListClause(clauses *[]string, args *[]any, column string, values []string) {
	cleaned := []string{}
	for _, value := range values {
		if text := strings.TrimSpace(value); text != "" {
			cleaned = append(cleaned, text)
		}
	}
	if len(cleaned) == 0 {
		return
	}
	placeholders := []string{}
	for _, value := range cleaned {
		*args = append(*args, value)
		placeholders = append(placeholders, fmt.Sprintf("$%d", len(*args)))
	}
	*clauses = append(*clauses, fmt.Sprintf("%s IN (%s)", column, strings.Join(placeholders, ", ")))
}

func singularField(query factQuery, field string) string {
	switch field {
	case "sessionId":
		return query.SessionID
	case "traceId":
		return query.TraceID
	case "messageId":
		return query.MessageID
	case "turnId":
		return query.TurnID
	case "status":
		return query.Status
	case "ownerUserId":
		return query.OwnerUserID
	case "conversationId":
		return query.ConversationID
	case "eventType":
		return query.EventType
	case "projectionStatus":
		return query.ProjectionStatus
	case "projectionHealth":
		return query.ProjectionHealth
	case "runId":
		return query.RunID
	case "commandId":
		return query.CommandID
	default:
		return ""
	}
}

func whereSQL(clauses []string) string {
	if len(clauses) == 0 {
		return ""
	}
	return " WHERE " + strings.Join(clauses, " AND ")
}

func emptyFacts() map[string][]any {
	return map[string][]any{
		"sessions":         []any{},
		"messageSummaries": []any{},
		"messages":         []any{},
		"parts":            []any{},
		"turns":            []any{},
		"traceEvents":      []any{},
		"checkpoints":      []any{},
	}
}

func factFamilySet(query factQuery) map[string]bool {
	all := []string{"sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"}
	selected := query.Families
	if len(selected) == 0 {
		selected = query.FactFamilies
	}
	result := map[string]bool{}
	if len(selected) == 0 {
		for _, family := range all {
			result[family] = true
		}
		return result
	}
	valid := map[string]bool{}
	for _, family := range all {
		valid[family] = true
	}
	for _, family := range selected {
		family = strings.TrimSpace(family)
		if valid[family] {
			result[family] = true
		}
	}
	if len(result) == 0 {
		for _, family := range all {
			result[family] = true
		}
	}
	return result
}

func factFamilyForTable(table string) string {
	switch table {
	case "workbench_sessions":
		return "sessions"
	case "workbench_messages":
		return "messages"
	case "workbench_parts":
		return "parts"
	case "workbench_turns":
		return "turns"
	case "workbench_trace_events":
		return "traceEvents"
	case "workbench_projection_checkpoints":
		return "checkpoints"
	default:
		return "unknown"
	}
}

func factOrder(query factQuery, family string) string {
	value := query.Order
	switch family {
	case "sessions":
		if query.SessionsOrder != "" {
			value = query.SessionsOrder
		}
	case "messages":
		if query.MessagesOrder != "" {
			value = query.MessagesOrder
		}
	case "parts":
		if query.PartsOrder != "" {
			value = query.PartsOrder
		}
	case "turns":
		if query.TurnsOrder != "" {
			value = query.TurnsOrder
		}
	case "checkpoints":
		if query.CheckpointsOrder != "" {
			value = query.CheckpointsOrder
		}
	}
	if value == "updated_desc" {
		return "updated_desc"
	}
	return "updated_asc"
}

func sessionSummarySelectClause() string {
	return strings.Join([]string{
		"session_id",
		"owner_user_id",
		"project_id",
		"conversation_id",
		"thread_id",
		"status",
		"last_trace_id",
		"projected_seq",
		"created_at",
		"updated_at",
	}, ", ")
}

func providerProfileFromSessionJSON(raw sql.NullString) string {
	if !raw.Valid || strings.TrimSpace(raw.String) == "" {
		return ""
	}
	var payload map[string]any
	if err := json.Unmarshal([]byte(raw.String), &payload); err != nil {
		return ""
	}
	if value := strings.TrimSpace(text(payload["providerProfile"])); value != "" {
		return value
	}
	if value := strings.TrimSpace(text(objectMap(payload["sessionJson"])["providerProfile"])); value != "" {
		return value
	}
	return ""
}

func boundedLimit(value int, fallback int, max int) int {
	if value <= 0 {
		return fallback
	}
	if max > 0 && value > max {
		return max
	}
	return value
}

func visibleSessions(rows []any, actor sessionActor) []map[string]any {
	result := []map[string]any{}
	for _, row := range rows {
		item := objectMap(row)
		if len(item) == 0 {
			continue
		}
		if strings.ToLower(text(item["status"])) == "archived" {
			continue
		}
		if text(item["ownerUserId"]) != strings.TrimSpace(actor.ID) {
			continue
		}
		result = append(result, item)
	}
	return result
}

func sessionSummary(session map[string]any, facts map[string][]any) map[string]any {
	sid := sessionID(session)
	if sid == "" {
		return nil
	}
	traceID := first(lastTraceID(session), latestTraceIDForSession(facts, sid))
	var projection map[string]any
	if traceID != "" {
		projection = projectionForTrace(facts, traceID)
	}
	turn := turnForTrace(facts, traceID)
	checkpoint := checkpointForTrace(facts, traceID)
	messages := messagesForSession(facts, sid)
	messageSummary := messageSummaryForSession(facts, sid)
	messageCount := len(messages)
	firstUserMessagePreview := firstUserPreview(messages)
	if len(messageSummary) > 0 {
		messageCount = intFromAny(messageSummary["messageCount"], messageCount)
		if preview := messageSummary["firstUserMessagePreview"]; preview != nil {
			firstUserMessagePreview = preview
		}
	}
	checkpointStatus := terminalStatus(text(checkpoint["status"]))
	turnStatus := normalizeStatus(turn["status"])
	status := normalizeStatus(first(checkpointStatus, turnStatus, text(session["status"])))
	timingSource := turn
	if checkpointStatus != "" {
		timingSource = checkpoint
	}
	timing := timingProjection(timingSource, status)
	var turnSummary any
	if len(turn) > 0 {
		turnSummary = map[string]any{"turnId": first(text(turn["turnId"]), traceID), "traceId": traceID, "status": status, "running": isRunning(status), "terminal": isTerminal(status), "eventCount": projection["lastProjectedSeq"], "projection": projection, "projectionStatus": projection["projectionStatus"], "projectionHealth": projection["projectionHealth"], "staleMs": projection["staleMs"], "blocker": projection["blocker"], "timing": timing, "startedAt": timing["startedAt"], "lastEventAt": timing["lastEventAt"], "finishedAt": timing["finishedAt"], "durationMs": timing["durationMs"], "updatedAt": updatedAt(turn)}
	}
	return map[string]any{"sessionId": sid, "threadId": nullableText(session["threadId"]), "agentId": first(text(session["agentId"]), "hwlab-code-agent"), "status": status, "running": isRunning(status), "terminal": isTerminal(status), "lastTraceId": nullableText(traceID), "projection": projection, "projectionStatus": projection["projectionStatus"], "projectionHealth": projection["projectionHealth"], "staleMs": projection["staleMs"], "blocker": projection["blocker"], "providerProfile": nullableText(first(text(session["providerProfile"]), text(objectMap(session["sessionJson"])["providerProfile"]))), "messageCount": messageCount, "firstUserMessagePreview": firstUserMessagePreview, "updatedAt": updatedAt(session), "turnSummary": turnSummary, "valuesRedacted": true}
}

func messageSummaryForSession(facts map[string][]any, sid string) map[string]any {
	for _, row := range facts["messageSummaries"] {
		item := objectMap(row)
		if text(item["sessionId"]) == sid {
			return item
		}
	}
	return map[string]any{}
}

func intFromAny(value any, fallback int) int {
	switch typed := value.(type) {
	case int:
		return typed
	case int64:
		return int(typed)
	case float64:
		return int(typed)
	case json.Number:
		if parsed, err := typed.Int64(); err == nil {
			return int(parsed)
		}
	}
	return fallback
}

func messagesForSession(facts map[string][]any, sid string) []map[string]any {
	result := []map[string]any{}
	for _, row := range facts["messages"] {
		item := objectMap(row)
		if text(item["sessionId"]) == sid {
			result = append(result, item)
		}
	}
	return result
}

func firstUserPreview(messages []map[string]any) any {
	for _, message := range messages {
		if text(message["role"]) != "user" {
			continue
		}
		if preview := text(message["textPreview"]); preview != "" {
			return preview
		}
		body := first(text(message["text"]), text(message["content"]), text(message["message"]))
		if body == "" {
			return nil
		}
		if len(body) > 240 {
			return body[:240]
		}
		return body
	}
	return nil
}

func projectionForTrace(facts map[string][]any, traceID string) map[string]any {
	checkpoint := checkpointForTrace(facts, traceID)
	if len(checkpoint) == 0 {
		return map[string]any{"projectionStatus": "unknown", "projectionHealth": "unknown", "lastProjectedSeq": nil, "sourceRunId": nil, "sourceCommandId": nil, "staleMs": nil, "blocker": nil, "updatedAt": nil, "valuesRedacted": true}
	}
	status := normalizeProjectionStatus(text(checkpoint["projectionStatus"]))
	health := normalizeProjectionHealth(text(checkpoint["projectionHealth"]), status)
	timing := timingProjection(checkpoint, normalizeStatus(checkpoint["status"]))
	return map[string]any{"projectionStatus": status, "projectionHealth": health, "lastProjectedSeq": seq(checkpoint), "sourceRunId": nullableText(first(text(checkpoint["runId"]), text(checkpoint["sourceRunId"]))), "sourceCommandId": nullableText(first(text(checkpoint["commandId"]), text(checkpoint["sourceCommandId"]))), "staleMs": nil, "blocker": firstAny(objectMap(checkpoint["diagnostic"])["blocker"], checkpoint["blocker"]), "timing": timing, "startedAt": timing["startedAt"], "lastEventAt": timing["lastEventAt"], "finishedAt": timing["finishedAt"], "durationMs": timing["durationMs"], "updatedAt": updatedAt(checkpoint), "valuesRedacted": true}
}

func timingProjection(record map[string]any, status string) map[string]any {
	source := objectMap(record["timing"])
	observedAt := time.Now().UTC().Format(time.RFC3339Nano)
	startedAt := timestamp(source["startedAt"])
	lastEventAt := timestamp(source["lastEventAt"])
	terminal := boolValue(record["terminal"]) || isTerminal(status)
	finishedAt := ""
	if terminal {
		finishedAt = timestamp(source["finishedAt"])
	}
	return map[string]any{"startedAt": nullableText(startedAt), "lastEventAt": nullableText(lastEventAt), "finishedAt": nullableText(finishedAt), "durationMs": elapsedMs(startedAt, first(finishedAt, observedAt)), "observedAt": nilIf(terminal, observedAt), "lastEventAgeMs": nilIf(terminal, elapsedMs(lastEventAt, observedAt)), "valuesRedacted": true}
}

func checkpointForTrace(facts map[string][]any, traceID string) map[string]any {
	if traceID == "" {
		return map[string]any{}
	}
	for _, row := range facts["checkpoints"] {
		item := objectMap(row)
		if text(item["traceId"]) == traceID {
			return item
		}
	}
	return map[string]any{}
}

func turnForTrace(facts map[string][]any, traceID string) map[string]any {
	if traceID == "" {
		return map[string]any{}
	}
	for _, row := range facts["turns"] {
		item := objectMap(row)
		if text(item["traceId"]) == traceID {
			return item
		}
	}
	return map[string]any{}
}

func latestTraceIDForSession(facts map[string][]any, sid string) string {
	for _, row := range facts["turns"] {
		item := objectMap(row)
		if text(item["sessionId"]) == sid && text(item["traceId"]) != "" {
			return text(item["traceId"])
		}
	}
	return ""
}

func mergeFacts(sets ...map[string][]any) map[string][]any {
	merged := emptyFacts()
	keys := map[string]string{"sessions": "sessionId", "messages": "messageId", "parts": "partId", "turns": "turnId", "traceEvents": "id", "checkpoints": "traceId"}
	for _, set := range sets {
		for family, idKey := range keys {
			seen := map[string]bool{}
			for _, row := range merged[family] {
				if id := text(objectMap(row)[idKey]); id != "" {
					seen[id] = true
				}
			}
			for _, row := range set[family] {
				id := text(objectMap(row)[idKey])
				if id != "" && seen[id] {
					continue
				}
				merged[family] = append(merged[family], row)
				if id != "" {
					seen[id] = true
				}
			}
		}
	}
	return merged
}

func sessionListContainsRoute(sessions []map[string]any, routeID string) bool {
	for _, session := range sessions {
		if sessionID(session) == routeID || text(session["conversationId"]) == routeID {
			return true
		}
	}
	return false
}

func sessionID(session map[string]any) string {
	return first(text(session["sessionId"]), text(session["id"]))
}
func lastTraceID(session map[string]any) string {
	return first(text(session["lastTraceId"]), text(session["traceId"]))
}
func updatedAt(record map[string]any) any {
	return nullableText(first(text(record["updatedAt"]), text(record["occurredAt"]), text(record["createdAt"])))
}
func mapSessions(sessions []map[string]any, fn func(map[string]any) string) []string {
	out := []string{}
	for _, session := range sessions {
		if value := fn(session); value != "" {
			out = append(out, value)
		}
	}
	return out
}
func uniqueStrings(values []string) []string {
	seen := map[string]bool{}
	out := []string{}
	for _, value := range values {
		if value != "" && !seen[value] {
			seen[value] = true
			out = append(out, value)
		}
	}
	return out
}
func cursorOffset(value string) int {
	text := strings.TrimSpace(value)
	if strings.HasPrefix(text, "idx:") {
		text = strings.TrimPrefix(text, "idx:")
	}
	parsed, err := strconv.Atoi(text)
	if err != nil || parsed < 0 {
		return 0
	}
	return parsed
}
func cursorOrNil(offset int) any {
	if offset > 0 {
		return fmt.Sprintf("idx:%d", offset)
	}
	return nil
}
func nextCursor(offset int, limit int, hasMore bool) any {
	if hasMore {
		return fmt.Sprintf("idx:%d", offset+limit)
	}
	return nil
}
func objectMap(value any) map[string]any {
	if item, ok := value.(map[string]any); ok {
		return item
	}
	return map[string]any{}
}
func text(value any) string {
	if value == nil {
		return ""
	}
	if stringValue, ok := value.(string); ok {
		return strings.TrimSpace(stringValue)
	}
	result := strings.TrimSpace(fmt.Sprint(value))
	if result == "<nil>" {
		return ""
	}
	return result
}
func nullableText(value any) any {
	text := text(value)
	if text == "" {
		return nil
	}
	return text
}
func boolValue(value any) bool { b, _ := value.(bool); return b }
func nilIf(condition bool, value any) any {
	if condition {
		return nil
	}
	return value
}
func firstAny(values ...any) any {
	for _, value := range values {
		if value != nil {
			return value
		}
	}
	return nil
}

func seq(record map[string]any) any {
	for _, key := range []string{"projectedSeq", "sourceSeq", "seq"} {
		switch value := record[key].(type) {
		case int64:
			if value >= 0 {
				return value
			}
		case int:
			if value >= 0 {
				return value
			}
		case float64:
			if value >= 0 {
				return int64(value)
			}
		}
	}
	return nil
}

func timestamp(value any) string {
	parsed, err := time.Parse(time.RFC3339Nano, text(value))
	if err == nil {
		return parsed.UTC().Format(time.RFC3339Nano)
	}
	parsed, err = time.Parse(time.RFC3339, text(value))
	if err == nil {
		return parsed.UTC().Format(time.RFC3339Nano)
	}
	return ""
}

func elapsedMs(startedAt string, endedAt string) any {
	start, err := time.Parse(time.RFC3339Nano, startedAt)
	if err != nil {
		return nil
	}
	end, err := time.Parse(time.RFC3339Nano, endedAt)
	if err != nil || end.Before(start) {
		return nil
	}
	return end.Sub(start).Milliseconds()
}

func normalizeStatus(value any) string {
	status := strings.ReplaceAll(strings.ToLower(text(value)), "_", "-")
	if status == "cancelled" {
		return "canceled"
	}
	if status == "" || status == "<nil>" {
		return "unknown"
	}
	return status
}
func terminalStatus(value string) string {
	status := normalizeStatus(value)
	if isTerminal(status) {
		return status
	}
	return ""
}
func isRunning(status string) bool {
	switch normalizeStatus(status) {
	case "running", "retrying", "pending", "queued", "accepted", "dispatching", "streaming", "processing", "busy":
		return true
	default:
		return false
	}
}
func isTerminal(status string) bool {
	switch normalizeStatus(status) {
	case "completed", "failed", "blocked", "timeout", "canceled", "stale", "thread-resume-failed", "interrupted", "expired":
		return true
	default:
		return false
	}
}
func normalizeProjectionStatus(value string) string {
	status := normalizeStatus(value)
	if status == "caughtup" {
		return "caught-up"
	}
	switch status {
	case "caught-up", "projecting", "blocked", "stalled", "unknown":
		return status
	default:
		return "unknown"
	}
}
func normalizeProjectionHealth(value string, projectionStatus string) string {
	health := normalizeStatus(value)
	if health == "healthy" && projectionStatus == "caught-up" {
		return "caught-up"
	}
	if health == "healthy" && projectionStatus == "projecting" {
		return "projecting"
	}
	switch health {
	case "caught-up", "projecting", "degraded", "stalled", "unavailable", "unknown":
		return health
	default:
		if projectionStatus == "caught-up" || projectionStatus == "projecting" {
			return projectionStatus
		}
		return "unknown"
	}
}

func configFromEnv() Config {
	port := first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_PORT"), os.Getenv("PORT"), "6671")
	return Config{
		Addr:           ":" + port,
		DatabaseURL:    first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_DB_URL"), os.Getenv("HWLAB_CLOUD_DB_URL")),
		QueryTimeout:   time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS", 10000)) * time.Millisecond,
		ReadinessDelay: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS", 2000)) * time.Millisecond,
		PoolMax:        envInt("HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX", 8),
		Cache:          cacheConfigFromEnv(),
	}
}

func (s *Server) persistenceSummary() map[string]any {
	return map[string]any{"ready": true, "durable": true, "adapter": "postgres", "serviceId": serviceID, "cache": s.cache.Summary(), "valuesRedacted": true}
}

func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool {
	defer r.Body.Close()
	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
	if err := decoder.Decode(target); err != nil {
		writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must be valid JSON")
		return false
	}
	return true
}

func writeQueryError(w http.ResponseWriter, err error) {
	log.Printf("%s query failed: %v", serviceID, err)
	errorPayload := map[string]any{
		"code":           "workbench_runtime_query_failed",
		"message":        "workbench runtime query failed",
		"retryable":      true,
		"transient":      true,
		"causeCode":      sqlErrorCode(err),
		"valuesRedacted": true,
	}
	if stage := queryErrorStage(err); stage != "" {
		errorPayload["stage"] = stage
		errorPayload["causeStage"] = stage
	}
	writeJSON(w, http.StatusServiceUnavailable, map[string]any{
		"ok":             false,
		"error":          errorPayload,
		"valuesRedacted": true,
	})
}

func writeAPIError(w http.ResponseWriter, status int, code string, message string) {
	writeJSON(w, status, map[string]any{"ok": false, "error": map[string]any{"code": code, "message": message, "valuesRedacted": true}, "valuesRedacted": true})
}

func methodNotAllowed(w http.ResponseWriter, allow string) {
	w.Header().Set("Allow", allow)
	writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
}

func writeJSON(w http.ResponseWriter, status int, value any) {
	w.Header().Set("Content-Type", "application/json; charset=utf-8")
	w.WriteHeader(status)
	_ = json.NewEncoder(w).Encode(value)
}

func envInt(name string, fallback int) int {
	value := strings.TrimSpace(os.Getenv(name))
	if value == "" {
		return fallback
	}
	parsed, err := strconv.Atoi(value)
	if err != nil || parsed <= 0 {
		return fallback
	}
	return parsed
}

func first(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return value
		}
	}
	return ""
}

func nullableString(value sql.NullString) any {
	if value.Valid {
		return value.String
	}
	return nil
}

func nullableStringValue(value sql.NullString) string {
	if value.Valid {
		return value.String
	}
	return ""
}

func nullableInt(value sql.NullInt64) any {
	if value.Valid {
		return value.Int64
	}
	return int64(0)
}

func nullableBool(value sql.NullBool) bool {
	return value.Valid && value.Bool
}

func nullableTime(value sql.NullTime) string {
	if value.Valid {
		return value.Time.UTC().Format(time.RFC3339Nano)
	}
	return ""
}

func nullableTimeText(value sql.NullString) string {
	if !value.Valid {
		return ""
	}
	raw := strings.TrimSpace(value.String)
	if raw == "" {
		return ""
	}
	parsed, err := time.Parse(time.RFC3339Nano, raw)
	if err == nil {
		return parsed.UTC().Format(time.RFC3339Nano)
	}
	return raw
}

type sqlStateError interface{ SQLState() string }

func sqlErrorCode(err error) string {
	if err == nil {
		return ""
	}
	var stateErr sqlStateError
	if errors.As(err, &stateErr) {
		return stateErr.SQLState()
	}
	return ""
}