//go:build runtime_api_integration

package runtimeapi

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/audit"
	featureauth "github.com/nexuscrm/auth"
	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/runtimeassembly"
	"github.com/nexuscrm/backend/internal/runtimeprocess"
	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	kernelauth "github.com/nexuscrm/kernel/pkg/auth"
	"github.com/nexuscrm/kernel/pkg/ports"
	domainrls "github.com/nexuscrm/kernel/pkg/rls"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedevents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/approvalir"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1"
	flowcommandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mailboxauthorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/mailboxauthorizationv1"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	accesssemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordaccessv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	recyclesemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recyclebinv1"
	reportsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/reportv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	"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"
)

const runtimeApprovalTestRecordID = "a01000000000001"

type outboundRepositoryTransactionRunner struct {
	manager *persistence.TransactionManager
}

func (r outboundRepositoryTransactionRunner) RunInPessimisticTransaction(
	ctx context.Context,
	fn func(context.Context) error,
) error {
	return r.manager.WithPessimisticTransaction(ctx, fn)
}

type outboundRepositoryResultCommitter struct {
	repository      *persistence.RuntimeOutboundIntentRepository
	projectionCalls *atomic.Int64
}

func (c outboundRepositoryResultCommitter) Commit(
	ctx context.Context,
	_ *catalogsnapshot.RuntimeMetadataView,
	scope persistence.RuntimeOutboundIntentScope,
	request outboundintentsemantic.CommitResultRequest,
) (bool, error) {
	return c.repository.CommitResultWithProjection(
		ctx, scope, request,
		func(context.Context, persistence.RuntimeOutboundIntentRow) error {
			c.projectionCalls.Add(1)
			return nil
		},
	)
}

func TestRuntimeAPIOutboundIntentCrossProcessTiDBRestartAndReplay(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIOutboundIntentTable(t, db)
	connection := database.NewConnectionFromDB(db)
	manager := persistence.NewTransactionManager(connection)
	repository := persistence.NewRuntimeOutboundIntentRepository(db, manager)
	runner := outboundRepositoryTransactionRunner{manager: manager}
	var projectionCalls atomic.Int64
	view := outboundIntentTestView(t, 7)
	current := commandConsumerIntegrationView(view)
	current.Receipts.SecurityEpoch = 1
	admission, err := NewOutboundIntentAdmission(t.Context(), OutboundIntentSelection{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID, RuntimeMetadataView: view,
	}, &catalogConsumerViewTestEngine{integrationView: current})
	if err != nil {
		t.Fatal(err)
	}
	scope := persistence.RuntimeOutboundIntentScope{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		AuthorizedWorkloadID: testWorkloadID,
	}
	intent := outboundIntentTestIntent(7)
	if err := runner.RunInPessimisticTransaction(t.Context(), func(txCtx context.Context) error {
		_, _, enqueueErr := repository.Enqueue(txCtx, scope, intent)
		return enqueueErr
	}); err != nil {
		t.Fatal(err)
	}
	config := ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-outbound-intent", outboundintentsemantic.CapabilityName,
		),
		OutboundIntentAdmission: admission, OutboundIntents: repository,
		OutboundResultCommitter: outboundRepositoryResultCommitter{
			repository: repository, projectionCalls: &projectionCalls,
		},
		OutboundIntentTransactions: runner,
		OutboundMailAttachments:    &outboundIntentTestFileEngine{},
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			outboundintentsemantic.CapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassFileStream: 1,
		},
	}
	start := func() (*runtimeprocess.Process, *grpc.ClientConn) {
		process, assembleErr := Assemble(config, &readerStub{})
		if assembleErr != nil {
			t.Fatal(assembleErr)
		}
		if startErr := process.Start(); startErr != nil {
			t.Fatal(startErr)
		}
		client, dialErr := grpc.NewClient(
			"passthrough:///"+process.GRPCAddress(),
			grpc.WithTransportCredentials(insecure.NewCredentials()),
		)
		if dialErr != nil {
			stopProcess(t, process)
			t.Fatal(dialErr)
		}
		return process, client
	}
	process, client := start()
	ctx, cancel := context.WithTimeout(outboundIntentProtocolContext(context.Background()), 3*time.Second)
	generationRequest := outboundintentsemantic.GenerationRequest{AfterGeneration: 0}
	wireGeneration, err := mapping.ToWireReadOutboundIntentGenerationRequest(ctx, generationRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireGenerationResponse, err := runtimev1.NewOutboundIntentServiceClient(client).
		ReadOutboundIntentGeneration(ctx, wireGeneration)
	if err != nil {
		t.Fatalf("read outbound intent generation: %v (%#v)", err,
			mapping.FromPlatformStatusError(err, outboundintentsemantic.CapabilityName))
	}
	generationResponse, err := mapping.FromWireReadOutboundIntentGenerationResponse(
		generationRequest,
		wireGenerationResponse,
	)
	if err != nil || generationResponse.Generation != 1 {
		t.Fatalf("generation=%#v err=%v, want 1", generationResponse, err)
	}
	listRequest := outboundintentsemantic.ListRequest{Limit: 4}
	wireList, err := mapping.ToWireListPendingOutboundIntentsRequest(ctx, listRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireListed, err := runtimev1.NewOutboundIntentServiceClient(client).
		ListPendingOutboundIntents(ctx, wireList)
	if err != nil {
		t.Fatalf("list pending outbound intents: %v (%#v)", err,
			mapping.FromPlatformStatusError(err, outboundintentsemantic.CapabilityName))
	}
	listed, err := mapping.FromWireListPendingOutboundIntentsResponse(listRequest, wireListed)
	if err != nil || len(listed.Intents) != 1 || listed.Intents[0].IntentID != intent.IntentID {
		t.Fatalf("listed=%#v err=%v", listed, err)
	}
	ack := outboundintentsemantic.AcknowledgeRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest,
		CatalogRevision: 7, Outcome: outboundintentsemantic.AdmissionAccepted,
		IntegrationCommandID: "integration-command-1",
	}
	wireAck, err := mapping.ToWireAcknowledgeOutboundIntentRequest(ctx, ack)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := runtimev1.NewOutboundIntentServiceClient(client).
		AcknowledgeOutboundIntent(ctx, wireAck); err != nil {
		t.Fatal(err)
	}
	commit := outboundintentsemantic.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest,
		CatalogRevision: 7, IntegrationCommandID: "integration-command-1", RemoteID: "remote-1",
	}
	wireCommit, err := mapping.ToWireCommitOutboundResultRequest(ctx, commit)
	if err != nil {
		t.Fatal(err)
	}
	committedWire, err := runtimev1.NewOutboundIntentServiceClient(client).
		CommitOutboundResult(ctx, wireCommit)
	if err != nil {
		t.Fatal(err)
	}
	committed, err := mapping.FromWireCommitOutboundResultResponse(commit, committedWire)
	if err != nil || committed.Duplicate {
		t.Fatalf("first commit=%#v err=%v", committed, err)
	}
	if got := projectionCalls.Load(); got != 1 {
		t.Fatalf("first commit projection calls = %d, want 1", got)
	}
	cancel()
	_ = client.Close()
	stopProcess(t, process)

	process, client = start()
	defer stopProcess(t, process)
	defer client.Close()
	replayCtx, replayCancel := context.WithTimeout(outboundIntentProtocolContext(context.Background()), 3*time.Second)
	defer replayCancel()
	replayedWire, err := runtimev1.NewOutboundIntentServiceClient(client).
		CommitOutboundResult(replayCtx, wireCommit)
	if err != nil {
		t.Fatal(err)
	}
	replayed, err := mapping.FromWireCommitOutboundResultResponse(commit, replayedWire)
	if err != nil || !replayed.Duplicate {
		t.Fatalf("replayed commit=%#v err=%v", replayed, err)
	}
	if got := projectionCalls.Load(); got != 1 {
		t.Fatalf("duplicate commit projection calls = %d, want 1", got)
	}
}

func TestRuntimeAPIProductionApprovalFieldUpdateRunsBeforeSaveFlow(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	// Keep the legacy reservation mechanism available so this test can only
	// observe zero sequence work when authoritative Runtime diffs take the new
	// cutover path. An ambient emergency-breaker setting must not make it green.
	t.Setenv("NEXUSCRM_FIELD_HISTORY_SEQUENCE_RESERVATION_ENABLED", "true")
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPIApprovalNotificationFlowSubstrate(t, db)
	createRuntimeAPIApprovalSubstrate(t, db)
	createRuntimeAPIApprovalAuthorizationFixture(t, db)

	capabilities := []string{
		approvalsemantic.ExecuteCapabilityName,
		mutationsemantic.CapabilityName,
	}
	snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilitiesAndMutationDefinitions(
		t,
		testOrgID,
		testInstallationID,
		capabilities,
		func(definition *approvalir.Definition) {
			for i := range definition.Steps {
				definition.Steps[i].Approvers[0].Reference = testPrincipalID
			}
			definition.InitialSubmissionActions = append(definition.InitialSubmissionActions, approvalir.Action{
				Name: "Run notification probe", Kind: approvalir.ActionKindFlow,
				Flow: &approvalir.FlowAction{APIName: approvalNotificationFlowAPIName},
			})
		},
		approvalBeforeSaveMutationDefinitions(t),
	)
	approvalView, err := catalogsnapshot.OpenRuntimeMetadataView(views[approvalsemantic.ExecuteCapabilityName])
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := approvalView.ActiveFlowVersion(approvalNotificationFlowAPIName); !ok {
		t.Fatal("Approval view is missing the notification-only Flow regression fixture")
	}
	mutationView, err := catalogsnapshot.OpenRuntimeMetadataView(views[mutationsemantic.CapabilityName])
	if err != nil {
		t.Fatal(err)
	}
	if _, ok := mutationView.ActiveFlowVersion(approvalNotificationFlowAPIName); ok {
		t.Fatal("Mutation view unexpectedly contains the notification-only Flow regression fixture")
	}
	process, err := AssembleProductionWithConnection(t.Context(), ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-approval-before-save", capabilities...,
		),
		ServingAdmission:     newApprovalServingFixture(t, snapshotArtifact, views),
		ServingEnabled:       true,
		TransportCredentials: insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			approvalsemantic.ExecuteCapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork: 1,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}, database.NewConnectionFromDB(db))
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection := openRuntimeAPIIntegrationGRPC(t, process.GRPCAddress())
	defer connection.Close()
	client := runtimev1.NewApprovalServiceClient(connection)
	var historySequenceBefore int64
	if err := db.QueryRow(`SELECT COALESCE(SUM(CurrentSequence), 0) FROM _System_FieldHistorySequence`).Scan(&historySequenceBefore); err != nil {
		t.Fatal(err)
	}

	response := executeApprovalIntegrationRequest(t, client, testPrincipalID, testSessionID, approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationSubmit, IdempotencyKey: "approval-before-save-submit",
		ObjectID: "WorkItem", TargetRecordID: runtimeApprovalTestRecordID,
	})
	if response.State != approvalsemantic.InstanceStatePending {
		t.Fatalf("Approval submit = %#v, want pending", response)
	}
	var statusValue, marker string
	if err := db.QueryRow(
		`SELECT Status, FlowMarker FROM WorkItem WHERE Id = ?`,
		runtimeApprovalTestRecordID,
	).Scan(&statusValue, &marker); err != nil {
		t.Fatal(err)
	}
	if statusValue != "Submitted" || marker != approvalBeforeSaveFlowMarker {
		t.Fatalf("Approval-owned DML saved Status/FlowMarker = %q/%q, want Submitted/%q", statusValue, marker, approvalBeforeSaveFlowMarker)
	}
	var notificationStatus string
	if err := db.QueryRow(
		`SELECT InterviewStatus FROM FlowInterview WHERE FlowVersionId = ? ORDER BY CreatedDate DESC LIMIT 1`,
		approvalNotificationFlowVersionID,
	).Scan(&notificationStatus); err != nil {
		t.Fatalf("read Approval notification Flow interview: %v", err)
	}
	if notificationStatus != models.FlowInterviewStatusFinished {
		t.Fatalf("Approval notification Flow interview status = %q, want %q", notificationStatus, models.FlowInterviewStatusFinished)
	}
	if len(response.PendingWorkItems) != 1 {
		t.Fatalf("Approval submit pending work items = %#v, want exactly one", response.PendingWorkItems)
	}
	firstApproval := executeApprovalIntegrationRequest(t, client, testPrincipalID, testSessionID, approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationApprove, IdempotencyKey: "approval-before-save-first-approve",
		WorkItemID: response.PendingWorkItems[0].WorkItemID,
	})
	if firstApproval.State != approvalsemantic.InstanceStatePending || len(firstApproval.PendingWorkItems) != 1 {
		t.Fatalf("first Approval step = %#v, want one pending second step", firstApproval)
	}
	terminalWorkItemID := firstApproval.PendingWorkItems[0].WorkItemID
	terminalApproval := executeApprovalIntegrationRequest(t, client, testPrincipalID, testSessionID, approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationApprove, IdempotencyKey: "approval-before-save-terminal-approve",
		WorkItemID: terminalWorkItemID,
	})
	if terminalApproval.State != approvalsemantic.InstanceStateApproved || len(terminalApproval.PendingWorkItems) != 0 {
		t.Fatalf("terminal Approval step = %#v, want approved with no pending work", terminalApproval)
	}
	var terminalAttempts, terminalCommits, terminalRollbacks int
	if err := db.QueryRow(`SELECT COUNT(*),
		COALESCE(SUM(Outcome = ?), 0), COALESCE(SUM(Outcome = ?), 0)
		FROM _System_ApprovalActionLog
		WHERE ActionEvent = ? AND WorkItemId = ?`,
		automation.ApprovalActionOutcomeCommitted,
		automation.ApprovalActionOutcomeRolledBack,
		automation.ApprovalActionRequestApprove,
		terminalWorkItemID,
	).Scan(&terminalAttempts, &terminalCommits, &terminalRollbacks); err != nil {
		t.Fatal(err)
	}
	if terminalAttempts != 1 || terminalCommits != 1 || terminalRollbacks != 0 {
		t.Fatalf("terminal Approval attempts/committed/rolled_back = %d/%d/%d, want exactly 1/1/0", terminalAttempts, terminalCommits, terminalRollbacks)
	}
	var historySequenceAfter int64
	if err := db.QueryRow(`SELECT COALESCE(SUM(CurrentSequence), 0) FROM _System_FieldHistorySequence`).Scan(&historySequenceAfter); err != nil {
		t.Fatal(err)
	}
	if historySequenceAfter != historySequenceBefore {
		rows, err := db.Query(`SELECT FieldId, CurrentSequence FROM _System_FieldHistorySequence ORDER BY FieldId`)
		if err != nil {
			t.Fatal(err)
		}
		defer rows.Close()
		sequences := map[string]int64{}
		for rows.Next() {
			var fieldID string
			var sequence int64
			if err := rows.Scan(&fieldID, &sequence); err != nil {
				t.Fatal(err)
			}
			sequences[fieldID] = sequence
		}
		if err := rows.Err(); err != nil {
			t.Fatal(err)
		}
		t.Fatalf("Runtime Approval changed field-history sequences: before=%d after=%d rows=%v", historySequenceBefore, historySequenceAfter, sequences)
	}
	var (
		eventCapabilityID string
		eventPayload      []byte
	)
	if err := db.QueryRow(`SELECT RuntimeCapabilityId, Payload
		FROM _System_RecordEventOutbox
		WHERE RecordKey = ? AND EventType = ?
		ORDER BY CreatedDate LIMIT 1`,
		"WorkItem/"+runtimeApprovalTestRecordID,
		string(sharedevents.RecordUpdated),
	).Scan(&eventCapabilityID, &eventPayload); err != nil {
		t.Fatal(err)
	}
	if eventCapabilityID != mutationsemantic.CapabilityName {
		t.Fatalf("Approval DML durable event capability = %q, want %q", eventCapabilityID, mutationsemantic.CapabilityName)
	}
	var durableEvent eventfeature.InternalRecordEventEnvelope
	if err := json.Unmarshal(eventPayload, &durableEvent); err != nil {
		t.Fatalf("decode Approval DML durable event: %v", err)
	}
	bindingSet, err := catalogsnapshot.OpenRuntimeWorkBindingSet(
		durableEvent.InternalMetadata["runtime_work_metadata_binding"],
	)
	if err != nil {
		t.Fatalf("open Approval DML durable work binding: %v", err)
	}
	wantMutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		t.Fatal(err)
	}
	if bindingSet.Primary != wantMutationBinding || bindingSet.FlowWork != nil {
		t.Fatalf("Approval DML durable binding = %#v, want Mutation primary %#v and no absent Flow sibling", bindingSet, wantMutationBinding)
	}
}

