package mapping

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"
	"time"

	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protowire"
	"google.golang.org/protobuf/proto"
)

func TestMutationMaximumStructuralCommandFitsReassemblyEnvelope(t *testing.T) {
	request := semantic.Request{CommandID: "cmd-wire-maximum", BatchMode: semantic.BatchIndependent}
	for operationIndex := 0; operationIndex < semantic.MaxOperations; operationIndex++ {
		operation := semantic.Operation{Kind: semantic.OperationCreate, ObjectID: "Object"}
		for fieldIndex := 0; fieldIndex < semantic.MaxFieldsPerOperation; fieldIndex++ {
			value := ""
			operation.Fields = append(operation.Fields, semantic.FieldValue{
				Name: fmt.Sprintf("F%03d", fieldIndex), Kind: semantic.ValueKindString, String: &value,
			})
		}
		request.Operations = append(request.Operations, operation)
	}
	remaining := semantic.MaxSemanticRequestBytes - semantic.SemanticRequestSize(request)
	if remaining <= 0 {
		t.Fatalf("structural request already exceeds semantic bound: %d", semantic.SemanticRequestSize(request))
	}
	payload := strings.Repeat("x", remaining)
	request.Operations[0].Fields[0].String = &payload
	if got := semantic.SemanticRequestSize(request); got != semantic.MaxSemanticRequestBytes {
		t.Fatalf("semantic size = %d, want %d", got, semantic.MaxSemanticRequestBytes)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	frames, err := EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	if got := frames[0].GetHeader().GetEncodedCommandBytes(); got > envelope.MaxMutationEncodedCommandBytes {
		t.Fatalf("encoded command = %d, cap = %d", got, envelope.MaxMutationEncodedCommandBytes)
	}
	for index, frame := range frames {
		if got := proto.Size(frame); got > envelope.MaxRequestWireBytes {
			t.Fatalf("frame %d = %d, cap = %d", index, got, envelope.MaxRequestWireBytes)
		}
	}

	oneOver := request
	oneOver.Operations = append([]semantic.Operation(nil), request.Operations...)
	oneOver.Operations[0] = request.Operations[0]
	oneOver.Operations[0].Fields = append([]semantic.FieldValue(nil), request.Operations[0].Fields...)
	oneOverPayload := payload + "x"
	oneOver.Operations[0].Fields[0].String = &oneOverPayload
	var failure *semantic.Failure
	if _, err := EncodeMutationFrames(ctx, oneOver); !errors.As(err, &failure) || failure.LimitKind != semantic.LimitSemanticRequest {
		t.Fatalf("one-over semantic bound = %#v", err)
	}

	tooManyFields := wireMutationRequest()
	for len(tooManyFields.Operations[0].Fields) <= semantic.MaxFieldsPerOperation {
		value := ""
		tooManyFields.Operations[0].Fields = append(tooManyFields.Operations[0].Fields, semantic.FieldValue{
			Name: fmt.Sprintf("X%03d", len(tooManyFields.Operations[0].Fields)), Kind: semantic.ValueKindString, String: &value,
		})
	}
	if _, err := EncodeMutationFrames(ctx, tooManyFields); !errors.As(err, &failure) || failure.LimitKind != semantic.LimitFieldCount {
		t.Fatalf("one-over field bound = %#v", err)
	}
}

func TestMutationFramesTypedRoundTripAndEnvelope(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	request := wireMutationRequest()
	large := strings.Repeat("x", 3*envelope.MaxMutationChunkDataBytes)
	request.Operations[0].Fields = append(request.Operations[0].Fields, semantic.FieldValue{Name: "Payload", Kind: semantic.ValueKindString, String: &large})
	frames, err := EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	if len(frames) < 4 {
		t.Fatalf("frames = %d, want header plus multiple chunks", len(frames))
	}
	decoder := new(MutationFrameDecoder)
	for i, frame := range frames {
		if got := proto.Size(frame); got > envelope.MaxRequestWireBytes {
			t.Fatalf("frame %d size = %d", i, got)
		}
		if err := decoder.Add(frame); err != nil {
			t.Fatalf("frame %d = %v", i, err)
		}
	}
	roundTrip, err := decoder.Finish(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(roundTrip, semantic.NormalizeRequest(request)) {
		t.Fatalf("round trip changed semantic request")
	}
}

func TestMutationFrameDecoderFailsClosed(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	frames, err := EncodeMutationFrames(ctx, wireMutationRequest())
	if err != nil {
		t.Fatal(err)
	}
	tests := []struct {
		name   string
		frames []*runtimev1.MutateRecordsRequest
	}{
		{name: "chunk before header", frames: frames[1:]},
		{name: "duplicate header", frames: append([]*runtimev1.MutateRecordsRequest{frames[0], frames[0]}, frames[1:]...)},
		{name: "missing chunk", frames: frames[:1]},
		{name: "nil frame", frames: []*runtimev1.MutateRecordsRequest{nil}},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			decoder := new(MutationFrameDecoder)
			for _, frame := range tc.frames {
				if decoder.Add(frame) != nil {
					break
				}
			}
			if _, err := decoder.Finish(ctx); err == nil {
				t.Fatal("malformed sequence accepted")
			}
		})
	}

	corrupt := cloneFrames(frames)
	corrupt[len(corrupt)-1].GetChunk().Data[0] ^= 1
	decoder := new(MutationFrameDecoder)
	for _, frame := range corrupt {
		if err := decoder.Add(frame); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := decoder.Finish(ctx); err == nil {
		t.Fatal("corrupt bytes accepted")
	}

	oversized := &runtimev1.MutateRecordsRequest{Frame: &runtimev1.MutateRecordsRequest_Chunk{Chunk: &runtimev1.MutationCommandChunk{Data: make([]byte, envelope.MaxMutationChunkDataBytes+1)}}}
	decoder = new(MutationFrameDecoder)
	if err := decoder.Add(frames[0]); err != nil {
		t.Fatal(err)
	}
	var failure *semantic.Failure
	if err := decoder.Add(oversized); !errors.As(err, &failure) || failure.LimitKind != semantic.LimitWireFrame {
		t.Fatalf("oversized chunk = %#v", err)
	}
}

func TestMutationMetadataOwnedInputExtensionsRoundTrip(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	request := wireMutationRequest()
	request.UseCase = semantic.MutationUseCaseCollection
	number, rawJSON := "1.2500e2", `{"z":2,"a":[0,false,"hidden"]}`
	request.Operations[0].Fields = append(request.Operations[0].Fields,
		semantic.FieldValue{Name: "Exact", Kind: semantic.ValueKindNumber, Number: &number},
		semantic.FieldValue{Name: "Payload", Kind: semantic.ValueKindJSON, JSON: &rawJSON},
	)
	frames, err := EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	decoder := new(MutationFrameDecoder)
	for _, frame := range frames {
		if err := decoder.Add(frame); err != nil {
			t.Fatal(err)
		}
	}
	round, err := decoder.Finish(ctx)
	if err != nil || !reflect.DeepEqual(round, semantic.NormalizeRequest(request)) {
		t.Fatalf("metadata-owned input round trip = %#v, %v", round, err)
	}
}

func TestMutationMapperRejectsUnknownSemanticMappingRevision(t *testing.T) {
	var failure *semantic.Failure
	if err := requireRecordMutationSemanticMappingRevision(struct{}{}); !errors.As(err, &failure) || failure.Category != semantic.FailureIncompatible {
		t.Fatalf("missing semantic mapping revision = %#v", err)
	}
	failure = nil
	if err := requireRecordMutationSemanticMappingRevision(mappingRevisionFixture(supportedRecordMutationSemanticMappingRevision + 1)); !errors.As(err, &failure) || failure.Category != semantic.FailureIncompatible {
		t.Fatalf("future semantic mapping revision = %#v", err)
	}
	failure = nil
	if err := requireRecordMutationSemanticMappingRevision(mappingRevisionFixture(supportedRecordMutationSemanticMappingRevision - 1)); !errors.As(err, &failure) || failure.Category != semantic.FailureIncompatible {
		t.Fatalf("stale semantic mapping revision = %#v", err)
	}
	if err := requireRecordMutationSemanticMappingRevision(mappingRevisionFixture(supportedRecordMutationSemanticMappingRevision)); err != nil {
		t.Fatalf("supported semantic mapping revision = %v", err)
	}
}

func TestMutationExternalIdentityRequiresReleasedCutover(t *testing.T) {
	if supportedRecordMutationSemanticMappingRevision != 7 {
		t.Fatalf("mutation mapper revision = %d, want 7", supportedRecordMutationSemanticMappingRevision)
	}
	if semantic.ProtocolMinor != 2 || semantic.SemanticMappingRevision != 7 {
		t.Fatalf("semantic release boundary = protocol 1.%d / mapping %d, want 1.2/7",
			semantic.ProtocolMinor, semantic.SemanticMappingRevision)
	}
}

func TestMutationExternalIdentityRoundTripsWithoutPayloadDuplication(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	value := "updated"
	request := semantic.Request{
		CommandID: "wire-external-upsert", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseCollection,
		Operations: []semantic.Operation{{
			Kind: semantic.OperationUpsert, ObjectID: "Entry",
			ExternalIdentity: &semantic.ExternalIdentity{FieldID: "ExternalKey", Value: "source:row:1"},
			Fields:           []semantic.FieldValue{{Name: "Name", Kind: semantic.ValueKindString, String: &value}},
		}},
	}
	frames, err := EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	decoder := new(MutationFrameDecoder)
	for _, frame := range frames {
		if err := decoder.Add(frame); err != nil {
			t.Fatal(err)
		}
	}
	round, err := decoder.Finish(ctx)
	if err != nil || !reflect.DeepEqual(round, request) {
		t.Fatalf("external identity round trip = %#v, %v", round, err)
	}
}

type mappingRevisionFixture uint32

func (revision mappingRevisionFixture) RecordMutationSemanticMappingRevision() uint32 {
	return uint32(revision)
}

func TestMutationDecoderRejectsUnknownWireUseCase(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	_, err := fromWireMutationCommand(ctx, &runtimev1.MutationCommand{UseCase: runtimev1.MutationUseCase(99)})
	var failure *semantic.Failure
	if !errors.As(err, &failure) || failure.Category != semantic.FailureIncompatible {
		t.Fatalf("unknown wire use case = %#v", err)
	}
}

func TestMutationPolicyDenialResultRoundTrips(t *testing.T) {
	request := semantic.Request{CommandID: "new-result-kind", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "WorkItem"}}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	response := semantic.Response{
		CommandID: request.CommandID, CommandDigest: digest, Receipts: mutationReceipts(), OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
		Results: []semantic.OperationResult{{Index: 0, Kind: semantic.OperationCreate, Errors: []semantic.OperationError{{Kind: semantic.ErrorPolicyDenied, AuthoredMessage: "Creation is closed.", Fields: []string{}}}}},
	}
	wireResponse, err := ToWireMutationResponse(request, response)
	if err != nil {
		t.Fatal(err)
	}
	round, err := FromWireMutationResponse(request, wireResponse)
	if err != nil || !reflect.DeepEqual(round, response) {
		t.Fatalf("policy denial round trip = %#v, %v", round, err)
	}
}

