package runtimeapi

import (
	"context"
	"encoding/json"
	"errors"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/telemetry"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/nexuscrm/shared/pkg/referenceidentity"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/queryv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
)

// RecordQueryEngine is the current L0 query execution seam. It returns the
// engine carrier (rather than the legacy neutral compatibility projection) so
// the Runtime adapter can preserve exact-decimal sidecars before serialization.
type RecordQueryEngine interface {
	QueryWithSpec(context.Context, queryspec.QuerySpec, *models.UserSession) (*models.QueryResult, error)
	QueryHistory(context.Context, contract.HistoryQueryRequest, *models.UserSession) (*contract.HistoryQueryResult, error)
}

func (s *QueryServer) QueryHistory(ctx context.Context, wireRequest *runtimev1.QueryHistoryRequest) (*runtimev1.QueryHistoryResponse, error) {
	trusted, ok := trustedCallFromContext(ctx)
	if !ok {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureUnauthenticated})
	}
	request, err := mapping.FromWireQueryHistoryRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	receipts, err := queryExecutionReceipts(ctx, s.receipts)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	receipts.CapabilityVersion = semantic.CapabilityName
	if request.MinCatalogRevision != nil && *request.MinCatalogRevision > receipts.CatalogRevision {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureStaleState, Receipts: &receipts, StaleAxis: semantic.StaleAxisCatalog, RequiredRevision: *request.MinCatalogRevision, ServedRevision: receipts.CatalogRevision})
	}
	// History is part of the same metadata-scoped query capability. Require the
	// request-bound Catalog view even though the current engine result does not
	// need it for response projection: otherwise a non-standard composition can
	// execute against repository/global metadata outside ServingAdmission.
	if _, ok := RuntimeMetadataViewFromContext(ctx); !ok {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureUnavailable})
	}
	user := trusted.UserSession()
	filter, err := queryPredicateFromSemanticForTrustedPrincipal(request.Filter, user.ID)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	repositoryCtx, sqlCounter := telemetry.WithSQLStatementCounter(ctx)
	result, err := s.query.QueryHistory(repositoryCtx, contract.HistoryQueryRequest{
		ParentObjectAPIName: request.ParentObjectID, SelectFields: append([]string(nil), request.SelectedFields...),
		ParentID: request.ParentRecordID, Filter: filter, Limit: int(request.Limit), Offset: int(request.Offset),
		SortDesc: request.SortDescending, ScanAll: request.IncludeDeleted, PageSize: int(request.PageSize), CursorOffset: int(request.CursorOffset),
	}, &user)
	s.metrics.observeRepository(recordsOperationQueryHistory, sqlCounter.Count())
	if err != nil {
		return nil, mapping.ToQueryStatusError(classifyQueryEngineError(err))
	}
	response := semantic.HistoryResponse{Done: true, Receipts: receipts}
	if result != nil {
		if result.TotalSize < 0 {
			return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureInternal})
		}
		response.TotalSize, response.Done = uint64(result.TotalSize), result.Done
		for _, row := range result.Rows {
			response.Rows = append(response.Rows, semantic.HistoryRow{ID: row.ID, ParentID: row.ParentID, Field: row.Field, OldValue: row.OldValue, NewValue: row.NewValue, DataType: row.DataType, CreatedDate: row.CreatedDate, CreatedByID: row.CreatedByID})
		}
	}
	wireResponse, err := mapping.ToWireQueryHistoryResponse(response)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	return wireResponse, nil
}

type QueryServer struct {
	runtimev1.UnimplementedRecordQueryServiceServer
	query    RecordQueryEngine
	receipts ReceiptSource
	metrics  *recordsMetrics
}

func NewQueryServer(query RecordQueryEngine, receipts ReceiptSource, metrics *recordsMetrics) (*QueryServer, error) {
	if query == nil {
		return nil, errors.New("runtime-api: query engine is required")
	}
	if metrics == nil {
		metrics = newRecordsMetrics(nil)
	}
	return &QueryServer{query: query, receipts: receipts, metrics: metrics}, nil
}