func TestRuntimeAPIProductionApprovalConcurrentSubmitsSerializePerTarget(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPIApprovalSubstrate(t, db)
	createRuntimeAPIApprovalAuthorizationFixture(t, db)

	capabilities := []string{
		approvalsemantic.ExecuteCapabilityName,
		approvalsemantic.ReadCapabilityName,
		mutationsemantic.CapabilityName,
	}
	snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilities(
		t, testOrgID, testInstallationID, capabilities,
		func(definition *approvalir.Definition) {
			for i := range definition.Steps {
				definition.Steps[i].Approvers[0].Reference = testPrincipalID
			}
		},
	)
	process, err := AssembleProductionWithConnection(t.Context(), ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-approval-concurrent", capabilities...,
		),
		ServingAdmission:     newApprovalServingFixture(t, snapshotArtifact, views),
		ServingEnabled:       true,
		TransportCredentials: insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			approvalsemantic.ExecuteCapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork: 2,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}, database.NewConnectionFromDB(db))
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection := openRuntimeAPIIntegrationGRPC(t, process.GRPCAddress())
	defer connection.Close()
	client := runtimev1.NewApprovalServiceClient(connection)

	type submitResult struct {
		response approvalsemantic.ExecuteResponse
		err      error
		rawErr   error
	}
	start := make(chan struct{})
	results := make(chan submitResult, 2)
	var ready sync.WaitGroup
	ready.Add(2)
	for _, key := range []string{"approval-concurrent-a", "approval-concurrent-b"} {
		request := approvalsemantic.ExecuteRequest{
			Operation: approvalsemantic.OperationSubmit, IdempotencyKey: key,
			ObjectID: "WorkItem", TargetRecordID: runtimeApprovalTestRecordID,
		}
		go func() {
			ready.Done()
			<-start
			ctx, cancel := context.WithTimeout(
				platformProtocolContext(context.Background(), approvalsemantic.ExecuteCapabilityName),
				4*time.Second,
			)
			defer cancel()
			wireRequest, mapErr := mapping.ToWireExecuteApprovalRequest(ctx, request)
			if mapErr != nil {
				results <- submitResult{err: mapErr}
				return
			}
			wireResponse, callErr := client.ExecuteApproval(ctx, wireRequest)
			if callErr != nil {
				results <- submitResult{err: mapping.FromPlatformStatusError(callErr, approvalsemantic.ExecuteCapabilityName), rawErr: callErr}
				return
			}
			response, mapErr := mapping.FromWireExecuteApprovalResponse(request, wireResponse)
			results <- submitResult{response: response, err: mapErr}
		}()
	}
	ready.Wait()
	close(start)

	succeeded, rejected := 0, 0
	for range 2 {
		result := <-results
		if result.err == nil {
			if result.response.State != approvalsemantic.InstanceStatePending || result.response.InstanceID == "" {
				t.Fatalf("successful concurrent submit = %#v", result.response)
			}
			succeeded++
			continue
		}
		var failure *semantic.Failure
		if !errors.As(result.err, &failure) || failure.Category != semantic.FailureInvalidRequest {
			t.Fatalf("rejected concurrent submit = %v (wire=%v), want invalid-request pending conflict", result.err, result.rawErr)
		}
		rejected++
	}
	if succeeded != 1 || rejected != 1 {
		t.Fatalf("concurrent submit outcomes: succeeded=%d rejected=%d, want 1/1", succeeded, rejected)
	}

	var pendingInstances, pendingWorkItems, durableInvocations int
	if err := db.QueryRow(`SELECT COUNT(*) FROM ProcessInstance WHERE TargetObjectId = ? AND Status = ?`, runtimeApprovalTestRecordID, constants.ApprovalStatusPending).Scan(&pendingInstances); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM ProcessInstanceWorkitem`).Scan(&pendingWorkItems); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_RuntimeApprovalInvocation`).Scan(&durableInvocations); err != nil {
		t.Fatal(err)
	}
	if pendingInstances != 1 || pendingWorkItems != 1 || durableInvocations != 1 {
		t.Fatalf("durable rows after concurrent submit: instances=%d workitems=%d invocations=%d, want 1/1/1", pendingInstances, pendingWorkItems, durableInvocations)
	}

	var sourceWorkItemID string
	if err := db.QueryRow(`SELECT Id FROM ProcessInstanceWorkitem LIMIT 1`).Scan(&sourceWorkItemID); err != nil {
		t.Fatal(err)
	}
	actionStart := make(chan struct{})
	actionResults := make(chan submitResult, 2)
	ready = sync.WaitGroup{}
	ready.Add(2)
	for _, key := range []string{"approval-concurrent-approve-a", "approval-concurrent-approve-b"} {
		request := approvalsemantic.ExecuteRequest{
			Operation: approvalsemantic.OperationApprove, IdempotencyKey: key,
			WorkItemID: sourceWorkItemID,
		}
		go func() {
			ready.Done()
			<-actionStart
			ctx, cancel := context.WithTimeout(
				platformProtocolContext(context.Background(), approvalsemantic.ExecuteCapabilityName),
				4*time.Second,
			)
			defer cancel()
			wireRequest, mapErr := mapping.ToWireExecuteApprovalRequest(ctx, request)
			if mapErr != nil {
				actionResults <- submitResult{err: mapErr}
				return
			}
			wireResponse, callErr := client.ExecuteApproval(ctx, wireRequest)
			if callErr != nil {
				actionResults <- submitResult{err: mapping.FromPlatformStatusError(callErr, approvalsemantic.ExecuteCapabilityName), rawErr: callErr}
				return
			}
			response, mapErr := mapping.FromWireExecuteApprovalResponse(request, wireResponse)
			actionResults <- submitResult{response: response, err: mapErr}
		}()
	}
	ready.Wait()
	close(actionStart)

	succeeded, rejected = 0, 0
	for range 2 {
		result := <-actionResults
		if result.err == nil {
			if result.response.State != approvalsemantic.InstanceStatePending || len(result.response.PendingWorkItems) != 1 {
				t.Fatalf("successful concurrent approve = %#v", result.response)
			}
			succeeded++
			continue
		}
		var failure *semantic.Failure
		if !errors.As(result.err, &failure) || (failure.Category != semantic.FailureMissing && failure.Category != semantic.FailureInvalidRequest) {
			t.Fatalf("rejected concurrent approve = %v (wire=%v), want missing/conflict", result.err, result.rawErr)
		}
		rejected++
	}
	if succeeded != 1 || rejected != 1 {
		t.Fatalf("concurrent approve outcomes: succeeded=%d rejected=%d, want 1/1", succeeded, rejected)
	}

	var processNodes, pendingNodes, actionInvocations int
	if err := db.QueryRow(`SELECT COUNT(*), SUM(NodeStatus = ?) FROM ProcessInstanceNode`, constants.ApprovalStatusPending).Scan(&processNodes, &pendingNodes); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_RuntimeApprovalInvocation`).Scan(&actionInvocations); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM ProcessInstanceWorkitem`).Scan(&pendingWorkItems); err != nil {
		t.Fatal(err)
	}
	if processNodes != 2 || pendingNodes != 1 || pendingWorkItems != 1 || actionInvocations != 2 {
		t.Fatalf("durable rows after concurrent approve: nodes=%d pending_nodes=%d workitems=%d invocations=%d, want 2/1/1/2", processNodes, pendingNodes, pendingWorkItems, actionInvocations)
	}
}

func TestRuntimeAPIProductionApprovalConcurrentUnanimousActionsSerializePerInstance(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPIApprovalSubstrate(t, db)
	createRuntimeAPIApprovalAuthorizationFixture(t, db)

	const secondPrincipalID = "approval-concurrent-user-2"
	const secondSessionID = "approval-concurrent-session-2"
	capabilities := []string{
		approvalsemantic.ExecuteCapabilityName,
		approvalsemantic.ReadCapabilityName,
		mutationsemantic.CapabilityName,
	}
	snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilities(
		t, testOrgID, testInstallationID, capabilities,
		func(definition *approvalir.Definition) {
			definition.Steps[0].Approvers = []approvalir.Approver{
				{Kind: approvalir.ApproverKindUser, Reference: testPrincipalID},
				{Kind: approvalir.ApproverKindUser, Reference: secondPrincipalID},
			}
			definition.Steps[0].MultipleApproverMode = approvalir.MultipleApproverModeUnanimous
			definition.Steps[1].Approvers[0].Reference = testPrincipalID
		},
	)
	securityAdmission := durableIntegrationSecurityAdmission(
		t, db, "replica-approval-unanimous", capabilities...,
	)
	if _, err := db.Exec(`INSERT INTO User
(Id, Username, Email, ProfileId, FirstName, LastName, IsActive, IsDeleted, FailedLoginAttempts, UserRoleId)
VALUES (?, 'approval-user-2', 'approval-user-2@example.test', 'profile-test', 'Approval', 'User Two', 1, 0, 0, 'role-test')`, secondPrincipalID); err != nil {
		t.Fatal(err)
	}
	insertRuntimeIntegrationSession(
		t, db, secondSessionID, secondPrincipalID,
		"Approval User Two", "approval-user-2@example.test",
	)

	process, err := AssembleProductionWithConnection(t.Context(), ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: securityAdmission,
		ServingAdmission:  newApprovalServingFixture(t, snapshotArtifact, views),
		ServingEnabled:    true, TransportCredentials: insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			approvalsemantic.ExecuteCapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork: 2,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}, database.NewConnectionFromDB(db))
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection := openRuntimeAPIIntegrationGRPC(t, process.GRPCAddress())
	defer connection.Close()
	client := runtimev1.NewApprovalServiceClient(connection)

	submitRequest := approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationSubmit, IdempotencyKey: "approval-unanimous-submit",
		ObjectID: "WorkItem", TargetRecordID: runtimeApprovalTestRecordID,
	}
	submitResponse := executeApprovalIntegrationRequest(
		t, client, testPrincipalID, testSessionID, submitRequest,
	)
	if submitResponse.State != approvalsemantic.InstanceStatePending || len(submitResponse.PendingWorkItems) != 2 {
		t.Fatalf("unanimous submit = %#v, want two pending work items", submitResponse)
	}

	workItemByActor := make(map[string]string, 2)
	for _, pending := range submitResponse.PendingWorkItems {
		workItemByActor[pending.ActorID] = pending.WorkItemID
	}
	if workItemByActor[testPrincipalID] == "" || workItemByActor[secondPrincipalID] == "" {
		t.Fatalf("unanimous work-item assignment = %#v", submitResponse.PendingWorkItems)
	}

	type actionResult struct {
		response approvalsemantic.ExecuteResponse
		err      error
	}
	start := make(chan struct{})
	results := make(chan actionResult, 2)
	var ready sync.WaitGroup
	ready.Add(2)
	for index, actor := range []struct {
		principalID string
		sessionID   string
	}{
		{principalID: testPrincipalID, sessionID: testSessionID},
		{principalID: secondPrincipalID, sessionID: secondSessionID},
	} {
		request := approvalsemantic.ExecuteRequest{
			Operation:      approvalsemantic.OperationApprove,
			IdempotencyKey: fmt.Sprintf("approval-unanimous-approve-%d", index+1),
			WorkItemID:     workItemByActor[actor.principalID],
		}
		go func() {
			ready.Done()
			<-start
			ctx, cancel := context.WithTimeout(
				approvalProtocolContextForPrincipal(context.Background(), approvalsemantic.ExecuteCapabilityName, actor.principalID, actor.sessionID),
				4*time.Second,
			)
			defer cancel()
			wireRequest, mapErr := mapping.ToWireExecuteApprovalRequest(ctx, request)
			if mapErr != nil {
				results <- actionResult{err: mapErr}
				return
			}
			wireResponse, callErr := client.ExecuteApproval(ctx, wireRequest)
			if callErr != nil {
				results <- actionResult{err: mapping.FromPlatformStatusError(callErr, approvalsemantic.ExecuteCapabilityName)}
				return
			}
			response, mapErr := mapping.FromWireExecuteApprovalResponse(request, wireResponse)
			results <- actionResult{response: response, err: mapErr}
		}()
	}
	ready.Wait()
	close(start)
	for range 2 {
		result := <-results
		if result.err != nil {
			t.Fatalf("concurrent unanimous approval failed: %v", result.err)
		}
		if result.response.State != approvalsemantic.InstanceStatePending {
			t.Fatalf("concurrent unanimous approval response = %#v", result.response)
		}
	}

	var processNodes, pendingNodes, pendingWorkItems, invocations int
	if err := db.QueryRow(`SELECT COUNT(*), SUM(NodeStatus = ?) FROM ProcessInstanceNode`, constants.ApprovalStatusPending).Scan(&processNodes, &pendingNodes); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM ProcessInstanceWorkitem`).Scan(&pendingWorkItems); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_RuntimeApprovalInvocation`).Scan(&invocations); err != nil {
		t.Fatal(err)
	}
	if processNodes != 2 || pendingNodes != 1 || pendingWorkItems != 1 || invocations != 3 {
		t.Fatalf("durable rows after concurrent unanimous actions: nodes=%d pending_nodes=%d workitems=%d invocations=%d, want 2/1/1/3", processNodes, pendingNodes, pendingWorkItems, invocations)
	}
}

func TestRuntimeAPIProductionApprovalExecutionCrossProcessRestart(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPIApprovalSubstrate(t, db)
	createRuntimeAPIApprovalAuthorizationFixture(t, db)

	capabilities := []string{
		approvalsemantic.ExecuteCapabilityName,
		approvalsemantic.ReadCapabilityName,
		mutationsemantic.CapabilityName,
	}
	snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilities(
		t, testOrgID, testInstallationID, capabilities,
		func(definition *approvalir.Definition) {
			for i := range definition.Steps {
				definition.Steps[i].Approvers[0].Reference = testPrincipalID
				definition.Steps[i].AllowDelegate = true
			}
		},
	)
	servingAdmission := newApprovalServingFixture(t, snapshotArtifact, views)
	securityAdmission := durableIntegrationSecurityAdmission(
		t, db, "replica-approval-real", capabilities...,
	)
	const reassignedPrincipalID = "user-approval-reassigned"
	if _, err := db.Exec(`INSERT INTO User
(Id, Username, Email, ProfileId, FirstName, LastName, IsActive, FailedLoginAttempts, UserRoleId)
VALUES (?, 'approval-reassigned', 'approval-reassigned@example.test', 'profile-test', 'Approval', 'Reassigned', 1, 0, 'role-test')`, reassignedPrincipalID); err != nil {
		t.Fatalf("prepare Approval reassignment principal: %v", err)
	}
	config := ProcessConfig{
		Version:           "integration",
		AdminAddress:      "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: securityAdmission,
		ServingAdmission:  servingAdmission,
		ServingEnabled:    true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			approvalsemantic.ExecuteCapabilityName: 5 * time.Second,
			approvalsemantic.ReadCapabilityName:    5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork: 1,
		},
		TransportCredentials:        insecure.NewCredentials(),
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}
	request := approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationSubmit, IdempotencyKey: "approval-production-1",
		ObjectID: "WorkItem", TargetRecordID: runtimeApprovalTestRecordID,
		Comments: "submit through private Platform endpoint",
	}

	first, firstAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, request,
	)
	second, secondAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, request,
	)
	if first.ExecutionID == "" || first.ExecutionID != second.ExecutionID ||
		first.InstanceID == "" || first.InstanceID != second.InstanceID ||
		first.State != approvalsemantic.InstanceStatePending ||
		len(first.PendingWorkItems) != 1 ||
		first.PendingWorkItems[0].WorkItemID == "" ||
		first.Receipts != second.Receipts {
		t.Fatalf("Approval restart replay = first %#v second %#v", first, second)
	}
	if first.Receipts.CapabilityVersion != approvalsemantic.ExecuteCapabilityName ||
		first.Receipts.CatalogRevision != 71 || first.Receipts.SchemaGeneration != 8 ||
		first.Receipts.EngineGeneration != 9 || first.Receipts.SecurityEpoch != 1 {
		t.Fatalf("Approval receipts = %#v", first.Receipts)
	}
	if firstAddress == "" || secondAddress == "" || firstAddress == secondAddress {
		t.Fatalf("Approval process restart addresses = %q, %q", firstAddress, secondAddress)
	}

	approveRequest := approvalsemantic.ExecuteRequest{
		Operation:      approvalsemantic.OperationApprove,
		IdempotencyKey: "approval-production-approve-1",
		WorkItemID:     first.PendingWorkItems[0].WorkItemID,
		Comments:       "approve through private Platform endpoint",
	}
	approved, approveAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, approveRequest,
	)
	approvedReplay, approveReplayAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, approveRequest,
	)
	if approved.ExecutionID == "" || approved.ExecutionID != approvedReplay.ExecutionID ||
		approved.InstanceID != first.InstanceID || approved.State != approvalsemantic.InstanceStatePending ||
		approved.SourceWorkItemID != approveRequest.WorkItemID || len(approved.PendingWorkItems) != 1 ||
		approved.PendingWorkItems[0].WorkItemID == "" || approved.Receipts != approvedReplay.Receipts ||
		approveAddress == "" || approveReplayAddress == "" || approveAddress == approveReplayAddress {
		t.Fatalf("Approval approve restart replay = first %#v second %#v addresses %q/%q", approved, approvedReplay, approveAddress, approveReplayAddress)
	}

	rejectRequest := approvalsemantic.ExecuteRequest{
		Operation:      approvalsemantic.OperationReject,
		IdempotencyKey: "approval-production-reject-1",
		WorkItemID:     approved.PendingWorkItems[0].WorkItemID,
		Comments:       "reject through private Platform endpoint",
	}
	rejected, rejectAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, rejectRequest,
	)
	rejectedReplay, rejectReplayAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, rejectRequest,
	)
	if rejected.ExecutionID == "" || rejected.ExecutionID != rejectedReplay.ExecutionID ||
		rejected.InstanceID != first.InstanceID || rejected.State != approvalsemantic.InstanceStateRejected ||
		rejected.SourceWorkItemID != rejectRequest.WorkItemID || len(rejected.PendingWorkItems) != 0 ||
		rejected.Receipts != rejectedReplay.Receipts || rejectAddress == "" || rejectReplayAddress == "" ||
		rejectAddress == rejectReplayAddress {
		t.Fatalf("Approval reject restart replay = first %#v second %#v addresses %q/%q", rejected, rejectedReplay, rejectAddress, rejectReplayAddress)
	}

	resubmitRequest := approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationSubmit, IdempotencyKey: "approval-production-2",
		ObjectID: "WorkItem", TargetRecordID: runtimeApprovalTestRecordID,
		Comments: "resubmit through private Platform endpoint",
	}
	resubmitted, resubmitAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, resubmitRequest,
	)
	resubmittedReplay, resubmitReplayAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, resubmitRequest,
	)
	if resubmitted.ExecutionID == "" || resubmitted.ExecutionID != resubmittedReplay.ExecutionID ||
		resubmitted.InstanceID == "" || resubmitted.InstanceID == first.InstanceID ||
		resubmitted.State != approvalsemantic.InstanceStatePending || len(resubmitted.PendingWorkItems) != 1 ||
		resubmitted.Receipts != resubmittedReplay.Receipts || resubmitAddress == "" ||
		resubmitReplayAddress == "" || resubmitAddress == resubmitReplayAddress {
		t.Fatalf("Approval resubmit restart replay = first %#v second %#v addresses %q/%q", resubmitted, resubmittedReplay, resubmitAddress, resubmitReplayAddress)
	}
	pending := runProductionApprovalRead(
		t, database.NewConnectionFromDB(db), config,
		approvalsemantic.ReadRequest{Kind: approvalsemantic.ReadPendingForCaller},
	)
	if len(pending.Pending) != 1 || pending.Pending[0].WorkItemID != resubmitted.PendingWorkItems[0].WorkItemID ||
		pending.Pending[0].InstanceID != resubmitted.InstanceID ||
		pending.Pending[0].DefinitionID != approvalTestDefinitionID ||
		pending.Pending[0].TargetRecordID != runtimeApprovalTestRecordID ||
		pending.Pending[0].ObjectID != "WorkItem" || pending.Pending[0].ActorID != testPrincipalID {
		t.Fatalf("Approval pending read projection = %#v", pending)
	}
	steps := runProductionApprovalRead(
		t, database.NewConnectionFromDB(db), config,
		approvalsemantic.ReadRequest{
			Kind: approvalsemantic.ReadStepDisplayForRecord, TargetRecordID: runtimeApprovalTestRecordID,
		},
	)
	if len(steps.Steps) != 2 ||
		steps.Steps[0].StepID != catalogsnapshot.ApprovalNodeStorageID(approvalTestDefinitionID, "FirstReview") ||
		steps.Steps[0].StepOrder != 1 || steps.Steps[0].StepName != "FirstReview" ||
		steps.Steps[0].StepLabel != "Catalog first" ||
		steps.Steps[1].StepID != catalogsnapshot.ApprovalNodeStorageID(approvalTestDefinitionID, "SecondReview") ||
		steps.Steps[1].StepOrder != 2 || steps.Steps[1].StepName != "SecondReview" ||
		steps.Steps[1].StepLabel != "Catalog second" {
		t.Fatalf("Approval step-display read projection = %#v", steps)
	}

	reassignRequest := approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationReassign, IdempotencyKey: "approval-production-reassign-1",
		WorkItemID: resubmitted.PendingWorkItems[0].WorkItemID, NextActorID: reassignedPrincipalID,
	}
	reassigned, reassignAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, reassignRequest,
	)
	reassignedReplay, reassignReplayAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, reassignRequest,
	)
	if reassigned.ExecutionID == "" || reassigned.ExecutionID != reassignedReplay.ExecutionID ||
		reassigned.InstanceID != resubmitted.InstanceID || reassigned.State != approvalsemantic.InstanceStatePending ||
		reassigned.SourceWorkItemID != reassignRequest.WorkItemID || len(reassigned.PendingWorkItems) != 1 ||
		reassigned.PendingWorkItems[0].WorkItemID != reassignRequest.WorkItemID ||
		reassigned.PendingWorkItems[0].ActorID != reassignedPrincipalID ||
		reassigned.Receipts != reassignedReplay.Receipts || reassignAddress == "" ||
		reassignReplayAddress == "" || reassignAddress == reassignReplayAddress {
		t.Fatalf("Approval reassign restart replay = first %#v second %#v addresses %q/%q", reassigned, reassignedReplay, reassignAddress, reassignReplayAddress)
	}

	recallRequest := approvalsemantic.ExecuteRequest{
		Operation: approvalsemantic.OperationRecall, IdempotencyKey: "approval-production-recall-1",
		WorkItemID: reassigned.PendingWorkItems[0].WorkItemID,
		Comments:   "recall through private Platform endpoint",
	}
	recalled, recallAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, recallRequest,
	)
	recalledReplay, recallReplayAddress := runProductionApprovalExecution(
		t, database.NewConnectionFromDB(db), config, recallRequest,
	)
	if recalled.ExecutionID == "" || recalled.ExecutionID != recalledReplay.ExecutionID ||
		recalled.InstanceID != resubmitted.InstanceID || recalled.State != approvalsemantic.InstanceStateWithdrawn ||
		recalled.SourceWorkItemID != recallRequest.WorkItemID || len(recalled.PendingWorkItems) != 0 ||
		recalled.Receipts != recalledReplay.Receipts || recallAddress == "" ||
		recallReplayAddress == "" || recallAddress == recallReplayAddress {
		t.Fatalf("Approval recall restart replay = first %#v second %#v addresses %q/%q", recalled, recalledReplay, recallAddress, recallReplayAddress)
	}

	latest := runProductionApprovalRead(
		t,
		database.NewConnectionFromDB(db),
		config,
		approvalsemantic.ReadRequest{
			Kind:           approvalsemantic.ReadLatestForRecord,
			TargetRecordID: runtimeApprovalTestRecordID,
		},
	)
	if latest.Latest == nil || latest.Latest.InstanceID != resubmitted.InstanceID ||
		latest.Latest.DefinitionID != approvalTestDefinitionID ||
		latest.Latest.State != approvalsemantic.InstanceStateWithdrawn ||
		latest.Receipts.CapabilityVersion != approvalsemantic.ReadCapabilityName ||
		latest.Receipts.CatalogRevision != 71 {
		t.Fatalf("Approval read projection = %#v", latest)
	}

	var (
		statusValue     string
		invocationCount int
		completedCount  int
		instanceCount   int
		workItemCount   int
		actionCount     int
	)
	if err := db.QueryRow(`SELECT
(SELECT Status FROM WorkItem WHERE Id = ?),
(SELECT COUNT(*) FROM _System_RuntimeApprovalInvocation),
(SELECT COUNT(*) FROM _System_RuntimeApprovalInvocation WHERE State = 'COMPLETED'),
(SELECT COUNT(*) FROM ProcessInstance WHERE TargetObjectId = ?),
(SELECT COUNT(*) FROM ProcessInstanceWorkitem),
(SELECT COUNT(*) FROM ProcessAction)`,
		runtimeApprovalTestRecordID, runtimeApprovalTestRecordID,
	).Scan(
		&statusValue,
		&invocationCount,
		&completedCount,
		&instanceCount,
		&workItemCount,
		&actionCount,
	); err != nil {
		t.Fatal(err)
	}
	if statusValue != "Reassigned" || invocationCount != 6 || completedCount != 6 ||
		instanceCount != 2 || workItemCount != 0 || actionCount != 0 {
		t.Fatalf(
			"Approval durable closure = status %q invocations %d/%d instances %d workitems %d live-actions %d",
			statusValue, completedCount, invocationCount, instanceCount, workItemCount, actionCount,
		)
	}
}