func TestMutationTransactionalPolicyOutcomesRoundTrip(t *testing.T) {
	created := true
	for name, test := range map[string]struct {
		mode    semantic.BatchMode
		results []semantic.OperationResult
	}{
		"independent": {
			mode: semantic.BatchIndependent,
			results: []semantic.OperationResult{
				{Index: 0, Kind: semantic.OperationCreate, RecordID: "work-created", Success: true, Created: &created, Errors: []semantic.OperationError{}},
				{Index: 1, Kind: semantic.OperationUpdate, RecordID: "work-1", Errors: []semantic.OperationError{{Kind: semantic.ErrorPolicyDenied, AuthoredMessage: "Updates are closed.", Fields: []string{"State"}}}},
			},
		},
		"atomic": {
			mode: semantic.BatchAtomic,
			results: []semantic.OperationResult{
				{Index: 0, Kind: semantic.OperationCreate, Errors: []semantic.OperationError{{Kind: semantic.ErrorAtomicRolledBack, Fields: []string{}}}},
				{Index: 1, Kind: semantic.OperationUpdate, RecordID: "work-1", Errors: []semantic.OperationError{{Kind: semantic.ErrorPolicyDenied, AuthoredMessage: "Updates are closed.", Fields: []string{"State"}}}},
			},
		},
	} {
		t.Run(name, func(t *testing.T) {
			request := semantic.Request{CommandID: "policy-recheck", BatchMode: test.mode, UseCase: semantic.MutationUseCaseCollection, Operations: []semantic.Operation{
				{Kind: semantic.OperationCreate, ObjectID: "WorkItem"},
				{Kind: semantic.OperationUpdate, ObjectID: "WorkItem", RecordID: "work-1"},
			}}
			digest, err := semantic.CommandDigest(request)
			if err != nil {
				t.Fatal(err)
			}
			response := semantic.Response{
				CommandID: request.CommandID, CommandDigest: digest, Results: test.results,
				Receipts: mutationReceipts(), OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
			}
			wireResponse, err := ToWireMutationResponse(request, response)
			if err != nil {
				t.Fatal(err)
			}
			round, err := FromWireMutationResponse(request, wireResponse)
			if err != nil || !reflect.DeepEqual(round, response) {
				t.Fatalf("transactional policy outcome round trip = %#v, %v", round, err)
			}
		})
	}
}