func (s *QueryServer) QueryRecords(ctx context.Context, wireRequest *runtimev1.QueryRecordsRequest) (*runtimev1.QueryRecordsResponse, error) {
	trusted, ok := trustedCallFromContext(ctx)
	if !ok {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureUnauthenticated})
	}
	request, err := mapping.FromWireQueryRecordsRequest(ctx, wireRequest)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	receipts, err := queryExecutionReceipts(ctx, s.receipts)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	receipts.CapabilityVersion = semantic.CapabilityName
	if request.MinCatalogRevision != nil && *request.MinCatalogRevision > receipts.CatalogRevision {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureStaleState, Receipts: &receipts, StaleAxis: semantic.StaleAxisCatalog, RequiredRevision: *request.MinCatalogRevision, ServedRevision: receipts.CatalogRevision})
	}
	view, ok := RuntimeMetadataViewFromContext(ctx)
	if !ok {
		return nil, mapping.ToQueryStatusError(&semantic.Failure{Category: semantic.FailureUnavailable})
	}
	user := trusted.UserSession()
	spec, err := querySpecFromSemanticForTrustedPrincipal(request, user.ID)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	queryCtx := referenceidentity.WithAuthority(ctx, view)
	if request.AggregateFetchLimit > 0 {
		queryCtx = queryspec.WithAggregateFetchLimit(queryCtx, int(request.AggregateFetchLimit))
	}
	repositoryCtx, sqlCounter := telemetry.WithSQLStatementCounter(queryCtx)
	result, err := s.query.QueryWithSpec(repositoryCtx, spec, &user)
	s.metrics.observeRepository(recordsOperationQuery, sqlCounter.Count())
	if err != nil {
		return nil, mapping.ToQueryStatusError(classifyQueryEngineError(err))
	}
	response, err := semanticQueryResponse(result, receipts, view, request.ObjectID)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	wireResponse, err := mapping.ToWireQueryRecordsResponse(response)
	if err != nil {
		return nil, mapping.ToQueryStatusError(err)
	}
	return wireResponse, nil
}