func TestRuntimeAPIRealQueryRepositoryCrossProcess(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	if _, err := db.Exec(`CREATE TABLE Issue (
        Id VARCHAR(255) PRIMARY KEY,
		OwnerId VARCHAR(255) NOT NULL,
		Subject VARCHAR(255), EmptySummary VARCHAR(255), Archived BOOLEAN, RetryCount BIGINT,
        Estimate DECIMAL(30,4), DueDate DATE, UpdatedAt DATETIME(6),
        ProjectId VARCHAR(255), OptionalValue VARCHAR(255), SecretNote VARCHAR(255)
    )`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO Issue
		(Id, OwnerId, Subject, EmptySummary, Archived, RetryCount, Estimate, DueDate, UpdatedAt, ProjectId, OptionalValue, SecretNote)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		"ISS-P3-1", testPrincipalID, "cross-process", "", false, 0, "12345678901234567890.1200",
		"2026-07-12", "2026-07-12 09:10:11.123456", "PRJ-1", nil, "hidden"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO Issue
		(Id, OwnerId, Subject, EmptySummary, Archived, RetryCount, Estimate, DueDate, UpdatedAt, ProjectId, OptionalValue, SecretNote)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		"ISS-P3-RLS", "different-owner", "masked", "", true, 1, "1.0000",
		"2026-07-12", "2026-07-12 09:10:11.123456", "PRJ-2", nil, "hidden"); err != nil {
		t.Fatal(err)
	}

	schema := runtimeAPIIntegrationSchema()
	queryRepo := &countingQueryRepository{QueryRepository: persistence.NewQueryRepository(db)}
	permissionRepo := &integrationPermissionRepo{fields: schema.Fields}
	permissionRepo.grantRead.Store(true)
	graph, err := runtimeassembly.NewRecordsReadRuntime(runtimeassembly.RecordsReadDependencies{
		DB: db, MetadataRepo: &integrationMetadataRepo{schema: schema},
		PermissionRepo: permissionRepo, QueryRepo: queryRepo, RecordRepo: persistence.NewRecordRepository(db),
	})
	if err != nil {
		t.Fatal(err)
	}
	securityAdmission := durableIntegrationSecurityAdmission(t, db, "replica-real", accesssemantic.CapabilityName, semantic.CapabilityName)
	process, err := Assemble(ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, ServingEnabled: true,
		RecordAccessMaxAdmissionLease: time.Second,
		WorkloadAuthority:             testWorkloadAuthority,
		SecurityAdmission:             securityAdmission, Receipts: ReceiptSourceFunc(BootstrapReceipts),
	}, graph.Records)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)

	connection, err := grpc.NewClient("passthrough:///"+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()
	ctx, cancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer cancel()
	wireResponse, err := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(ctx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-P3-1",
		SelectedRootFields: []string{"UpdatedAt", "SecretNote", "OptionalValue", "ProjectId", "DueDate", "Estimate", "RetryCount", "Archived", "EmptySummary", "Subject"},
	})
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if queryRepo.calls.Load() != 1 {
		t.Fatalf("database query calls = %d, want 1", queryRepo.calls.Load())
	}
	byName := make(map[string]semantic.Field, len(response.Fields))
	names := make([]string, 0, len(response.Fields))
	for _, field := range response.Fields {
		byName[field.Name] = field
		names = append(names, field.Name)
	}
	if want := []string{"Archived", "DueDate", "EmptySummary", "Estimate", "OptionalValue", "ProjectId", "RetryCount", "Subject", "UpdatedAt"}; !slices.Equal(names, want) {
		t.Fatalf("field order = %v, want %v", names, want)
	}
	if len(byName) != 9 || byName["Archived"].Boolean == nil || *byName["Archived"].Boolean ||
		byName["RetryCount"].Integer == nil || *byName["RetryCount"].Integer != 0 ||
		byName["EmptySummary"].String == nil || *byName["EmptySummary"].String != "" ||
		byName["OptionalValue"].Null != true || byName["Estimate"].Decimal == nil ||
		byName["Estimate"].Decimal.Coefficient != "123456789012345678901200" || byName["Estimate"].Decimal.Scale != 4 ||
		byName["Subject"].String == nil || *byName["Subject"].String != "cross-process" ||
		byName["DueDate"].Date == nil || *byName["DueDate"].Date != "2026-07-12" ||
		byName["UpdatedAt"].DateTime == nil || *byName["UpdatedAt"].DateTime != "2026-07-12T09:10:11.123456Z" ||
		byName["ProjectId"].Reference == nil || *byName["ProjectId"].Reference != "PRJ-1" {
		t.Fatalf("typed response = %#v", response.Fields)
	}
	if _, exists := byName["SecretNote"]; exists {
		t.Fatal("FLS-hidden field crossed the process boundary")
	}
	if got := permissionRepo.fieldObjectLoads.Load(); got != 1 {
		t.Fatalf("object-bounded field permission loads = %d, want 1", got)
	}
	if got := permissionRepo.fieldPointLoads.Load(); got != 0 {
		t.Fatalf("point field permission loads = %d, want 0", got)
	}

	missingCtx, missingCancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer missingCancel()
	missing, missingErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(missingCtx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-MISSING", SelectedRootFields: []string{"Subject"},
	})
	if missing != nil || failureCategory(missingErr) != semantic.FailureMissing {
		t.Fatalf("missing result=%#v error=%v", missing, missingErr)
	}
	if queryRepo.calls.Load() != 2 {
		t.Fatalf("query calls after missing = %d, want 2", queryRepo.calls.Load())
	}

	rlsCtx, rlsCancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer rlsCancel()
	rlsMasked, rlsErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(rlsCtx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-P3-RLS", SelectedRootFields: []string{"Subject"},
	})
	if rlsMasked != nil || failureCategory(rlsErr) != semantic.FailureMissing {
		t.Fatalf("RLS result=%#v error=%v", rlsMasked, rlsErr)
	}
	if queryRepo.calls.Load() != 3 {
		t.Fatalf("query calls after RLS-masked read = %d, want 3", queryRepo.calls.Load())
	}

	accessCtx, accessCancel := context.WithTimeout(recordAccessProtocolContext(context.Background()), time.Second)
	defer accessCancel()
	accessRequest := accesssemantic.Request{ObjectID: stringPointer("Issue"), RecordIDs: []string{"ISS-P3-RLS", "ISS-P3-1", "ISS-MISSING", "ISS-P3-1"}}
	accessWire, accessErr := runtimev1.NewRecordsServiceClient(connection).EvaluateRecordAccess(accessCtx, &runtimev1.EvaluateRecordAccessRequest{
		ObjectId: accessRequest.ObjectID, RecordIds: append([]string(nil), accessRequest.RecordIDs...),
	})
	if accessErr != nil {
		t.Fatal(accessErr)
	}
	accessResponse, err := mapping.FromWireEvaluateRecordAccessResponse(accessRequest, accessWire)
	if err != nil {
		t.Fatal(err)
	}
	if len(accessResponse.Results) != 1 || accessResponse.Results[0].RecordID != "ISS-P3-1" ||
		!accessResponse.Results[0].Access.Read || !accessResponse.Results[0].Access.Update || !accessResponse.Results[0].Access.Delete || accessResponse.Results[0].Access.Locked ||
		accessResponse.Receipts.CapabilityVersion != accesssemantic.CapabilityName {
		t.Fatalf("record-access response = %#v", accessResponse)
	}

	permissionRepo.grantRead.Store(false)
	forbiddenCtx, forbiddenCancel := context.WithTimeout(protocolContext(context.Background()), time.Second)
	defer forbiddenCancel()
	forbidden, forbiddenErr := runtimev1.NewRecordsServiceClient(connection).ReadSelectedRoot(forbiddenCtx, &runtimev1.ReadSelectedRootRequest{
		ObjectId: "Issue", RecordId: "ISS-P3-1", SelectedRootFields: []string{"Subject"},
	})
	if forbidden != nil || failureCategory(forbiddenErr) != semantic.FailureForbidden {
		t.Fatalf("forbidden result=%#v error=%v", forbidden, forbiddenErr)
	}
	if queryRepo.calls.Load() != 3 {
		t.Fatalf("object denial reached repository: %d queries", queryRepo.calls.Load())
	}
}

