//go:build integration_service_integration

package integrationservice

import (
	"bufio"
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/ed25519"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/tls"
	"crypto/x509"
	"crypto/x509/pkix"
	"database/sql"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"io"
	"math/big"
	"net"
	"net/http"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"reflect"
	"strings"
	"sync"
	"syscall"
	"testing"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/backend/internal/runtimeapi"
	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	connectorinbound "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inbound"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/runtimehandoff"
	"github.com/pingcap-inc/nexuscrm-connectors/features/namedcredential"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	flowcommandsemanticv3 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv3"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	"google.golang.org/grpc/credentials"
	"google.golang.org/grpc/metadata"
)

const (
	m2RuntimeChildModeEnv   = "NEXUSCRM_M2_RUNTIME_CHILD"
	m2RuntimeReadyFileEnv   = "NEXUSCRM_M2_RUNTIME_READY_FILE"
	m2RuntimeEventsFileEnv  = "NEXUSCRM_M2_RUNTIME_EVENTS_FILE"
	m2RuntimeViewFileEnv    = "NEXUSCRM_M2_RUNTIME_VIEW_FILE"
	m2RuntimeDatabaseDSNEnv = "NEXUSCRM_M2_RUNTIME_DATABASE_DSN"
	m2RuntimeCertFileEnv    = "NEXUSCRM_M2_RUNTIME_CERT_FILE"
	m2RuntimeKeyFileEnv     = "NEXUSCRM_M2_RUNTIME_KEY_FILE"
	m2RuntimeWorkloadToken  = "workload-secret"
	m2RuntimeServerName     = "runtime.test"
	m2RuntimeChildTestName  = "TestM2IntegrationPlatformRuntimeChildProcess"
	m2RuntimeProcessVersion = "m2-cross-process"
)

type m2RuntimeChildReady struct {
	PID          int    `json:"pid"`
	GRPCAddress  string `json:"grpcAddress"`
	AdminAddress string `json:"adminAddress"`
}

type m2RuntimeEvent struct {
	PID     int                     `json:"pid"`
	Method  string                  `json:"method"`
	Command commandsemantic.Request `json:"command,omitempty"`
}

type m2RuntimeChild struct {
	command *exec.Cmd
	wait    chan error
	logs    strings.Builder
	address string
	admin   string
	stopped bool
	mu      sync.Mutex
}

type m2ApplicationResultFixture struct {
	Pending  applicationresultsemantic.GetResultRequest
	Terminal applicationresultsemantic.GetResultRequest
	Foreign  applicationresultsemantic.GetResultRequest
}

type m2RuntimeTransactionOwner struct {
	*persistence.TransactionManager
}

func (o m2RuntimeTransactionOwner) RunInPessimisticTransaction(
	ctx context.Context,
	fn func(context.Context) error,
) error {
	_, err := o.RunInPessimisticTransactionWithOutcome(ctx, fn)
	return err
}

func (o m2RuntimeTransactionOwner) RunInPessimisticTransactionWithOutcome(
	ctx context.Context,
	fn func(context.Context) error,
) (ports.TransactionCompletionOutcome, error) {
	var completion ports.TransactionCompletionOutcome
	err := o.WithPessimisticRetry(ctx, func(txCtx context.Context) error {
		ports.RegisterTransactionCompletionHook(
			txCtx,
			func(outcome ports.TransactionCompletionOutcome) { completion = outcome },
		)
		return fn(txCtx)
	}, 3)
	return completion, err
}

