package runtimeapi

import (
	"context"
	"errors"
	"io"
	"net"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/apierr"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	identitysemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/identityv1"
	layoutauthoringsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/layoutauthoringv2"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	attachmentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordattachmentsv1"
	defaultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recorddefaultsv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	recordviewsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordviewv1"
	recyclemutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinmutationv1"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	relatedsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/relatedcountsv1"
	reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

func TestRuntimeAPIProcessGeneratedClientCrossesAuthoritativeAdapter(t *testing.T) {
	falseValue, zero, empty := false, int64(0), ""
	reader := &readerStub{record: &records.SelectedRootRecord{RecordID: "ISS-1", Fields: []records.SelectedRootField{
		{Name: "Archived", Kind: records.SelectedRootBoolean, Boolean: &falseValue},
		{Name: "EmptySummary", Kind: records.SelectedRootString, String: &empty},
		{Name: "Optional", Kind: records.SelectedRootString, Null: true},
		{Name: "RetryCount", Kind: records.SelectedRootInteger, Integer: &zero},
	}}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(func(context.Context) (semantic.Receipts, error) {
			return semantic.Receipts{
				ProtocolVersion: "1.0", CatalogRevision: 7, SchemaGeneration: 8,
				EngineGeneration: 9, SecurityEpoch: 10, CapabilityVersion: semantic.CapabilityName,
			}, nil
		}),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusOK)

	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	client := runtimev1.NewRecordsServiceClient(connection)
	ctx, cancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer cancel()
	wireResponse, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-1",
		SelectedRootFields: []string{"RetryCount", "Optional", "Archived", "EmptySummary"},
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if response.RecordID != "ISS-1" || response.Receipts.CatalogRevision != 7 {
		t.Fatalf("response = %#v", response)
	}
	names := make([]string, len(response.Fields))
	for i := range response.Fields {
		names[i] = response.Fields[i].Name
	}
	if !reflect.DeepEqual(names, []string{"Archived", "EmptySummary", "Optional", "RetryCount"}) {
		t.Fatalf("fields = %#v", response.Fields)
	}
	if reader.calls != 1 || reader.userID != testPrincipalID || !reader.requestCachesInstalled {
		t.Fatalf("adapter calls=%d user=%q requestCachesInstalled=%v", reader.calls, reader.userID, reader.requestCachesInstalled)
	}
}

func TestRuntimeAPIProcessPublishesPermissionStateCapabilitiesAndReadiness(t *testing.T) {
	var permissionStateReady atomic.Bool
	permissionStateReady.Store(true)
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		PermissionStateCapabilities: []string{
			"permission_state.v1",
			"permission_state.lineage=01010101010101010101010101010101",
		},
		PermissionStateReady: permissionStateReady.Load,
		Receipts: ReceiptSourceFunc(func(context.Context) (semantic.Receipts, error) {
			return semantic.Receipts{
				ProtocolVersion: "1.0", CatalogRevision: 7, SchemaGeneration: 8,
				EngineGeneration: 9, SecurityEpoch: 10, CapabilityVersion: semantic.CapabilityName,
			}, nil
		}),
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusOK)
	body := readRuntimeMetrics(t, "http://"+process.AdminAddress()+processhttp.CapabilitiesPath)
	for _, fact := range []string{
		`"permission_state.lineage=01010101010101010101010101010101"`,
		`"permission_state.v1"`,
	} {
		if !strings.Contains(body, fact) {
			t.Fatalf("capabilities missing %q: %s", fact, body)
		}
	}
	permissionStateReady.Store(false)
	assertRuntimeProbe(
		t,
		"http://"+process.AdminAddress()+processhttp.ReadyPath,
		http.StatusServiceUnavailable,
	)
}

func TestRuntimeAPIProcessRequiresOnlyEnabledPlatformResourceClasses(t *testing.T) {
	base := ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		ReportEngine: &reportDefinitionTestEngine{},
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			reportsemantic.CapabilityName: time.Second,
		},
	}
	if _, err := Assemble(base, &readerStub{}); err == nil ||
		!strings.Contains(err.Error(), "requires an explicit resource-class admission limit") {
		t.Fatalf("missing class limit error = %v", err)
	}

	unused := base
	unused.PlatformResourceClassAdmissionLimits = map[runtimegrpc.ResourceClass]int{
		runtimegrpc.ResourceClassHeavyWork:  1,
		runtimegrpc.ResourceClassFileStream: 1,
	}
	if _, err := Assemble(unused, &readerStub{}); err == nil ||
		!strings.Contains(err.Error(), "cannot configure an unused Platform class") {
		t.Fatalf("unused class limit error = %v", err)
	}

	exact := base
	exact.PlatformResourceClassAdmissionLimits = map[runtimegrpc.ResourceClass]int{
		runtimegrpc.ResourceClassHeavyWork: 1,
	}
	process, err := Assemble(exact, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if process == nil {
		t.Fatal("exact enabled Platform resource class did not assemble")
	}
}