func TestRuntimeAPIProductionCommandAcceptanceCrossProcessRestart(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIPlatformCommandTable(t, db)
	// Durable acceptance remains available while the role-scoped executor is
	// stopped or restarting. Deliberately do not create a global work-family
	// owner here: it cannot prove this AppInstallation has an active Worker.

	runtimeView := commandFlowAcceptanceTestView(
		t,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	current.Receipts.SecurityEpoch = 1
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	commandConsumerAdmission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		&catalogConsumerViewTestEngine{integrationView: current},
	)
	if err != nil {
		t.Fatal(err)
	}
	securityAdmission := durableIntegrationSecurityAdmission(
		t, db, "replica-command-real", commandsemantic.CapabilityName,
	)
	connection := database.NewConnectionFromDB(db)
	config := ProcessConfig{
		Version:                  "integration",
		AdminAddress:             "127.0.0.1:0",
		GRPCAddress:              "127.0.0.1:0",
		AdmissionLimit:           DefaultAdmissionLimit,
		WorkloadAuthority:        testWorkloadAuthority,
		SecurityAdmission:        securityAdmission,
		ServingAdmission:         &ServingAdmission{},
		CommandConsumerAdmission: commandConsumerAdmission,
		ServingEnabled:           true,
		TransportCredentials:     insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			commandsemantic.CapabilityName: time.Second,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}
	revision := uint64(17)
	payload, err := flowcommandsemantic.Encode(flowcommandsemantic.Payload{
		DefinitionRef:     "flow.integration-ingress",
		DefinitionVersion: "flow-version-1",
		CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID: "command-production-1", IdempotencyKey: "request-production-1",
		CommandKind: flowcommandsemantic.CommandKind, PartitionKey: "tenant-partition-1",
		PayloadFormatVersion: flowcommandsemantic.PayloadFormatVersion, Payload: payload,
		PayloadDigest:      commandsemantic.ComputePayloadDigest(payload),
		MinCatalogRevision: &revision,
	}

	first, firstAddress := runProductionCommandAcceptance(t, connection, config, request)
	if first.Duplicate {
		t.Fatalf("first AcceptCommand duplicate = true")
	}
	if first.CommandID != request.CommandID ||
		first.Receipts.CapabilityVersion != commandsemantic.CapabilityName ||
		first.Receipts.CatalogRevision != revision ||
		first.Receipts.SchemaGeneration != 8 ||
		first.Receipts.EngineGeneration != 9 ||
		first.Receipts.SecurityEpoch != 1 {
		t.Fatalf("first AcceptCommand response = %#v", first)
	}

	second, secondAddress := runProductionCommandAcceptance(t, connection, config, request)
	if !second.Duplicate {
		t.Fatalf("restart AcceptCommand duplicate = false")
	}
	if second.CommandID != first.CommandID || second.Receipts != first.Receipts {
		t.Fatalf("restart response = %#v, first = %#v", second, first)
	}
	if firstAddress == "" || secondAddress == "" {
		t.Fatalf("runtime-api gRPC addresses = %q, %q", firstAddress, secondAddress)
	}

	var (
		storedOrgID, storedInstallationID, storedWorkloadID, storedCapabilityID string
		storedBinding                                                           string
		storedPayload                                                           []byte
		storedReceipts                                                          []byte
		storedState                                                             string
	)
	if err := db.QueryRow(`SELECT OrgId, AppInstallationId, CallerWorkloadId,
		       RuntimeCapabilityId, MetadataBinding, Payload, Receipts, State
		FROM _System_RuntimePlatformCommand
		WHERE CommandId = ? AND IdempotencyKey = ?`,
		request.CommandID, request.IdempotencyKey,
	).Scan(
		&storedOrgID, &storedInstallationID, &storedWorkloadID, &storedCapabilityID,
		&storedBinding, &storedPayload, &storedReceipts, &storedState,
	); err != nil {
		t.Fatal(err)
	}
	if storedOrgID != testOrgID || storedInstallationID != testInstallationID ||
		storedWorkloadID != testWorkloadID || storedCapabilityID != commandsemantic.CapabilityName {
		t.Fatalf("stored execution authority = %q/%q/%q/%q",
			storedOrgID, storedInstallationID, storedWorkloadID, storedCapabilityID)
	}
	binding, err := catalogsnapshot.OpenRuntimeWorkBinding(storedBinding)
	if err != nil {
		t.Fatal(err)
	}
	if err := binding.VerifyView(runtimeView, catalogsnapshot.ProjectionCommandAcceptanceV1); err != nil {
		t.Fatalf("stored execution binding: %v", err)
	}
	if !bytes.Equal(storedPayload, payload) {
		t.Fatalf("stored opaque payload = %x, want %x", storedPayload, payload)
	}
	if storedState != persistence.RuntimePlatformCommandAccepted {
		t.Fatalf("stored state = %q, want %q", storedState, persistence.RuntimePlatformCommandAccepted)
	}
	var receipts commandsemantic.Receipts
	if err := json.Unmarshal(storedReceipts, &receipts); err != nil {
		t.Fatal(err)
	}
	if receipts != first.Receipts {
		t.Fatalf("stored receipts = %#v, response receipts = %#v", receipts, first.Receipts)
	}
	var count int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_RuntimePlatformCommand`).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 1 {
		t.Fatalf("durable command rows after restart replay = %d, want 1", count)
	}
}

func TestRuntimeAPIProductionV2CommandAcceptanceUsesExactCatalogAuthority(
	t *testing.T,
) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIPlatformCommandTable(t, db)

	runtimeView := commandFlowAcceptanceTestViewVersion(
		t,
		flowcommandsemanticv2.PayloadFormatVersion,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemanticv2.PayloadFormatVersion
	current.Receipts.SecurityEpoch = 1
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	commandConsumerAdmission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		&catalogConsumerViewTestEngine{integrationView: current},
	)
	if err != nil {
		t.Fatal(err)
	}
	connection := database.NewConnectionFromDB(db)
	config := ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-command-v2-real", commandsemantic.CapabilityName,
		),
		ServingAdmission:         &ServingAdmission{},
		CommandConsumerAdmission: commandConsumerAdmission,
		ServingEnabled:           true,
		TransportCredentials:     insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			commandsemantic.CapabilityName: time.Second,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}
	revision := uint64(17)
	payload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef:     "flow.integration-ingress",
		DefinitionVersion: "flow-version-1",
		CanonicalInputs:   []byte(`{"recordId":"record-1"}`),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "contacts", SourceID: "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID:            "command-production-v2",
		IdempotencyKey:       "request-production-v2",
		CommandKind:          flowcommandsemantic.CommandKind,
		PartitionKey:         "tenant-partition-1",
		PayloadFormatVersion: flowcommandsemanticv2.PayloadFormatVersion,
		Payload:              payload, PayloadDigest: commandsemantic.ComputePayloadDigest(payload),
		MinCatalogRevision: &revision,
	}
	response, _ := runProductionCommandAcceptance(t, connection, config, request)
	if response.Duplicate || response.CommandID != request.CommandID ||
		response.Receipts.CatalogRevision != revision {
		t.Fatalf("V2 AcceptCommand response = %#v", response)
	}
	var storedVersion uint32
	var storedPayload []byte
	if err := db.QueryRow(`SELECT PayloadFormatVersion, Payload
		FROM _System_RuntimePlatformCommand WHERE CommandId = ?`,
		request.CommandID,
	).Scan(&storedVersion, &storedPayload); err != nil {
		t.Fatal(err)
	}
	if storedVersion != flowcommandsemanticv2.PayloadFormatVersion ||
		!bytes.Equal(storedPayload, payload) {
		t.Fatalf("stored V2 command = version %d payload %x", storedVersion, storedPayload)
	}
}

func TestRuntimeAPIProductionV3CommandAcceptanceUsesExactCatalogAuthority(
	t *testing.T,
) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIPlatformCommandTable(t, db)

	runtimeView := commandFlowAcceptanceTestViewVersion(
		t,
		flowcommandsemanticv3.PayloadFormatVersion,
		[]catalogsnapshot.CommandExecutionBinding{{
			AuthorizedWorkloadID: testWorkloadID,
			DefinitionRef:        "flow.integration-ingress",
			DefinitionVersion:    "flow-version-1",
			ContextNamespace:     "connector.test",
		}},
	)
	current := commandConsumerIntegrationView(runtimeView)
	current.Bindings[0].RuntimeCommand.PayloadFormatVersion =
		flowcommandsemanticv3.PayloadFormatVersion
	current.Receipts.SecurityEpoch = 1
	current.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(current)
	commandConsumerAdmission, err := NewCommandConsumerAdmission(
		t.Context(),
		CommandConsumerSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, RuntimeMetadataView: runtimeView,
		},
		&catalogConsumerViewTestEngine{integrationView: current},
	)
	if err != nil {
		t.Fatal(err)
	}
	connection := database.NewConnectionFromDB(db)
	config := ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-command-v3-real", commandsemantic.CapabilityName,
		),
		ServingAdmission:         &ServingAdmission{},
		CommandConsumerAdmission: commandConsumerAdmission,
		ServingEnabled:           true,
		TransportCredentials:     insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			commandsemantic.CapabilityName: time.Second,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}
	revision := uint64(17)
	canonicalInputs := []byte(`{"recordId":"record-1"}`)
	inputDigest, err := flowcommandsemanticv3.ComputeInputDigest(
		canonicalInputs,
		flowcommandsemanticv3.SuppressExternalEffects,
	)
	if err != nil {
		t.Fatal(err)
	}
	payload, err := flowcommandsemanticv3.Encode(flowcommandsemanticv3.Payload{
		ExecutionRef: applicationresultsemantic.ExecutionRef{
			ExecutionID:        "connector-execution-production-v3",
			PackGenerationID:   "pack-generation-production-v3",
			ServingEpochDigest: "sha256:" + strings.Repeat("1", 64),
			DefinitionRef:      "flow.integration-ingress", DefinitionVersion: "flow-version-1",
			InputDigest: inputDigest,
		},
		CanonicalInputs:     canonicalInputs,
		ExternalEffectsMode: flowcommandsemanticv3.SuppressExternalEffects,
	})
	if err != nil {
		t.Fatal(err)
	}
	request := commandsemantic.Request{
		CommandID:            "command-production-v3",
		IdempotencyKey:       "request-production-v3",
		CommandKind:          flowcommandsemantic.CommandKind,
		PartitionKey:         "tenant-partition-1",
		PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
		Payload:              payload,
		PayloadDigest:        commandsemantic.ComputePayloadDigest(payload),
		MinCatalogRevision:   &revision,
	}
	response, _ := runProductionCommandAcceptance(t, connection, config, request)
	if response.Duplicate || response.CommandID != request.CommandID ||
		response.Receipts.CatalogRevision != revision {
		t.Fatalf("V3 AcceptCommand response = %#v", response)
	}
	var storedVersion uint32
	var storedPayload []byte
	if err := db.QueryRow(`SELECT PayloadFormatVersion, Payload
		FROM _System_RuntimePlatformCommand WHERE CommandId = ?`,
		request.CommandID,
	).Scan(&storedVersion, &storedPayload); err != nil {
		t.Fatal(err)
	}
	if storedVersion != flowcommandsemanticv3.PayloadFormatVersion ||
		!bytes.Equal(storedPayload, payload) {
		t.Fatalf("stored V3 command = version %d payload %x", storedVersion, storedPayload)
	}
}

func TestRuntimeAPIProductionFlowExecutionCrossProcessRestart(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPIFlowAuthorizationFixture(t, db)

	snapshotArtifact, viewArtifact, mutationArtifact := flowServerTestArtifactsForMutation(t, true)
	runtimeView, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	mutationView, err := catalogsnapshot.OpenRuntimeMetadataView(mutationArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appView := flowServerTestAppView(runtimeView)
	servingAdmission, appView := newFlowMutationServingFixture(
		t,
		snapshotArtifact,
		viewArtifact,
		mutationArtifact,
		appView,
	)
	for _, statement := range []string{
		`CREATE TABLE WorkItem (
			Id VARCHAR(191) PRIMARY KEY,
			Status VARCHAR(191) NULL,
			IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
			LastModifiedById VARCHAR(191) NULL,
			LastModifiedDate DATETIME(6) NULL
		)`,
		`INSERT INTO WorkItem (Id, Status) VALUES ('a99000000000001', 'before')`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
	catalog := &catalogConsumerViewTestEngine{appViewV2: appView}
	securityAdmission := durableIntegrationSecurityAdmission(
		t,
		db,
		"replica-flow-real",
		catalogsemantic.AppMetadataV2CapabilityName,
		flowsemantic.CapabilityName,
	)
	resourceLimits := map[runtimegrpc.ResourceClass]int{}
	if class, ok := runtimegrpc.PlatformCapabilityResourceClass(
		flowsemantic.CapabilityName,
	); !ok {
		t.Fatal("Flow capability has no host-owned resource classification")
	} else if class != runtimegrpc.ResourceClassShortUnary {
		resourceLimits[class] = 1
	}
	config := ProcessConfig{
		Version:           "integration",
		AdminAddress:      "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: securityAdmission,
		ServingAdmission:  servingAdmission,
		ServingEnabled:    true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			flowsemantic.CapabilityName: 10 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: resourceLimits,
		CatalogConsumerViews:                 catalog,
		TransportCredentials:                 insecure.NewCredentials(),
		PermissionStateCapabilities:          []string{"permission_state.v1"},
		PermissionStateReady:                 func() bool { return true },
	}
	request := flowsemantic.Request{
		DefinitionRef: "AppOwnedAction", DefinitionVersion: "1",
		IdempotencyKey:  "flow-production-1",
		CanonicalInputs: []byte(`{"$Record":{"Id":"a99000000000001","Status":"before"}}`),
	}

	first, firstAddress := runProductionFlowExecution(
		t,
		database.NewConnectionFromDB(db),
		config,
		request,
	)
	second, secondAddress := runProductionFlowExecution(
		t,
		database.NewConnectionFromDB(db),
		config,
		request,
	)
	if first.ExecutionID == "" ||
		first.ExecutionID != second.ExecutionID ||
		!bytes.Equal(first.CanonicalOutputs, second.CanonicalOutputs) ||
		first.Receipts != second.Receipts {
		t.Fatalf("Flow restart replay = first %#v second %#v", first, second)
	}
	if first.Receipts.CapabilityVersion != flowsemantic.CapabilityName ||
		first.Receipts.CatalogRevision != 17 ||
		first.Receipts.SchemaGeneration != 8 ||
		first.Receipts.EngineGeneration != 9 ||
		first.Receipts.SecurityEpoch != 1 {
		t.Fatalf("Flow receipts = %#v", first.Receipts)
	}
	if firstAddress == "" || secondAddress == "" {
		t.Fatalf("runtime-api gRPC addresses = %q, %q", firstAddress, secondAddress)
	}
	var (
		rowCount       int
		completedCount int
		minResultBytes int
	)
	if err := db.QueryRow(`SELECT COUNT(*), SUM(State = 'COMPLETED'),
       MIN(OCTET_LENGTH(ResultPayload))
FROM _System_RuntimeFlowInvocation`).Scan(
		&rowCount,
		&completedCount,
		&minResultBytes,
	); err != nil {
		t.Fatal(err)
	}
	if rowCount != 1 ||
		completedCount != 1 ||
		minResultBytes == 0 {
		t.Fatalf(
			"durable Flow invocations = count %d completed %d min-result-bytes %d",
			rowCount,
			completedCount,
			minResultBytes,
		)
	}
	var (
		statusValue       string
		eventID           string
		eventCapabilityID string
		eventPayload      []byte
		eventCount        int
	)
	if err := db.QueryRow(`SELECT
(SELECT Status FROM WorkItem WHERE Id = 'a99000000000001'),
(SELECT Id FROM _System_RecordEventOutbox WHERE RecordKey = 'WorkItem/a99000000000001' ORDER BY CreatedDate LIMIT 1),
(SELECT RuntimeCapabilityId FROM _System_RecordEventOutbox WHERE RecordKey = 'WorkItem/a99000000000001' ORDER BY CreatedDate LIMIT 1),
(SELECT Payload FROM _System_RecordEventOutbox WHERE RecordKey = 'WorkItem/a99000000000001' ORDER BY CreatedDate LIMIT 1),
(SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE RecordKey = 'WorkItem/a99000000000001')`).Scan(
		&statusValue,
		&eventID,
		&eventCapabilityID,
		&eventPayload,
		&eventCount,
	); err != nil {
		t.Fatal(err)
	}
	if statusValue != "after" {
		t.Fatalf("Flow DML status = %q, want after", statusValue)
	}
	if eventID == "" {
		t.Fatal("Flow DML did not produce a durable record event")
	}
	if eventCount != 1 {
		t.Fatalf("Flow restart replay produced %d durable record events, want 1", eventCount)
	}
	if eventCapabilityID != mutationsemantic.CapabilityName {
		t.Fatalf(
			"Flow DML durable event capability = %q, want %q for the record-event Worker",
			eventCapabilityID,
			mutationsemantic.CapabilityName,
		)
	}
	var durableEvent eventfeature.InternalRecordEventEnvelope
	if err := json.Unmarshal(eventPayload, &durableEvent); err != nil {
		t.Fatalf("decode Flow DML durable event: %v", err)
	}
	bindingSet, err := catalogsnapshot.OpenRuntimeWorkBindingSet(
		durableEvent.InternalMetadata["runtime_work_metadata_binding"],
	)
	if err != nil {
		t.Fatalf("open Flow DML durable work binding: %v", err)
	}
	wantMutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		t.Fatal(err)
	}
	wantFlowBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(runtimeView)
	if err != nil {
		t.Fatal(err)
	}
	if bindingSet.Primary != wantMutationBinding {
		t.Fatalf("Flow DML primary binding = %#v, want mutation %#v", bindingSet.Primary, wantMutationBinding)
	}
	if bindingSet.FlowWork == nil || *bindingSet.FlowWork != wantFlowBinding {
		t.Fatalf("Flow DML auxiliary binding = %#v, want Flow %#v", bindingSet.FlowWork, wantFlowBinding)
	}
}

func TestRuntimeAPIProductionCatalogActionCrossProcessRestart(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPICatalogActionAuthorizationFixture(t, db)

	snapshotArtifact, viewArtifact, mutationArtifact := flowServerTestArtifactsForMutation(t, true)
	runtimeView, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	for _, statement := range []string{
		`CREATE TABLE WorkItem (
			Id VARCHAR(191) PRIMARY KEY,
			Status VARCHAR(191) NULL,
			IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
			LastModifiedById VARCHAR(191) NULL,
			LastModifiedDate DATETIME(6) NULL
		)`,
		`INSERT INTO WorkItem (Id, Status) VALUES ('a99000000000001', 'before')`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
	appView, actionComponent := flowServerTestAppViewWithCatalogAction(t, runtimeView, true)
	actionComponent.CanonicalBehavior = []byte(strings.Replace(
		string(actionComponent.CanonicalBehavior),
		`"apiName":"input","dataType":"string"`,
		`"apiName":"$Record","dataType":"sobject"`,
		1,
	))
	actionComponent.ContentDigest, err = appauthority.ComponentDigest(
		actionComponent.ComponentKind,
		actionComponent.ComponentFormatVersion,
		actionComponent.CanonicalBehavior,
	)
	if err != nil {
		t.Fatal(err)
	}
	for index := range appView.AppComponents {
		if appView.AppComponents[index].ComponentID == actionComponent.ComponentID {
			appView.AppComponents[index] = actionComponent
		}
	}
	servingAdmission, appView := newFlowMutationServingFixture(
		t, snapshotArtifact, viewArtifact, mutationArtifact, appView,
	)
	catalog := &catalogConsumerViewTestEngine{appViewV2: appView}
	securityAdmission := durableIntegrationSecurityAdmission(
		t,
		db,
		"replica-catalog-action-real",
		catalogsemantic.AppMetadataV2CapabilityName,
		flowsemantic.CapabilityName,
	)
	resourceLimits := map[runtimegrpc.ResourceClass]int{}
	if class, ok := runtimegrpc.PlatformCapabilityResourceClass(
		flowsemantic.CapabilityName,
	); !ok {
		t.Fatal("Flow capability has no host-owned resource classification")
	} else if class != runtimegrpc.ResourceClassShortUnary {
		resourceLimits[class] = 1
	}
	config := ProcessConfig{
		Version:           "integration",
		AdminAddress:      "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: securityAdmission,
		ServingAdmission:  servingAdmission,
		ServingEnabled:    true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			flowsemantic.CapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: resourceLimits,
		CatalogConsumerViews:                 catalog,
		TransportCredentials:                 insecure.NewCredentials(),
		PermissionStateCapabilities:          []string{"permission_state.v1"},
		PermissionStateReady:                 func() bool { return true },
	}
	request := flowsemantic.CatalogActionRequest{
		IdempotencyKey: "catalog-action-production-1",
		CanonicalInputs: []byte(
			`{"$Record":{"Id":"a99000000000001","Status":"before"}}`,
		),
		CatalogAction: flowsemantic.CatalogActionBinding{
			ComponentRef: actionComponent.ComponentID,
			ActionRef:    "runBounded",
		},
	}

	first, firstAddress := runProductionCatalogActionExecution(
		t, database.NewConnectionFromDB(db), config, request,
	)
	second, secondAddress := runProductionCatalogActionExecution(
		t, database.NewConnectionFromDB(db), config, request,
	)
	if first.ExecutionID == "" || first.ExecutionID != second.ExecutionID ||
		!bytes.Equal(first.CanonicalOutputs, second.CanonicalOutputs) ||
		first.Receipts != second.Receipts || firstAddress == "" || secondAddress == "" {
		t.Fatalf(
			"Catalog action restart replay = first %#v second %#v addresses %q/%q",
			first, second, firstAddress, secondAddress,
		)
	}
	var statusValue string
	var invocationCount, eventCount int
	if err := db.QueryRow(`SELECT
(SELECT Status FROM WorkItem WHERE Id = 'a99000000000001'),
(SELECT COUNT(*) FROM _System_RuntimeFlowInvocation),
(SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE RecordKey = 'WorkItem/a99000000000001')`).Scan(
		&statusValue,
		&invocationCount,
		&eventCount,
	); err != nil {
		t.Fatal(err)
	}
	if statusValue != "after" || invocationCount != 1 || eventCount != 1 {
		t.Fatalf(
			"Catalog action durable effect = status %q invocations %d events %d",
			statusValue, invocationCount, eventCount,
		)
	}
}

func TestRuntimeAPIProductionCatalogActionPreviewIsNoEffectAndRevisionBound(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFlowSubstrate(t, db)
	createRuntimeAPICatalogActionAuthorizationFixture(t, db)

	snapshotArtifact, viewArtifact, mutationArtifact := flowServerTestArtifactsForMutation(t, false)
	runtimeView, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appView, actionComponent := flowServerTestAppViewWithPreviewCatalogAction(t, runtimeView)
	servingAdmission, appView := newFlowMutationServingFixture(
		t, snapshotArtifact, viewArtifact, mutationArtifact, appView,
	)
	resourceLimits := map[runtimegrpc.ResourceClass]int{}
	if class, ok := runtimegrpc.PlatformCapabilityResourceClass(
		flowsemantic.CapabilityName,
	); !ok {
		t.Fatal("Flow capability has no host-owned resource classification")
	} else if class != runtimegrpc.ResourceClassShortUnary {
		resourceLimits[class] = 1
	}
	config := ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(
			t, db, "replica-catalog-action-preview-real",
			catalogsemantic.AppMetadataV2CapabilityName, flowsemantic.CapabilityName,
		),
		ServingAdmission: servingAdmission,
		ServingEnabled:   true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			flowsemantic.CapabilityName: 5 * time.Second,
		},
		PlatformResourceClassAdmissionLimits: resourceLimits,
		CatalogConsumerViews:                 &catalogConsumerViewTestEngine{appViewV2: appView},
		TransportCredentials:                 insecure.NewCredentials(),
		PermissionStateCapabilities:          []string{"permission_state.v1"},
		PermissionStateReady:                 func() bool { return true },
	}
	revision := uint64(17)
	requests := []flowsemantic.CatalogActionPreviewRequest{
		{
			Mode:               flowsemantic.CatalogActionPreviewCreateRecords,
			CanonicalInputs:    []byte(`{"recordId":"a99000000000001"}`),
			MinCatalogRevision: &revision,
			CatalogAction: flowsemantic.CatalogActionBinding{
				ComponentRef: actionComponent.ComponentID, ActionRef: "runBounded",
			},
		},
		{
			Mode:     flowsemantic.CatalogActionPreviewOutputs,
			RecordID: "a99000000000001", MinCatalogRevision: &revision,
			CatalogAction: flowsemantic.CatalogActionBinding{
				ComponentRef: actionComponent.ComponentID, ActionRef: "runBounded",
			},
		},
	}
	responses, address := runProductionCatalogActionPreviews(
		t, database.NewConnectionFromDB(db), config, requests,
	)
	if address == "" || len(responses) != len(requests) {
		t.Fatalf("preview endpoint/responses = %q/%d", address, len(responses))
	}
	for index, response := range responses {
		if response.Receipts.CatalogRevision != revision ||
			response.Receipts.SchemaGeneration == 0 ||
			response.Receipts.EngineGeneration == 0 ||
			response.Receipts.SecurityEpoch == 0 {
			t.Fatalf("preview %d receipts = %#v", index, response.Receipts)
		}
		var result map[string]any
		if err := json.Unmarshal(response.CanonicalResult, &result); err != nil ||
			result["isSuccess"] != true {
			t.Fatalf("preview %d result = %s, error %v", index, response.CanonicalResult, err)
		}
	}
	var invocations int
	if err := db.QueryRow(`SELECT COUNT(*) FROM _System_RuntimeFlowInvocation`).Scan(
		&invocations,
	); err != nil {
		t.Fatal(err)
	}
	if invocations != 0 {
		t.Fatalf("no-effect previews created %d durable invocation rows", invocations)
	}
}

func newFlowMutationServingFixture(
	t *testing.T,
	snapshotArtifact catalogsnapshot.Artifact,
	flowArtifact catalogsnapshot.Artifact,
	mutationArtifact catalogsnapshot.Artifact,
	appView catalogsemantic.AppMetadataViewV2,
) (*ServingAdmission, catalogsemantic.AppMetadataViewV2) {
	t.Helper()
	appReleaseDigest := catalogsnapshot.ComputeDigest([]byte("app-release"))
	serviceReleaseDigest := catalogsnapshot.ComputeDigest([]byte("service-release"))
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appMetadataArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: testInstallationID,
			CapabilityID:      catalogsemantic.AppMetadataV2CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	type capability struct {
		appID   string
		runtime string
		view    catalogsnapshot.Artifact
	}
	capabilities := []capability{
		{
			appID:   "io.nexuscrm.sfcrm.flow-execution.v1",
			runtime: flowsemantic.CapabilityName,
			view:    flowArtifact,
		},
		{
			appID:   "io.nexuscrm.sfcrm.records-mutation.v1",
			runtime: mutationsemantic.CapabilityName,
			view:    mutationArtifact,
		},
		{
			appID:   "io.nexuscrm.sfcrm.app-metadata.v2",
			runtime: catalogsemantic.AppMetadataV2CapabilityName,
			view:    appMetadataArtifact,
		},
	}
	installation := ServingInstallation{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		ActiveReleaseDigest: string(appReleaseDigest), CatalogRevision: 17,
	}
	bindings := make(map[string]ServingHandlerBinding, len(capabilities))
	selections := make([]ServingSelection, 0, len(capabilities))
	for _, entry := range capabilities {
		if err := store.Put(t.Context(), entry.view); err != nil {
			t.Fatal(err)
		}
		receipt := servingReceipt(
			t,
			snapshotArtifact,
			entry.view,
			17,
			appReleaseDigest,
			serviceReleaseDigest,
			3,
			entry.appID,
			entry.runtime,
		)
		if err := store.Put(t.Context(), receipt); err != nil {
			t.Fatal(err)
		}
		bindings[entry.appID] = ServingHandlerBinding{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			AppReleaseDigest: string(appReleaseDigest), CapabilityID: entry.appID,
			WorkloadID: testWorkloadID, ServiceReleaseDigest: string(serviceReleaseDigest),
			OwnershipEpoch: 3, CompatibilityReceiptDigest: string(receipt.Digest()),
		}
		selections = append(selections, ServingSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			ExpectedWorkloadID: testWorkloadID,
			AppCapabilityID:    entry.appID, RuntimeCapabilityID: entry.runtime,
			ExpectedServiceReleaseDigest:       serviceReleaseDigest,
			ExpectedOwnershipEpoch:             3,
			ExpectedCompatibilityReceiptDigest: receipt.Digest(),
			LocalCompatibility: catalogsnapshot.Compatibility{
				SchemaGeneration: 8, EngineGeneration: 9,
			},
			SecurityEpoch: 1,
		})
	}
	appView.AppReleaseDigest = string(appReleaseDigest)
	appView.Receipts.SecurityEpoch = 1
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	state, err := NewServingState(ServingStateConfig{
		Store: store,
		Catalog: func(
			_ context.Context,
			installationID string,
			capabilityID string,
		) (ServingInstallation, ServingHandlerBinding, bool, error) {
			binding, ok := bindings[capabilityID]
			if installationID != testInstallationID || !ok {
				return ServingInstallation{}, ServingHandlerBinding{}, false, nil
			}
			return installation, binding, true, nil
		},
		AppMetadata: &flowServerTestCatalog{view: appView, requireDeadline: true},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := state.ReconcileAll(t.Context(), selections); err != nil {
		t.Fatal(err)
	}
	admission, err := NewServingAdmission(state)
	if err != nil {
		t.Fatal(err)
	}
	return admission, appView
}

func TestSQLMailboxAuthorizationHandleStoreMintsWithTiDBDatabaseClock(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIMailboxAuthorizationTable(t, db)
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	store, err := NewSQLMailboxAuthorizationHandleStore(
		db,
		&runtimeMailboxAuthorizationTransactionOwner{transactions: transactions},
	)
	if err != nil {
		t.Fatal(err)
	}
	record := mailboxAuthorizationHandle{
		HandleDigest: digestMailboxAuthorizationHandle("opaque-callback-handle"),
		OrgID:        "org-1", AppInstallationID: "installation-1",
		AuthorizedSubjectID: "user-1", SessionID: "session-1", AuthEpoch: 7,
		PermissionFloor: mailboxPermissionFloor(t, 11),
		CatalogRevision: 41, SecurityEpoch: 9,
	}
	if err := store.Mint(t.Context(), record, mailboxAuthorizationTTL); err != nil {
		t.Fatalf("Mint() against TiDB: %v", err)
	}

	var persisted mailboxAuthorizationHandle
	duplicate, err := store.ConsumeAuthorized(
		t.Context(),
		record.HandleDigest,
		func(_ context.Context, current mailboxAuthorizationHandle) error {
			persisted = current
			return nil
		},
	)
	if err != nil || duplicate {
		t.Fatalf("ConsumeAuthorized() after Mint() = duplicate:%t err:%v", duplicate, err)
	}
	if persisted.HandleDigest != record.HandleDigest ||
		persisted.OrgID != record.OrgID ||
		persisted.AppInstallationID != record.AppInstallationID ||
		persisted.AuthorizedSubjectID != record.AuthorizedSubjectID ||
		persisted.SessionID != record.SessionID ||
		persisted.AuthEpoch != record.AuthEpoch ||
		persisted.CatalogRevision != record.CatalogRevision ||
		persisted.SecurityEpoch != record.SecurityEpoch {
		t.Fatalf("persisted mailbox authorization binding = %#v", persisted)
	}
	if comparison, compareErr := persisted.PermissionFloor.Compare(record.PermissionFloor); compareErr != nil || comparison != 0 {
		t.Fatalf("persisted permission floor comparison = %d, %v", comparison, compareErr)
	}
	if got := persisted.ExpiresAt.Sub(persisted.CreatedAt); got != mailboxAuthorizationTTL {
		t.Fatalf("database-clock expiry interval = %s, want %s", got, mailboxAuthorizationTTL)
	}
}

func TestRuntimeAPIProductionCompositionBuildsPlatformAdapters(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPIFileUploadOperationTable(t, db)
	createRuntimeAPIMailboxAuthorizationTable(t, db)
	createRuntimeAPIFileAuthorizationFixture(t, db)
	fileStorageRoot := t.TempDir()
	t.Setenv("FILE_STORAGE_PROVIDER", "filesystem")
	t.Setenv("FILE_STORAGE_ROOT", fileStorageRoot)
	t.Setenv("FILE_STORAGE_LEGACY_UPLOAD_ROOT", t.TempDir())

	snapshotArtifact, viewArtifact := servingArtifactsForObjects(
		t, 17, filessemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordMutationV1,
		runtimeAPIFileProductionObjects(),
	)
	fixture := newServingFixtureForArtifacts(
		t, 17, "io.nexuscrm.integration.search.v1",
		filessemantic.CapabilityName, snapshotArtifact, viewArtifact,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	mailboxView := validCatalogIntegrationBindingView(17)
	mailboxView.Receipts.SecurityEpoch = 1
	mailboxView.Lineage.ViewDigest =
		catalogsemantic.ComputeIntegrationBindingViewDigest(mailboxView)
	mailboxAdmission, err := NewMailboxAuthorizationAdmission(
		t.Context(),
		MailboxAuthorizationSelection{
			OrgID: mailboxView.OrgID, AppInstallationID: mailboxView.AppInstallationID,
			WorkloadID: mailboxView.AuthorizedWorkloadID,
		},
		&catalogConsumerViewTestEngine{integrationView: mailboxView},
	)
	if err != nil {
		t.Fatal(err)
	}
	securityAdmission := durableIntegrationSecurityAdmission(
		t, db, "replica-platform-adapters",
		reportsemantic.CapabilityName, mailboxauthorizationsemantic.CapabilityName,
		filessemantic.CapabilityName, searchsemantic.CapabilityName,
	)
	process, err := AssembleProductionWithConnection(t.Context(), ProcessConfig{
		Version:           "integration",
		AdminAddress:      "127.0.0.1:0",
		GRPCAddress:       "127.0.0.1:0",
		AdmissionLimit:    DefaultAdmissionLimit,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: securityAdmission,
		ServingAdmission:  fixture.admission,
		MailboxAuthorization: &MailboxAuthorizationProcessConfig{
			Admission: mailboxAdmission, RequiredPermission: MailboxPermissionRequirement{CustomPermission: "SendRecordEmail"},
		},
		ServingEnabled:       true,
		TransportCredentials: insecure.NewCredentials(),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			searchsemantic.CapabilityName:               time.Second,
			reportsemantic.CapabilityName:               time.Second,
			filessemantic.CapabilityName:                time.Second,
			mailboxauthorizationsemantic.CapabilityName: time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassHeavyWork:  1,
			runtimegrpc.ResourceClassFileStream: 1,
		},
		PermissionStateCapabilities: []string{"permission_state.v1"},
		PermissionStateReady:        func() bool { return true },
	}, database.NewConnectionFromDB(db))
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	if process.GRPCAddress() == "" {
		t.Fatal("production Platform adapters did not start on the Runtime gRPC endpoint")
	}
	connection, err := grpc.NewClient(
		"passthrough:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(
			context.Background(), filessemantic.CapabilityName,
		),
		time.Second,
	)
	defer cancel()
	filesClient := runtimev1.NewAuthorizedFilesServiceClient(connection)
	stream, err := filesClient.CreateFile(ctx)
	if err != nil {
		t.Fatal(err)
	}
	header, err := mapping.ToWireCreateFileChunk(
		filessemantic.CreateChunk{Header: &filessemantic.CreateHeader{
			IdempotencyKey: "production-files-atomic-1",
			ParentObjectID: "Widget", ParentRecordID: "widget-1",
			FileName: "proof.txt", ContentType: "text/plain",
			ContentLength: 5,
		}},
	)
	if err != nil {
		t.Fatal(err)
	}
	data, err := mapping.ToWireCreateFileChunk(
		filessemantic.CreateChunk{Data: []byte("proof")},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := stream.Send(header); err != nil {
		t.Fatal(err)
	}
	if err := stream.Send(data); err != nil {
		t.Fatal(err)
	}
	wireResponse, err := stream.CloseAndRecv()
	if err != nil {
		var journalID, journalState, errorCode string
		diagnosticErr := db.QueryRow(`SELECT Id, State, COALESCE(ErrorCode, '')
FROM _System_FileUploadOperation
ORDER BY CreatedDate DESC LIMIT 1`).Scan(
			&journalID, &journalState, &errorCode,
		)
		t.Fatalf(
			"production Files RPC = %v; journal id=%q state=%q code=%q diagnostic=%v",
			err, journalID, journalState, errorCode, diagnosticErr,
		)
	}
	response, err := mapping.FromWireCreateFileResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if response.FileID == "" || response.FileName != "proof.txt" ||
		response.Size != 5 {
		t.Fatalf("production Files response = %#v", response)
	}
	var journalState string
	var metadataCount int
	if err := db.QueryRow(`SELECT State
FROM _System_FileUploadOperation WHERE SystemFileId = ?`,
		response.FileID,
	).Scan(&journalState); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM _System_File WHERE Id = ?`,
		response.FileID,
	).Scan(&metadataCount); err != nil {
		t.Fatal(err)
	}
	if journalState != persistence.FileUploadOperationCompleted ||
		metadataCount != 1 {
		t.Fatalf(
			"production Files atomic facts = journal %q metadata %d",
			journalState, metadataCount,
		)
	}
	var storageKey string
	if err := db.QueryRow(
		`SELECT StorageKey FROM _System_File WHERE Id = ?`,
		response.FileID,
	).Scan(&storageKey); err != nil {
		t.Fatal(err)
	}
	if _, err := os.Stat(filepath.Join(fileStorageRoot, storageKey)); err != nil {
		t.Fatalf("committed object is unavailable: %v", err)
	}

	openRequest, err := mapping.ToWireOpenFileRequest(
		ctx, filessemantic.OpenRequest{FileID: response.FileID},
	)
	if err != nil {
		t.Fatal(err)
	}
	openStream, err := filesClient.OpenFile(ctx, openRequest)
	if err != nil {
		t.Fatal(err)
	}
	var opened bytes.Buffer
	for {
		wireChunk, recvErr := openStream.Recv()
		if errors.Is(recvErr, io.EOF) {
			break
		}
		if recvErr != nil {
			t.Fatal(recvErr)
		}
		chunk, mapErr := mapping.FromWireOpenFileChunk(wireChunk)
		if mapErr != nil {
			t.Fatal(mapErr)
		}
		opened.Write(chunk.Data)
	}
	if opened.String() != "proof" {
		t.Fatalf("opened file body = %q, want proof", opened.String())
	}

	deleteRequest, err := mapping.ToWireDeleteFileRequest(
		ctx, filessemantic.DeleteRequest{FileID: response.FileID},
	)
	if err != nil {
		t.Fatal(err)
	}
	deleteResponse, err := filesClient.DeleteFile(ctx, deleteRequest)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := mapping.FromWireDeleteFileResponse(deleteResponse); err != nil {
		t.Fatal(err)
	}
	var deletedStatus string
	if err := db.QueryRow(
		`SELECT Status FROM _System_File WHERE Id = ?`,
		response.FileID,
	).Scan(&deletedStatus); err != nil {
		t.Fatal(err)
	}
	if deletedStatus != "Deleted" {
		t.Fatalf("deleted file status = %q, want Deleted", deletedStatus)
	}
	if _, err := os.Stat(filepath.Join(fileStorageRoot, storageKey)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("deleted object still exists or returned unexpected error: %v", err)
	}
}