// TestM2IntegrationHandoffCrossesSeparateRuntimeProcess proves the exact
// Integration-owned production handoff crosses a TLS gRPC process boundary.
// Runtime command execution semantics remain covered by the real Runtime API
// and Worker TiDB suites; this test owns the missing consumer/process proof.
func TestM2IntegrationHandoffCrossesSeparateRuntimeProcess(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("M2 Integration cross-process proof requires the declared TiDB environment")
	}
	db, cleanup := openIntegrationScratchDB(t)
	defer cleanup()
	createCanonicalSystemSchema(t, db)
	seedIntegrationWorkItemCutover(t, db)

	const (
		ownerPack   = "connector-test"
		packInstall = "connector-install-1"
		workID      = "work-m2-cross-process"
		eventID     = "event-m2-cross-process"
		leaseToken  = "lease-m2-cross-process"
	)
	digest := "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)

	config := processTestConfig()
	config.OrgID = "org-test"
	config.AppInstallationID = "01900000-0000-7000-8000-000000000001"
	config.AuthorizedWorkloadID = "integration-service"
	view := productionIntegrationBindingView(
		config, ownerPack, packInstall, digest, "flow-v1",
	)
	runtimeView, _ := m2RuntimeConsumerViews(t, view)
	view.Lineage.SourceSnapshotDigest = string(runtimeView.SourceSnapshotDigest())
	view.Receipts.SecurityEpoch = 1
	view.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(view)
	if err := catalogsemantic.ValidateIntegrationBindingView(view); err != nil {
		t.Fatalf("M2 Integration binding fixture: %v", err)
	}
	seedIntegrationServingGeneration(
		t, db, ownerPack, packInstall, digest, view.Bindings,
	)
	t.Setenv(
		namedcredential.MasterKeyEnvVar,
		"AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
	)
	resultFixture := seedM2ApplicationResults(t, db, config)

	testDir := t.TempDir()
	eventsFile := filepath.Join(testDir, "runtime-events.jsonl")
	viewFile := filepath.Join(testDir, "integration-view.json")
	caFile, certFile, keyFile := writeM2RuntimeTLSFiles(t, testDir)
	writeM2JSON(t, viewFile, view)
	runtimeDSN := m2RuntimeDatabaseDSN(t, db)

	// First prove the independently deployable Integration composition opens
	// ingress only after reading its exact authority from another OS process.
	startupChild := startM2RuntimeChild(
		t, viewFile, certFile, keyFile, eventsFile, runtimeDSN,
	)
	startupClient := dialM2RuntimeClient(
		t, startupChild.address, caFile, config.AppInstallationID,
	)
	process, err := assembleProductionWithDatabase(
		t.Context(), m2RuntimeProcessVersion, config, db,
		startupClient, &fakeDatabase{},
	)
	if err != nil {
		t.Fatalf(
			"%v\nRuntime events: %#v\nRuntime child:\n%s",
			err, readM2RuntimeEventsFile(eventsFile), startupChild.logs.String(),
		)
	}
	if err := process.Start(t.Context()); err != nil {
		t.Fatal(err)
	}
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", 200)
	shutdownCtx, cancelShutdown := context.WithTimeout(
		context.Background(), 10*time.Second,
	)
	if err := process.Shutdown(shutdownCtx); err != nil {
		cancelShutdown()
		t.Fatal(err)
	}
	cancelShutdown()
	startupChild.stop(t)
	startupEvents := readM2RuntimeEvents(t, eventsFile)
	if countM2RuntimeEvents(startupEvents, "GetIntegrationBindingView") == 0 ||
		countM2RuntimeEvents(startupEvents, "AcceptCommand") != 0 {
		t.Fatalf("Integration startup Runtime events = %#v", startupEvents)
	}

	now := time.Now().UTC().Truncate(time.Microsecond)
	if _, err := db.ExecContext(t.Context(), `
INSERT INTO _System_ConnectorWorkItem
       (WorkId, EventId, InstalledPackId, WorkType, QueueClass, State,
        RunAfter, Attempt, LeaseOwner, LeaseToken, LeaseUntil, SourceLaneKey,
        Generation, LaneOrdinal, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, 'connector_event', 'default', 'processing', ?, 1,
        'integration-worker', ?, ?, 'connector-test:source-1', 1, 1, ?, ?)`,
		workID, eventID, packInstall, now, leaseToken, now.Add(time.Minute), now, now,
	); err != nil {
		t.Fatal(err)
	}

	scope := connectorinbound.RuntimeCommandHandoffScope{
		WorkItemID: workID, EventID: eventID, LeaseToken: leaseToken,
		Namespace: ownerPack, InstalledPackID: packInstall,
		EventKind: "thing.updated", SourcePartition: "source-1", RouteVersion: 1,
	}
	request := connectorinbound.RuntimeCommandHandoffRequest{
		Scope: scope, DefinitionRef: "BoundedInbound", DefinitionVersion: "flow-v1",
		Inputs: map[string]any{"eventID": eventID, "sourceID": "source-1"},
	}

	firstChild := startM2RuntimeChild(
		t, viewFile, certFile, keyFile, eventsFile, runtimeDSN,
	)
	firstClient := dialM2RuntimeClient(
		t, firstChild.address, caFile, config.AppInstallationID,
	)
	assertM2RuntimeCapability(t, firstChild.admin)
	firstHandoff := newM2RuntimeHandoff(t, db, firstClient, config)
	firstCtx, cancelFirst := context.WithTimeout(t.Context(), 5*time.Second)
	if err := firstHandoff.AcceptRuntimeCommand(firstCtx, request); err != nil {
		cancelFirst()
		t.Fatal(err)
	}
	cancelFirst()
	firstPending := readM2ApplicationResult(t, firstClient, resultFixture.Pending)
	firstTerminal := readM2ApplicationResult(
		t,
		firstClient,
		resultFixture.Terminal,
	)
	assertM2ApplicationResults(t, resultFixture, firstPending, firstTerminal)
	wrongRef := resultFixture.Terminal
	wrongRef.ExecutionRefDigest = "sha256:" + strings.Repeat("f", 64)
	assertM2ApplicationResultMissing(t, firstClient, wrongRef)
	assertM2ApplicationResultMissing(t, firstClient, resultFixture.Foreign)
	pendingBytes, err := json.Marshal(firstPending)
	if err != nil {
		t.Fatal(err)
	}
	terminalBytes, err := json.Marshal(firstTerminal)
	if err != nil {
		t.Fatal(err)
	}
	if err := firstClient.Close(); err != nil {
		t.Fatal(err)
	}
	firstChild.stop(t)
	firstEvents := readM2RuntimeEvents(t, eventsFile)
	firstCommands := m2RuntimeCommands(firstEvents)
	if len(firstCommands) != 1 {
		t.Fatalf("first Runtime commands = %#v", firstCommands)
	}
	assertM2CommandSharesPlatformEndpoint(t, firstEvents, firstCommands[0], view)

	// A restarted Integration/Runtime pair must replay the immutable receipt,
	// not rebuild command bytes from changed provider input.
	request.Inputs["sourceID"] = "changed-provider-value"
	restartChild := startM2RuntimeChild(
		t, viewFile, certFile, keyFile, eventsFile, runtimeDSN,
	)
	restartClient := dialM2RuntimeClient(
		t, restartChild.address, caFile, config.AppInstallationID,
	)
	restartHandoff := newM2RuntimeHandoff(t, db, restartClient, config)
	restartTerminal := readM2ApplicationResult(
		t,
		restartClient,
		resultFixture.Terminal,
	)
	assertM2ApplicationResultBytes(
		t,
		"second Runtime restart terminal",
		terminalBytes,
		restartTerminal,
	)
	restartPending := readM2ApplicationResult(t, restartClient, resultFixture.Pending)
	assertM2ApplicationResultBytes(
		t,
		"second Runtime restart pending",
		pendingBytes,
		restartPending,
	)
	assertM2ApplicationResults(t, resultFixture, restartPending, restartTerminal)
	restartCtx, cancelRestart := context.WithTimeout(t.Context(), 5*time.Second)
	found, err := restartHandoff.ResumePreparedRuntimeCommand(restartCtx, scope)
	cancelRestart()
	if err != nil || !found {
		t.Fatalf("resume prepared Runtime command = found %v, error %v", found, err)
	}
	restartEvents := readM2RuntimeEvents(t, eventsFile)
	commandsAfterRestart := m2RuntimeCommands(restartEvents)
	if len(commandsAfterRestart) != 2 ||
		!reflect.DeepEqual(commandsAfterRestart[0], commandsAfterRestart[1]) {
		t.Fatalf("Runtime commands across restart = %#v", commandsAfterRestart)
	}
	assertM2RuntimeRestartedOnOnePlatformEndpoint(t, restartEvents)

	var receiptCount int
	if err := db.QueryRowContext(t.Context(), `
SELECT COUNT(*) FROM _System_ConnectorRuntimeHandoffReceipt WHERE WorkId = ?`,
		workID,
	).Scan(&receiptCount); err != nil {
		t.Fatal(err)
	}
	if receiptCount != 1 {
		t.Fatalf("Integration immutable handoff receipts = %d, want 1", receiptCount)
	}
	var commandCount int
	if err := db.QueryRowContext(t.Context(), `
SELECT COUNT(*) FROM _System_RuntimePlatformCommand WHERE CommandId = ?`,
		commandsAfterRestart[0].CommandID,
	).Scan(&commandCount); err != nil {
		t.Fatal(err)
	}
	if commandCount != 1 {
		t.Fatalf("Runtime durable commands after restart replay = %d, want 1", commandCount)
	}

	canceled, cancel := context.WithCancel(context.Background())
	cancel()
	if found, err := restartHandoff.ResumePreparedRuntimeCommand(canceled, scope); err == nil || !found {
		t.Fatalf("canceled resume = found %v, error %v", found, err)
	}
	if got := len(m2RuntimeCommands(readM2RuntimeEvents(t, eventsFile))); got != 2 {
		t.Fatalf("canceled handoff reached Runtime: commands=%d", got)
	}

	// With Runtime stopped, the exact frozen owner remains retriable. There is
	// no process-local Flow execution or legacy fallback path to report success.
	restartChild.stop(t)
	if err := restartClient.Close(); err != nil {
		t.Fatal(err)
	}
	unavailableCtx, cancelUnavailable := context.WithTimeout(
		context.Background(), time.Second,
	)
	_, unavailableErr := restartHandoff.ResumePreparedRuntimeCommand(
		unavailableCtx, scope,
	)
	cancelUnavailable()
	var handoffFailure *connectorinbound.RuntimeCommandHandoffFailure
	if unavailableErr == nil || !errors.As(unavailableErr, &handoffFailure) ||
		!handoffFailure.Retriable {
		t.Fatalf("stopped Runtime handoff error = %v", unavailableErr)
	}
	if got := len(m2RuntimeCommands(readM2RuntimeEvents(t, eventsFile))); got != 2 {
		t.Fatalf("stopped Runtime created a fallback command: commands=%d", got)
	}
}