func queryExecutionReceipts(ctx context.Context, source ReceiptSource) (semantic.Receipts, error) {
	if scope, ok := runtimeExecutionScopeFromContext(ctx); ok {
		return scope.receipts, nil
	}
	if source == nil {
		return semantic.Receipts{}, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	return source.Receipts(ctx)
}

func querySpecFromSemantic(request semantic.Request) (queryspec.QuerySpec, error) {
	return querySpecFromSemanticForTrustedPrincipal(request, "")
}

func querySpecFromSemanticForTrustedPrincipal(request semantic.Request, principalID string) (queryspec.QuerySpec, error) {
	filter, err := queryPredicateFromSemanticForTrustedPrincipal(request.Filter, principalID)
	if err != nil {
		return queryspec.QuerySpec{}, err
	}
	having, err := queryPredicateFromSemanticForContext(request.Having, queryPredicateContextHaving, principalID)
	if err != nil {
		return queryspec.QuerySpec{}, err
	}
	selections, err := querySelectionsFromSemantic(request.Selections)
	if err != nil {
		return queryspec.QuerySpec{}, err
	}
	result := queryspec.QuerySpec{ObjectAPIName: request.ObjectID, SelectFields: selections, Filter: filter, Limit: int(request.Limit), Offset: int(request.Offset), SkipTotalSize: request.SkipTotalSize, ScanAll: request.IncludeDeleted, GroupBy: request.GroupBy, Having: having}
	for _, order := range request.OrderBy {
		direction, err := querySortDirectionFromSemantic(order.Direction)
		if err != nil {
			return queryspec.QuerySpec{}, err
		}
		result.OrderBy = append(result.OrderBy, queryspec.SortCriterion{Field: order.Field, Direction: direction})
	}
	if len(result.OrderBy) > 0 {
		result.SortField = result.OrderBy[0].Field
		result.SortDirection = result.OrderBy[0].Direction
	}
	for _, sub := range request.Subqueries {
		predicate, err := queryPredicateFromSemanticForTrustedPrincipal(sub.Filter, principalID)
		if err != nil {
			return queryspec.QuerySpec{}, err
		}
		item := queryspec.SubQuerySpec{RelationshipName: sub.RelationshipName, SelectFields: append([]string(nil), sub.SelectedFields...), Filter: predicate, Limit: int(sub.Limit)}
		for _, order := range sub.OrderBy {
			direction, err := querySortDirectionFromSemantic(order.Direction)
			if err != nil {
				return queryspec.QuerySpec{}, err
			}
			item.OrderBy = append(item.OrderBy, queryspec.SortCriterion{Field: order.Field, Direction: direction})
		}
		result.SubQueries = append(result.SubQueries, item)
	}
	return result, nil
}

// querySelectionsFromSemantic is the only bridge to the current in-process
// queryspec aggregate carrier. The cross-process contract is closed and typed;
// these strings are reconstructed locally for the existing engine and never
// accepted from the caller.
func querySelectionsFromSemantic(values []semantic.Selection) ([]string, error) {
	result := make([]string, 0, len(values))
	for _, value := range values {
		switch value.Kind {
		case semantic.SelectionField:
			result = append(result, value.Field)
		case semantic.SelectionAggregate:
			function, err := queryAggregateFromSemantic(value.Aggregate)
			if err != nil {
				return nil, err
			}
			selection := function + "(" + value.Field + ")"
			if value.Alias != "" {
				selection += " " + value.Alias
			}
			result = append(result, selection)
		default:
			return nil, invalidQuerySemantic()
		}
	}
	return result, nil
}

func queryAggregateFromSemantic(value semantic.AggregateFunction) (string, error) {
	switch value {
	case semantic.AggregateCount:
		return "COUNT", nil
	case semantic.AggregateSum:
		return "SUM", nil
	case semantic.AggregateAverage:
		return "AVG", nil
	case semantic.AggregateMinimum:
		return "MIN", nil
	case semantic.AggregateMaximum:
		return "MAX", nil
	default:
		return "", invalidQuerySemantic()
	}
}

func querySortDirectionFromSemantic(value semantic.SortDirection) (string, error) {
	switch value {
	case semantic.SortAscending:
		return "ASC", nil
	case semantic.SortDescending:
		return "DESC", nil
	default:
		return "", invalidQuerySemantic()
	}
}

func queryComparisonOperatorFromSemantic(value semantic.ComparisonOperator) (string, error) {
	switch value {
	case semantic.ComparisonEqual:
		return "==", nil
	case semantic.ComparisonNotEqual:
		return "!=", nil
	case semantic.ComparisonGreaterThan:
		return ">", nil
	case semantic.ComparisonGreaterOrEqual:
		return ">=", nil
	case semantic.ComparisonLessThan:
		return "<", nil
	case semantic.ComparisonLessOrEqual:
		return "<=", nil
	case semantic.ComparisonIn:
		return "in", nil
	case semantic.ComparisonNotIn:
		return "not in", nil
	case semantic.ComparisonContains:
		return "contains", nil
	case semantic.ComparisonStartsWith:
		return "startsWith", nil
	case semantic.ComparisonEndsWith:
		return "endsWith", nil
	case semantic.ComparisonLike:
		return "LIKE", nil
	case semantic.ComparisonNotLike:
		return "NOT LIKE", nil
	default:
		return "", invalidQuerySemantic()
	}
}

type queryPredicateContext uint8

const (
	queryPredicateContextFilter queryPredicateContext = iota
	queryPredicateContextHaving
)

// queryFunctionFromSemantic is the executable semantic-to-engine translation
// boundary, not a mirror of queryv1's closed wire vocabulary. The current
// engine compiles WHERE/history/subquery predicates through the formula SQL
// walker, which does not implement queryv1's function set. HAVING is the only
// predicate context with a proven subset: one-field SQL aggregates. Reject all
// other plans before QueryWithSpec so a known semantic function can never fall
// through to a database error or an INTERNAL response.
func queryFunctionFromSemantic(value semantic.Operand, predicateContext queryPredicateContext) (string, error) {
	if predicateContext != queryPredicateContextHaving || len(value.Args) != 1 ||
		value.Args[0].Kind != semantic.OperandField || value.Args[0].Field == "" {
		return "", invalidQuerySemantic()
	}
	switch value.Function {
	case semantic.FunctionCount:
		return "COUNT", nil
	case semantic.FunctionSum:
		return "SUM", nil
	case semantic.FunctionAverage:
		return "AVG", nil
	case semantic.FunctionMinimum:
		return "MIN", nil
	case semantic.FunctionMaximum:
		return "MAX", nil
	default:
		return "", invalidQuerySemantic()
	}
}

func invalidQuerySemantic() *semantic.Failure {
	return &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentQueryPlan}
}

func queryPredicateFromSemantic(value semantic.Predicate) (queryspec.Predicate, error) {
	return queryPredicateFromSemanticForTrustedPrincipal(value, "")
}

func queryPredicateFromSemanticForTrustedPrincipal(value semantic.Predicate, principalID string) (queryspec.Predicate, error) {
	return queryPredicateFromSemanticForContext(value, queryPredicateContextFilter, principalID)
}