func TestRuntimeAPIProductionSearchScopeOrderUsesTiDBAuthority(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	createRuntimeAPISearchScopeAuthorizationFixture(t, db)

	objects := runtimeAPISearchScopeObjects()
	snapshotArtifact, viewArtifact := servingArtifactsForObjects(
		t, 17, searchsemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordQueryV1,
		objects,
	)
	fixture := newServingFixtureForArtifacts(
		t, 17, "io.nexuscrm.integration.search-scope.v1",
		searchsemantic.CapabilityName, snapshotArtifact, viewArtifact,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	securityAdmission := durableIntegrationSecurityAdmission(
		t, db, "replica-search-scope", searchsemantic.CapabilityName,
	)
	process, err := AssembleProductionWithConnection(
		t.Context(),
		ProcessConfig{
			Version:              "integration",
			AdminAddress:         "127.0.0.1:0",
			GRPCAddress:          "127.0.0.1:0",
			AdmissionLimit:       DefaultAdmissionLimit,
			WorkloadAuthority:    testWorkloadAuthority,
			SecurityAdmission:    securityAdmission,
			ServingAdmission:     fixture.admission,
			ServingEnabled:       true,
			TransportCredentials: insecure.NewCredentials(),
			PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
				searchsemantic.CapabilityName: time.Second,
			},
			PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
				runtimegrpc.ResourceClassHeavyWork: 1,
			},
			PermissionStateCapabilities: []string{"permission_state.v1"},
			PermissionStateReady:        func() bool { return true },
		},
		database.NewConnectionFromDB(db),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)

	connection, err := grpc.NewClient(
		"passthrough:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(
			context.Background(), searchsemantic.CapabilityName,
		),
		time.Second,
	)
	defer cancel()
	revision := uint64(17)
	request := searchsemantic.ScopeOrderRequest{
		ObjectIDs:          []string{"Widget", "Gadget", "Hidden", "ReadGated", "PolicyGated"},
		MinCatalogRevision: &revision,
	}
	wireRequest, err := mapping.ToWireSearchScopeOrderRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRecordSearchServiceClient(
		connection,
	).ReadSearchScopeOrder(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireSearchScopeOrderResponse(
		request, wireResponse,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !slices.Equal(response.ObjectIDs, []string{"Gadget", "Widget"}) ||
		response.Receipts.CatalogRevision != 17 ||
		response.Receipts.SecurityEpoch != 1 {
		t.Fatalf("production Search scope response = %#v", response)
	}
}

func runProductionCommandAcceptance(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	request commandsemantic.Request,
) (commandsemantic.Response, string) {
	t.Helper()
	process, err := AssembleProductionWithConnection(t.Context(), config, connection)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	address := process.GRPCAddress()
	defer stopProcess(t, process)

	grpcConnection, err := grpc.NewClient(
		"passthrough:///"+address,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer grpcConnection.Close()

	ctx, cancel := context.WithTimeout(commandProtocolContext(context.Background()), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireAcceptCommandRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewCommandAcceptanceServiceClient(grpcConnection).AcceptCommand(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireAcceptCommandResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}

	// The existing Records service and the new coarse Command service are
	// registered on this same gRPC listener. A command-authorized call is
	// rejected by Records admission, but must not be routed to an unimplemented
	// second endpoint.
	_, recordsErr := runtimev1.NewRecordsServiceClient(grpcConnection).ReadSelectedRoot(
		ctx,
		&runtimev1.ReadSelectedRootRequest{
			ObjectId: "WorkItem", RecordId: "not-called",
			SelectedRootFields: []string{"Id"},
		},
	)
	if recordsErr == nil || status.Code(recordsErr) == codes.Unimplemented {
		t.Fatalf("same-endpoint Records probe = %v, want registered service rejection", recordsErr)
	}
	return response, address
}

func runProductionFlowExecution(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	request flowsemantic.Request,
) (flowsemantic.Response, string) {
	t.Helper()
	ctx, cancel := context.WithTimeout(
		flowProtocolContext(context.Background()),
		10*time.Second,
	)
	defer cancel()
	wireRequest, err := mapping.ToWireExecuteFlowRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	return runProductionFlowWireExecution(t, connection, config, ctx, wireRequest)
}

func runProductionCatalogActionExecution(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	request flowsemantic.CatalogActionRequest,
) (flowsemantic.Response, string) {
	t.Helper()
	ctx, cancel := context.WithTimeout(
		flowProtocolContext(context.Background()), 4*time.Second,
	)
	defer cancel()
	wireRequest, err := mapping.ToWireExecuteCatalogActionRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	return runProductionFlowWireExecution(t, connection, config, ctx, wireRequest)
}

func runProductionCatalogActionPreviews(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	requests []flowsemantic.CatalogActionPreviewRequest,
) ([]flowsemantic.CatalogActionPreviewResponse, string) {
	t.Helper()
	process, err := AssembleProductionWithConnection(t.Context(), config, connection)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	address := process.GRPCAddress()
	defer stopProcess(t, process)
	grpcConnection := openRuntimeAPIIntegrationGRPC(t, address)
	defer grpcConnection.Close()

	responses := make([]flowsemantic.CatalogActionPreviewResponse, 0, len(requests))
	for _, request := range requests {
		ctx, cancel := context.WithTimeout(
			flowProtocolContext(context.Background()), 4*time.Second,
		)
		wireRequest, err := mapping.ToWirePreviewCatalogActionRequest(ctx, request)
		if err != nil {
			cancel()
			t.Fatal(err)
		}
		wireResponse, err := runtimev1.NewFlowExecutionServiceClient(
			grpcConnection,
		).PreviewCatalogAction(ctx, wireRequest)
		cancel()
		if err != nil {
			t.Fatal(err)
		}
		response, err := mapping.FromWirePreviewCatalogActionResponse(request, wireResponse)
		if err != nil {
			t.Fatal(err)
		}
		responses = append(responses, response)
	}
	return responses, address
}

func runProductionFlowWireExecution(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	ctx context.Context,
	wireRequest *runtimev1.ExecuteFlowRequest,
) (flowsemantic.Response, string) {
	t.Helper()
	process, err := AssembleProductionWithConnection(t.Context(), config, connection)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	address := process.GRPCAddress()
	defer stopProcess(t, process)

	grpcConnection, err := grpc.NewClient(
		"passthrough:///"+address,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer grpcConnection.Close()

	wireResponse, err := runtimev1.NewFlowExecutionServiceClient(
		grpcConnection,
	).ExecuteFlow(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireExecuteFlowResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}

	_, recordsErr := runtimev1.NewRecordsServiceClient(
		grpcConnection,
	).ReadSelectedRoot(
		ctx,
		&runtimev1.ReadSelectedRootRequest{
			ObjectId: "WorkItem", RecordId: "not-called",
			SelectedRootFields: []string{"Id"},
		},
	)
	if recordsErr == nil || status.Code(recordsErr) == codes.Unimplemented {
		t.Fatalf(
			"same-endpoint Records probe = %v, want registered service rejection",
			recordsErr,
		)
	}
	return response, address
}

func runProductionApprovalExecution(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	request approvalsemantic.ExecuteRequest,
) (approvalsemantic.ExecuteResponse, string) {
	t.Helper()
	process, err := AssembleProductionWithConnection(t.Context(), config, connection)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	address := process.GRPCAddress()
	defer stopProcess(t, process)
	grpcConnection := openRuntimeAPIIntegrationGRPC(t, address)
	defer grpcConnection.Close()
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), approvalsemantic.ExecuteCapabilityName),
		4*time.Second,
	)
	defer cancel()
	wireRequest, err := mapping.ToWireExecuteApprovalRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewApprovalServiceClient(grpcConnection).ExecuteApproval(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ExecuteApproval: %v (semantic: %v)", err, mapping.FromPlatformStatusError(err, approvalsemantic.ExecuteCapabilityName))
	}
	response, err := mapping.FromWireExecuteApprovalResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	return response, address
}

func executeApprovalIntegrationRequest(
	t *testing.T,
	client runtimev1.ApprovalServiceClient,
	principalID string,
	sessionID string,
	request approvalsemantic.ExecuteRequest,
) approvalsemantic.ExecuteResponse {
	t.Helper()
	ctx, cancel := context.WithTimeout(
		approvalProtocolContextForPrincipal(context.Background(), approvalsemantic.ExecuteCapabilityName, principalID, sessionID),
		4*time.Second,
	)
	defer cancel()
	wireRequest, err := mapping.ToWireExecuteApprovalRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := client.ExecuteApproval(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ExecuteApproval: %v (semantic: %v)", err, mapping.FromPlatformStatusError(err, approvalsemantic.ExecuteCapabilityName))
	}
	response, err := mapping.FromWireExecuteApprovalResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	return response
}

func approvalProtocolContextForPrincipal(ctx context.Context, capability, principalID, sessionID string) context.Context {
	ctx = platformWorkloadProtocolContext(ctx, capability)
	claims := testClaimsFor(time.Now(), capability)
	claims.DelegationID = "delegation-" + principalID
	claims.PrincipalID = principalID
	claims.SessionID = sessionID
	return withTestDelegation(ctx, mustSignTestDelegation(claims))
}

func runProductionApprovalRead(
	t *testing.T,
	connection *database.TiDBConnection,
	config ProcessConfig,
	request approvalsemantic.ReadRequest,
) approvalsemantic.ReadResponse {
	t.Helper()
	process, err := AssembleProductionWithConnection(t.Context(), config, connection)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	grpcConnection := openRuntimeAPIIntegrationGRPC(t, process.GRPCAddress())
	defer grpcConnection.Close()
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), approvalsemantic.ReadCapabilityName),
		4*time.Second,
	)
	defer cancel()
	wireRequest, err := mapping.ToWireReadApprovalRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewApprovalServiceClient(grpcConnection).ReadApproval(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ReadApproval: %v (semantic: %v)", err, mapping.FromPlatformStatusError(err, approvalsemantic.ReadCapabilityName))
	}
	response, err := mapping.FromWireReadApprovalResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	return response
}