func TestMutationObjectNotFoundPreflightRoundTrips(t *testing.T) {
	request := semantic.Request{CommandID: "missing-object", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseCollection, Operations: []semantic.Operation{
		{Kind: semantic.OperationCreate, ObjectID: "KnownItem"},
		{Kind: semantic.OperationCreate, ObjectID: "UnknownItem"},
	}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	response := semantic.Response{
		CommandID: request.CommandID, CommandDigest: digest, Receipts: mutationReceipts(), OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
		Results: []semantic.OperationResult{
			{Index: 0, Kind: semantic.OperationCreate, Errors: []semantic.OperationError{{Kind: semantic.ErrorProcessingHalted, Fields: []string{}}}},
			{Index: 1, Kind: semantic.OperationCreate, Errors: []semantic.OperationError{{Kind: semantic.ErrorObjectNotFound, Fields: []string{}}}},
		},
	}
	wireResponse, err := ToWireMutationResponse(request, response)
	if err != nil {
		t.Fatal(err)
	}
	round, err := FromWireMutationResponse(request, wireResponse)
	if err != nil || !reflect.DeepEqual(round, response) {
		t.Fatalf("object-not-found preflight round trip = %#v, %v", round, err)
	}
}

func TestMutationCreateFromSourceGuardRoundTrips(t *testing.T) {
	request := semantic.Request{CommandID: "source-guard", BatchMode: semantic.BatchIndependent, UseCase: semantic.MutationUseCaseSingle, Operations: []semantic.Operation{{Kind: semantic.OperationCreate, ObjectID: "QuoteProduct"}}}
	digest, err := semantic.CommandDigest(request)
	if err != nil {
		t.Fatal(err)
	}
	response := semantic.Response{
		CommandID: request.CommandID, CommandDigest: digest, Receipts: mutationReceipts(), OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
		Results: []semantic.OperationResult{{Index: 0, Kind: semantic.OperationCreate, Errors: []semantic.OperationError{{
			Kind: semantic.ErrorInvalidOperation, Fields: []string{}, CreateFromSourceGuard: &semantic.CreateFromSourceGuard{
				ObjectID: "QuoteProduct", ActionIDs: []string{"addProduct", "createFromQuote"},
			},
		}}}},
	}
	wireResponse, err := ToWireMutationResponse(request, response)
	if err != nil {
		t.Fatal(err)
	}
	round, err := FromWireMutationResponse(request, wireResponse)
	if err != nil || !reflect.DeepEqual(round, response) {
		t.Fatalf("source guard round trip = %#v, %v", round, err)
	}
}

func TestMutationResponseAndStatusRoundTrip(t *testing.T) {
	request := wireMutationRequest()
	shape, err := semantic.CommandShapeFor(request)
	if err != nil {
		t.Fatal(err)
	}
	created := true
	response := semantic.Response{
		CommandID: request.CommandID, CommandDigest: shape.CommandDigest,
		Results: []semantic.OperationResult{
			{Index: 0, Kind: semantic.OperationCreate, RecordID: "record-1", Success: true, Created: &created, Errors: []semantic.OperationError{}},
			{Index: 1, Kind: semantic.OperationDelete, RecordID: "hidden", Errors: []semantic.OperationError{{Kind: semantic.ErrorMissingOrDenied, Fields: []string{}}}},
		},
		Receipts: mutationReceipts(), OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
	}
	wireResponse, err := ToWireMutationResponse(request, response)
	if err != nil {
		t.Fatal(err)
	}
	roundResponse, err := FromWireMutationResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(roundResponse, response) {
		t.Fatalf("response round trip = %#v", roundResponse)
	}

	statusRequest, err := semantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	for _, candidate := range []semantic.StatusResponse{
		{CommandID: request.CommandID, CommandDigest: shape.CommandDigest, State: semantic.CommandAccepted},
		{CommandID: request.CommandID, CommandDigest: shape.CommandDigest, State: semantic.CommandExecuting},
		{CommandID: request.CommandID, CommandDigest: shape.CommandDigest, State: semantic.CommandCompleted, Results: response.Results, Receipts: ptrReceipts(response.Receipts), OutcomeExpiresAtUTC: response.OutcomeExpiresAtUTC},
	} {
		wireStatus, err := ToWireMutationStatusResponse(shape, statusRequest, candidate)
		if err != nil {
			t.Fatal(err)
		}
		roundStatus, err := FromWireMutationStatusResponse(shape, statusRequest, wireStatus)
		if err != nil || !reflect.DeepEqual(roundStatus, candidate) {
			t.Fatalf("status round trip = %#v, %v", roundStatus, err)
		}
	}
}

func TestMutationUnknownOptionalFieldsAndSafeStatus(t *testing.T) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	frames, err := EncodeMutationFrames(ctx, wireMutationRequest())
	if err != nil {
		t.Fatal(err)
	}
	decoder := new(MutationFrameDecoder)
	for _, frame := range frames {
		if err := decoder.Add(frame); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := decoder.Finish(ctx); err != nil {
		t.Fatal(err)
	}

	semanticStatus, err := semantic.StatusRequestFor(wireMutationRequest())
	if err != nil {
		t.Fatal(err)
	}
	statusRequest, err := ToWireMutationStatusRequest(ctx, semanticStatus)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := proto.Marshal(statusRequest)
	if err != nil {
		t.Fatal(err)
	}
	raw = protowire.AppendTag(raw, 100, protowire.BytesType)
	raw = protowire.AppendString(raw, "future-optional")
	var decoded runtimev1.GetMutationStatusRequest
	if err := proto.Unmarshal(raw, &decoded); err != nil {
		t.Fatal(err)
	}
	if _, err := FromWireMutationStatusRequest(ctx, &decoded); err != nil {
		t.Fatal(err)
	}

	for _, tc := range []struct {
		failure *semantic.Failure
		code    codes.Code
	}{
		{&semantic.Failure{Category: semantic.FailureInvalidRequest, InvalidComponent: semantic.InvalidComponentCommandID}, codes.InvalidArgument},
		{&semantic.Failure{Category: semantic.FailureMissing}, codes.NotFound},
		{&semantic.Failure{Category: semantic.FailureResourceExhausted, LimitKind: semantic.LimitOperationCount, Maximum: semantic.MaxOperations}, codes.ResourceExhausted},
		{&semantic.Failure{Category: semantic.FailureUnavailable}, codes.Unavailable},
	} {
		toStatus := ToRecordMutationStatusError
		fromStatus := FromRecordMutationStatusError
		if tc.failure.Category == semantic.FailureMissing {
			toStatus = ToRecordMutationReconciliationStatusError
			fromStatus = FromRecordMutationReconciliationStatusError
		}
		wireErr := toStatus(tc.failure)
		if status.Code(wireErr) != tc.code {
			t.Fatalf("%s status = %s", tc.failure.Category, status.Code(wireErr))
		}
		var round *semantic.Failure
		if err := fromStatus(wireErr); !errors.As(err, &round) || round.Category != tc.failure.Category {
			t.Fatalf("%s round = %#v", tc.failure.Category, err)
		}
	}
	if got := status.Code(ToRecordMutationStatusError(&semantic.Failure{Category: semantic.FailureMissing})); got != codes.Internal {
		t.Fatalf("mutate accepted top-level MISSING: %s", got)
	}
}

func wireMutationRequest() semantic.Request {
	falseValue := false
	zero := int64(0)
	empty := ""
	return semantic.Request{CommandID: "cmd-wire-001", BatchMode: semantic.BatchIndependent, Operations: []semantic.Operation{
		{Kind: semantic.OperationCreate, ObjectID: "WorkItem", Fields: []semantic.FieldValue{
			{Name: "Zero", Kind: semantic.ValueKindInteger, Integer: &zero},
			{Name: "Empty", Kind: semantic.ValueKindString, String: &empty},
			{Name: "False", Kind: semantic.ValueKindBoolean, Boolean: &falseValue},
			{Name: "Null", Kind: semantic.ValueKindNull, Null: true},
		}},
		{Kind: semantic.OperationDelete, RecordID: "hidden"},
	}}
}

func mutationReceipts() semantic.Receipts {
	return semantic.Receipts{ProtocolVersion: "1.0", CatalogRevision: 41, SchemaGeneration: 7, EngineGeneration: 9, SecurityEpoch: 13, CapabilityVersion: semantic.CapabilityName}
}

func ptrReceipts(value semantic.Receipts) *semantic.Receipts { return &value }

func cloneFrames(frames []*runtimev1.MutateRecordsRequest) []*runtimev1.MutateRecordsRequest {
	out := make([]*runtimev1.MutateRecordsRequest, len(frames))
	for i, frame := range frames {
		out[i] = proto.Clone(frame).(*runtimev1.MutateRecordsRequest)
	}
	return out
}