func queryPredicateFromSemanticForContext(value semantic.Predicate, predicateContext queryPredicateContext, principalID string) (queryspec.Predicate, error) {
	kind, err := queryPredicateKindFromSemantic(value.Kind)
	if err != nil {
		return queryspec.Predicate{}, err
	}
	result := queryspec.Predicate{Kind: kind, Field: value.Field, UnsupportedReason: ""}
	if value.Kind == semantic.PredicateComparison || value.Kind == semantic.PredicateFieldComparison {
		result.Op, err = queryComparisonOperatorFromSemantic(value.Operator)
		if err != nil {
			return queryspec.Predicate{}, err
		}
	}
	if value.Value != nil {
		result.Value, err = queryScalarValue(*value.Value)
		if err != nil {
			return queryspec.Predicate{}, err
		}
	}
	result.Left, err = queryOperandFromSemanticForContext(value.Left, predicateContext, principalID)
	if err != nil {
		return queryspec.Predicate{}, err
	}
	result.Right, err = queryOperandFromSemanticForContext(value.Right, predicateContext, principalID)
	if err != nil {
		return queryspec.Predicate{}, err
	}
	result.Operand, err = queryOperandFromSemanticForContext(value.Operand, predicateContext, principalID)
	if err != nil {
		return queryspec.Predicate{}, err
	}
	for _, child := range value.Children {
		mapped, err := queryPredicateFromSemanticForContext(child, predicateContext, principalID)
		if err != nil {
			return queryspec.Predicate{}, err
		}
		result.Children = append(result.Children, mapped)
	}
	if value.CorrelatedMembership != nil {
		inner, err := queryPredicateFromSemanticForContext(value.CorrelatedMembership.InnerFilter, predicateContext, principalID)
		if err != nil {
			return queryspec.Predicate{}, err
		}
		m := value.CorrelatedMembership
		result.CorrelatedMembership = &queryspec.CorrelatedMembership{ID: m.ID, OuterField: m.OuterField, Negate: m.Negate, InnerObjectAPIName: m.InnerObjectID, InnerSelectField: m.InnerSelectField, InnerFilter: inner, InnerLimit: int(m.InnerLimit), InnerScanAll: m.InnerIncludeDeleted}
	}
	return result, nil
}

func queryOperandFromSemantic(value semantic.Operand) (queryspec.Operand, error) {
	return queryOperandFromSemanticForContext(value, queryPredicateContextFilter, "")
}

func queryOperandFromSemanticForContext(value semantic.Operand, predicateContext queryPredicateContext, principalID string) (queryspec.Operand, error) {
	if value.Kind == semantic.OperandTrustedContext {
		if value.TrustedContext != semantic.TrustedContextPrincipalID {
			return queryspec.Operand{}, invalidQuerySemantic()
		}
		if principalID == "" {
			return queryspec.Operand{}, &semantic.Failure{Category: semantic.FailureUnauthenticated}
		}
		return queryspec.LiteralOperand(principalID), nil
	}
	kind, err := queryOperandKindFromSemantic(value.Kind)
	if err != nil {
		return queryspec.Operand{}, err
	}
	result := queryspec.Operand{Kind: kind, Field: value.Field}
	if value.Kind == semantic.OperandFunction {
		result.Function, err = queryFunctionFromSemantic(value, predicateContext)
		if err != nil {
			return queryspec.Operand{}, err
		}
	}
	if value.Literal != nil {
		literal, err := queryScalarValue(*value.Literal)
		if err != nil {
			return queryspec.Operand{}, err
		}
		result.Value = literal
	}
	for _, arg := range value.Args {
		mapped, err := queryOperandFromSemanticForContext(arg, predicateContext, principalID)
		if err != nil {
			return queryspec.Operand{}, err
		}
		result.Args = append(result.Args, mapped)
	}
	for _, element := range value.Elements {
		mapped, err := queryOperandFromSemanticForContext(element, predicateContext, principalID)
		if err != nil {
			return queryspec.Operand{}, err
		}
		result.Elements = append(result.Elements, mapped)
	}
	return result, nil
}

func queryPredicateKindFromSemantic(kind semantic.PredicateKind) (queryspec.PredicateKind, error) {
	switch kind {
	case semantic.PredicateNone:
		return queryspec.PredicateNone, nil
	case semantic.PredicateAnd:
		return queryspec.PredicateAnd, nil
	case semantic.PredicateOr:
		return queryspec.PredicateOr, nil
	case semantic.PredicateNot:
		return queryspec.PredicateNot, nil
	case semantic.PredicateExpression:
		return queryspec.PredicateExpression, nil
	case semantic.PredicateComparison:
		return queryspec.PredicateComparison, nil
	case semantic.PredicateFieldComparison:
		return queryspec.PredicateFieldComparison, nil
	case semantic.PredicateCorrelatedMembership:
		return queryspec.PredicateCorrelatedMembership, nil
	default:
		return queryspec.PredicateNone, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentPredicate}
	}
}

func queryOperandKindFromSemantic(kind semantic.OperandKind) (queryspec.OperandKind, error) {
	switch kind {
	case semantic.OperandNone:
		return queryspec.OperandNone, nil
	case semantic.OperandField:
		return queryspec.OperandField, nil
	case semantic.OperandLiteral:
		return queryspec.OperandLiteral, nil
	case semantic.OperandList:
		return queryspec.OperandList, nil
	case semantic.OperandFunction:
		return queryspec.OperandFunction, nil
	default:
		return queryspec.OperandNone, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentPredicate}
	}
}