func TestRuntimeAPIProcessRecordAccessCrossesAuthoritativeAdapter(t *testing.T) {
	object := "WorkItem"
	reader := &readerStub{access: map[string]contract.RecordAccess{
		"W-2": {Read: true, Locked: true},
		"W-1": {Read: true, Update: true, Delete: true},
	}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecordAccessMaxAdmissionLease: time.Second,
		WorkloadAuthority:             testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(func(context.Context) (semantic.Receipts, error) {
			return semantic.Receipts{CatalogRevision: 7, SchemaGeneration: 8, EngineGeneration: 9, SecurityEpoch: 10}, nil
		}),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(recordAccessProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireResponse, err := runtimev1.NewRecordsServiceClient(connection).EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{
		ObjectId: &object, RecordIds: []string{"W-2", "W-1", "W-2"},
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireEvaluateRecordAccessResponse(accesssemantic.Request{ObjectID: &object, RecordIDs: []string{"W-2", "W-1", "W-2"}}, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if reader.accessCalls != 1 || reader.userID != testPrincipalID || reader.accessObject == nil || *reader.accessObject != object ||
		!reflect.DeepEqual(reader.accessIDs, []string{"W-2", "W-1", "W-2"}) {
		t.Fatalf("adapter calls=%d user=%q object=%v ids=%v", reader.accessCalls, reader.userID, reader.accessObject, reader.accessIDs)
	}
	if len(response.Results) != 2 || response.Results[0].RecordID != "W-1" || response.Results[1].RecordID != "W-2" ||
		!response.Results[0].Access.Delete || !response.Results[1].Access.Locked || response.Receipts.CapabilityVersion != accesssemantic.CapabilityName || response.Receipts.CatalogRevision != 7 {
		t.Fatalf("response = %#v", response)
	}
	metrics := readRuntimeMetrics(t, "http://"+process.AdminAddress()+"/metrics")
	for _, fact := range []string{
		`nexuscrm_runtime_records_repository_calls_total{operation="evaluate_record_access"} 1`,
		`nexuscrm_runtime_records_sql_statements_total{operation="evaluate_record_access"} 0`,
	} {
		if !strings.Contains(metrics, fact) {
			t.Fatalf("Runtime records metrics missing %q:\n%s", fact, metrics)
		}
	}
}

func TestRuntimeAPIProcessRecordViewUsesBoundCapabilityAndLeavesHistoricResolverUnimplemented(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.sfcrm.ui.record-view-read.v1",
		recordviewsemantic.CapabilityName, catalogsnapshot.ProjectionRecordViewV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engine := &recordViewServerEngineStub{result: &records.RuntimeRecordViewResult{
		RecordID: "widget-1", Root: models.SObject{constants.FieldID: "widget-1"},
	}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecordViewMaxAdmissionLease: time.Second, RecordView: engine,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t), ServingAdmission: fixture.admission,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	request := recordviewsemantic.ReadRequest{
		ObjectID: "Widget", Selector: recordviewsemantic.Selector{Kind: recordviewsemantic.SelectorRecordID, RecordID: "widget-1"}, Projection: recordviewsemantic.ProjectionRoot,
	}
	ctx, cancel := context.WithTimeout(recordViewProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireReadRecordViewRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRecordViewReadServiceClient(connection).ReadRecordView(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireReadRecordViewResponse(request, wireResponse)
	if err != nil || response.RecordID != "widget-1" || response.Receipts.CapabilityVersion != recordviewsemantic.CapabilityName || engine.calls != 1 {
		t.Fatalf("response=%#v err=%v calls=%d", response, err, engine.calls)
	}
	_, historicErr := runtimev1.NewRecordViewReadServiceClient(connection).ResolveRecordAttachmentSources(ctx, &runtimev1.ResolveRecordAttachmentSourcesRequest{})
	if status.Code(historicErr) != codes.Unimplemented || engine.calls != 1 {
		t.Fatalf("historic attachment error=%v calls=%d", historicErr, engine.calls)
	}
}

func TestRuntimeAPIProcessAttachmentSourceRequiresItsOwnDelegatedCapability(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.sfcrm.ui.record-view-attachments.v1",
		attachmentsemantic.CapabilityName, catalogsnapshot.ProjectionRecordViewV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engine := &attachmentSourceServerEngineStub{results: []attachmentsemantic.SourceResult{{
		Index: 0, Outcome: attachmentsemantic.SourceResolved, FileID: "file-1", FileName: "spec.pdf",
	}}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		AttachmentSourceMaxAdmissionLease: time.Second, AttachmentSources: engine,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t), ServingAdmission: fixture.admission,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	request := attachmentsemantic.ResolveRequest{
		ParentObjectID: "Widget", ParentRecordID: "widget-1", ResolverID: "attachment",
		Sources: []attachmentsemantic.Source{{SourceFieldID: "DocumentLink", SourceValue: "ABCDEF123456", SourceValueOffset: 3}},
	}
	ctx, cancel := context.WithTimeout(attachmentSourceProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireResolveAttachmentSourcesRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRecordAttachmentSourceResolutionServiceClient(connection).ResolveAttachmentSources(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireResolveAttachmentSourcesResponse(request, wireResponse)
	if err != nil || len(response.Results) != 1 || response.Results[0].Outcome != attachmentsemantic.SourceResolved ||
		response.Receipts.CapabilityVersion != attachmentsemantic.CapabilityName || engine.calls != 1 {
		t.Fatalf("response=%#v err=%v calls=%d", response, err, engine.calls)
	}

	wrongCtx := metadata.AppendToOutgoingContext(context.Background(),
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(attachmentsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(attachmentsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, attachmentsemantic.CapabilityName,
	)
	wrongCtx = withTestDelegation(wrongCtx, mustSignTestDelegation(testClaimsFor(time.Now(), recordviewsemantic.CapabilityName)))
	wrongCtx, wrongCancel := context.WithTimeout(wrongCtx, time.Second)
	defer wrongCancel()
	_, wrongErr := runtimev1.NewRecordAttachmentSourceResolutionServiceClient(connection).ResolveAttachmentSources(wrongCtx, wireRequest)
	if status.Code(wrongErr) != codes.PermissionDenied || engine.calls != 1 {
		t.Fatalf("wrong delegation error=%v calls=%d", wrongErr, engine.calls)
	}
}

func TestRuntimeAPIProcessRecordMutationUsesOneTrustedStreamAndBoundReceipts(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.sfcrm.records.mutate.v1",
		mutationsemantic.CapabilityName, catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	executor := &recordMutationExecutorStub{}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission: fixture.admission,
		RecordMutation:   executor, RecordMutationAttemptCap: time.Second,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)

	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	empty := ""
	request := mutationsemantic.Request{
		CommandID: "command-1", BatchMode: mutationsemantic.BatchAtomic,
		Operations: []mutationsemantic.Operation{{
			Kind: mutationsemantic.OperationUpdate, ObjectID: "Widget", RecordID: "widget-1",
			Fields: []mutationsemantic.FieldValue{{Name: "Name", Kind: mutationsemantic.ValueKindString, String: &empty}},
		}},
	}
	ctx, cancel := context.WithTimeout(recordMutationProtocolContext(context.Background()), time.Second)
	defer cancel()
	frames, err := mapping.EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	stream, err := runtimev1.NewRecordMutationsServiceClient(connection).MutateRecords(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, frame := range frames {
		if err := stream.Send(frame); err != nil {
			t.Fatal(err)
		}
	}
	wireResponse, err := stream.CloseAndRecv()
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireMutationResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if executor.calls != 1 || executor.userID != testPrincipalID || len(executor.scopeDigest) != 64 ||
		executor.receipts.CapabilityVersion != mutationsemantic.CapabilityName || executor.receipts.CatalogRevision != 7 ||
		!executor.requestCachesInstalled || len(response.Results) != 1 || !response.Results[0].Success ||
		response.Results[0].RecordID != "widget-1" {
		t.Fatalf("executor=%+v response=%+v", executor, response)
	}
	statusRequest, err := mutationsemantic.StatusRequestFor(request)
	if err != nil {
		t.Fatal(err)
	}
	wireStatusRequest, err := mapping.ToWireMutationStatusRequest(ctx, statusRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireStatus, err := runtimev1.NewRecordMutationsServiceClient(connection).GetMutationStatus(ctx, wireStatusRequest)
	if err != nil {
		t.Fatal(err)
	}
	shape, err := mutationsemantic.CommandShapeFor(request)
	if err != nil {
		t.Fatal(err)
	}
	statusResponse, err := mapping.FromWireMutationStatusResponse(shape, statusRequest, wireStatus)
	if err != nil {
		t.Fatal(err)
	}
	if executor.statusCalls != 1 || executor.statusScope != executor.scopeDigest ||
		statusResponse.State != mutationsemantic.CommandCompleted || statusResponse.CommandDigest != response.CommandDigest {
		t.Fatalf("status executor=%+v response=%+v", executor, statusResponse)
	}

	// Stream admission must establish the same trusted capability boundary as a
	// unary Runtime call. Neither a mismatched delegation nor a missing one may
	// reach the durable command executor merely because frames are streamed.
	wrongCapabilityCtx := metadata.AppendToOutgoingContext(context.Background(),
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)
	wrongCapabilityCtx = withTestDelegation(wrongCapabilityCtx, mustSignTestDelegation(testClaimsFor(time.Now(), semantic.CapabilityName)))
	wrongCapabilityCtx, wrongCapabilityCancel := context.WithTimeout(wrongCapabilityCtx, time.Second)
	defer wrongCapabilityCancel()
	wrongCapabilityStream, err := runtimev1.NewRecordMutationsServiceClient(connection).MutateRecords(wrongCapabilityCtx)
	if err != nil {
		t.Fatal(err)
	}
	_, err = wrongCapabilityStream.CloseAndRecv()
	if status.Code(err) != codes.PermissionDenied || executor.calls != 1 {
		t.Fatalf("mismatched delegation error=%v executor calls=%d, want permission denied before execution", err, executor.calls)
	}

	missingDelegationCtx := metadata.AppendToOutgoingContext(context.Background(),
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)
	missingDelegationCtx, missingDelegationCancel := context.WithTimeout(missingDelegationCtx, time.Second)
	defer missingDelegationCancel()
	missingDelegationStream, err := runtimev1.NewRecordMutationsServiceClient(connection).MutateRecords(missingDelegationCtx)
	if err != nil {
		t.Fatal(err)
	}
	_, err = missingDelegationStream.CloseAndRecv()
	if status.Code(err) != codes.Unauthenticated || executor.calls != 1 {
		t.Fatalf("missing delegation error=%v executor calls=%d, want unauthenticated before execution", err, executor.calls)
	}

	staleRequest := request
	staleRequest.CommandID = "command-stale"
	staleRequest.MinCatalogRevision = uint64Pointer(8)
	staleFrames, err := mapping.EncodeMutationFrames(ctx, staleRequest)
	if err != nil {
		t.Fatal(err)
	}
	staleStream, err := runtimev1.NewRecordMutationsServiceClient(connection).MutateRecords(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, frame := range staleFrames {
		if err := staleStream.Send(frame); err != nil {
			t.Fatal(err)
		}
	}
	_, err = staleStream.CloseAndRecv()
	var failure *mutationsemantic.Failure
	if !errors.As(mapping.FromRecordMutationStatusError(err), &failure) ||
		failure.Category != mutationsemantic.FailureStaleState ||
		failure.RequiredRevision != 8 || failure.ServedRevision != 7 || executor.calls != 1 {
		t.Fatalf("stale result = %#v (%v), executor calls = %d", failure, err, executor.calls)
	}
}

func TestRuntimeAPIProcessRelatedCountCrossesAuthoritativeAdapters(t *testing.T) {
	reader := &readerStub{access: map[string]contract.RecordAccess{"P-1": {Read: true}}}
	query := &relatedCountQueryStub{results: map[string]*models.QueryResult{"WorkItem": {TotalSize: 3}}}
	fixture := newServingFixtureForCapability(t, 7, "io.nexuscrm.sfcrm.related-list-counts.v1", relatedsemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1)
	relatedPrepared, err := fixture.state.prepare(t.Context(), fixture.selection, newServingArtifactSet(fixture.state.store))
	if err != nil {
		t.Fatal(err)
	}
	accessPrepared := *relatedPrepared
	accessPrepared.selection.RuntimeCapabilityID = accesssemantic.CapabilityName
	accessPrepared.view = &catalogsnapshot.RuntimeMetadataView{}
	fixture.state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		relatedsemantic.CapabilityName: relatedPrepared,
		accesssemantic.CapabilityName:  &accessPrepared,
	}})
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RelatedCountMaxAdmissionLease: time.Second,
		QueryEngine:                   query,
		WorkloadAuthority:             testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission: fixture.admission,
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()

	request := relatedsemantic.Request{Counts: []relatedsemantic.CountSpec{{
		ParentRecordID: "P-1", ChildObjectID: "WorkItem", RelationshipField: "WorkspaceId",
	}}}
	ctx, cancel := context.WithTimeout(relatedCountProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireCountRelatedRecordsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRelatedRecordCountServiceClient(connection).CountRelatedRecords(ctx, wireRequest)
	if err != nil {
		grpcStatus := status.Convert(err)
		t.Fatalf("CountRelatedRecords = code %s message %q details %#v semantic %#v read_access_calls=%d full_access_calls=%d serving_ready=%v", grpcStatus.Code(), grpcStatus.Message(), grpcStatus.Details(), mapping.FromRelatedCountStatusError(err), reader.readAccessCalls, reader.accessCalls, fixture.admission.Ready())
	}
	response, err := mapping.FromWireCountRelatedRecordsResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Results) != 1 || response.Results[0].Outcome != relatedsemantic.CountOutcomeCounted || response.Results[0].Count != 3 ||
		response.Receipts.CapabilityVersion != relatedsemantic.CapabilityName || response.Receipts.CatalogRevision != 7 {
		t.Fatalf("response = %#v", response)
	}
	if reader.readAccessCalls != 1 || reader.accessCalls != 0 || reader.accessObject != nil || !reflect.DeepEqual(reader.accessIDs, []string{"P-1"}) {
		t.Fatalf("read access calls=%d full access calls=%d object=%v ids=%v", reader.readAccessCalls, reader.accessCalls, reader.accessObject, reader.accessIDs)
	}
	query.mu.Lock()
	defer query.mu.Unlock()
	if len(query.calls) != 1 || query.calls[0].ObjectAPIName != "WorkItem" || query.calls[0].Filter.Field != "WorkspaceId" || query.calls[0].Filter.Value != "P-1" {
		t.Fatalf("query calls = %#v", query.calls)
	}
}

func TestRuntimeAPIProcessCreateDefaultPreviewCrossesAuthoritativeAdapter(t *testing.T) {
	ownerID, ownerName, next := testPrincipalID, "Runtime User", int64(901)
	preview := &createDefaultPreviewEngineStub{fields: []defaultsemantic.Field{
		{
			Name: "AssignedPrincipalId", Kind: defaultsemantic.ValueReference, Reference: &ownerID,
			DisplayValue: &ownerName, Provenance: defaultsemantic.ProvenanceCurrentPrincipal,
		},
		{
			Name: "Sequence", Kind: defaultsemantic.ValueInteger, Integer: &next,
			Provenance: defaultsemantic.ProvenanceSiblingMaxPlusOne,
		},
	}}
	reader := &createDefaultPreviewProcessReader{readerStub: &readerStub{}, preview: preview}
	fixture := newServingFixtureForCapability(
		t, 7, "io.nexuscrm.fixture.ui.record-defaults.v1", defaultsemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit:                        2,
		ServingEnabled:                        true,
		CreateDefaultPreviewMaxAdmissionLease: time.Second,
		WorkloadAuthority:                     testWorkloadAuthority,
		SecurityAdmission:                     testSecurityAdmission(t),
		ServingAdmission:                      fixture.admission,
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient(
		"dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()

	parentField, parentID, minimum := "WorkspaceId", "workspace-1", uint64(7)
	request := defaultsemantic.Request{
		ObjectID: "Widget", ParentFieldID: &parentField, ParentRecordID: &parentID,
		MinCatalogRevision: &minimum,
	}
	ctx, cancel := context.WithTimeout(createDefaultPreviewProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWirePreviewCreateDefaultsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRecordCreateDefaultPreviewServiceClient(connection).
		PreviewCreateDefaults(ctx, wireRequest)
	if err != nil {
		grpcStatus := status.Convert(err)
		t.Fatalf(
			"PreviewCreateDefaults = code %s message %q details %#v semantic %#v serving_ready=%v request=%#v user=%#v",
			grpcStatus.Code(), grpcStatus.Message(), grpcStatus.Details(),
			mapping.FromCreateDefaultPreviewStatusError(err), fixture.admission.Ready(), preview.request, preview.user,
		)
	}
	response, err := mapping.FromWirePreviewCreateDefaultsResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if preview.user == nil || preview.user.ID != testPrincipalID ||
		!reflect.DeepEqual(preview.request, request) {
		t.Fatalf("preview request=%#v user=%#v", preview.request, preview.user)
	}
	if response.Receipts.CapabilityVersion != defaultsemantic.CapabilityName ||
		response.Receipts.CatalogRevision != 7 || !reflect.DeepEqual(response.Fields, preview.fields) {
		t.Fatalf("response=%#v", response)
	}
}

func TestRuntimeAPIProcessRecycleBinCrossesAuthoritativeAdapter(t *testing.T) {
	reader := &readerStub{recycleBin: &contract.RecycleBinListResult{
		Items: []contract.RecycleBinItem{{
			ID: "row-1", RecordID: "row-1", RecordName: "Visible", ObjectAPIName: "WorkItem",
			DeletedBy: "Test User", DeletedDate: "2026-07-22T04:00:00Z",
		}},
		FailedObjects: []contract.RecycleBinFailure{{
			ObjectAPIName: "RestrictedItem", ErrorCode: string(contract.RecycleBinFailureCodeForbidden), Message: "record read is forbidden",
		}},
	}}
	fixture := newServingFixtureForCapability(t, 7, "io.nexuscrm.sfcrm.recycle-bin-read.v1", recyclesemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecycleBinMaxAdmissionLease: time.Second,
		RecycleBinLister:            reader,
		WorkloadAuthority:           testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission: fixture.admission,
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()

	ctx, cancel := context.WithTimeout(recycleBinProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireResponse, err := runtimev1.NewRecycleBinReadServiceClient(connection).ListRecycleBin(ctx, &runtimev1.ListRecycleBinRequest{
		Scope: runtimev1.RecycleBinScope_RECYCLE_BIN_SCOPE_ALL,
	})
	if err != nil {
		t.Fatalf("ListRecycleBin = %v (%#v)", err, mapping.FromRecycleBinStatusError(err))
	}
	response, err := mapping.FromWireListRecycleBinResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if reader.recycleCalls != 1 || reader.recycleScope != contract.RecycleBinScopeAll || reader.userID != testPrincipalID {
		t.Fatalf("adapter calls=%d scope=%q user=%q", reader.recycleCalls, reader.recycleScope, reader.userID)
	}
	if len(response.Items) != 1 || response.Items[0].RecordID != "row-1" || len(response.FailedObjects) != 1 ||
		response.FailedObjects[0].Code != recyclesemantic.ObjectFailureForbidden ||
		response.Receipts.CapabilityVersion != recyclesemantic.CapabilityName || response.Receipts.CatalogRevision != 7 {
		t.Fatalf("response = %#v", response)
	}
	metrics := readRuntimeMetrics(t, "http://"+process.AdminAddress()+"/metrics")
	if !strings.Contains(metrics, `nexuscrm_runtime_records_repository_calls_total{operation="list_recycle_bin"} 1`) {
		t.Fatalf("Runtime recycle-bin metrics missing:\n%s", metrics)
	}
}

func TestRuntimeAPIProcessRecycleBinMutationsCrossAuthoritativeAdapter(t *testing.T) {
	reader := &readerStub{}
	fixture := newServingFixtureForCapability(t, 7, "io.nexuscrm.sfcrm.recycle-bin-mutation.v1", recyclemutationsemantic.CapabilityName, catalogsnapshot.ProjectionRecordMutationV1)
	recyclePrepared, err := fixture.state.prepare(t.Context(), fixture.selection, newServingArtifactSet(fixture.state.store))
	if err != nil {
		t.Fatal(err)
	}
	// This test isolates the transport adapter, so install the already-proven
	// ordinary Mutation lifecycle sibling directly. ReconcileAll dependency and
	// exact-view behavior have dedicated admission coverage.
	mutationPrepared := *recyclePrepared
	mutationPrepared.selection.RuntimeCapabilityID = mutationsemantic.CapabilityName
	_, mutationArtifact := servingArtifacts(t, 7, mutationsemantic.CapabilityName, catalogsnapshot.ProjectionRecordMutationV1)
	mutationPrepared.view, err = catalogsnapshot.OpenRuntimeMetadataView(mutationArtifact)
	if err != nil {
		t.Fatal(err)
	}
	fixture.state.prepared.Store(&preparedServingSet{byCapability: map[string]*preparedServingState{
		recyclemutationsemantic.CapabilityName: recyclePrepared,
		mutationsemantic.CapabilityName:        &mutationPrepared,
	}})
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecycleBinMutationMaxAdmissionLease: time.Second,
		RecycleBinMutator:                   reader,
		WorkloadAuthority:                   testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission: fixture.admission,
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()

	ctx, cancel := context.WithTimeout(recycleBinMutationProtocolContext(context.Background()), time.Second)
	defer cancel()
	client := runtimev1.NewRecycleBinMutationServiceClient(connection)
	restoreRequest := recyclemutationsemantic.Request{RecordID: "row-1"}
	wireRestore, err := mapping.ToWireRestoreRecycleBinRequest(ctx, restoreRequest)
	if err != nil {
		t.Fatal(err)
	}
	restoreResponse, err := client.RestoreRecycleBin(ctx, wireRestore)
	if err != nil {
		t.Fatalf("RestoreRecycleBin = %v (%#v), adapter calls=%d", err, mapping.FromRecycleBinMutationStatusError(err), reader.recycleRestoreCalls)
	}
	if _, err := mapping.FromWireRestoreRecycleBinResponse(restoreRequest, restoreResponse); err != nil {
		t.Fatal(err)
	}
	purgeRequest := recyclemutationsemantic.Request{RecordID: "row-2"}
	wirePurge, err := mapping.ToWirePurgeRecycleBinRequest(ctx, purgeRequest)
	if err != nil {
		t.Fatal(err)
	}
	purgeResponse, err := client.PurgeRecycleBin(ctx, wirePurge)
	if err != nil {
		t.Fatalf("PurgeRecycleBin = %v (%#v)", err, mapping.FromRecycleBinMutationStatusError(err))
	}
	if _, err := mapping.FromWirePurgeRecycleBinResponse(purgeRequest, purgeResponse); err != nil {
		t.Fatal(err)
	}
	if reader.recycleRestoreCalls != 1 || reader.recyclePurgeCalls != 1 || reader.recycleRecordID != "row-2" || reader.userID != testPrincipalID {
		t.Fatalf("adapter calls=%d/%d record=%q user=%q", reader.recycleRestoreCalls, reader.recyclePurgeCalls, reader.recycleRecordID, reader.userID)
	}
	readOnlyCtx, cancelReadOnly := context.WithTimeout(context.Background(), time.Second)
	defer cancelReadOnly()
	readOnlyCtx = metadata.AppendToOutgoingContext(readOnlyCtx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(recyclemutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(recyclemutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, recyclemutationsemantic.CapabilityName,
	)
	readOnlyCtx = withTestDelegation(readOnlyCtx, mustSignTestDelegation(testClaimsFor(time.Now(), recyclesemantic.CapabilityName)))
	_, denied := client.RestoreRecycleBin(readOnlyCtx, wireRestore)
	var deniedFailure *recyclemutationsemantic.Failure
	if !errors.As(mapping.FromRecycleBinMutationStatusError(denied), &deniedFailure) ||
		deniedFailure.Category != recyclemutationsemantic.FailureForbidden ||
		reader.recycleRestoreCalls != 1 || reader.recyclePurgeCalls != 1 {
		t.Fatalf("read-only mutation error=%#v adapter calls=%d/%d", denied, reader.recycleRestoreCalls, reader.recyclePurgeCalls)
	}
	metrics := readRuntimeMetrics(t, "http://"+process.AdminAddress()+"/metrics")
	for _, metric := range []string{
		`nexuscrm_runtime_records_repository_calls_total{operation="restore_recycle_bin"} 1`,
		`nexuscrm_runtime_records_repository_calls_total{operation="purge_recycle_bin"} 1`,
	} {
		if !strings.Contains(metrics, metric) {
			t.Fatalf("Runtime recycle-bin mutation metric %q missing:\n%s", metric, metrics)
		}
	}
}

func TestRuntimeAPIProductionAdmissionBindsReceiptsAndFailsBeforeAdapter(t *testing.T) {
	fixture := newServingFixture(t, 7)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	reader := &readerStub{access: map[string]contract.RecordAccess{"W-1": {Read: true}}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecordAccessMaxAdmissionLease: time.Second,
		WorkloadAuthority:             testWorkloadAuthority,
		SecurityAdmission:             testSecurityAdmission(t),
		ServingAdmission:              fixture.admission,
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	client := runtimev1.NewRecordsServiceClient(connection)

	ctx, cancel := context.WithTimeout(recordAccessProtocolContext(context.Background()), time.Second)
	wireResponse, callErr := client.EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{RecordIds: []string{"W-1"}})
	cancel()
	if callErr != nil {
		t.Fatal(callErr)
	}
	response, err := mapping.FromWireEvaluateRecordAccessResponse(accesssemantic.Request{RecordIDs: []string{"W-1"}}, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if reader.accessCalls != 1 || response.Receipts.CatalogRevision != 7 || response.Receipts.SchemaGeneration != 8 ||
		response.Receipts.EngineGeneration != 9 || response.Receipts.SecurityEpoch != 1 {
		t.Fatalf("reader calls=%d receipts=%+v", reader.accessCalls, response.Receipts)
	}

	fixture.catalogFound = false
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err == nil {
		t.Fatal("inactive Catalog installation remained serving")
	}
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusServiceUnavailable)
	blockedCtx, blockedCancel := context.WithTimeout(recordAccessProtocolContext(context.Background()), time.Second)
	blocked, blockedErr := client.EvaluateRecordAccess(blockedCtx, &runtimev1.EvaluateRecordAccessRequest{RecordIds: []string{"W-1"}})
	blockedCancel()
	var failure *accesssemantic.Failure
	if blocked != nil || !errors.As(mapping.FromRecordAccessStatusError(blockedErr), &failure) || failure.Category != accesssemantic.FailureUnavailable {
		t.Fatalf("blocked response=%#v error=%v failure=%#v", blocked, blockedErr, failure)
	}
	if reader.accessCalls != 1 {
		t.Fatalf("failed serving admission reached adapter: %d calls", reader.accessCalls)
	}
	metrics := readRuntimeMetrics(t, "http://"+process.AdminAddress()+"/metrics")
	if !strings.Contains(metrics, `nexuscrm_runtime_records_repository_calls_total{operation="evaluate_record_access"} 1`) {
		t.Fatalf("failed admission changed repository metric:\n%s", metrics)
	}
}

func TestRuntimeAPIProcessRecordAccessRequiresItsOwnDelegatedCapability(t *testing.T) {
	reader := &readerStub{access: map[string]contract.RecordAccess{"W-1": {Read: true}}}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		RecordAccessMaxAdmissionLease: time.Second,
		WorkloadAuthority:             testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(BootstrapReceipts),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)

	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx := metadata.AppendToOutgoingContext(context.Background(),
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(accesssemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(accesssemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, accesssemantic.CapabilityName,
	)
	// The transport metadata names record-access, but the signed authority fact
	// grants only selected-root. SecurityAdmission must reject before Records.
	ctx = withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), semantic.CapabilityName)))
	ctx, cancel := context.WithTimeout(ctx, time.Second)
	defer cancel()
	_, callErr := runtimev1.NewRecordsServiceClient(connection).EvaluateRecordAccess(ctx, &runtimev1.EvaluateRecordAccessRequest{
		RecordIds: []string{"W-1"},
	})
	if status.Code(callErr) != codes.PermissionDenied || reader.accessCalls != 0 {
		t.Fatalf("error=%v adapter calls=%d, want permission denied before adapter", callErr, reader.accessCalls)
	}
}

func TestRuntimeAPIProcessRegistersIdentityExchangeInTheSamePolicyChain(t *testing.T) {
	var calls atomic.Int64
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts:           ReceiptSourceFunc(BootstrapReceipts),
		IdentityAttemptCap: time.Second,
		IdentityExchange: func(ctx context.Context, request identitysemantic.Request) (identitysemantic.Response, error) {
			calls.Add(1)
			if caller, ok := runtimegrpc.Caller(ctx); !ok || caller != testWorkloadID {
				t.Fatalf("caller = %q, %v", caller, ok)
			}
			if bearer, ok := runtimegrpc.EndUserBearer(ctx); !ok || bearer != "opaque-user-token" {
				t.Fatalf("bearer = %q, %v", bearer, ok)
			}
			if request.AppInstallationID != testInstallationID || !reflect.DeepEqual(request.RequestedCapabilities, []string{semantic.CapabilityName}) {
				t.Fatalf("request = %#v", request)
			}
			return identitysemantic.Response{OpaqueDelegation: "header.payload.signature", ExpiresAtUnixSecond: time.Now().Add(time.Minute).Unix()}, nil
		},
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)

	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	wireClient := runtimev1.NewIdentityContextServiceClient(connection)
	client, err := runtimegrpc.NewIdentityExchangeClient(wireClient.Exchange)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	response, err := client.Exchange(ctx, "opaque-user-token", identitysemantic.Request{
		AppInstallationID: testInstallationID, RequestedCapabilities: []string{semantic.CapabilityName},
	})
	if err != nil {
		t.Fatal(err)
	}
	if response.OpaqueDelegation != "header.payload.signature" || calls.Load() != 1 {
		t.Fatalf("response = %#v, calls = %d", response, calls.Load())
	}
}

func TestRuntimeAPIRequiresIdentityExchangeAndAttemptCapTogether(t *testing.T) {
	for _, config := range []ProcessConfig{
		{IdentityExchange: func(context.Context, identitysemantic.Request) (identitysemantic.Response, error) {
			return identitysemantic.Response{}, nil
		}},
		{IdentityAttemptCap: time.Second},
	} {
		if process, err := Assemble(config, &readerStub{}); err == nil || process != nil {
			t.Fatalf("Assemble(%+v) = %#v, %v; want rejected", config, process, err)
		}
	}
}

func TestRuntimeAPIMapsStableOutcomesBeforeAndAfterAdapter(t *testing.T) {
	tests := []struct {
		name     string
		reader   *readerStub
		revision *uint64
		want     semantic.FailureCategory
	}{
		{name: "missing", reader: &readerStub{err: apierr.NotFound("record not found")}, want: semantic.FailureMissing},
		{name: "forbidden", reader: &readerStub{err: apierr.Forbidden("denied")}, want: semantic.FailureForbidden},
		{name: "stale-before-adapter", reader: &readerStub{}, revision: uint64Pointer(2), want: semantic.FailureStaleState},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			process, err := Assemble(ProcessConfig{
				Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
				AdmissionLimit: 1, ServingEnabled: true,
				WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
				Receipts: ReceiptSourceFunc(BootstrapReceipts),
			}, tc.reader)
			if err != nil {
				t.Fatal(err)
			}
			if err := process.Start(); err != nil {
				t.Fatal(err)
			}
			defer stopProcess(t, process)
			connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
			if err != nil {
				t.Fatal(err)
			}
			defer connection.Close()
			ctx, cancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
			defer cancel()
			result, callErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
				ObjectId: "Issue", RecordId: "ISS-1", SelectedRootFields: []string{"Subject"}, MinCatalogRevision: tc.revision,
			})
			if result != nil {
				t.Fatalf("result = %#v", result)
			}
			var failure *semantic.Failure
			if decoded := mapping.FromStatusError(callErr); !errors.As(decoded, &failure) || failure.Category != tc.want {
				t.Fatalf("error = %v decoded=%#v, want %s", callErr, decoded, tc.want)
			}
			if tc.name == "stale-before-adapter" && tc.reader.calls != 0 {
				t.Fatalf("stale request reached adapter: %d", tc.reader.calls)
			}
		})
	}
}

func TestProductionBootstrapProcessIsNotReadyWithoutP6Authority(t *testing.T) {
	process, err := Assemble(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		WorkloadAuthority: runtimegrpc.AuthorityFunc(RejectingAuthority), SecurityAdmission: NewRejectingSecurityAdmission(),
		Receipts: ReceiptSourceFunc(BootstrapReceipts), ServingEnabled: false,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.LivePath, http.StatusOK)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusServiceUnavailable)
}

func TestRejectingSecurityAdmissionCannotBeMadeReadyByServingFlag(t *testing.T) {
	process, err := Assemble(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: NewRejectingSecurityAdmission(),
		Receipts: ReceiptSourceFunc(BootstrapReceipts), ServingEnabled: true,
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusServiceUnavailable)
}

func TestDeployableBootstrapStartsWithoutDatabaseOrServingAuthority(t *testing.T) {
	t.Setenv("TIDB_HOST", "database-must-not-be-opened")
	process, err := AssembleNonServingBootstrap(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0", AdmissionLimit: 2,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.LivePath, http.StatusOK)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusServiceUnavailable)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+"/debug/pprof/goroutine", http.StatusOK)

	connection, err := grpc.NewClient("dns:///"+process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer cancel()
	response, callErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-1", SelectedRootFields: []string{"Subject"},
	})
	if response != nil || failureCategoryForTest(callErr) != semantic.FailureUnauthenticated {
		t.Fatalf("bootstrap response=%#v error=%v", response, callErr)
	}
}

func TestDeployableBootstrapKeepsNetworkAdminDiagnosticsFree(t *testing.T) {
	process, err := AssembleNonServingBootstrap(ProcessConfig{
		Version: "test", AdminAddress: "0.0.0.0:0", GRPCAddress: "127.0.0.1:0", AdmissionLimit: 2,
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	_, port, err := net.SplitHostPort(process.AdminAddress())
	if err != nil {
		t.Fatal(err)
	}
	admin := "http://" + net.JoinHostPort("127.0.0.1", port)
	assertRuntimeProbe(t, admin+processhttp.LivePath, http.StatusOK)
	assertRuntimeProbe(t, admin+processhttp.ReadyPath, http.StatusServiceUnavailable)
	assertRuntimeProbe(t, admin+"/debug/pprof/goroutine", http.StatusNotFound)

	if _, err := AssembleNonServingBootstrap(ProcessConfig{
		Version: "test", AdminAddress: "0.0.0.0:9091", GRPCAddress: "127.0.0.1:9090",
		AdmissionLimit: 2, ProfilingEnabled: true,
	}); err == nil {
		t.Fatal("continuous profiling was allowed on a network admin listener")
	}
}

func TestAdmissionConfigurationHasExactReplicaBoundary(t *testing.T) {
	for _, tc := range []struct {
		raw  string
		want int
		ok   bool
	}{
		{"", 16, true},
		{"1", 1, true},
		{"16", 16, true},
		{"64", 64, true},
		{"0", 0, false},
		{"65", 0, false},
		{"not-a-number", 0, false},
	} {
		got, err := ParseAdmissionLimit(tc.raw)
		if (err == nil) != tc.ok || got != tc.want {
			t.Fatalf("ParseAdmissionLimit(%q) = %d, %v; want %d, ok=%v", tc.raw, got, err, tc.want, tc.ok)
		}
	}
}

func TestRuntimeAPIProcessAdmissionIsFailFast(t *testing.T) {
	value := "accepted"
	reader := &blockingReader{
		entered: make(chan struct{}, 2), release: make(chan struct{}),
		record: &records.SelectedRootRecord{RecordID: "ISS-1", Fields: []records.SelectedRootField{{
			Name: "Subject", Kind: records.SelectedRootString, String: &value,
		}}},
	}
	process, err := Assemble(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(BootstrapReceipts),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	dialCtx, dialCancel := context.WithTimeout(context.Background(), time.Second)
	defer dialCancel()
	connection, err := grpc.DialContext(dialCtx, process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	client := runtimev1.NewRecordsServiceClient(connection)
	done := make(chan error, 2)
	for range 2 {
		go func() {
			ctx, cancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
			defer cancel()
			_, err := client.ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
				ObjectId: "Issue", RecordId: "ISS-1", SelectedRootFields: []string{"Subject"},
			})
			done <- err
		}()
	}
	for range 2 {
		select {
		case <-reader.entered:
		case <-time.After(time.Second):
			t.Fatal("admitted calls did not reach reader")
		}
	}
	thirdCtx, thirdCancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer thirdCancel()
	started := time.Now()
	third, thirdErr := client.ReadSelectedRoot(thirdCtx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-3", SelectedRootFields: []string{"Subject"},
	})
	if third != nil || failureCategoryForTest(thirdErr) != semantic.FailureResourceExhausted {
		t.Fatalf("third result=%#v error=%v", third, thirdErr)
	}
	if time.Since(started) > 250*time.Millisecond {
		t.Fatalf("load shedding queued for %v", time.Since(started))
	}
	if reader.calls.Load() != 2 {
		t.Fatalf("overload reached reader: %d calls", reader.calls.Load())
	}
	close(reader.release)
	for range 2 {
		if err := <-done; err != nil {
			t.Fatalf("admitted call failed: %v", err)
		}
	}
}

func TestRuntimeAPIProcessPropagatesDeadlineIntoAdapter(t *testing.T) {
	reader := &blockingReader{entered: make(chan struct{}, 1), release: make(chan struct{}), canceled: make(chan error, 1)}
	process, err := Assemble(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 1, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(BootstrapReceipts),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	dialCtx, dialCancel := context.WithTimeout(context.Background(), time.Second)
	defer dialCancel()
	connection, err := grpc.DialContext(dialCtx, process.GRPCAddress(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock())
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(protocolContext(context.Background()), 25*time.Millisecond)
	defer cancel()
	result, callErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-1", SelectedRootFields: []string{"Subject"},
	})
	if result != nil || failureCategoryForTest(callErr) != semantic.FailureDeadlineExceeded {
		t.Fatalf("deadline result=%#v error=%v", result, callErr)
	}
	select {
	case adapterErr := <-reader.canceled:
		if adapterErr == nil {
			t.Fatal("adapter observed nil cancellation")
		}
	case <-time.After(time.Second):
		t.Fatal("deadline did not propagate into adapter")
	}
}

func TestRuntimeAPIGracefulStopDeliversCompleteResponse(t *testing.T) {
	largeValue := strings.Repeat("x", 900_000)
	reader := &blockingReader{
		entered: make(chan struct{}, 1), release: make(chan struct{}),
		record: &records.SelectedRootRecord{RecordID: "ISS-LARGE", Fields: []records.SelectedRootField{{
			Name: "Body", Kind: records.SelectedRootString, String: &largeValue,
		}}},
	}
	process, err := Assemble(ProcessConfig{
		Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 1, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		Receipts: ReceiptSourceFunc(BootstrapReceipts),
	}, reader)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	dialCtx, dialCancel := context.WithTimeout(context.Background(), time.Second)
	defer dialCancel()
	connection, err := grpc.DialContext(dialCtx, process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock(),
		grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes)),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	type callResult struct {
		response *runtimev1.ReadSelectedRootResponse
		err      error
	}
	callDone := make(chan callResult, 1)
	go func() {
		ctx, cancel := context.WithTimeout(protocolContext(context.Background()), 2*time.Second)
		defer cancel()
		response, err := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
			ObjectId: "Issue", RecordId: "ISS-LARGE", SelectedRootFields: []string{"Body"},
		})
		callDone <- callResult{response: response, err: err}
	}()
	select {
	case <-reader.entered:
	case <-time.After(time.Second):
		t.Fatal("request did not enter Runtime adapter")
	}
	stopDone := make(chan error, 1)
	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		stopDone <- process.Stop(ctx)
	}()
	deadline := time.Now().Add(time.Second)
	for process.Ready() && time.Now().Before(deadline) {
		time.Sleep(time.Millisecond)
	}
	if process.Ready() {
		t.Fatal("process did not leave readiness before shutdown")
	}
	close(reader.release)
	result := <-callDone
	if result.err != nil {
		t.Fatalf("in-flight client RPC failed during graceful stop: %v", result.err)
	}
	semanticResponse, err := mapping.FromWireResponse(result.response)
	if err != nil {
		t.Fatal(err)
	}
	if len(semanticResponse.Fields) != 1 || semanticResponse.Fields[0].String == nil || *semanticResponse.Fields[0].String != largeValue {
		t.Fatal("client did not receive the complete response")
	}
	if err := <-stopDone; err != nil {
		t.Fatalf("graceful stop failed: %v", err)
	}
}

type blockingReader struct {
	calls    atomic.Int64
	entered  chan struct{}
	release  chan struct{}
	canceled chan error
	record   *records.SelectedRootRecord
}

func (r *blockingReader) ReadSelectedRecord(ctx context.Context, _, _ string, _ []string, _ *models.UserSession) (*records.SelectedRootRecord, error) {
	r.calls.Add(1)
	r.entered <- struct{}{}
	select {
	case <-r.release:
		return r.record, nil
	case <-ctx.Done():
		if r.canceled != nil {
			r.canceled <- ctx.Err()
		}
		return nil, ctx.Err()
	}
}

func (*blockingReader) EvaluateRecordAccess(context.Context, *string, []string, *models.UserSession) (map[string]contract.RecordAccess, error) {
	return map[string]contract.RecordAccess{}, nil
}

func failureCategoryForTest(err error) semantic.FailureCategory {
	var failure *semantic.Failure
	if errors.As(mapping.FromStatusError(err), &failure) {
		return failure.Category
	}
	return ""
}

type recordMutationExecutorStub struct {
	calls                  int
	scopeDigest            string
	userID                 string
	receipts               mutationsemantic.Receipts
	requestCachesInstalled bool
	statusCalls            int
	statusScope            string
	request                mutationsemantic.Request
	response               mutationsemantic.Response
}

func (s *recordMutationExecutorStub) Mutate(ctx context.Context, scope string, request mutationsemantic.Request, receipts mutationsemantic.Receipts, user *models.UserSession) (mutationsemantic.Response, error) {
	s.calls++
	s.scopeDigest = scope
	s.receipts = receipts
	s.request = request
	s.requestCachesInstalled = permissions.WithAccessInputCache(ctx) == ctx && records.EnsureRecordReadCache(ctx) == ctx
	if user != nil {
		s.userID = user.ID
	}
	digest, err := mutationsemantic.CommandDigest(request)
	if err != nil {
		return mutationsemantic.Response{}, err
	}
	results := make([]mutationsemantic.OperationResult, len(request.Operations))
	for i, operation := range request.Operations {
		results[i] = mutationsemantic.OperationResult{
			Index: uint32(i), Kind: operation.Kind, RecordID: operation.RecordID,
			Success: true, Errors: []mutationsemantic.OperationError{},
		}
	}
	s.response = mutationsemantic.Response{
		CommandID: request.CommandID, CommandDigest: digest, Results: results, Receipts: receipts,
		OutcomeExpiresAtUTC: "2026-07-20T12:30:00Z",
	}
	return s.response, nil
}

func (s *recordMutationExecutorStub) Status(_ context.Context, scope string, request mutationsemantic.StatusRequest) (mutationsemantic.CommandShape, mutationsemantic.StatusResponse, error) {
	s.statusCalls++
	s.statusScope = scope
	shape, err := mutationsemantic.CommandShapeFor(s.request)
	if err != nil {
		return mutationsemantic.CommandShape{}, mutationsemantic.StatusResponse{}, err
	}
	if request.CommandID != shape.CommandID || request.CommandDigest != shape.CommandDigest {
		return mutationsemantic.CommandShape{}, mutationsemantic.StatusResponse{}, errors.New("status request does not match accepted command")
	}
	receipts := s.response.Receipts
	return shape, mutationsemantic.StatusResponse{
		CommandID: s.response.CommandID, CommandDigest: s.response.CommandDigest,
		State: mutationsemantic.CommandCompleted, Results: s.response.Results,
		Receipts: &receipts, OutcomeExpiresAtUTC: s.response.OutcomeExpiresAtUTC,
	}, nil
}

type readerStub struct {
	record                 *records.SelectedRootRecord
	err                    error
	calls                  int
	userID                 string
	access                 map[string]contract.RecordAccess
	accessErr              error
	accessCalls            int
	readAccessCalls        int
	accessObject           *string
	accessIDs              []string
	recycleBin             *contract.RecycleBinListResult
	recycleErr             error
	recycleCalls           int
	recycleScope           contract.RecycleBinScope
	recycleRestoreCalls    int
	recyclePurgeCalls      int
	recycleRecordID        string
	requestCachesInstalled bool
}

type createDefaultPreviewProcessReader struct {
	*readerStub
	preview *createDefaultPreviewEngineStub
}

func (reader *createDefaultPreviewProcessReader) PreviewCreateDefaults(
	ctx context.Context,
	request defaultsemantic.Request,
	user *models.UserSession,
) ([]defaultsemantic.Field, error) {
	return reader.preview.PreviewCreateDefaults(ctx, request, user)
}

func (r *readerStub) ReadSelectedRecord(ctx context.Context, _, _ string, _ []string, user *models.UserSession) (*records.SelectedRootRecord, error) {
	r.calls++
	r.requestCachesInstalled = permissions.WithAccessInputCache(ctx) == ctx && records.EnsureRecordReadCache(ctx) == ctx
	if user != nil {
		r.userID = user.ID
	}
	return r.record, r.err
}

func (r *readerStub) EvaluateRecordAccess(_ context.Context, object *string, ids []string, user *models.UserSession) (map[string]contract.RecordAccess, error) {
	r.accessCalls++
	return r.recordAccessResult(object, ids, user)
}

func (r *readerStub) EvaluateRecordReadAccess(_ context.Context, object *string, ids []string, user *models.UserSession) (map[string]contract.RecordAccess, error) {
	r.readAccessCalls++
	return r.recordAccessResult(object, ids, user)
}

func (r *readerStub) recordAccessResult(object *string, ids []string, user *models.UserSession) (map[string]contract.RecordAccess, error) {
	if user != nil {
		r.userID = user.ID
	}
	if object != nil {
		value := *object
		r.accessObject = &value
	}
	r.accessIDs = append([]string(nil), ids...)
	return r.access, r.accessErr
}

func (r *readerStub) ListRecycleBin(_ context.Context, user *models.UserSession, scope contract.RecycleBinScope) (*contract.RecycleBinListResult, error) {
	r.recycleCalls++
	r.recycleScope = scope
	if user != nil {
		r.userID = user.ID
	}
	return r.recycleBin, r.recycleErr
}

func (r *readerStub) RestoreRecycleBinRecord(_ context.Context, recordID string, user *models.UserSession) error {
	r.recycleRestoreCalls++
	r.recycleRecordID = recordID
	if user != nil {
		r.userID = user.ID
	}
	return r.recycleErr
}

func (r *readerStub) PurgeRecycleBinRecord(_ context.Context, recordID string, user *models.UserSession) error {
	r.recyclePurgeCalls++
	r.recycleRecordID = recordID
	if user != nil {
		r.userID = user.ID
	}
	return r.recycleErr
}

func protocolContext(ctx context.Context) context.Context {
	return protocolContextWithDelegation(ctx, testDelegationToken)
}

func recordAccessProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(accesssemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(accesssemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, accesssemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), accesssemantic.CapabilityName)))
}

func recordViewProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(recordviewsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(recordviewsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, recordviewsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), recordviewsemantic.CapabilityName)))
}

func attachmentSourceProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(attachmentsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(attachmentsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, attachmentsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), attachmentsemantic.CapabilityName)))
}

func relatedCountProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(relatedsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(relatedsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, relatedsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), relatedsemantic.CapabilityName)))
}

func createDefaultPreviewProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(defaultsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(defaultsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, defaultsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), defaultsemantic.CapabilityName)))
}

func recycleBinProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(recyclesemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(recyclesemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, recyclesemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), recyclesemantic.CapabilityName)))
}

func recycleBinMutationProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(recyclemutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(recyclemutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, recyclemutationsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), recyclemutationsemantic.CapabilityName)))
}

func recordMutationProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)
	return withTestDelegation(ctx, mustSignTestDelegation(testClaimsFor(time.Now(), mutationsemantic.CapabilityName)))
}

func protocolContextWithDelegation(ctx context.Context, delegation string) context.Context {
	ctx = metadata.AppendToOutgoingContext(ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(semantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, semantic.CapabilityName,
	)
	if delegation != "" {
		ctx = withTestDelegation(ctx, delegation)
	}
	return ctx
}

func uint64Pointer(value uint64) *uint64 { return &value }

func stopProcess(t *testing.T, process interface{ Stop(context.Context) error }) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := process.Stop(ctx); err != nil {
		t.Fatal(err)
	}
}