func seedM2ApplicationResults(
	t *testing.T,
	db *sql.DB,
	config Config,
) m2ApplicationResultFixture {
	t.Helper()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(
		t.Context(),
		persistence.RuntimeWorkOwnershipActivation{
			OperationID:         "activate-m2-application-result-owner",
			WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
			DeploymentID:        "m2-application-result-owner",
			WorkerReleaseDigest: "sha256:" + strings.Repeat("d", 64),
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	workScope := persistence.RuntimeWorkScope{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}
	trustedScope := persistence.RuntimePlatformCommandTrustedScope{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		CallerWorkloadID:    config.AuthorizedWorkloadID,
		RuntimeCapabilityID: commandsemantic.CapabilityName,
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	store, err := persistence.NewRuntimePlatformCommandExecutionStore(
		db,
		m2RuntimeTransactionOwner{TransactionManager: transactions},
		owner,
		workScope,
		30*time.Second,
		3,
	)
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewRuntimePlatformCommandRepository(db)
	terminal := acceptM2ApplicationResultCommand(
		t,
		repository,
		workScope,
		trustedScope,
		"m2-application-result-terminal",
	)
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	lease, found, err := store.Claim(t.Context())
	if err != nil || !found || lease.Row.CommandID != terminal.CommandID {
		t.Fatalf(
			"M2 Runtime result Claim() = %#v, found %v, error %v",
			lease,
			found,
			err,
		)
	}
	if err := store.ExecuteApplication(
		t.Context(),
		lease,
		func(context.Context) (persistence.RuntimePlatformCommandApplicationCompletion, error) {
			return persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome: applicationresultsemantic.OutcomeApplied,
				Body: json.RawMessage(
					`{"commandId":"m2-application-result-terminal","status":"applied"}`,
				),
			}, nil
		},
	); err != nil {
		t.Fatal(err)
	}
	pending := acceptM2ApplicationResultCommand(
		t,
		repository,
		workScope,
		trustedScope,
		"m2-application-result-pending",
	)
	foreignScope := trustedScope
	foreignScope.CallerWorkloadID += "-foreign"
	foreign := acceptM2ApplicationResultCommand(
		t, repository, workScope, foreignScope, "m2-application-result-foreign",
	)
	return m2ApplicationResultFixture{
		Pending: pending, Terminal: terminal, Foreign: foreign,
	}
}

func acceptM2ApplicationResultCommand(
	t *testing.T,
	repository *persistence.RuntimePlatformCommandRepository,
	workScope persistence.RuntimeWorkScope,
	trustedScope persistence.RuntimePlatformCommandTrustedScope,
	commandID string,
) applicationresultsemantic.GetResultRequest {
	t.Helper()
	var inputDigest, executionRefDigest string
	switch commandID {
	case "m2-application-result-terminal":
		inputDigest = "sha256:5836df18151da5e4cd78eeada33dcd3ec42946961f25120b4f7efdf5f60e4fbe"
		executionRefDigest = "sha256:884b3c4056b27e10f5bacfc0068a1c0a177a728d72f26f88821412be53b461ed"
	case "m2-application-result-pending":
		inputDigest = "sha256:a3856a7503371adb54c2c1fd3c7d30d3a83ba0b72e20ee62a012619d4661a603"
		executionRefDigest = "sha256:082477c58a7fe1fc08aeabadcc482b9b52eff4963e26c7529e5cc23cdc613aab"
	case "m2-application-result-foreign":
		inputDigest = "sha256:8bb95d804121a66e20c318b54a9e2d64c19afc2bff61426495a4d0de21f2ca02"
		executionRefDigest = "sha256:2dbaa4cf2bf7d264776ac2420105f452c80ecf50979edf7f2f87c74bd9dce431"
	default:
		t.Fatalf("unknown M2 ApplicationResult fixture %q", commandID)
	}
	payload := fmt.Appendf(nil, `{"canonicalInputs":{"commandId":%q},"executionRef":{"definitionRef":"BoundedInbound","definitionVersion":"flow-v3","executionId":%q,"inputDigest":%q,"packGenerationId":"pack-generation-m2","servingEpochDigest":"sha256:%s"},"externalEffectsMode":"SUPPRESS_EXTERNAL_EFFECTS"}`,
		commandID, "execution-"+commandID, inputDigest, strings.Repeat("1", 64))
	executionScopeDigest, err := workScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	trustedScopeDigest, err := trustedScope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	_, duplicate, err := repository.Accept(
		t.Context(),
		persistence.RuntimePlatformCommandAcceptance{
			RowID:                commandID + "-row",
			OrgID:                workScope.OrgID,
			AppInstallationID:    workScope.AppInstallationID,
			CallerWorkloadID:     trustedScope.CallerWorkloadID,
			RuntimeCapabilityID:  workScope.RuntimeCapabilityID,
			MetadataBinding:      `{"binding":"m2-runtime-result"}`,
			ExecutionScopeDigest: executionScopeDigest,
			ScopeDigest:          trustedScopeDigest,
			CommandID:            commandID,
			IdempotencyKey:       commandID + "-idempotency",
			CommandKind:          flowcommandsemanticv3.CommandKind,
			PayloadFormatVersion: flowcommandsemanticv3.PayloadFormatVersion,
			RequestDigest:        "sha256:" + strings.Repeat("2", 64),
			Payload:              payload,
			Receipts:             []byte(`{"catalogRevision":41,"securityEpoch":1}`),
		},
	)
	if err != nil || duplicate {
		t.Fatalf("M2 Runtime Accept(%s) duplicate=%v error=%v", commandID, duplicate, err)
	}
	return applicationresultsemantic.GetResultRequest{
		CommandID: commandID, ExecutionRefDigest: executionRefDigest,
	}
}

func readM2ApplicationResult(
	t *testing.T,
	client *runtimegrpc.RuntimePlatformClient,
	request applicationresultsemantic.GetResultRequest,
) applicationresultsemantic.GetResultResponse {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	response, err := client.GetApplicationResult(ctx, request)
	if err != nil {
		t.Fatalf("M2 Runtime GetApplicationResult(%s): %v", request.CommandID, err)
	}
	return response
}

func assertM2ApplicationResultMissing(
	t *testing.T,
	client *runtimegrpc.RuntimePlatformClient,
	request applicationresultsemantic.GetResultRequest,
) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	_, err := client.GetApplicationResult(ctx, request)
	var failure *applicationresultsemantic.Failure
	if !errors.As(err, &failure) || failure.Category != commonsemantic.FailureMissing {
		t.Fatalf("M2 Runtime result miss for %#v = %v, want semantic MISSING", request, err)
	}
}

func assertM2ApplicationResults(
	t *testing.T,
	fixture m2ApplicationResultFixture,
	pending applicationresultsemantic.GetResultResponse,
	terminal applicationresultsemantic.GetResultResponse,
) {
	t.Helper()
	if pending.CommandID != fixture.Pending.CommandID ||
		pending.ExecutionRefDigest != fixture.Pending.ExecutionRefDigest ||
		pending.State != applicationresultsemantic.ResultPending || pending.Result != nil {
		t.Fatalf("M2 pending result = %#v for %#v", pending, fixture.Pending)
	}
	if terminal.CommandID != fixture.Terminal.CommandID ||
		terminal.ExecutionRefDigest != fixture.Terminal.ExecutionRefDigest ||
		terminal.State != applicationresultsemantic.ResultTerminal || terminal.Result == nil ||
		terminal.Result.Outcome != applicationresultsemantic.OutcomeApplied ||
		applicationresultsemantic.ValidateApplicationResult(*terminal.Result) != nil {
		t.Fatalf("M2 terminal result = %#v for %#v", terminal, fixture.Terminal)
	}
}

func assertM2ApplicationResultBytes(
	t *testing.T,
	stage string,
	want []byte,
	response applicationresultsemantic.GetResultResponse,
) {
	t.Helper()
	got, err := json.Marshal(response)
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(got, want) {
		t.Fatalf("M2 %s result bytes = %s, want %s", stage, got, want)
	}
}

func assertM2RuntimeCapability(t *testing.T, adminAddress string) {
	t.Helper()
	ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
	defer cancel()
	request, err := http.NewRequestWithContext(
		ctx,
		http.MethodGet,
		"http://"+adminAddress+processhttp.CapabilitiesPath,
		nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	response, err := http.DefaultClient.Do(request)
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	body, err := io.ReadAll(response.Body)
	if err != nil {
		t.Fatal(err)
	}
	if response.StatusCode != http.StatusOK ||
		!bytes.Contains(body, []byte(`"`+applicationresultsemantic.CapabilityName+`"`)) {
		t.Fatalf(
			"M2 Runtime capabilities status/body = %d/%s, want %q",
			response.StatusCode,
			body,
			applicationresultsemantic.CapabilityName,
		)
	}
}

func newM2RuntimeHandoff(
	t *testing.T,
	db *sql.DB,
	client platformClient,
	config Config,
) *runtimehandoff.Handoff {
	t.Helper()
	view, err := readPlatformAuthority(t.Context(), client, config)
	if err != nil {
		t.Fatal(err)
	}
	store, err := runtimehandoff.NewStore(db)
	if err != nil {
		t.Fatal(err)
	}
	handoff, err := runtimehandoff.NewHandoff(
		newPinnedPlatformClient(client, view), store,
		runtimehandoff.Config{
			OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
			AuthorizedWorkloadID: config.AuthorizedWorkloadID,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := handoff.VerifyReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	return handoff
}

func TestM2IntegrationPlatformRuntimeChildProcess(t *testing.T) {
	if os.Getenv(m2RuntimeChildModeEnv) != "1" {
		t.Skip("M2 Runtime child is launched by the parent integration test")
	}
	var view catalogsemantic.IntegrationBindingView
	readM2JSON(t, os.Getenv(m2RuntimeViewFileEnv), &view)
	runtimeView, outboundRuntimeView := m2RuntimeConsumerViews(t, view)
	if string(runtimeView.SourceSnapshotDigest()) != view.Lineage.SourceSnapshotDigest {
		t.Fatalf(
			"Runtime command/Catalog source digest = %q/%q",
			runtimeView.SourceSnapshotDigest(), view.Lineage.SourceSnapshotDigest,
		)
	}
	certificate, err := tls.LoadX509KeyPair(
		os.Getenv(m2RuntimeCertFileEnv), os.Getenv(m2RuntimeKeyFileEnv),
	)
	if err != nil {
		t.Fatal(err)
	}
	db, err := sql.Open("mysql", os.Getenv(m2RuntimeDatabaseDSNEnv))
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	if err := db.PingContext(t.Context()); err != nil {
		t.Fatal(err)
	}
	engine := &m2RuntimeEngine{
		view: view, eventsFile: os.Getenv(m2RuntimeEventsFileEnv),
		commands: persistence.NewRuntimePlatformCommandRepository(db),
	}
	catalogAdmission, err := runtimeapi.NewCatalogConsumerAdmission(
		[]runtimeapi.CatalogConsumerSelection{{
			OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
			WorkloadID:   view.AuthorizedWorkloadID,
			CapabilityID: catalogsemantic.IntegrationBindingCapabilityName,
		}},
	)
	if err != nil {
		t.Fatal(err)
	}
	commandAdmission, err := runtimeapi.NewCommandConsumerAdmission(
		t.Context(),
		runtimeapi.CommandConsumerSelection{
			OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
			WorkloadID: view.AuthorizedWorkloadID, RuntimeMetadataView: runtimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	outboundAdmission, err := runtimeapi.NewOutboundIntentAdmission(
		t.Context(),
		runtimeapi.OutboundIntentSelection{
			OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
			WorkloadID: view.AuthorizedWorkloadID, RuntimeMetadataView: outboundRuntimeView,
		},
		engine,
	)
	if err != nil {
		t.Fatal(err)
	}
	applicationResultAdmission, err := runtimeapi.NewApplicationResultAdmission(
		runtimeapi.ApplicationResultSelection{
			AppInstallationID: view.AppInstallationID,
			WorkloadID:        view.AuthorizedWorkloadID,
		},
		func(context.Context, string, string) (persistence.RuntimeSecurityIssuanceState, error) {
			now := time.Now()
			return persistence.RuntimeSecurityIssuanceState{
				StateFound: true, BindingFound: true, BindingActive: true,
				OrgID: view.OrgID, Capabilities: []string{applicationresultsemantic.CapabilityName},
				SecurityEpoch: 1, LeaseFound: true, LeaseSecurityEpoch: 1,
				LeaseExpiresAt: now.Add(time.Hour), AuthorityNow: now,
			}, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	transactionOwner := m2RuntimeTransactionOwner{
		TransactionManager: persistence.NewTransactionManager(database.NewConnectionFromDB(db)),
	}
	process, err := runtimeapi.Assemble(runtimeapi.ProcessConfig{
		Version: m2RuntimeProcessVersion, AdminAddress: "127.0.0.1:0",
		GRPCAddress: "127.0.0.1:0", AdmissionLimit: 3,
		WorkloadAuthority: func(ctx context.Context, _ string) (runtimegrpc.VerifiedCaller, error) {
			values, _ := metadata.FromIncomingContext(ctx)
			authorization := values.Get("authorization")
			if len(authorization) != 1 || authorization[0] != "Bearer "+m2RuntimeWorkloadToken {
				return "", errors.New("M2 Runtime child: invalid workload credential")
			}
			return runtimegrpc.VerifiedCaller(view.AuthorizedWorkloadID), nil
		},
		SecurityAdmission:          m2RuntimeSecurityAdmission(t, view),
		CatalogConsumerAdmission:   catalogAdmission,
		CommandConsumerAdmission:   commandAdmission,
		OutboundIntentAdmission:    outboundAdmission,
		ApplicationResultAdmission: applicationResultAdmission,
		CatalogConsumerViews:       engine,
		CommandEngine:              engine,
		OutboundIntents:            engine,
		OutboundResultCommitter:    engine,
		OutboundIntentTransactions: transactionOwner,
		ApplicationResults:         engine,
		ServingEnabled:             true,
		TransportCredentials: credentials.NewTLS(&tls.Config{
			MinVersion: tls.VersionTLS13, Certificates: []tls.Certificate{certificate},
		}),
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			catalogsemantic.IntegrationBindingCapabilityName: time.Second,
			commandsemantic.CapabilityName:                   time.Second,
			outboundintentsemantic.CapabilityName:            time.Second,
			applicationresultsemantic.CapabilityName:         time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassCatalogWatch: 1,
		},
	}, &m2RecordsReader{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	if !process.Ready() {
		t.Fatal("assembled Runtime process did not become ready")
	}
	writeM2JSON(t, os.Getenv(m2RuntimeReadyFileEnv), m2RuntimeChildReady{
		PID: os.Getpid(), GRPCAddress: process.GRPCAddress(),
		AdminAddress: process.AdminAddress(),
	})

	stop := make(chan os.Signal, 1)
	signal.Notify(stop, os.Interrupt, syscall.SIGTERM)
	defer signal.Stop(stop)
	<-stop
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := process.Stop(shutdownCtx); err != nil {
		t.Fatal(err)
	}
}

type m2RuntimeEngine struct {
	view       catalogsemantic.IntegrationBindingView
	eventsFile string
	commands   *persistence.RuntimePlatformCommandRepository
	mu         sync.Mutex
}

func (s *m2RuntimeEngine) GetIntegrationBindingView(
	ctx context.Context,
	_ runtimeapi.CatalogConsumerScope,
	_ catalogsemantic.GetRequest,
) (catalogsemantic.IntegrationBindingView, error) {
	if err := ctx.Err(); err != nil {
		return catalogsemantic.IntegrationBindingView{}, err
	}
	s.appendEvent(m2RuntimeEvent{PID: os.Getpid(), Method: "GetIntegrationBindingView"})
	return s.view, nil
}

func (s *m2RuntimeEngine) Accept(
	ctx context.Context,
	input persistence.RuntimePlatformCommandAcceptance,
) (persistence.RuntimePlatformCommandRow, bool, error) {
	row, duplicate, err := s.commands.Accept(ctx, input)
	if err != nil {
		return row, duplicate, err
	}
	revision := s.view.Lineage.CatalogRevision
	s.appendEvent(m2RuntimeEvent{PID: os.Getpid(), Method: "AcceptCommand", Command: commandsemantic.Request{
		CommandID: input.CommandID, IdempotencyKey: input.IdempotencyKey,
		CommandKind: input.CommandKind, PartitionKey: input.PartitionKey,
		PayloadFormatVersion: input.PayloadFormatVersion,
		Payload:              append([]byte(nil), input.Payload...),
		PayloadDigest:        commandsemantic.ComputePayloadDigest(input.Payload),
		MinCatalogRevision:   &revision,
	}})
	return row, duplicate, nil
}

func (s *m2RuntimeEngine) GetApplicationResult(
	ctx context.Context,
	scope persistence.RuntimePlatformCommandTrustedScope,
	request applicationresultsemantic.GetResultRequest,
) (applicationresultsemantic.GetResultResponse, error) {
	return s.commands.GetApplicationResult(ctx, scope, request)
}

func (s *m2RuntimeEngine) ListPending(
	ctx context.Context,
	_ persistence.RuntimeOutboundIntentScope,
	_ uint64,
	_ uint32,
) ([]persistence.RuntimeOutboundIntentRow, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return nil, nil
}

func (s *m2RuntimeEngine) ReadGeneration(
	ctx context.Context,
	_ persistence.RuntimeOutboundIntentScope,
) (uint64, error) {
	if err := ctx.Err(); err != nil {
		return 0, err
	}
	return 0, nil
}

func (s *m2RuntimeEngine) Acknowledge(
	context.Context,
	persistence.RuntimeOutboundIntentScope,
	outboundintentsemantic.AcknowledgeRequest,
) (persistence.RuntimeOutboundIntentRow, bool, error) {
	return persistence.RuntimeOutboundIntentRow{}, false, persistence.ErrRuntimeOutboundIntentNotFound
}

func (s *m2RuntimeEngine) Commit(
	context.Context,
	*catalogsnapshot.RuntimeMetadataView,
	persistence.RuntimeOutboundIntentScope,
	outboundintentsemantic.CommitResultRequest,
) (bool, error) {
	return false, persistence.ErrRuntimeOutboundIntentNotFound
}

func (s *m2RuntimeEngine) appendEvent(event m2RuntimeEvent) {
	s.mu.Lock()
	defer s.mu.Unlock()
	file, err := os.OpenFile(s.eventsFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
	if err != nil {
		panic(err)
	}
	defer file.Close()
	if err := json.NewEncoder(file).Encode(event); err != nil {
		panic(err)
	}
}

func (s *m2RuntimeEngine) GetAppMetadataView(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataView, error) {
	return catalogsemantic.AppMetadataView{}, errors.New("M2 Runtime fixture: AppMetadataView is not selected")
}

func (s *m2RuntimeEngine) NextAppMetadataView(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.WatchRequest,
) (catalogsemantic.AppMetadataView, error) {
	return catalogsemantic.AppMetadataView{}, errors.New("M2 Runtime fixture: AppMetadataView watch is not selected")
}

func (s *m2RuntimeEngine) GetAppMetadataViewV2(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV2, error) {
	return catalogsemantic.AppMetadataViewV2{}, errors.New("M2 Runtime fixture: AppMetadataViewV2 is not selected")
}

func (s *m2RuntimeEngine) NextAppMetadataViewV2(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.WatchRequest,
) (catalogsemantic.AppMetadataViewV2, error) {
	return catalogsemantic.AppMetadataViewV2{}, errors.New("M2 Runtime fixture: AppMetadataViewV2 watch is not selected")
}

func (s *m2RuntimeEngine) GetAppMetadataViewV3(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV3, error) {
	return catalogsemantic.AppMetadataViewV3{}, errors.New("M2 Runtime fixture: AppMetadataViewV3 is not selected")
}

func (s *m2RuntimeEngine) NextAppMetadataViewV3(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.WatchRequest,
) (catalogsemantic.AppMetadataViewV3, error) {
	return catalogsemantic.AppMetadataViewV3{}, errors.New("M2 Runtime fixture: AppMetadataViewV3 watch is not selected")
}

func (s *m2RuntimeEngine) NextIntegrationBindingView(
	context.Context, runtimeapi.CatalogConsumerScope, catalogsemantic.WatchRequest,
) (catalogsemantic.IntegrationBindingView, error) {
	return catalogsemantic.IntegrationBindingView{}, errors.New("M2 Runtime fixture: IntegrationBindingView watch is not selected")
}

type m2RecordsReader struct{}

func (*m2RecordsReader) ReadSelectedRecord(
	context.Context, string, string, []string, *models.UserSession,
) (*records.SelectedRootRecord, error) {
	return nil, errors.New("M2 Runtime fixture: records capability is not selected")
}

func (*m2RecordsReader) EvaluateRecordAccess(
	context.Context, *string, []string, *models.UserSession,
) (map[string]contract.RecordAccess, error) {
	return nil, errors.New("M2 Runtime fixture: record access capability is not selected")
}

func m2RuntimeSecurityAdmission(
	t *testing.T,
	view catalogsemantic.IntegrationBindingView,
) *runtimeapi.SecurityAdmission {
	t.Helper()
	now := time.Now
	state, err := runtimeapi.NewRuntimeSecurityState(t.Context(), runtimeapi.RuntimeSecurityStateConfig{
		Now: now,
		Replica: func(context.Context) (runtimeapi.RuntimeReplicaSecurityState, bool, error) {
			return runtimeapi.RuntimeReplicaSecurityState{
				OrgID: view.OrgID, RuntimeAudience: "m2-runtime",
				SecurityEpoch: 1, LeaseSecurityEpoch: 1,
				LeaseExpiresAt: now().Add(time.Hour),
			}, true, nil
		},
		Admission: func(_ context.Context, claims delegationv1.Claims) (runtimeapi.RuntimeAdmissionSecurityState, error) {
			installationFound := claims.AppInstallationID == view.AppInstallationID
			return runtimeapi.RuntimeAdmissionSecurityState{
				StateFound: true, InstallationFound: installationFound,
				InstallationActive: installationFound,
				Capabilities: []string{
					catalogsemantic.IntegrationBindingCapabilityName,
					commandsemantic.CapabilityName,
					applicationresultsemantic.CapabilityName,
					outboundintentsemantic.CapabilityName,
				},
				SecurityEpoch: 1, LeaseFound: true, LeaseSecurityEpoch: 1,
				LeaseExpiresAt: now().Add(time.Hour),
			}, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	privateKey := ed25519.NewKeyFromSeed(bytes.Repeat([]byte{0x51}, ed25519.SeedSize))
	verifier, err := delegationv1.NewVerifier(delegationv1.VerifierConfig{
		Keys: []delegationv1.VerificationKey{{
			SigningKeyID: "m2-test-key", PublicKey: privateKey.Public().(ed25519.PublicKey),
			State: delegationv1.KeyStateCurrent,
		}},
		SupportedCapabilities: []string{
			catalogsemantic.IntegrationBindingCapabilityName,
			commandsemantic.CapabilityName,
			applicationresultsemantic.CapabilityName,
			outboundintentsemantic.CapabilityName,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	admission, err := runtimeapi.NewSecurityAdmission(runtimeapi.SecurityAdmissionConfig{
		Verifier: verifier, State: state,
		Bindings: func(_ context.Context, caller runtimegrpc.VerifiedCaller) (runtimeapi.RuntimeCallerBinding, bool, error) {
			return runtimeapi.RuntimeCallerBinding{CallerWorkloadID: view.AuthorizedWorkloadID},
				caller == runtimegrpc.VerifiedCaller(view.AuthorizedWorkloadID), nil
		},
		Sessions: func(context.Context, string, string) (*runtimeapi.RuntimeSessionState, error) {
			return nil, errors.New("M2 Runtime fixture: workload consumer cannot open a user session")
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	return admission
}

func m2RuntimeConsumerViews(
	t *testing.T,
	view catalogsemantic.IntegrationBindingView,
) (*catalogsnapshot.RuntimeMetadataView, *catalogsnapshot.RuntimeMetadataView) {
	t.Helper()
	var commandBinding *catalogsemantic.RuntimeCommandBinding
	contextNamespace := ""
	for index := range view.Bindings {
		if view.Bindings[index].RuntimeCommand != nil {
			commandBinding = view.Bindings[index].RuntimeCommand
			contextNamespace = view.Bindings[index].OwnerPack
			break
		}
	}
	if commandBinding == nil {
		t.Fatal("M2 Runtime fixture has no command binding")
	}
	revision := catalogsnapshot.CatalogRevision(view.Lineage.CatalogRevision)
	componentID := "component.m2.integration-command"
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: view.OrgID,
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision:  revision,
			SchemaGeneration: catalogsnapshot.SchemaGeneration(view.Lineage.SchemaGeneration),
			EngineGeneration: catalogsnapshot.EngineGeneration(view.Lineage.EngineGeneration),
		},
		Objects: []catalogsnapshot.ObjectInput{{
			CatalogRevision: revision,
			Metadata: &models.ObjectMetadata{
				ID: "object-m2-work-item", APIName: "WorkItem",
				Label: "Work Item", PluralLabel: "Work Items",
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{{
					ID: "field-m2-work-item-id", APIName: constants.FieldID,
					Label: "ID", Type: constants.FieldTypeString,
					PhysicalName: "id", IsSystem: true,
				}},
			},
		}},
		CommandRegistrations: []catalogsnapshot.CommandRegistrationInput{{
			CatalogRevision: revision, ComponentID: componentID, DefinitionVersion: "1",
			ContentDigest:        catalogsnapshot.ComputeDigest([]byte("m2.integration-command/v1")),
			CommandKind:          commandBinding.CommandKind,
			PayloadFormatVersion: commandBinding.PayloadFormatVersion,
			ExecutionBindings: []catalogsnapshot.CommandExecutionBinding{{
				AuthorizedWorkloadID: view.AuthorizedWorkloadID,
				DefinitionRef:        commandBinding.DefinitionRef,
				DefinitionVersion:    commandBinding.DefinitionVersion,
				ContextNamespace:     contextNamespace,
			}},
		}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{
				CatalogRevision: revision, AppInstallationID: view.AppInstallationID,
				CapabilityID:           commandsemantic.CapabilityName,
				Profile:                catalogsnapshot.ProjectionCommandAcceptanceV1,
				DefinitionComponentIDs: []string{componentID},
			},
			{
				CatalogRevision: revision, AppInstallationID: view.AppInstallationID,
				CapabilityID:   outboundintentsemantic.CapabilityName,
				Profile:        catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{"WorkItem"},
			},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	commandArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: view.AppInstallationID, CapabilityID: commandsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	runtimeView, err := catalogsnapshot.OpenRuntimeMetadataView(commandArtifact)
	if err != nil {
		t.Fatal(err)
	}
	if err := runtimeView.VerifySource(snapshot); err != nil {
		t.Fatal(err)
	}
	outboundArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: view.AppInstallationID,
		CapabilityID:      outboundintentsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	outboundRuntimeView, err := catalogsnapshot.OpenRuntimeMetadataView(outboundArtifact)
	if err != nil {
		t.Fatal(err)
	}
	if err := outboundRuntimeView.VerifySource(snapshot); err != nil {
		t.Fatal(err)
	}
	return runtimeView, outboundRuntimeView
}

func m2RuntimeDatabaseDSN(t *testing.T, db *sql.DB) string {
	t.Helper()
	var databaseName string
	if err := db.QueryRowContext(t.Context(), "SELECT DATABASE()").Scan(&databaseName); err != nil {
		t.Fatal(err)
	}
	port := os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	config := mysql.Config{
		User: os.Getenv("TIDB_USER"), Passwd: os.Getenv("TIDB_PASSWORD"),
		Net: "tcp", Addr: net.JoinHostPort(os.Getenv("TIDB_HOST"), port),
		DBName: databaseName, ParseTime: true, Loc: time.UTC, AllowNativePasswords: true,
	}
	return config.FormatDSN()
}

func startM2RuntimeChild(
	t *testing.T,
	viewFile, certFile, keyFile, eventsFile, databaseDSN string,
) *m2RuntimeChild {
	t.Helper()
	readyFile := filepath.Join(
		t.TempDir(), fmt.Sprintf("runtime-ready-%d.json", time.Now().UnixNano()),
	)
	command := exec.Command(
		os.Args[0], "-test.run=^"+m2RuntimeChildTestName+"$", "-test.v",
	)
	command.Env = replaceM2ChildEnvironment(
		os.Environ(),
		m2RuntimeChildModeEnv+"=1",
		m2RuntimeReadyFileEnv+"="+readyFile,
		m2RuntimeEventsFileEnv+"="+eventsFile,
		m2RuntimeViewFileEnv+"="+viewFile,
		m2RuntimeDatabaseDSNEnv+"="+databaseDSN,
		m2RuntimeCertFileEnv+"="+certFile,
		m2RuntimeKeyFileEnv+"="+keyFile,
	)
	child := &m2RuntimeChild{command: command, wait: make(chan error, 1)}
	command.Stdout, command.Stderr = &child.logs, &child.logs
	if err := command.Start(); err != nil {
		t.Fatal(err)
	}
	go func() { child.wait <- command.Wait() }()
	t.Cleanup(func() { child.stop(t) })

	deadline := time.Now().Add(20 * time.Second)
	for time.Now().Before(deadline) {
		var ready m2RuntimeChildReady
		if data, err := os.ReadFile(readyFile); err == nil &&
			json.Unmarshal(data, &ready) == nil && ready.GRPCAddress != "" &&
			ready.AdminAddress != "" {
			if ready.PID == 0 || ready.PID == os.Getpid() {
				t.Fatalf("M2 Runtime child PID = %d, parent PID = %d", ready.PID, os.Getpid())
			}
			child.address = ready.GRPCAddress
			child.admin = ready.AdminAddress
			return child
		}
		select {
		case err := <-child.wait:
			child.mu.Lock()
			child.stopped = true
			child.mu.Unlock()
			t.Fatalf("M2 Runtime child exited before ready: %v\n%s", err, child.logs.String())
		default:
		}
		time.Sleep(25 * time.Millisecond)
	}
	t.Fatalf("M2 Runtime child did not become ready\n%s", child.logs.String())
	return nil
}

func (c *m2RuntimeChild) stop(t *testing.T) {
	t.Helper()
	if c == nil {
		return
	}
	c.mu.Lock()
	if c.stopped {
		c.mu.Unlock()
		return
	}
	c.stopped = true
	c.mu.Unlock()
	if err := c.command.Process.Signal(syscall.SIGTERM); err != nil {
		t.Fatalf("signal M2 Runtime child: %v", err)
	}
	select {
	case err := <-c.wait:
		if err != nil {
			t.Fatalf("M2 Runtime child exit: %v\n%s", err, c.logs.String())
		}
	case <-time.After(10 * time.Second):
		_ = c.command.Process.Kill()
		t.Fatalf("M2 Runtime child did not drain\n%s", c.logs.String())
	}
}

func dialM2RuntimeClient(
	t *testing.T,
	address, caFile, appInstallationID string,
) *runtimegrpc.RuntimePlatformClient {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	client, err := runtimegrpc.DialPlatformRuntime(ctx, runtimegrpc.RuntimeClientConfig{
		Target: "passthrough:///" + address, AppInstallationID: appInstallationID,
		WorkloadToken: m2RuntimeWorkloadToken, CAFile: caFile,
		ServerName: m2RuntimeServerName, AttemptTimeout: time.Second,
	})
	if err != nil {
		t.Fatal(err)
	}
	return client
}

func writeM2RuntimeTLSFiles(
	t *testing.T,
	dir string,
) (caFile, certFile, keyFile string) {
	t.Helper()
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now()
	template := &x509.Certificate{
		SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: m2RuntimeServerName},
		NotBefore: now.Add(-time.Minute), NotAfter: now.Add(time.Hour),
		KeyUsage:    x509.KeyUsageDigitalSignature,
		ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
		DNSNames:    []string{m2RuntimeServerName},
	}
	der, err := x509.CreateCertificate(
		rand.Reader, template, template, &key.PublicKey, key,
	)
	if err != nil {
		t.Fatal(err)
	}
	keyDER, err := x509.MarshalPKCS8PrivateKey(key)
	if err != nil {
		t.Fatal(err)
	}
	certificatePEM := pem.EncodeToMemory(
		&pem.Block{Type: "CERTIFICATE", Bytes: der},
	)
	keyPEM := pem.EncodeToMemory(
		&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER},
	)
	caFile = filepath.Join(dir, "runtime-ca.pem")
	certFile = filepath.Join(dir, "runtime-cert.pem")
	keyFile = filepath.Join(dir, "runtime-key.pem")
	for path, content := range map[string][]byte{
		caFile: certificatePEM, certFile: certificatePEM, keyFile: keyPEM,
	} {
		if err := os.WriteFile(path, content, 0o600); err != nil {
			t.Fatal(err)
		}
	}
	return caFile, certFile, keyFile
}

func writeM2JSON(t *testing.T, path string, value any) {
	t.Helper()
	data, err := json.Marshal(value)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatal(err)
	}
}

func readM2JSON(t *testing.T, path string, target any) {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	if err := json.Unmarshal(data, target); err != nil {
		t.Fatal(err)
	}
}

func readM2RuntimeEvents(t *testing.T, path string) []m2RuntimeEvent {
	t.Helper()
	events := readM2RuntimeEventsFile(path)
	if events == nil {
		t.Fatalf("read M2 Runtime events from %s", path)
	}
	return events
}

func readM2RuntimeEventsFile(path string) []m2RuntimeEvent {
	file, err := os.Open(path)
	if errors.Is(err, os.ErrNotExist) {
		return []m2RuntimeEvent{}
	}
	if err != nil {
		return nil
	}
	defer file.Close()
	events := []m2RuntimeEvent{}
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		var event m2RuntimeEvent
		if json.Unmarshal(scanner.Bytes(), &event) != nil {
			return nil
		}
		events = append(events, event)
	}
	if scanner.Err() != nil {
		return nil
	}
	return events
}

func m2RuntimeCommands(events []m2RuntimeEvent) []commandsemantic.Request {
	commands := []commandsemantic.Request{}
	for _, event := range events {
		if event.Method == "AcceptCommand" {
			commands = append(commands, event.Command)
		}
	}
	return commands
}

func countM2RuntimeEvents(events []m2RuntimeEvent, method string) int {
	count := 0
	for _, event := range events {
		if event.Method == method {
			count++
		}
	}
	return count
}

func assertM2CommandSharesPlatformEndpoint(
	t *testing.T,
	events []m2RuntimeEvent,
	command commandsemantic.Request,
	view catalogsemantic.IntegrationBindingView,
) {
	t.Helper()
	if command.MinCatalogRevision == nil ||
		*command.MinCatalogRevision != view.Lineage.CatalogRevision {
		t.Fatalf(
			"Runtime command Catalog revision = %v, want %d",
			command.MinCatalogRevision, view.Lineage.CatalogRevision,
		)
	}
	commandPID := 0
	for _, event := range events {
		if event.Method == "AcceptCommand" && event.Command.CommandID == command.CommandID {
			commandPID = event.PID
			break
		}
	}
	if commandPID == 0 || commandPID == os.Getpid() {
		t.Fatalf("Runtime command PID = %d, parent PID = %d", commandPID, os.Getpid())
	}
	for _, event := range events {
		if event.Method == "GetIntegrationBindingView" && event.PID == commandPID {
			return
		}
	}
	t.Fatalf(
		"Runtime command PID %d did not serve its Catalog authority on the same Platform endpoint: %#v",
		commandPID, events,
	)
}

func assertM2RuntimeRestartedOnOnePlatformEndpoint(
	t *testing.T,
	events []m2RuntimeEvent,
) {
	t.Helper()
	commandPIDs := map[int]struct{}{}
	catalogPIDs := map[int]struct{}{}
	for _, event := range events {
		switch event.Method {
		case "AcceptCommand":
			commandPIDs[event.PID] = struct{}{}
		case "GetIntegrationBindingView":
			catalogPIDs[event.PID] = struct{}{}
		}
	}
	if len(commandPIDs) != 2 {
		t.Fatalf("Runtime command owners across restart = %v, want two", commandPIDs)
	}
	for pid := range commandPIDs {
		if pid == 0 || pid == os.Getpid() {
			t.Fatalf("Runtime command owner PID = %d, parent PID = %d", pid, os.Getpid())
		}
		if _, found := catalogPIDs[pid]; !found {
			t.Fatalf("Runtime command owner PID %d did not serve Catalog on the same endpoint", pid)
		}
	}
}

func replaceM2ChildEnvironment(base []string, assignments ...string) []string {
	replacements := make(map[string]string, len(assignments))
	for _, assignment := range assignments {
		key, _, _ := strings.Cut(assignment, "=")
		replacements[key] = assignment
	}
	result := make([]string, 0, len(base)+len(assignments))
	for _, entry := range base {
		key, _, _ := strings.Cut(entry, "=")
		if _, replaced := replacements[key]; replaced {
			continue
		}
		result = append(result, entry)
	}
	for _, assignment := range assignments {
		result = append(result, assignment)
	}
	return result
}