func openRuntimeAPIIntegrationGRPC(t *testing.T, address string) *grpc.ClientConn {
	t.Helper()
	connection, err := grpc.NewClient(
		"passthrough:///"+address,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	return connection
}

func commandProtocolContext(ctx context.Context) context.Context {
	return metadata.AppendToOutgoingContext(
		ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(commandsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(commandsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, commandsemantic.CapabilityName,
	)
}

func flowProtocolContext(ctx context.Context) context.Context {
	ctx = metadata.AppendToOutgoingContext(
		ctx,
		envelope.ProtocolMajorMetadataKey,
		strconv.FormatUint(uint64(flowsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey,
		strconv.FormatUint(uint64(flowsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey,
		flowsemantic.CapabilityName,
	)
	claims := testClaimsFor(time.Now(), flowsemantic.CapabilityName)
	claims.Capabilities = []string{
		catalogsemantic.AppMetadataV2CapabilityName,
		flowsemantic.CapabilityName,
	}
	return withTestDelegation(ctx, mustSignTestDelegation(claims))
}

func createRuntimeAPIPlatformCommandTable(t *testing.T, db *sql.DB) {
	t.Helper()
	const tableName = "_System_RuntimePlatformCommand"
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, definition := range definitions {
		if definition.TableName != tableName {
			continue
		}
		if err := persistence.NewSchemaRepository(db, nil).CreatePhysicalTable(
			t.Context(),
			definition,
		); err != nil {
			t.Fatalf("create canonical Runtime Platform command table: %v", err)
		}
		return
	}
	t.Fatalf("canonical schema is missing %s", tableName)
}

func createRuntimeAPIOutboundIntentTable(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		"_System_RuntimeOutboundIntent":           false,
		"_System_RuntimeOutboundIntentGeneration": false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create canonical Runtime outbound intent table %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for tableName, found := range wanted {
		if !found {
			t.Fatalf("canonical schema is missing %s", tableName)
		}
	}
}

func outboundIntentProtocolContext(ctx context.Context) context.Context {
	return metadata.AppendToOutgoingContext(
		ctx,
		envelope.ProtocolMajorMetadataKey,
		strconv.FormatUint(uint64(outboundintentsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey,
		strconv.FormatUint(uint64(outboundintentsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey,
		outboundintentsemantic.CapabilityName,
	)
}

func createRuntimeAPIFlowSubstrate(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		"_System_AuditLog":                      false,
		"_System_AuditOutbox":                   false,
		"_System_FieldHistorySequence":          false,
		"_System_RecordEventOutbox":             false,
		"_System_RollupDelta":                   false,
		"_System_RuntimeRecordCommand":          false,
		"_System_RuntimeRecordCommandOperation": false,
		"_System_RuntimeFlowInvocation":         false,
		persistence.RuntimeWorkOwnershipTable:   false,
		"UserRole":                              false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(
			t.Context(),
			definition,
		); err != nil {
			t.Fatalf("create canonical %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
	for index, family := range []string{
		audit.AuditHistoryFanoutWorkFamily,
		eventfeature.RuntimeRecordEventDispatchWorkFamily,
		records.RuntimeRollupDeltaWorkFamily,
		persistence.RuntimeRecordCommandLifecycleWorkFamily,
	} {
		if _, err := db.Exec(`INSERT INTO _System_RuntimeWorkOwnership
(WorkFamily, DeploymentId, WorkerReleaseDigest, OwnershipEpoch, IsActive, OperationId)
VALUES (?, ?, ?, 1, 1, ?)`,
			family,
			"runtime-api-integration",
			"sha256:"+strings.Repeat("a", 64),
			fmt.Sprintf("runtime-api-integration-%d", index),
		); err != nil {
			t.Fatalf("activate work family %s: %v", family, err)
		}
	}
}

func createRuntimeAPIApprovalNotificationFlowSubstrate(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		constants.TableFlowDefinition: false,
		constants.TableFlow:           false,
		constants.TableFlowInterview:  false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create canonical %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
	if _, err := db.Exec(`INSERT INTO FlowDefinition (Id, DeveloperName, MasterLabel) VALUES (?, ?, ?)`,
		approvalNotificationFlowDefinitionID,
		approvalNotificationFlowAPIName,
		"Approval notification probe",
	); err != nil {
		t.Fatalf("seed Approval notification FlowDefinition: %v", err)
	}
	if _, err := db.Exec(`INSERT INTO Flow (Id, DefinitionId) VALUES (?, ?)`,
		approvalNotificationFlowVersionID,
		approvalNotificationFlowDefinitionID,
	); err != nil {
		t.Fatalf("seed Approval notification Flow version: %v", err)
	}
}

func createRuntimeAPIApprovalSubstrate(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		"_System_ParentLane":                   false,
		"_System_ApprovalActionLog":            false,
		constants.TableProcessDefinition:       false,
		constants.TableProcessNode:             false,
		constants.TableProcessAction:           false,
		constants.TableProcessInstance:         false,
		constants.TableProcessInstanceHistory:  false,
		constants.TableProcessInstanceNode:     false,
		constants.TableProcessInstanceWorkitem: false,
		"_System_RecycleBin":                   false,
		"_System_ApprovalActionOutbox":         false,
		"_System_ApprovalSubmitLineage":        false,
		"_System_RuntimeApprovalInvocation":    false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create canonical %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
}

func createRuntimeAPIApprovalAuthorizationFixture(t *testing.T, db *sql.DB) {
	t.Helper()
	createRuntimeAPIFileAuthorizationFixture(t, db)
	statements := []struct {
		query string
		args  []any
	}{
		{
			query: `CREATE TABLE _System_RecordShare (
				Id VARCHAR(255) PRIMARY KEY,
				ObjectApiName VARCHAR(255) NOT NULL,
				RecordId VARCHAR(255) NOT NULL,
				ShareWithUserId VARCHAR(255) NOT NULL,
				AccessLevel VARCHAR(32) NOT NULL,
				Reason VARCHAR(64) NOT NULL,
				IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
				CreatedById VARCHAR(255) NULL,
				LastModifiedById VARCHAR(255) NULL,
				CreatedDate DATETIME(6) NOT NULL,
				LastModifiedDate DATETIME(6) NOT NULL
			)`,
		},
		{
			query: `CREATE TABLE WorkItem (
				Id VARCHAR(191) PRIMARY KEY,
				Status VARCHAR(191) NULL,
				FlowMarker VARCHAR(191) NULL,
				IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
				LastModifiedById VARCHAR(191) NULL,
				LastModifiedDate DATETIME(6) NULL
			)`,
		},
		{
			query: `INSERT INTO WorkItem (Id, Status) VALUES (?, 'before')`,
			args:  []any{runtimeApprovalTestRecordID},
		},
		{
			query: `INSERT INTO ProcessDefinition
				(Id, Name, DeveloperName, Description, State, TableEnumOrId, LockType,
				 EntryFilter, FinalApprovalRecordLock, FinalRejectionRecordLock,
				 AllowRecall, IsActive, CreatedDate, LastModifiedDate, Metadata,
				 ProcessOrder, OwnerPack)
			VALUES (?, 'conflicting live label', 'PrimaryReview', 'conflicting live description',
				'Active', 'WorkItem', 'Unlock', 'FALSE', 0, 0, 1, 1, NOW(6), NOW(6),
				'{"source":"live-conflict"}', 999, '')`,
			args: []any{approvalTestStorageID},
		},
		{
			query: `INSERT INTO ProcessNode
				(Id, Name, DeveloperName, ProcessDefinitionId, Description, IsDeleted,
				 CreatedDate, LastModifiedDate, ApproverType, ApproverUserId,
				 AllowComments, RequireComments, StepOrder, RejectBehavior)
			VALUES
				('node-first', 'conflicting live first', 'FirstReview', ?, '', 0,
				 NOW(6), NOW(6), 'User', 'stale-live-user', 1, 0, 1, 'RejectRequest'),
				('node-second', 'conflicting live second', 'SecondReview', ?, '', 0,
				 NOW(6), NOW(6), 'User', 'stale-live-user', 1, 0, 2, 'RejectRequest')`,
			args: []any{approvalTestStorageID, approvalTestStorageID},
		},
		{
			query: `INSERT INTO ObjectPermissions
				(ParentId, SobjectType, PermissionsRead, PermissionsEdit)
			VALUES ('profile-permissions', 'workitem', 1, 1)`,
		},
		{
			query: `INSERT INTO FieldPermissions
				(ParentId, SobjectType, Field, PermissionsRead, PermissionsEdit)
			VALUES
				('profile-permissions', 'workitem', 'Id', 1, 0),
				('profile-permissions', 'workitem', 'Status', 1, 1),
				('profile-permissions', 'workitem', 'FlowMarker', 1, 1)`,
		},
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement.query, statement.args...); err != nil {
			t.Fatalf("prepare Runtime Approval fixture: %v", err)
		}
	}
}

func createRuntimeAPIFlowAuthorizationFixture(t *testing.T, db *sql.DB) {
	t.Helper()
	createRuntimeAPIFileAuthorizationFixture(t, db)
	for _, statement := range []string{
		`CREATE TABLE ProcessDefinition (
			Id VARCHAR(255) PRIMARY KEY,
			LockType VARCHAR(64) NULL
		)`,
		`CREATE TABLE ProcessInstance (
			Id VARCHAR(255) PRIMARY KEY,
			ProcessDefinitionId VARCHAR(255) NOT NULL,
			TargetObjectId VARCHAR(255) NOT NULL,
			Status VARCHAR(64) NOT NULL,
			CreatedDate DATETIME(6) NOT NULL
		)`,
		`INSERT INTO ObjectPermissions
			(ParentId, SobjectType, PermissionsRead, PermissionsEdit)
			VALUES ('profile-permissions', 'workitem', 1, 1)`,
		`INSERT INTO FieldPermissions
			(ParentId, SobjectType, Field, PermissionsRead, PermissionsEdit)
			VALUES
			('profile-permissions', 'workitem', 'Id', 1, 0),
			('profile-permissions', 'workitem', 'Status', 1, 1)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
}

func createRuntimeAPICatalogActionAuthorizationFixture(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		constants.TableProfile:                       false,
		constants.TablePermissionSet:                 false,
		constants.TableObjectPermissions:             false,
		constants.TableFieldPermissions:              false,
		constants.TableProfileCustomPermission:       false,
		constants.TablePermissionSetCustomPermission: false,
		constants.TablePermission:                    false,
		constants.TablePermissionSetPermission:       false,
		constants.TableCustomPermission:              false,
		constants.TableProfileAppVisibility:          false,
		constants.TablePermissionSetAppVisibility:    false,
		constants.TableProfileTabSetting:             false,
		constants.TablePermissionSetTabSetting:       false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, required := wanted[definition.TableName]; !required {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf(
				"create canonical Catalog-action authority table %s: %v",
				definition.TableName, err,
			)
		}
		wanted[definition.TableName] = true
	}
	for table, created := range wanted {
		if !created {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
	now := time.Now().UTC()
	statements := []struct {
		query string
		args  []any
	}{
		{
			query: `CREATE TABLE PermissionSetAssignment (
				PermissionSetId VARCHAR(255) NOT NULL,
				AssigneeId VARCHAR(255) NOT NULL,
				IsDeleted BOOL NOT NULL DEFAULT 0
			)`,
		},
		{
			query: `CREATE TABLE PermissionSetGroup (
				Id VARCHAR(255) PRIMARY KEY,
				IsActive BOOL NOT NULL,
				IsDeleted BOOL NOT NULL DEFAULT 0
			)`,
		},
		{
			query: `CREATE TABLE PermissionSetGroupComponent (
				PermissionSetId VARCHAR(255) NOT NULL,
				GroupId VARCHAR(255) NOT NULL
			)`,
		},
		{
			query: `CREATE TABLE PermissionSetGroupAssignment (
				GroupId VARCHAR(255) NOT NULL,
				AssigneeId VARCHAR(255) NOT NULL,
				IsDeleted BOOL NOT NULL DEFAULT 0
			)`,
		},
		{
			query: `CREATE TABLE MutingPermissionSet (
				Id VARCHAR(255) PRIMARY KEY,
				GroupId VARCHAR(255) NOT NULL
			)`,
		},
		{
			query: `CREATE TABLE _System_MutingPermissionSetPermission (
				MutingPermissionSetId VARCHAR(255) NOT NULL,
				PermissionApiName VARCHAR(255) NOT NULL
			)`,
		},
		{
			query: `INSERT INTO Profile
(Id, Name, IsActive, IsSystem, UserType, IsDeleted, CreatedDate, LastModifiedDate)
VALUES ('profile-test', 'Runtime Catalog Action', 1, 0, 'Standard', 0, ?, ?)`,
			args: []any{now, now},
		},
		{
			query: `INSERT INTO PermissionSet
(Id, Name, Label, IsActive, IsCustom, IsDeleted, ProfileId, IsOwnedByProfile, CreatedDate, LastModifiedDate)
VALUES ('profile-permissions', 'Runtime_Catalog_Action', 'Runtime Catalog Action', 1, 0, 0, 'profile-test', 1, ?, ?)`,
			args: []any{now, now},
		},
		{
			query: `INSERT INTO _System_Permission
(Id, ApiName, Label, Category, IsSystem, IsActive, CreatedDate, LastModifiedDate)
VALUES ('permission-run-bounded', 'RunBoundedAction', 'Run Bounded Action', 'runtime', 1, 1, ?, ?)`,
			args: []any{now, now},
		},
		{
			query: `INSERT INTO _System_PermissionSetPermission
(Id, PermissionSetId, PermissionApiName, IsEnabled, CreatedDate, LastModifiedDate)
VALUES ('grant-run-bounded', 'profile-permissions', 'RunBoundedAction', 1, ?, ?)`,
			args: []any{now, now},
		},
		{
			query: `INSERT INTO ObjectPermissions
(Id, ParentId, SobjectType, PermissionsRead, PermissionsEdit, IsDeleted)
VALUES ('object-workitem', 'profile-permissions', 'workitem', 1, 1, 0)`,
		},
		{
			query: `INSERT INTO FieldPermissions
(Id, ParentId, SobjectType, Field, PermissionsRead, PermissionsEdit, IsDeleted)
VALUES
('field-workitem-id', 'profile-permissions', 'workitem', 'Id', 1, 0, 0),
('field-workitem-status', 'profile-permissions', 'workitem', 'Status', 1, 1, 0)`,
		},
		{
			query: `CREATE TABLE ProcessDefinition (
			Id VARCHAR(255) PRIMARY KEY,
			LockType VARCHAR(64) NULL
		)`,
		},
		{
			query: `CREATE TABLE ProcessInstance (
			Id VARCHAR(255) PRIMARY KEY,
			ProcessDefinitionId VARCHAR(255) NOT NULL,
			TargetObjectId VARCHAR(255) NOT NULL,
			Status VARCHAR(64) NOT NULL,
			CreatedDate DATETIME(6) NOT NULL
		)`,
		},
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement.query, statement.args...); err != nil {
			t.Fatalf("prepare Catalog-action authorization fixture: %v", err)
		}
	}
}

func createRuntimeAPIFileUploadOperationTable(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	wanted := map[string]bool{
		constants.TableFile:           false,
		"_System_FileUploadOperation": false,
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(
			t.Context(), definition,
		); err != nil {
			t.Fatalf("create canonical %s: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
}

func createRuntimeAPIMailboxAuthorizationTable(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, definition := range definitions {
		if definition.TableName != constants.TableRuntimeMailboxAuthorizationHandle {
			continue
		}
		if err := persistence.NewSchemaRepository(db, nil).CreatePhysicalTable(
			t.Context(),
			definition,
		); err != nil {
			t.Fatalf("create canonical Runtime mailbox authorization table: %v", err)
		}
		return
	}
	t.Fatalf(
		"canonical schema is missing %s",
		constants.TableRuntimeMailboxAuthorizationHandle,
	)
}

func createRuntimeAPIFileAuthorizationFixture(t *testing.T, db *sql.DB) {
	t.Helper()
	statements := []string{
		`CREATE TABLE Widget (Id VARCHAR(255) PRIMARY KEY)`,
		`INSERT INTO Widget (Id) VALUES ('widget-1')`,
		`CREATE TABLE PermissionSet (
			Id VARCHAR(255) PRIMARY KEY,
			ProfileId VARCHAR(255) NULL,
			IsOwnedByProfile BOOL NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetAssignment (
			PermissionSetId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetGroup (
			Id VARCHAR(255) PRIMARY KEY,
			IsActive BOOL NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetGroupComponent (
			PermissionSetId VARCHAR(255) NOT NULL,
			GroupId VARCHAR(255) NOT NULL
		)`,
		`CREATE TABLE PermissionSetGroupAssignment (
			GroupId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE ObjectPermissions (
			ParentId VARCHAR(255) NOT NULL,
			SobjectType VARCHAR(255) NOT NULL,
			PermissionsRead BOOL NOT NULL DEFAULT 0,
			PermissionsCreate BOOL NOT NULL DEFAULT 0,
			PermissionsEdit BOOL NOT NULL DEFAULT 0,
			PermissionsDelete BOOL NOT NULL DEFAULT 0,
			PermissionsViewAllRecords BOOL NOT NULL DEFAULT 0,
			PermissionsModifyAllRecords BOOL NOT NULL DEFAULT 0,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE FieldPermissions (
			ParentId VARCHAR(255) NOT NULL,
			SobjectType VARCHAR(255) NOT NULL,
			Field VARCHAR(255) NOT NULL,
			PermissionsRead BOOL NOT NULL DEFAULT 0,
			PermissionsEdit BOOL NOT NULL DEFAULT 0,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`INSERT INTO PermissionSet
			(Id, ProfileId, IsOwnedByProfile, IsDeleted)
			VALUES ('profile-permissions', 'profile-test', 1, 0)`,
		`INSERT INTO ObjectPermissions
			(ParentId, SobjectType, PermissionsRead, PermissionsCreate, PermissionsDelete)
			VALUES
			('profile-permissions', 'widget', 1, 0, 0),
			('profile-permissions', '_system_file', 1, 1, 1)`,
		`INSERT INTO FieldPermissions
			(ParentId, SobjectType, Field, PermissionsRead)
			VALUES ('profile-permissions', 'widget', 'id', 1)`,
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement); err != nil {
			t.Fatalf("prepare production Files authorization fixture: %v", err)
		}
	}
}

func createRuntimeAPISearchScopeAuthorizationFixture(
	t *testing.T,
	db *sql.DB,
) {
	t.Helper()
	createRuntimeAPISearchScopeAuthorityTables(t, db)
	statements := []string{
		`CREATE TABLE Widget (Id VARCHAR(255) PRIMARY KEY)`,
		`CREATE TABLE Gadget (Id VARCHAR(255) PRIMARY KEY)`,
		`CREATE TABLE Hidden (Id VARCHAR(255) PRIMARY KEY)`,
		`CREATE TABLE ReadGated (Id VARCHAR(255) PRIMARY KEY)`,
		`CREATE TABLE PolicyGated (Id VARCHAR(255) PRIMARY KEY)`,
		`CREATE TABLE PermissionSetAssignment (
			PermissionSetId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
		`CREATE TABLE PermissionSetGroup (
			Id VARCHAR(255) PRIMARY KEY,
			IsActive BOOL NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
		`CREATE TABLE PermissionSetGroupComponent (
			PermissionSetId VARCHAR(255) NOT NULL,
			GroupId VARCHAR(255) NOT NULL
		) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
		`CREATE TABLE PermissionSetGroupAssignment (
			GroupId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		) DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`,
		`INSERT INTO Profile (Id, Name)
			VALUES ('profile-test', 'Search Scope Test')`,
		`INSERT INTO PermissionSet
			(Id, Name, Label, ProfileId, IsOwnedByProfile, IsDeleted)
			VALUES
			('profile-permissions', 'Search_Scope_Profile', 'Search Scope Profile', 'profile-test', 1, 0)`,
		`INSERT INTO ObjectPermissions
			(Id, ParentId, SobjectType, PermissionsRead)
		 VALUES
			('object-permission-widget', 'profile-permissions', 'widget', 1),
			('object-permission-gadget', 'profile-permissions', 'gadget', 1),
			('object-permission-hidden', 'profile-permissions', 'hidden', 0),
			('object-permission-read-gated', 'profile-permissions', 'readgated', 1),
			('object-permission-policy-gated', 'profile-permissions', 'policygated', 1)`,
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement); err != nil {
			t.Fatalf("prepare production Search scope fixture: %v", err)
		}
	}
	if _, err := db.Exec(
		`INSERT INTO _System_UserObjectInteraction
			(Id, UserId, ObjectApiName, InteractionCount, LastAccessedAt, IsPinned, CreatedDate, LastModifiedDate)
			VALUES
			('interaction-hidden', ?, 'hidden', 100, CURRENT_TIMESTAMP(6), 1, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6)),
			('interaction-read-gated', ?, 'readgated', 90, CURRENT_TIMESTAMP(6), 1, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6)),
			('interaction-policy-gated', ?, 'policygated', 80, CURRENT_TIMESTAMP(6), 1, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6)),
			('interaction-gadget', ?, 'gadget', 20, CURRENT_TIMESTAMP(6), 0, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6)),
			('interaction-widget', ?, 'widget', 10, CURRENT_TIMESTAMP(6), 0, CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6))`,
		testPrincipalID, testPrincipalID, testPrincipalID, testPrincipalID,
		testPrincipalID,
	); err != nil {
		t.Fatalf("prepare production Search scope MRU: %v", err)
	}
}

func createRuntimeAPISearchScopeAuthorityTables(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := map[string]bool{
		constants.TableProfile:                       false,
		constants.TablePermissionSet:                 false,
		constants.TableObjectPermissions:             false,
		constants.TableMutingPermissionSet:           false,
		constants.TableProfileCustomPermission:       false,
		constants.TablePermissionSetCustomPermission: false,
		constants.TableUserRole:                      false,
		constants.TablePermission:                    false,
		constants.TablePermissionSetPermission:       false,
		constants.TableMutingPermissionSetPermission: false,
		constants.TableCustomPermission:              false,
		constants.TableUserObjectInteraction:         false,
	}
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, required := wanted[definition.TableName]; !required {
			continue
		}
		if err := repository.CreatePhysicalTable(
			t.Context(), definition,
		); err != nil {
			t.Fatalf(
				"create canonical Search scope authority table %s: %v",
				definition.TableName, err,
			)
		}
		wanted[definition.TableName] = true
	}
	for table, created := range wanted {
		if !created {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
}

func runtimeAPISearchScopeObjects() []*models.ObjectMetadata {
	result := make([]*models.ObjectMetadata, 0, 5)
	for _, name := range []string{
		"Widget", "Gadget", "Hidden", "ReadGated", "PolicyGated",
	} {
		object := &models.ObjectMetadata{
			ID: "object-" + strings.ToLower(name), APIName: name,
			Label: name, PluralLabel: name + "s",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{{
				ID:      "field-" + strings.ToLower(name) + "-id",
				APIName: constants.FieldID, Label: "ID",
				Type: constants.FieldTypeString, PhysicalName: "Id",
				IsSystem: true,
			}},
		}
		switch name {
		case "ReadGated":
			object.RequiredReadPermission = "ReadSearchGatedObject"
		case "PolicyGated":
			object.AccessPolicy = &models.ObjectAccessPolicy{
				Mode: "derived",
				ObjectRead: &models.AccessPolicyExpr{
					Kind: "systemPermission", APIName: "ReadPolicyGatedObject",
				},
			}
		}
		result = append(result, object)
	}
	return result
}

func runtimeAPIFileProductionObjects() []*models.ObjectMetadata {
	field := func(name string, fieldType models.FieldType) models.FieldMetadata {
		isSystemField := strings.EqualFold(name, constants.FieldID) ||
			strings.EqualFold(name, constants.FieldCreatedByID) ||
			strings.EqualFold(name, constants.FieldCreatedDate) ||
			strings.EqualFold(name, constants.FieldLastModifiedDate)
		return models.FieldMetadata{
			ID: "field-file-" + strings.ToLower(name), APIName: name,
			Label: name, Type: fieldType, PhysicalName: name,
			IsSystem: isSystemField, Createable: !isSystemField,
			Updateable: !isSystemField,
		}
	}
	return []*models.ObjectMetadata{
		{
			ID: "object-widget", APIName: "Widget", Label: "Widget",
			PluralLabel: "Widgets", SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{
				field("Id", constants.FieldTypeString),
			},
		},
		{
			ID: "object-system-file", APIName: constants.TableFile,
			Label: "File", PluralLabel: "Files", IsSystem: true,
			TableType:    constants.TableTypeSystemCore,
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{
				field("Id", constants.FieldTypeString),
				field("ParentId", constants.FieldTypeString),
				field("ParentObjectApiName", constants.FieldTypeString),
				field("Name", constants.FieldTypeString),
				field("OriginalFileName", constants.FieldTypeString),
				field("Extension", constants.FieldTypeString),
				field("MimeType", constants.FieldTypeString),
				field("SizeBytes", constants.FieldTypeInt),
				field("StoragePath", constants.FieldTypeTextarea),
				field("StorageProvider", constants.FieldTypeString),
				field("StorageBucket", constants.FieldTypeString),
				field("StorageKey", constants.FieldTypeString),
				field("Status", constants.FieldTypeString),
				field("SourceType", constants.FieldTypeString),
				field("AttachmentKind", constants.FieldTypeString),
				field("Checksum", constants.FieldTypeString),
				field("DeletedById", constants.FieldTypeString),
				field("DeletedDate", constants.FieldTypeDatetime),
				field("CreatedById", constants.FieldTypeString),
				field("CreatedDate", constants.FieldTypeDatetime),
				field("LastModifiedDate", constants.FieldTypeDatetime),
			},
		},
	}
}

func servingArtifactsForObjects(
	t *testing.T,
	revision uint64,
	runtimeCapabilityID string,
	profile catalogsnapshot.ProjectionProfile,
	objects []*models.ObjectMetadata,
) (catalogsnapshot.Artifact, catalogsnapshot.Artifact) {
	t.Helper()
	input := catalogsnapshot.CompileInput{
		OrgID: testOrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision:  catalogsnapshot.CatalogRevision(revision),
			SchemaGeneration: 8, EngineGeneration: 9,
		},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision:   catalogsnapshot.CatalogRevision(revision),
			AppInstallationID: testInstallationID,
			CapabilityID:      runtimeCapabilityID, Profile: profile,
		}},
	}
	for _, object := range objects {
		input.Objects = append(input.Objects, catalogsnapshot.ObjectInput{
			CatalogRevision: catalogsnapshot.CatalogRevision(revision),
			Metadata:        object,
		})
		input.Projections[0].ObjectAPINames = append(
			input.Projections[0].ObjectAPINames, object.APIName,
		)
	}
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: testInstallationID,
			CapabilityID:      runtimeCapabilityID,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact
}

func TestRuntimeAPIRealRecycleBinRepositoryCrossProcess(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_api_integration requires the declared TiDB environment")
	}
	db, cleanup := openRuntimeAPIScratchDB(t)
	defer cleanup()
	if _, err := db.Exec(`CREATE TABLE Issue (
		Id VARCHAR(255) PRIMARY KEY, OwnerId VARCHAR(255) NOT NULL,
		IsDeleted BOOLEAN NOT NULL DEFAULT 0, Subject VARCHAR(255),
		CreatedDate DATETIME(6), LastModifiedDate DATETIME(6)
	)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO Issue
		(Id, OwnerId, IsDeleted, Subject, CreatedDate, LastModifiedDate)
		VALUES (?, ?, 1, ?, ?, ?)`, "ISS-P3-DELETED", testPrincipalID, "deleted issue",
		"2026-01-01 00:00:00.000000", "2026-07-21 00:00:00.000000"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`CREATE TABLE _System_RecycleBin (
		Id VARCHAR(255) PRIMARY KEY, RecordId VARCHAR(255) NOT NULL, ObjectApiName VARCHAR(255) NOT NULL,
		RecordName VARCHAR(255), DeletedBy VARCHAR(255), DeletedById VARCHAR(255), DeletedDate DATETIME(6) NOT NULL
	)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_RecycleBin
		(Id, RecordId, ObjectApiName, RecordName, DeletedBy, DeletedById, DeletedDate)
		VALUES (?, ?, ?, ?, ?, ?, ?)`, "tombstone-1", "ISS-P3-DELETED", "Issue", "deleted issue",
		"Runtime User", testPrincipalID, "2026-07-22 04:00:00.000000"); err != nil {
		t.Fatal(err)
	}

	queryRepo := &countingQueryRepository{QueryRepository: persistence.NewQueryRepository(db)}
	schema := runtimeRecycleBinIntegrationSchema()
	permissionRepo := &integrationPermissionRepo{fields: schema.Fields}
	permissionRepo.grantRead.Store(true)
	graph, err := runtimeassembly.NewServingRecordsReadRuntime(runtimeassembly.RecordsReadDependencies{
		DB:             db,
		PermissionRepo: permissionRepo, QueryRepo: queryRepo, RecordRepo: persistence.NewRecordRepository(db),
	}, RuntimeMetadataViewFromContext)
	if err != nil {
		t.Fatal(err)
	}
	fixture := newServingFixtureForCapabilityAndObject(t, 7, "io.nexuscrm.sfcrm.recycle-bin-read.v1", recyclesemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1, schema)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	process, err := Assemble(ProcessConfig{
		Version: "integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: DefaultAdmissionLimit, ServingEnabled: true,
		RecycleBinMaxAdmissionLease: time.Second, RecycleBinLister: graph.Records,
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: durableIntegrationSecurityAdmission(t, db, "replica-recycle-bin", recyclesemantic.CapabilityName),
		ServingAdmission:  fixture.admission,
	}, graph.Records)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	connection, err := grpc.NewClient("passthrough:///"+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()
	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_MINE,
	})
	if err != nil {
		t.Fatalf("ListRecycleBin = %v (%#v)", err, mapping.FromRecycleBinStatusError(err))
	}
	response, err := mapping.FromWireListRecycleBinResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Items) != 1 || response.Items[0].RecordID != "ISS-P3-DELETED" ||
		response.Items[0].RecordName != "deleted issue" || response.Items[0].ObjectID != "Issue" ||
		response.Items[0].DeletedBy != "Runtime User" || response.Items[0].DeletedDate != "2026-07-22T04:00:00Z" ||
		response.Items[0].CreatedDate != "2026-01-01T00:00:00Z" || response.Items[0].LastModifiedDate != "2026-07-21T00:00:00Z" ||
		response.Receipts.CapabilityVersion != recyclesemantic.CapabilityName || response.Receipts.CatalogRevision != 7 {
		t.Fatalf("response = %#v", response)
	}
	if queryRepo.calls.Load() != 1 {
		t.Fatalf("database visibility-query calls = %d, want one query for one object", queryRepo.calls.Load())
	}
}

func durableIntegrationSecurityAdmission(t *testing.T, db *sql.DB, replicaID string, capabilities ...string) *SecurityAdmission {
	t.Helper()
	statements := []string{
		`CREATE TABLE User (Id VARCHAR(255) PRIMARY KEY, Username VARCHAR(255) NOT NULL, Email VARCHAR(255) NOT NULL, ProfileId VARCHAR(255) NOT NULL, FirstName VARCHAR(255) NULL, LastName VARCHAR(255) NULL, IsActive BOOL NOT NULL, IsDeleted BOOL NOT NULL DEFAULT 0, FailedLoginAttempts INT NOT NULL DEFAULT 0, LockedUntil DATETIME NULL, UserRoleId VARCHAR(255) NULL)`,
		`CREATE TABLE _System_Session (Id VARCHAR(255) PRIMARY KEY, UserId VARCHAR(255) NOT NULL, Token VARBINARY(3000) NOT NULL, CurrentBearerToken VARBINARY(3000) NULL, ExpiresAt DATETIME NOT NULL, IpAddress VARCHAR(255), UserAgent VARCHAR(255), IsRevoked BOOL NOT NULL, AuthEpoch DECIMAL(20,0) NOT NULL, ControlProofHash CHAR(64) NULL, LastActivity DATETIME NOT NULL, Scope TEXT NULL, ConnectedAppId VARCHAR(255) NULL, GrantType VARCHAR(50) NULL, CreatedDate DATETIME NOT NULL, LastModifiedDate DATETIME NOT NULL)`,
		`CREATE TABLE _System_ConnectedApp (Id VARCHAR(255) PRIMARY KEY, IsActive BOOL NOT NULL)`,
		`CREATE TABLE _System_ConnectedAppDeclaration (ConnectedAppId VARCHAR(255), InstalledPackName VARCHAR(255), IsDeleted BOOL NOT NULL DEFAULT 0)`,
		`CREATE TABLE _System_OAuthDynamicClientRegistration (ConnectedAppId VARCHAR(255), ScopeOwnerPack VARCHAR(255), IsDeleted BOOL NOT NULL DEFAULT 0)`,
		`CREATE TABLE _System_PermissionStateVersion (SingletonKey TINYINT PRIMARY KEY, Lineage BINARY(16) NOT NULL, Revision DECIMAL(20,0) NOT NULL, EnforcementMode VARCHAR(16) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`,
		`CREATE TABLE _System_RuntimeSecurityState (OrgId VARCHAR(255) PRIMARY KEY, RuntimeAudience VARCHAR(255) NOT NULL, SecurityEpoch DECIMAL(20,0) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`,
		`CREATE TABLE _System_RuntimeSecurityBinding (Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, AppInstallationId VARCHAR(255) NOT NULL, CallerWorkloadId VARCHAR(1024) NOT NULL, CallerWorkloadDigest CHAR(64) NOT NULL, Capabilities JSON NOT NULL, IsActive BOOL NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL, INDEX idx_installation (OrgId, AppInstallationId, CallerWorkloadDigest))`,
		`CREATE TABLE _System_RuntimeSecurityLease (RuntimeReplicaId VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, SecurityEpoch DECIMAL(20,0) NOT NULL, ExpiresAt DATETIME(6) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL)`,
		`CREATE TABLE _System_RuntimeDelegationRevocation (Id VARCHAR(255) PRIMARY KEY, OrgId VARCHAR(255) NOT NULL, DelegationId VARCHAR(255) NOT NULL, ExpiresAt DATETIME(6) NOT NULL, CreatedDate DATETIME(6) NOT NULL, LastModifiedDate DATETIME(6) NOT NULL, UNIQUE KEY ux_org_delegation (OrgId, DelegationId))`,
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
	now := time.Now().UTC()
	if _, err := db.Exec(`INSERT INTO User (Id, Username, Email, ProfileId, FirstName, LastName, IsActive, FailedLoginAttempts, UserRoleId) VALUES (?, ?, ?, ?, ?, ?, 1, 0, ?)`, testPrincipalID, "runtime-user", "runtime@example.test", "profile-test", "Runtime", "User", "role-test"); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_ConnectedApp (Id, IsActive) VALUES ('connected-app-test', 1)`); err != nil {
		t.Fatal(err)
	}
	insertRuntimeIntegrationSession(
		t, db, testSessionID, testPrincipalID,
		"Runtime User", "runtime@example.test",
	)
	if _, err := db.Exec(
		`INSERT INTO _System_PermissionStateVersion (SingletonKey, Lineage, Revision, EnforcementMode, CreatedDate, LastModifiedDate) VALUES (1, ?, 1, 'dark', ?, ?)`,
		[]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
		now,
		now,
	); err != nil {
		t.Fatal(err)
	}

	securityRepository := persistence.NewRuntimeSecurityRepository(db)
	if err := securityRepository.ActivateProjection(context.Background(), 0, persistence.RuntimeSecurityProjection{
		OrgID: testOrgID, RuntimeAudience: testRuntimeAudience, SecurityEpoch: 1,
		Bindings: []persistence.RuntimeSecurityBinding{{
			OrgID: testOrgID, AppInstallationID: testInstallationID, CallerWorkloadID: testWorkloadID,
			Capabilities: append([]string(nil), capabilities...), Active: true,
		}},
	}); err != nil {
		t.Fatal(err)
	}
	if err := securityRepository.IssueLease(context.Background(), testOrgID, replicaID, 1, now.Add(time.Minute)); err != nil {
		t.Fatal(err)
	}
	state, err := NewRuntimeSecurityState(context.Background(), RuntimeSecurityStateConfig{
		Replica: func(ctx context.Context) (RuntimeReplicaSecurityState, bool, error) {
			stored, found, err := securityRepository.ReadReplicaSecurityContext(ctx, replicaID)
			return RuntimeReplicaSecurityState{OrgID: stored.OrgID, RuntimeAudience: stored.RuntimeAudience, SecurityEpoch: stored.SecurityEpoch, LeaseSecurityEpoch: stored.LeaseSecurityEpoch, LeaseExpiresAt: stored.LeaseExpiresAt}, found, err
		},
		Admission: func(ctx context.Context, claims delegationv1.Claims) (RuntimeAdmissionSecurityState, error) {
			stored, err := securityRepository.ReadAdmissionState(ctx, replicaID, claims.OrgID, claims.AppInstallationID, claims.CallerWorkloadID, claims.DelegationID, time.Now())
			return RuntimeAdmissionSecurityState{StateFound: stored.StateFound, InstallationFound: stored.InstallationFound, InstallationActive: stored.InstallationActive, Capabilities: stored.Capabilities, SecurityEpoch: stored.SecurityEpoch, LeaseFound: stored.LeaseFound, LeaseSecurityEpoch: stored.LeaseSecurityEpoch, LeaseExpiresAt: stored.LeaseExpiresAt, DelegationRevoked: stored.DelegationRevoked}, err
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	authService := featureauth.NewAuthService(persistence.NewUserRepository(db), persistence.NewSessionRepository(db), nil)
	if _, err := authService.AdmitRuntimeAuthorizingSession(
		context.Background(), testSessionID, testPrincipalID,
	); err != nil {
		t.Fatalf("durable Runtime security fixture cannot admit its session: %v", err)
	}
	admission, err := NewSecurityAdmission(SecurityAdmissionConfig{
		Verifier: testVerifier(t), State: state,
		Sessions: func(ctx context.Context, sessionID, principalID string) (*RuntimeSessionState, error) {
			validated, err := authService.AdmitRuntimeAuthorizingSession(ctx, sessionID, principalID)
			if err != nil {
				return nil, err
			}
			return &RuntimeSessionState{
				User: validated.User, AuthEpoch: validated.AuthEpoch,
				Subject: validated.Subject, PermissionFloor: validated.PermissionFloor,
			}, nil
		},
		Bindings: func(_ context.Context, caller runtimegrpc.VerifiedCaller) (RuntimeCallerBinding, bool, error) {
			return RuntimeCallerBinding{CallerWorkloadID: testWorkloadID}, caller == testWorkloadID, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	return admission
}

func insertRuntimeIntegrationSession(
	t *testing.T,
	db *sql.DB,
	sessionID string,
	principalID string,
	name string,
	email string,
) {
	t.Helper()
	now := time.Now().UTC()
	roleID := "role-test"
	roleName := "Runtime role"
	bearer, _, err := kernelauth.GenerateTokenForSession(
		kernelauth.UserSession{
			ID: principalID, Name: name, Email: email,
			ProfileId: "profile-test", ProfileName: "Runtime profile",
			RoleId: &roleID, RoleName: &roleName,
		},
		sessionID,
		now,
		time.Hour,
	)
	if err != nil {
		t.Fatalf("generate durable Runtime session bearer: %v", err)
	}
	connectedAppID := "connected-app-test"
	if err := persistence.NewSessionRepository(db).InsertSession(t.Context(), &models.SystemSession{
		ID: sessionID, UserID: principalID, Token: bearer,
		ExpiresAt: now.Add(time.Hour), LastActivity: now,
		IPAddress: "127.0.0.1", UserAgent: "integration", AuthEpoch: 1,
		ConnectedAppID: &connectedAppID,
	}); err != nil {
		t.Fatal(err)
	}
}

type countingQueryRepository struct {
	ports.QueryRepository
	calls atomic.Int64
}

func (r *countingQueryRepository) Find(ctx context.Context, schema *models.ObjectMetadata, request models.QueryRequest, result *domainrls.Result, fields []string) (*models.QueryResult, error) {
	r.calls.Add(1)
	return r.QueryRepository.Find(ctx, schema, request, result, fields)
}

type integrationMetadataRepo struct {
	ports.MetadataRepository
	schema *models.ObjectMetadata
}

func (r *integrationMetadataRepo) GetAllSchemas(context.Context) ([]*models.ObjectMetadata, error) {
	return []*models.ObjectMetadata{r.schema}, nil
}

func (r *integrationMetadataRepo) GetAllRelationships(context.Context) ([]*models.Relationship, error) {
	return []*models.Relationship{}, nil
}

func (*integrationMetadataRepo) GetSharingRules(context.Context, string) ([]*models.SharingRule, error) {
	return []*models.SharingRule{}, nil
}

func (r *integrationMetadataRepo) GetSchemaByAPIName(_ context.Context, name string) (*models.ObjectMetadata, error) {
	if strings.EqualFold(name, r.schema.APIName) {
		return r.schema, nil
	}
	return nil, nil
}

type integrationPermissionRepo struct {
	ports.PermissionRepository
	grantRead        atomic.Bool
	fieldObjectLoads atomic.Int64
	fieldPointLoads  atomic.Int64
	fields           []models.FieldMetadata
}

func (*integrationPermissionRepo) GetAllRoles(context.Context) ([]*models.UserRole, error) {
	return []*models.UserRole{}, nil
}

func (r *integrationPermissionRepo) LoadEffectiveObjectPermission(context.Context, *models.UserSession, string) (*models.ObjectPermissions, error) {
	return &models.ObjectPermissions{PermissionsRead: r.grantRead.Load(), PermissionsEdit: true, PermissionsDelete: true}, nil
}

func (r *integrationPermissionRepo) LoadEffectiveFieldPermission(_ context.Context, _ *models.UserSession, _, field string) (*models.FieldPermissions, error) {
	r.fieldPointLoads.Add(1)
	return &models.FieldPermissions{PermissionsRead: !strings.EqualFold(field, "SecretNote")}, nil
}

func (r *integrationPermissionRepo) ListEffectiveFieldPermissionsForUserObject(_ context.Context, _, _, object string) ([]models.FieldPermissions, error) {
	r.fieldObjectLoads.Add(1)
	if !strings.EqualFold(object, "Issue") {
		return nil, nil
	}
	permissions := make([]models.FieldPermissions, 0, len(r.fields))
	for _, field := range r.fields {
		permissions = append(permissions, models.FieldPermissions{
			SobjectType:     "Issue",
			Field:           field.APIName,
			PermissionsRead: !strings.EqualFold(field.APIName, "SecretNote"),
		})
	}
	return permissions, nil
}

func (*integrationPermissionRepo) GetEffectiveSystemPermissions(context.Context, string, string) (map[string]bool, error) {
	return map[string]bool{}, nil
}

func (*integrationPermissionRepo) GetUserGroupIDs(context.Context, string) ([]string, error) {
	return nil, nil
}

func (*integrationPermissionRepo) GetManualShareRecordIDs(context.Context, string, string) ([]string, error) {
	return nil, nil
}

func (*integrationPermissionRepo) GetTeamMemberRecordIDs(context.Context, string, string) ([]string, error) {
	return nil, nil
}

func (*integrationPermissionRepo) IsUserInGroup(context.Context, string, string) (bool, error) {
	return false, nil
}

func (*integrationPermissionRepo) GetManualShareAccessLevels(context.Context, string, string, string) ([]string, error) {
	return nil, nil
}

func (*integrationPermissionRepo) GetTeamMemberAccessLevel(context.Context, string, string, string) (*string, error) {
	return nil, nil
}

func (*integrationPermissionRepo) GetPendingApprovalLockInfo(context.Context, string) (string, string, error) {
	return "", "", nil
}

func (*integrationPermissionRepo) HasPendingApprovalWorkitemForUser(context.Context, string, string) (bool, error) {
	return false, nil
}

func stringPointer(value string) *string { return &value }

func runtimeAPIIntegrationSchema() *models.ObjectMetadata {
	return &models.ObjectMetadata{
		APIName: "Issue", SharingModel: constants.SharingModelPrivate,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true},
			{APIName: constants.FieldOwnerID, Type: constants.FieldTypeReference, IsSystem: true},
			{APIName: "Archived", Type: constants.FieldTypeBoolean},
			{APIName: "DueDate", Type: constants.FieldTypeDate},
			{APIName: "EmptySummary", Type: constants.FieldTypeString},
			{APIName: "Estimate", Type: constants.FieldTypeDouble},
			{APIName: "OptionalValue", Type: constants.FieldTypeString},
			{APIName: "ProjectId", Type: constants.FieldTypeReference},
			{APIName: "RetryCount", Type: constants.SchemaFieldType("integer")},
			{APIName: "SecretNote", Type: constants.FieldTypeString},
			{APIName: "Subject", Type: constants.FieldTypeString},
			{APIName: "UpdatedAt", Type: constants.FieldTypeDatetime},
		},
	}
}

func runtimeRecycleBinIntegrationSchema() *models.ObjectMetadata {
	return &models.ObjectMetadata{
		ID: "object-issue", APIName: "Issue", Label: "Issue", PluralLabel: "Issues", SharingModel: constants.SharingModelPrivate,
		Fields: []models.FieldMetadata{
			{ID: "field-issue-id", APIName: constants.FieldID, Label: "ID", PhysicalName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true},
			{ID: "field-issue-owner", APIName: constants.FieldOwnerID, Label: "Owner", PhysicalName: constants.FieldOwnerID, Type: constants.FieldTypeReference, IsSystem: true},
			{ID: "field-issue-deleted", APIName: constants.FieldIsDeleted, Label: "Deleted", PhysicalName: constants.FieldIsDeleted, Type: constants.FieldTypeBoolean, IsSystem: true},
			{ID: "field-issue-created", APIName: constants.FieldCreatedDate, Label: "Created", PhysicalName: constants.FieldCreatedDate, Type: constants.FieldTypeDatetime, IsSystem: true},
			{ID: "field-issue-modified", APIName: constants.FieldLastModifiedDate, Label: "Modified", PhysicalName: constants.FieldLastModifiedDate, Type: constants.FieldTypeDatetime, IsSystem: true},
			{ID: "field-issue-subject", APIName: "Subject", Label: "Subject", PhysicalName: "Subject", Type: constants.FieldTypeString, IsNameField: true},
		},
	}
}

func openRuntimeAPIScratchDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	config := mysql.NewConfig()
	config.Net = "tcp"
	config.Addr = net.JoinHostPort(os.Getenv("TIDB_HOST"), envOrTest("TIDB_PORT", "4000"))
	config.User = envOrTest("TIDB_USER", "root")
	config.Passwd = os.Getenv("TIDB_PASSWORD")
	config.ParseTime = true
	admin, err := sql.Open("mysql", config.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	name := fmt.Sprintf("nexuscrm_runtime_api_p3_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil {
		admin.Close()
		t.Fatal(err)
	}
	config.DBName = name
	db, err := sql.Open("mysql", config.FormatDSN())
	if err != nil {
		t.Fatal(err)
	}
	if err := db.Ping(); err != nil {
		t.Fatal(err)
	}
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
	}
}

func envOrTest(key, fallback string) string {
	if value := os.Getenv(key); value != "" {
		return value
	}
	return fallback
}

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