func queryScalarValue(value semantic.Scalar) (any, error) {
	switch {
	case value.Null:
		return nil, nil
	case value.String != nil:
		return *value.String, nil
	case value.Boolean != nil:
		return *value.Boolean, nil
	case value.Integer != nil:
		return *value.Integer, nil
	case value.Unsigned != nil:
		return *value.Unsigned, nil
	case value.Double != nil:
		return *value.Double, nil
	case value.Decimal != nil:
		text, err := decimalText(value.Decimal.Coefficient, value.Decimal.Scale)
		if err != nil {
			return nil, err
		}
		return queryspec.NumericLiteral(text), nil
	case value.Date != nil:
		return *value.Date, nil
	case value.DateTime != nil:
		return *value.DateTime, nil
	default:
		return nil, &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentPredicate}
	}
}

func decimalText(coefficient string, scale uint32) (string, error) {
	if uint64(scale) > uint64(semantic.MaxSemanticRequestBytes) || len(coefficient) > semantic.MaxSemanticRequestBytes {
		return "", &semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitSemanticRequest, Maximum: semantic.MaxSemanticRequestBytes}
	}
	negative := strings.HasPrefix(coefficient, "-")
	digits := strings.TrimPrefix(coefficient, "-")
	if scale > 0 {
		if len(digits) <= int(scale) {
			padding := int(scale) - len(digits) + 1
			if len(digits)+padding > semantic.MaxSemanticRequestBytes {
				return "", &semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitSemanticRequest, Maximum: semantic.MaxSemanticRequestBytes}
			}
			digits = strings.Repeat("0", padding) + digits
		}
		pivot := len(digits) - int(scale)
		digits = digits[:pivot] + "." + digits[pivot:]
	}
	if negative {
		return "-" + digits, nil
	}
	return digits, nil
}