func assertRuntimeProbe(t *testing.T, url string, want int) {
	t.Helper()
	client := http.Client{Timeout: time.Second}
	response, err := client.Get(url)
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	if response.StatusCode != want {
		t.Fatalf("GET %s = %d, want %d", url, response.StatusCode, want)
	}
}

func readRuntimeMetrics(t *testing.T, url string) string {
	t.Helper()
	client := http.Client{Timeout: time.Second}
	response, err := client.Get(url)
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		t.Fatalf("GET %s = %d", url, response.StatusCode)
	}
	body, err := io.ReadAll(response.Body)
	if err != nil {
		t.Fatal(err)
	}
	return string(body)
}

type capabilityLayoutAuthoringEngine struct{ LayoutAuthoringV2Engine }

func TestRuntimeAPIProcessAdvertisesLayoutAuthoringOnlyWhenEnabled(t *testing.T) {
	for _, enabled := range []bool{false, true} {
		t.Run(strconv.FormatBool(enabled), func(t *testing.T) {
			config := ProcessConfig{
				Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
				AdmissionLimit: 2, ServingEnabled: true,
				WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
			}
			if enabled {
				admission, err := NewLayoutAuthoringV2Admission(LayoutAuthoringV2Selection{OrgID: "org", AppInstallationID: "app", WorkloadID: "workload"})
				if err != nil {
					t.Fatal(err)
				}
				config.LayoutAuthoringV2 = &capabilityLayoutAuthoringEngine{}
				config.LayoutAuthoringV2Admission = admission
				config.EffectivePermissions = &sessionScopedEffectivePermissionEngine{}
				config.PlatformCapabilityMaxAdmissionLeases = map[string]time.Duration{layoutauthoringsemantic.CapabilityName: time.Second}
				config.PlatformResourceClassAdmissionLimits = map[runtimegrpc.ResourceClass]int{runtimegrpc.ResourceClassHeavyWork: 1}
			}
			process, err := Assemble(config, &readerStub{})
			if err != nil {
				t.Fatal(err)
			}
			if err := process.Start(); err != nil {
				t.Fatal(err)
			}
			defer stopProcess(t, process)
			body := readRuntimeMetrics(t, "http://"+process.AdminAddress()+processhttp.CapabilitiesPath)
			if strings.Contains(body, `"`+layoutauthoringsemantic.CapabilityName+`"`) != enabled {
				t.Fatalf("enabled=%v capabilities=%s", enabled, body)
			}
		})
	}
}