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"

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

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

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)
	}
	server := NewServer(config, db)
	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 NewServer(config Config, db *sql.DB) *Server {
	server := &Server{config: config, db: db, mux: http.NewServeMux()}
	server.routes()
	return server
}

func (s *Server) routes() {
	s.mux.HandleFunc("/health/live", s.handleLive)
	s.mux.HandleFunc("/health/ready", s.handleReady)
	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) handleLive(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "live", "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", "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()
	facts, err := s.queryFacts(ctx, query)
	if err != nil {
		writeQueryError(w, err)
		return
	}
	count := 0
	for _, rows := range facts {
		count += len(rows)
	}
	writeJSON(w, http.StatusOK, map[string]any{
		"contractVersion": "workbench-runtime-facts-v1",
		"facts":           facts,
		"count":           count,
		"persistence":     s.persistenceSummary(),
		"valuesRedacted":  true,
	})
}

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 := boundedLimit(query.Limit, 20, 100)
	offset := query.Offset
	if offset <= 0 {
		offset = cursorOffset(query.Cursor)
	}
	if offset < 0 {
		offset = 0
	}
	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 (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 {
		bySession, err := s.queryFacts(ctx, factQuery{SessionIDs: sessionIDs, Families: []string{"messages", "turns"}, Limit: 1000, MessagesOrder: "updated_asc", TurnsOrder: "updated_desc"})
		if err != nil {
			return nil, err
		}
		facts = mergeFacts(facts, bySession)
	}
	if len(traceIDs) > 0 {
		byTrace, err := s.queryFacts(ctx, factQuery{TraceIDs: traceIDs, Families: []string{"checkpoints"}, Limit: 1000, 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, map[string]any{
		"db.system":          "postgresql",
		"db.operation.name":  "SELECT",
		"db.sql.table":       table,
		"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 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) querySessionSummaries(ctx context.Context, stage string, sqlText string, args []any) ([]any, error) {
	result := []any{}
	err := s.withOtelInternalSpan(ctx, stage, map[string]any{
		"db.system":          "postgresql",
		"db.operation.name":  "SELECT",
		"db.sql.table":       "workbench_sessions",
		"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 sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, sessionJSON sql.NullString
			var createdAt, updatedAt sql.NullString
			var projectedSeq, sourceSeq sql.NullInt64
			var terminal, sealed sql.NullBool
			if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &sessionJSON); 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),
				"sourceSeq":      nullableInt(sourceSeq),
				"sourceEventId":  nullableString(sourceEventID),
				"terminal":       nullableBool(terminal),
				"sealed":         nullableBool(sealed),
				"createdAt":      nullableTimeText(createdAt),
				"updatedAt":      first(nullableTimeText(updatedAt), nullableTimeText(createdAt)),
				"valuesPrinted":  false,
				"valuesRedacted": true,
			}
			if providerProfile := providerProfileFromSessionJSON(sessionJSON); providerProfile != "" {
				item["providerProfile"] = providerProfile
				item["sessionJson"] = map[string]any{"providerProfile": providerProfile, "valuesRedacted": true, "secretMaterialStored": false}
			}
			result = append(result, item)
		}
		return rows.Err()
	})
	if err != nil {
		return nil, wrapQueryStage(stage, err)
	}
	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{},
		"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",
		"source_seq",
		"source_event_id",
		"terminal",
		"sealed",
		"created_at",
		"updated_at",
		"session_json",
	}, ", ")
}

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)
	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": len(messages), "firstUserMessagePreview": firstUserPreview(messages), "updatedAt": updatedAt(session), "turnSummary": turnSummary, "valuesRedacted": true}
}

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),
	}
}

func (s *Server) persistenceSummary() map[string]any {
	return map[string]any{"ready": true, "durable": true, "adapter": "postgres", "serviceId": serviceID, "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 ""
}