func semanticQueryResponse(result *models.QueryResult, receipts semantic.Receipts, view *catalogsnapshot.RuntimeMetadataView, defaultObject string) (semantic.Response, error) {
	response := semantic.Response{Done: true, Receipts: receipts}
	if result == nil {
		return response, nil
	}
	if result.TotalSize < 0 {
		return semantic.Response{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
	response.TotalSize, response.TotalSizeUnknown, response.Done, response.Aggregate = uint64(result.TotalSize), result.TotalSizeUnknown, result.Done, result.IsAggregate
	for _, row := range result.Records {
		record, err := semanticQueryRecord(row, view, defaultObject, 1)
		if err != nil {
			return semantic.Response{}, err
		}
		response.Records = append(response.Records, record)
	}
	return response, nil
}

func semanticQueryRecord(row models.SObject, view *catalogsnapshot.RuntimeMetadataView, defaultObject string, depth int) (semantic.Record, error) {
	if depth > semantic.MaxNestingDepth {
		return semantic.Record{}, &semantic.Failure{
			Category:  semantic.FailureResourceExhausted,
			LimitKind: semantic.LimitQueryDepth,
			Maximum:   uint64(semantic.MaxNestingDepth),
		}
	}
	objectID := models.RecordObjectType(row)
	if objectID == "" {
		objectID = defaultObject
	}
	var schema *models.ObjectMetadata
	if view != nil && objectID != "" {
		var ok bool
		schema, ok = view.Object(objectID)
		if !ok {
			// A materialized record that names an object outside the request-bound
			// metadata closure cannot be projected safely. Never fall back to
			// untyped strings or a global/repository metadata read path.
			return semantic.Record{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
	}
	row, err := normalizeQueryRelationshipCarriers(row, view, objectID)
	if err != nil {
		return semantic.Record{}, err
	}
	keys := make([]string, 0, len(row))
	for key := range row {
		if !models.IsRecordMetadataKey(key) {
			keys = append(keys, key)
		}
	}
	sort.Strings(keys)
	record := semantic.Record{ObjectID: objectID, Fields: make([]semantic.Field, 0, len(keys))}
	for _, key := range keys {
		var field *models.FieldMetadata
		if schema != nil {
			field = models.FindField(schema, key)
		}
		if exact, ok := schemautil.ExactDecimalForField(row, key); ok {
			coefficient, scale, valid := exact.CoefficientAndScale()
			if !valid {
				return semantic.Record{}, &semantic.Failure{Category: semantic.FailureInternal}
			}
			record.Fields = append(record.Fields, semantic.Field{Name: key, Value: semantic.Value{Kind: semantic.ValueDecimal, Decimal: &semantic.Decimal{Coefficient: coefficient, Scale: scale}}})
			continue
		}
		value, err := semanticQueryValue(row[key], field, view, depth)
		if err != nil {
			return semantic.Record{}, err
		}
		record.Fields = append(record.Fields, semantic.Field{Name: key, Value: value})
	}
	return record, nil
}

// normalizeQueryRelationshipCarriers translates the current engine's
// metadata-derived <reference-field>_Name compatibility carrier into a neutral
// nested record before it crosses the Runtime boundary. The request-bound
// RuntimeMetadataView is the only relationship authority: suffix shape alone
// never invents an edge. Raw fields are retained because the existing L1 wire
// projection still owns their compatibility behavior.
func normalizeQueryRelationshipCarriers(row models.SObject, view *catalogsnapshot.RuntimeMetadataView, objectID string) (models.SObject, error) {
	if row == nil || view == nil || objectID == "" {
		return row, nil
	}
	relationships := view.RelationshipsForChild(objectID)
	if len(relationships) == 0 {
		return row, nil
	}
	var normalized models.SObject
	copyRow := func() models.SObject {
		if normalized == nil {
			normalized = make(models.SObject, len(row)+len(relationships))
			for key, value := range row {
				normalized[key] = value
			}
		}
		return normalized
	}
	for _, relationship := range relationships {
		if relationship == nil || relationship.FieldAPIName == "" || relationship.RelationshipName == "" || relationship.ParentObjectAPIName == "" {
			continue
		}
		nameValue, hasName := row[relationship.FieldAPIName+"_Name"]
		name, nameIsString := nameValue.(string)
		if !hasName || !nameIsString {
			continue
		}
		nested, err := neutralQueryRelationshipMap(row[relationship.RelationshipName])
		if err != nil {
			return nil, &semantic.Failure{Category: semantic.FailureInternal}
		}
		if nested == nil {
			nested = map[string]interface{}{}
		}
		models.StampObjectType(models.SObject(nested), relationship.ParentObjectAPIName)
		if id, ok := row[relationship.FieldAPIName].(string); ok {
			nested[constants.FieldID] = id
		}
		nested[constants.FieldName] = name
		copyRow()[relationship.RelationshipName] = nested
	}
	if normalized == nil {
		return row, nil
	}
	return normalized, nil
}

func neutralQueryRelationshipMap(value any) (map[string]interface{}, error) {
	if value == nil {
		return nil, nil
	}
	var source map[string]interface{}
	switch typed := value.(type) {
	case map[string]interface{}:
		source = typed
	case models.SObject:
		source = map[string]interface{}(typed)
	case contract.Record:
		source = map[string]interface{}(typed)
	default:
		return nil, errors.New("query relationship carrier is not an object")
	}
	result := make(map[string]interface{}, len(source)+3)
	for key, item := range source {
		result[key] = item
	}
	return result, nil
}

func semanticQueryValue(raw any, field *models.FieldMetadata, view *catalogsnapshot.RuntimeMetadataView, depth int) (semantic.Value, error) {
	if field != nil && models.FieldCanonicalType(field) == constants.FieldTypeJson {
		return semanticAnonymousStructuredValue(raw, depth+1)
	}
	if raw == nil {
		return semantic.Value{Kind: semanticKindForField(field), Null: true}, nil
	}
	switch value := raw.(type) {
	case string:
		kind := semanticKindForField(field)
		switch kind {
		case semantic.ValueDate:
			return semantic.Value{Kind: kind, Date: &value}, nil
		case semantic.ValueDateTime:
			return semantic.Value{Kind: kind, DateTime: &value}, nil
		case semantic.ValueReference:
			return semantic.Value{Kind: kind, Reference: &value}, nil
		default:
			return semantic.Value{Kind: semantic.ValueString, String: &value}, nil
		}
	case []byte:
		text := string(value)
		return semantic.Value{Kind: semantic.ValueString, String: &text}, nil
	case bool:
		return semantic.Value{Kind: semantic.ValueBoolean, Boolean: &value}, nil
	case int:
		converted := int64(value)
		return semantic.Value{Kind: semantic.ValueInteger, Integer: &converted}, nil
	case int8:
		converted := int64(value)
		return semantic.Value{Kind: semantic.ValueInteger, Integer: &converted}, nil
	case int16:
		converted := int64(value)
		return semantic.Value{Kind: semantic.ValueInteger, Integer: &converted}, nil
	case int32:
		converted := int64(value)
		return semantic.Value{Kind: semantic.ValueInteger, Integer: &converted}, nil
	case int64:
		return semantic.Value{Kind: semantic.ValueInteger, Integer: &value}, nil
	case uint:
		converted := uint64(value)
		return semantic.Value{Kind: semantic.ValueUnsigned, Unsigned: &converted}, nil
	case uint8:
		converted := uint64(value)
		return semantic.Value{Kind: semantic.ValueUnsigned, Unsigned: &converted}, nil
	case uint16:
		converted := uint64(value)
		return semantic.Value{Kind: semantic.ValueUnsigned, Unsigned: &converted}, nil
	case uint32:
		converted := uint64(value)
		return semantic.Value{Kind: semantic.ValueUnsigned, Unsigned: &converted}, nil
	case uint64:
		return semantic.Value{Kind: semantic.ValueUnsigned, Unsigned: &value}, nil
	case float32:
		converted := float64(value)
		return semantic.Value{Kind: semantic.ValueDouble, Double: &converted}, nil
	case float64:
		return semantic.Value{Kind: semantic.ValueDouble, Double: &value}, nil
	case json.Number:
		if integer, err := value.Int64(); err == nil {
			return semantic.Value{Kind: semantic.ValueInteger, Integer: &integer}, nil
		}
		decimal, err := strconv.ParseFloat(string(value), 64)
		if err != nil {
			return semantic.Value{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
		return semantic.Value{Kind: semantic.ValueDouble, Double: &decimal}, nil
	case schemautil.ExactDecimal:
		coefficient, scale, ok := value.CoefficientAndScale()
		if !ok {
			return semantic.Value{}, &semantic.Failure{Category: semantic.FailureInternal}
		}
		return semantic.Value{Kind: semantic.ValueDecimal, Decimal: &semantic.Decimal{Coefficient: coefficient, Scale: scale}}, nil
	case time.Time:
		normalized := value.UTC().Format(time.RFC3339Nano)
		if semanticKindForField(field) == semantic.ValueDate {
			date := value.Format("2006-01-02")
			return semantic.Value{Kind: semantic.ValueDate, Date: &date}, nil
		}
		return semantic.Value{Kind: semantic.ValueDateTime, DateTime: &normalized}, nil
	case map[string]interface{}:
		if !models.IsNeutralRecordMap(value) {
			return semanticAnonymousStructuredValue(value, depth+1)
		}
		record, err := semanticQueryRecord(models.SObject(value), view, "", depth+1)
		if err != nil {
			return semantic.Value{}, err
		}
		return semantic.Value{Kind: semantic.ValueObject, Object: &record}, nil
	case []models.SObject:
		values := make([]semantic.Value, 0, len(value))
		for _, row := range value {
			// []SObject is the engine's explicit nested-record carrier. Unlike an
			// anonymous map value, every row must therefore carry the object identity
			// stamped by the engine owner that resolved/materialized it. Without that
			// identity this adapter cannot safely recover metadata-derived field types.
			if models.RecordObjectType(row) == "" {
				return semantic.Value{}, &semantic.Failure{Category: semantic.FailureInternal}
			}
			record, err := semanticQueryRecord(row, view, "", depth+1)
			if err != nil {
				return semantic.Value{}, err
			}
			values = append(values, semantic.Value{Kind: semantic.ValueObject, Object: &record})
		}
		return semantic.Value{Kind: semantic.ValueList, List: &values}, nil
	case []interface{}:
		values := make([]semantic.Value, 0, len(value))
		for _, item := range value {
			mapped, err := semanticQueryValue(item, nil, view, depth+1)
			if err != nil {
				return semantic.Value{}, err
			}
			values = append(values, mapped)
		}
		return semantic.Value{Kind: semantic.ValueList, List: &values}, nil
	case []string:
		values := make([]semantic.Value, 0, len(value))
		for i := range value {
			item := value[i]
			values = append(values, semantic.Value{Kind: semantic.ValueString, String: &item})
		}
		return semantic.Value{Kind: semantic.ValueList, List: &values}, nil
	default:
		return semantic.Value{}, &semantic.Failure{Category: semantic.FailureInternal}
	}
}

// semanticAnonymousStructuredValue projects one metadata-declared structured
// value without interpreting its object keys as record metadata. The existing
// recursive query carrier already represents objects, lists, scalars and null;
// an empty ObjectID is the neutral discriminator for an anonymous value object.
func semanticAnonymousStructuredValue(raw any, depth int) (semantic.Value, error) {
	if depth > semantic.MaxNestingDepth {
		return semantic.Value{}, &semantic.Failure{
			Category:  semantic.FailureResourceExhausted,
			LimitKind: semantic.LimitQueryDepth,
			Maximum:   uint64(semantic.MaxNestingDepth),
		}
	}
	if raw == nil {
		return semantic.Value{Kind: semantic.ValueString, Null: true}, nil
	}
	switch value := raw.(type) {
	case map[string]interface{}:
		keys := make([]string, 0, len(value))
		for key := range value {
			keys = append(keys, key)
		}
		sort.Strings(keys)
		object := semantic.Record{Fields: make([]semantic.Field, 0, len(keys))}
		for _, key := range keys {
			mapped, err := semanticAnonymousStructuredValue(value[key], depth+1)
			if err != nil {
				return semantic.Value{}, err
			}
			object.Fields = append(object.Fields, semantic.Field{Name: key, Value: mapped})
		}
		return semantic.Value{Kind: semantic.ValueObject, Object: &object}, nil
	case []interface{}:
		items := make([]semantic.Value, 0, len(value))
		for _, item := range value {
			mapped, err := semanticAnonymousStructuredValue(item, depth+1)
			if err != nil {
				return semantic.Value{}, err
			}
			items = append(items, mapped)
		}
		return semantic.Value{Kind: semantic.ValueList, List: &items}, nil
	default:
		return semanticQueryValue(raw, nil, nil, depth)
	}
}

func semanticKindForField(field *models.FieldMetadata) semantic.ValueKind {
	if field == nil {
		return semantic.ValueString
	}
	typeName := models.FieldCanonicalType(field)
	if field.Type == constants.FieldTypeFormula && field.ReturnType != nil {
		typeName = constants.NormalizeFieldType(string(*field.ReturnType))
	}
	switch typeName {
	case constants.FieldTypeBoolean:
		return semantic.ValueBoolean
	case constants.FieldTypeInt:
		return semantic.ValueInteger
	case constants.FieldTypeDouble:
		return semantic.ValueDouble
	case constants.FieldTypeCurrency, constants.FieldTypePercent:
		return semantic.ValueDecimal
	case constants.FieldTypeDate:
		return semantic.ValueDate
	case constants.FieldTypeDatetime:
		return semantic.ValueDateTime
	case constants.FieldTypeReference, constants.FieldTypeMasterDetail:
		return semantic.ValueReference
	default:
		return semantic.ValueString
	}
}

func classifyQueryEngineError(err error) error {
	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
		return err
	}
	var semanticFailure *semantic.Failure
	if errors.As(err, &semanticFailure) {
		return semanticFailure
	}
	if failure, ok := classifyRetriableEngineCondition(err); ok {
		return failure
	}
	var denied *contract.FieldDeniedByFLSError
	if errors.As(err, &denied) {
		return &semantic.Failure{Category: semantic.FailureForbidden}
	}
	var historyFailure *contract.HistoryQueryError
	if errors.As(err, &historyFailure) {
		switch historyFailure.Kind {
		case contract.HistoryQueryErrorObjectPermissionDenied, contract.HistoryQueryErrorRecordVisibilityDenied:
			return &semantic.Failure{Category: semantic.FailureForbidden}
		case contract.HistoryQueryErrorMalformed:
			return &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentQueryPlan}
		case contract.HistoryQueryErrorUnavailable:
			return &semantic.Failure{Category: semantic.FailureUnavailable}
		default:
			return &semantic.Failure{Category: semantic.FailureInternal}
		}
	}
	var notFound *appErrors.NotFoundError
	if errors.As(err, &notFound) {
		// QueryWithSpec uses this exact engine error when either the root object
		// or a structured correlated-membership object identifier does not exist.
		// Other not-found resources indicate an internal metadata/execution gap
		// and must not be misreported as a client object-identifier failure.
		if strings.EqualFold(strings.TrimSpace(notFound.Resource), "Object") {
			return &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentObjectID}
		}
		return &semantic.Failure{Category: semantic.FailureInternal}
	}
	var internal interface{ InternalCondition() bool }
	if errors.As(err, &internal) && internal.InternalCondition() {
		return &semantic.Failure{Category: semantic.FailureInternal}
	}
	var validation *appErrors.ValidationError
	if errors.As(err, &validation) && validation.Field == "SelectFields" {
		return &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentSelection}
	}
	var statuser interface{ HTTPStatus() int }
	if errors.As(err, &statuser) {
		switch statuser.HTTPStatus() {
		case http.StatusBadRequest:
			return &semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentQueryPlan}
		case http.StatusUnauthorized:
			return &semantic.Failure{Category: semantic.FailureUnauthenticated}
		case http.StatusForbidden:
			return &semantic.Failure{Category: semantic.FailureForbidden}
		}
	}
	return &semantic.Failure{Category: semantic.FailureInternal}
}