//go:build runtime_worker_integration

package runtimeworker

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"runtime/debug"
	"strings"
	"sync"
	"testing"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/automation"
	recordmutation "github.com/nexuscrm/backend/internal/runtimeassembly/recordmutation"
	"github.com/nexuscrm/backend/internal/runtimeconfig"
	"github.com/nexuscrm/backend/internal/runtimepermissionstate"
	"github.com/nexuscrm/backend/internal/runtimeprocess"
	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/files"
	"github.com/nexuscrm/kernel/pkg/ports"
	"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/nexuscrm/storage"
	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	flowcommandsemanticv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
)

const integrationWorkerDigest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
const integrationWorkerSourceRevision = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
const integrationScheduledAppCapability = "io.nexuscrm.test.scheduled-flow.v1"

type runtimeWorkerScheduledRunTransactionOwner struct {
	*persistence.TransactionManager
}

func (o runtimeWorkerScheduledRunTransactionOwner) 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
}

type platformCommandIntegrationExecutor struct {
	transactions *persistence.TransactionManager
	mu           *sync.Mutex
	attempts     map[string]int
}

func (executor platformCommandIntegrationExecutor) ExecutePlatformCommand(
	ctx context.Context,
	command PlatformCommandExecution,
) ([]byte, error) {
	if executor.transactions == nil {
		return nil, errors.New("test Platform-command executor has no transaction authority")
	}
	tx, ok := executor.transactions.ExtractTx(ctx).(*sql.Tx)
	if !ok || tx == nil {
		return nil, errors.New("test Platform-command executor ran outside the Worker-owned transaction")
	}
	if strings.Contains(command.CommandID, "permanent") {
		return nil, &PlatformCommandExecutionError{
			Cause:     errors.New("test payload is permanently invalid"),
			Permanent: true,
		}
	}
	if strings.Contains(command.CommandID, "retry") {
		if executor.mu == nil || executor.attempts == nil {
			return nil, errors.New("test retry executor has no attempt authority")
		}
		executor.mu.Lock()
		executor.attempts[command.CommandID]++
		attempt := executor.attempts[command.CommandID]
		executor.mu.Unlock()
		if attempt == 1 {
			return nil, errors.New("test executor is temporarily unavailable")
		}
	}
	if _, err := tx.ExecContext(
		ctx,
		`INSERT INTO RuntimePlatformCommandEffect
(CommandId, CommandKind, PayloadFormatVersion, Payload)
VALUES (?, ?, ?, ?)`,
		command.CommandID,
		command.CommandKind,
		command.PayloadFormatVersion,
		command.Payload,
	); err != nil {
		return nil, err
	}
	if strings.Contains(command.CommandID, "oversized") {
		return make([]byte, (1<<20)+1), nil
	}
	return []byte(`{"status":"applied"}`), nil
}

func TestRuntimeWorkerRealTiDBFenceRecoveryAndLifecycle(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}

	auditOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-audit-one", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
		DeploymentID: "runtime-audit-one", WorkerReleaseDigest: integrationWorkerDigest,
	})
	firstOutbox := insertAuditOutbox(t, db, "ISS-D2B-1")
	process, workerDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision, integrationWorkerConfig(auditOwner), permissionStateFactory)
	if err != nil {
		t.Fatal(err)
	}
	assertAuditCounts(t, db, firstOutbox, 1, 0)
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	assertWorkerOperationalSurface(t, process, auditOwner.WorkFamily)
	waitForAuditCounts(t, db, firstOutbox, 0, 1)

	// Transfer the durable family while the queue is empty. Process health must
	// continuously follow the durable authority rather than waiting for another
	// work item to happen to exercise a transactional claim fence.
	newOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-audit-two", WorkFamily: auditOwner.WorkFamily,
		DeploymentID: "runtime-audit-two", WorkerReleaseDigest: integrationWorkerDigest,
		ExpectedOwnershipEpoch: auditOwner.OwnershipEpoch,
	})
	select {
	case err := <-waitProcess(process):
		if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
			t.Fatalf("old worker exit = %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("idle old worker did not stop after ownership transfer")
	}
	secondOutbox := insertAuditOutbox(t, db, "ISS-D2B-2")
	assertAuditCounts(t, db, secondOutbox, 1, 0)
	stopWorker(t, process, workerDB)

	recovery, recoveryDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision, integrationWorkerConfig(newOwner), permissionStateFactory)
	if err != nil {
		t.Fatal(err)
	}
	if err := recovery.Start(); err != nil {
		_ = recoveryDB.Close()
		t.Fatal(err)
	}
	waitForAuditCounts(t, db, secondOutbox, 0, 1)
	stopWorker(t, recovery, recoveryDB)

	lifecycleOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-command-lifecycle", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
		DeploymentID: "runtime-command-lifecycle", WorkerReleaseDigest: integrationWorkerDigest,
	})
	receipts, err := json.Marshal(semantic.Receipts{
		ProtocolVersion: "1.0", CatalogRevision: 7, SchemaGeneration: 2,
		EngineGeneration: 3, SecurityEpoch: 11, CapabilityVersion: semantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeRecordCommand
(Id, ScopeDigest, CommandId, RequestDigest, BatchMode, OperationCount, State, Receipts, ResultPayload, OutcomeExpiresAt, OutcomeRetentionMicros)
VALUES ('command-accepted', REPEAT('c',64), 'command-accepted-1', CONCAT('sha256:', REPEAT('d',64)), 'INDEPENDENT', 1, 'ACCEPTED', ?, NULL, UTC_TIMESTAMP(6) - INTERVAL 1 SECOND, 3600000000)`, receipts); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeRecordCommandOperation
(Id, CommandRowId, OperationIndex, OperationKind, OperationObjectId, BoundRecordId, State, ResultPayload)
VALUES ('operation-accepted', 'command-accepted', 0, 'UPDATE', 'Issue', 'issue-accepted', 'PENDING', NULL)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeRecordCommand
(Id, ScopeDigest, CommandId, RequestDigest, BatchMode, OperationCount, State, Receipts, ResultPayload, OutcomeExpiresAt, OutcomeRetentionMicros)
VALUES ('command-expired', REPEAT('a',64), 'command-1', CONCAT('sha256:', REPEAT('b',64)), 'INDEPENDENT', 1, 'COMPLETED', '{}', '{}', UTC_TIMESTAMP(6) - INTERVAL 1 SECOND, 3600000000)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO _System_RuntimeRecordCommandOperation
(Id, CommandRowId, OperationIndex, OperationKind, State, ResultPayload)
VALUES ('operation-expired', 'command-expired', 0, 'UPDATE', 'COMPLETED', '{}')`); err != nil {
		t.Fatal(err)
	}
	lifecycle, lifecycleDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision, integrationWorkerConfig(lifecycleOwner), permissionStateFactory)
	if err != nil {
		t.Fatal(err)
	}
	if err := lifecycle.Start(); err != nil {
		_ = lifecycleDB.Close()
		t.Fatal(err)
	}
	deadline := time.Now().Add(5 * time.Second)
	for {
		var commands, operations, acceptedCommands, acceptedOperations int
		err := db.QueryRow(`SELECT
(SELECT COUNT(*) FROM _System_RuntimeRecordCommand WHERE Id = 'command-expired'),
(SELECT COUNT(*) FROM _System_RuntimeRecordCommandOperation WHERE Id = 'operation-expired'),
(SELECT COUNT(*) FROM _System_RuntimeRecordCommand WHERE Id = 'command-accepted' AND State = 'COMPLETED' AND ResultPayload IS NOT NULL AND OutcomeExpiresAt > UTC_TIMESTAMP(6)),
(SELECT COUNT(*) FROM _System_RuntimeRecordCommandOperation WHERE Id = 'operation-accepted' AND State = 'COMPLETED' AND ResultPayload IS NOT NULL)`).
			Scan(&commands, &operations, &acceptedCommands, &acceptedOperations)
		if err == nil && commands == 0 && operations == 0 && acceptedCommands == 1 && acceptedOperations == 1 {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf("command lifecycle did not converge expired=%d/%d accepted=%d/%d err=%v", commands, operations, acceptedCommands, acceptedOperations, err)
		}
		time.Sleep(20 * time.Millisecond)
	}
	var commandResult, operationResult []byte
	if err := db.QueryRow(`SELECT c.ResultPayload, o.ResultPayload
FROM _System_RuntimeRecordCommand c JOIN _System_RuntimeRecordCommandOperation o ON o.CommandRowId = c.Id
WHERE c.Id = 'command-accepted' AND o.Id = 'operation-accepted'`).Scan(&commandResult, &operationResult); err != nil {
		t.Fatal(err)
	}
	var response semantic.Response
	var result semantic.OperationResult
	if err := json.Unmarshal(commandResult, &response); err != nil {
		t.Fatal(err)
	}
	if err := json.Unmarshal(operationResult, &result); err != nil {
		t.Fatal(err)
	}
	if response.CommandID != "command-accepted-1" || len(response.Results) != 1 ||
		response.Results[0].Index != result.Index || response.Results[0].Kind != result.Kind ||
		response.Results[0].RecordID != result.RecordID || result.Success || len(result.Errors) != 1 ||
		len(response.Results[0].Errors) != 1 || result.Errors[0].Kind != semantic.ErrorInternal ||
		response.Results[0].Errors[0].Kind != semantic.ErrorInternal {
		t.Fatalf("terminalized accepted command response=%#v operation=%#v", response, result)
	}
	stopWorker(t, lifecycle, lifecycleDB)
}

func TestRuntimeFileUploadLifecycleWorkerRealTiDBComposition(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID:         "activate-file-upload-lifecycle",
		WorkFamily:          files.RuntimeFileUploadLifecycleWorkFamily,
		DeploymentID:        "runtime-file-upload-lifecycle",
		WorkerReleaseDigest: integrationWorkerDigest,
	})

	storageRoot := t.TempDir()
	t.Setenv("FILE_STORAGE_PROVIDER", storage.ProviderFilesystem)
	t.Setenv("FILE_STORAGE_ROOT", storageRoot)
	objectStore, err := storage.NewStoreFromEnv()
	if err != nil {
		t.Fatal(err)
	}
	storageAuthorityDigest, err := storage.ConfigAuthorityDigest(storage.LoadConfigFromEnv())
	if err != nil {
		t.Fatal(err)
	}
	const objectKey = "system-file/Widget/widget-1/file-orphan/.attempts/upload-owner/blob"
	if _, err := objectStore.PutObject(
		t.Context(), objectKey, strings.NewReader("orphan"),
		storage.PutObjectOptions{ContentType: "text/plain"},
	); err != nil {
		t.Fatal(err)
	}
	journal := persistence.NewFileUploadOperationRepository(db)
	const headerHash = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	accepted, acquired, err := journal.Accept(
		t.Context(),
		persistence.FileUploadOperationAcceptance{
			ID: "file-orphan", IdempotencyKey: "file-orphan",
			RequestHash: headerHash, OwnerToken: "upload-owner",
			ObjectKey: objectKey, SystemFileID: "file-orphan",
			StorageProvider:        storage.ProviderFilesystem,
			StorageBucket:          storage.DefaultFilesystemBucket,
			StorageAuthorityDigest: storageAuthorityDigest,
			ParentID:               "widget-1", ParentObjectAPIName: "Widget",
			LeaseUntil: time.Now().UTC().Add(time.Minute),
			Now:        time.Now().UTC(),
		},
	)
	if err != nil || !acquired {
		t.Fatalf("accept orphan upload owner=%v error=%v", acquired, err)
	}
	contentDigest := sha256.Sum256([]byte("orphan"))
	completeHash := headerHash + "|sha256:" + hex.EncodeToString(contentDigest[:])
	if err := journal.MarkBlobStored(
		t.Context(), accepted.ID, accepted.OwnerToken, headerHash,
		completeHash, objectKey, time.Now().UTC().Add(-time.Second),
	); err != nil {
		t.Fatal(err)
	}
	const preparedObjectKey = "system-file/Widget/widget-1/file-prepared/.attempts/prepared-owner/blob"
	prepared, acquired, err := journal.Accept(
		t.Context(),
		persistence.FileUploadOperationAcceptance{
			ID: "file-prepared", IdempotencyKey: "file-prepared",
			RequestHash: headerHash, OwnerToken: "prepared-owner",
			ObjectKey: preparedObjectKey, SystemFileID: "file-prepared",
			StorageProvider:        storage.ProviderFilesystem,
			StorageBucket:          storage.DefaultFilesystemBucket,
			StorageAuthorityDigest: storageAuthorityDigest,
			ParentID:               "widget-1", ParentObjectAPIName: "Widget",
			LeaseUntil: time.Now().UTC().Add(-time.Minute),
			Now:        time.Now().UTC().Add(-2 * time.Minute),
		},
	)
	if err != nil || !acquired {
		t.Fatalf("accept prepared upload owner=%v error=%v", acquired, err)
	}
	if _, err := objectStore.PutObject(
		t.Context(),
		preparedObjectKey,
		strings.NewReader("prepared"),
		storage.PutObjectOptions{ContentType: "text/plain"},
	); err != nil {
		t.Fatal(err)
	}

	process, workerDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision,
		integrationWorkerConfig(owner), permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	assertWorkerOperationalSurface(t, process, owner.WorkFamily)
	deadline := time.Now().Add(5 * time.Second)
	for {
		var rows int
		queryErr := db.QueryRow(
			`SELECT COUNT(*) FROM _System_FileUploadOperation WHERE Id = ?`,
			accepted.ID,
		).Scan(&rows)
		_, objectErr := objectStore.HeadObject(t.Context(), objectKey)
		if queryErr == nil && rows == 0 &&
			errors.Is(objectErr, storage.ErrObjectNotFound) {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"file lifecycle worker did not clean orphan: rows=%d query=%v object=%v",
				rows, queryErr, objectErr,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
	deadline = time.Now().Add(5 * time.Second)
	for {
		var state, errorCode string
		queryErr := db.QueryRow(`SELECT State, COALESCE(ErrorCode, '')
FROM _System_FileUploadOperation WHERE Id = ?`, prepared.ID).Scan(
			&state,
			&errorCode,
		)
		_, objectErr := objectStore.HeadObject(t.Context(), preparedObjectKey)
		if queryErr == nil &&
			state == persistence.FileUploadOperationFailed &&
			errorCode == "PREPARED_CLEANUP_QUIET" &&
			errors.Is(objectErr, storage.ErrObjectNotFound) {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"file lifecycle worker did not enter prepared quiet period: state=%q code=%q query=%v object=%v",
				state,
				errorCode,
				queryErr,
				objectErr,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
	if _, err := objectStore.PutObject(
		t.Context(),
		preparedObjectKey,
		strings.NewReader("late-put"),
		storage.PutObjectOptions{ContentType: "text/plain"},
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET ExpiresAt = DATE_SUB(UTC_TIMESTAMP(6), INTERVAL 1 SECOND)
WHERE Id = ?`, prepared.ID); err != nil {
		t.Fatal(err)
	}
	deadline = time.Now().Add(5 * time.Second)
	for {
		var rows int
		queryErr := db.QueryRow(
			`SELECT COUNT(*) FROM _System_FileUploadOperation WHERE Id = ?`,
			prepared.ID,
		).Scan(&rows)
		_, objectErr := objectStore.HeadObject(t.Context(), preparedObjectKey)
		if queryErr == nil && rows == 0 &&
			errors.Is(objectErr, storage.ErrObjectNotFound) {
			break
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"file lifecycle worker did not finish prepared cleanup: rows=%d query=%v object=%v",
				rows,
				queryErr,
				objectErr,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
	stopWorker(t, process, workerDB)

	repairCandidate, acquired, err := journal.Accept(
		t.Context(),
		persistence.FileUploadOperationAcceptance{
			ID: "file-repair", IdempotencyKey: "file-repair",
			RequestHash: headerHash, OwnerToken: "repair-owner",
			ObjectKey:    "system-file/Widget/widget-1/file-repair/.attempts/repair-owner/blob",
			SystemFileID: "file-repair", ParentID: "widget-1",
			StorageProvider:        storage.ProviderFilesystem,
			StorageBucket:          storage.DefaultFilesystemBucket,
			StorageAuthorityDigest: storageAuthorityDigest,
			ParentObjectAPIName:    "Widget",
			LeaseUntil:             time.Now().UTC().Add(time.Minute),
			Now:                    time.Now().UTC(),
		},
	)
	if err != nil || !acquired {
		t.Fatalf("accept repair candidate owner=%v error=%v", acquired, err)
	}
	const repairError = "committed_blob_missing"
	if _, err := db.Exec(`UPDATE _System_FileUploadOperation
SET State = 'FAILED', ErrorCode = 'BLOB_REPAIR_POISONED',
    ErrorMessage = ?, RetryCount = 1, ExpiresAt = NULL
WHERE Id = ?`, repairError, repairCandidate.ID); err != nil {
		t.Fatal(err)
	}
	repairDigest := sha256.Sum256([]byte(repairError))
	if err := repairProduction(
		context.Background(), integrationWorkerSourceRevision,
		integrationWorkerConfig(owner),
		RepairRequest{Poison: persistence.RuntimePoisonRepair{
			RowID: repairCandidate.ID, ExpectedRetryCount: 1,
			ExpectedLastErrorSHA256: hex.EncodeToString(repairDigest[:]),
			RepairID:                "repair-file-upload", Reason: "restore exact immutable blob",
		}},
		permissionStateFactory,
	); err != nil {
		t.Fatal(err)
	}
	var repairedState, repairedID string
	var repairCount int
	if err := db.QueryRow(`SELECT State, RepairId, RepairCount
FROM _System_FileUploadOperation WHERE Id = ?`, repairCandidate.ID).Scan(
		&repairedState, &repairedID, &repairCount,
	); err != nil {
		t.Fatal(err)
	}
	if repairedState != persistence.FileUploadOperationBlobStored ||
		repairedID != "repair-file-upload" || repairCount != 1 {
		t.Fatalf(
			"RepairProduction Files result = state %q repair %q count %d",
			repairedState, repairedID, repairCount,
		)
	}
}

func TestRuntimeMetadataWorkersRealTiDBExactArtifactsRepairAndEffects(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}

	snapshotArtifact, viewArtifact, view, _, _ := compileWorkerMutationArtifacts(t)
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		t.Fatal(err)
	}

	eventOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-record-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
		DeploymentID: "runtime-record-events", WorkerReleaseDigest: integrationWorkerDigest,
	})
	rollupOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-rollup-delta", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
		DeploymentID: "runtime-rollup-delta", WorkerReleaseDigest: integrationWorkerDigest,
	})
	eventID := enqueueBoundRecordEvent(t, db, view)
	expiredProcessedEventID := insertExpiredProcessedRecordEvent(t, db, binding)
	eventConfig := integrationMetadataWorkerConfig(eventOwner, storeRoot, view)
	eventConfig.Role.MaxAttempts = 1
	assertMetadataWorkerAssemblyRequiresOwner(
		t, eventConfig, audit.AuditHistoryFanoutWorkFamily, permissionStateFactory,
	)
	_ = activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-audit-producer", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
		DeploymentID: "runtime-audit-producer", WorkerReleaseDigest: integrationWorkerDigest,
	})
	assertMetadataWorkerAssemblyRequiresOwner(
		t, eventConfig, persistence.RuntimeRecordCommandLifecycleWorkFamily, permissionStateFactory,
	)
	_ = activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-command-lifecycle-producer", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
		DeploymentID: "runtime-command-lifecycle-producer", WorkerReleaseDigest: integrationWorkerDigest,
	})
	// Merely shipping B1 does not enable it. Without a scheduled-flow owner,
	// the existing record-event role must still start and process unrelated
	// durable events.
	if err := persistence.VerifyActiveRuntimeWorkFamily(
		context.Background(),
		db,
		automation.RuntimeScheduledFlowWorkFamily,
	); !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("scheduled-flow owner before B1 activation = %v, want disabled", err)
	}
	eventProcess, eventDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision, eventConfig, permissionStateFactory)
	if err != nil {
		t.Fatal(err)
	}
	if err := eventProcess.Start(); err != nil {
		_ = eventDB.Close()
		t.Fatal(err)
	}
	retryCount, lastError := waitForRecordEventStatus(t, db, eventID, "failed")
	waitForRecordEventDeletion(t, db, expiredProcessedEventID)
	stopWorker(t, eventProcess, eventDB)
	if retryCount != 1 || lastError == "" {
		t.Fatalf("record-event poison retry=%d error=%q", retryCount, lastError)
	}

	// The operator restores the exact content named by the durable row, then
	// compare-and-set repair requeues that same obligation. No latest selector
	// or rewritten queue binding is involved.
	if err := store.Put(context.Background(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	if err := store.Put(context.Background(), viewArtifact); err != nil {
		t.Fatal(err)
	}
	errorDigest := sha256.Sum256([]byte(lastError))
	if err := repairProduction(context.Background(), integrationWorkerSourceRevision, eventConfig, RepairRequest{Poison: persistence.RuntimePoisonRepair{
		RowID: eventID, ExpectedRetryCount: retryCount, ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID: "repair-record-event", Reason: "restore exact immutable Catalog artifacts",
	}}, permissionStateFactory); err != nil {
		t.Fatal(err)
	}
	recoveredProcess, recoveredDB, err := assembleProduction(context.Background(), integrationWorkerSourceRevision, eventConfig, permissionStateFactory)
	if err != nil {
		t.Fatal(err)
	}
	if err := recoveredProcess.Start(); err != nil {
		_ = recoveredDB.Close()
		t.Fatal(err)
	}
	waitForRecordEventStatus(t, db, eventID, "processed")
	newEventOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "transfer-record-events", WorkFamily: eventOwner.WorkFamily,
		DeploymentID: "runtime-record-events-next", WorkerReleaseDigest: integrationWorkerDigest,
		ExpectedOwnershipEpoch: eventOwner.OwnershipEpoch,
	})
	select {
	case err := <-waitProcess(recoveredProcess):
		if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
			t.Fatalf("stale metadata worker exit=%v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("idle metadata worker did not stop after ownership transfer")
	}
	stopWorker(t, recoveredProcess, recoveredDB)
	_ = newEventOwner

	prepareRollupRecords(t, db)
	insertBoundRollupDelta(t, db, binding)
	rollupProcess, rollupDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, integrationMetadataWorkerConfig(rollupOwner, storeRoot, view),
		permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := rollupProcess.Start(); err != nil {
		_ = rollupDB.Close()
		t.Fatal(err)
	}
	waitForRollupEffect(t, db, 2)
	stopWorker(t, rollupProcess, rollupDB)
}

func TestRuntimePlatformCommandRepairRealTiDBComposition(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID:         "activate-platform-command-repair",
		WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "runtime-platform-command-repair",
		WorkerReleaseDigest: integrationWorkerDigest,
	})
	view := commandExecutionTestView(t)
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(),
		RuntimeCapabilityID: view.CapabilityID(),
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewRuntimePlatformCommandRepository(db)
	row, duplicate, err := repository.Accept(t.Context(), persistence.RuntimePlatformCommandAcceptance{
		RowID: "platform-command-repair-row", OrgID: scope.OrgID,
		AppInstallationID: scope.AppInstallationID, CallerWorkloadID: "integration-repair",
		RuntimeCapabilityID: scope.RuntimeCapabilityID, MetadataBinding: encodedBinding,
		ExecutionScopeDigest: executionScopeDigest,
		ScopeDigest:          strings.Repeat("9", 64),
		CommandID:            "platform-command-repair", IdempotencyKey: "platform-command-repair-request",
		CommandKind: "runtime.test", PayloadFormatVersion: 1,
		RequestDigest: "sha256:" + strings.Repeat("8", 64),
		Payload:       []byte(`{"value":1}`), Receipts: []byte(`{"CatalogRevision":7,"SecurityEpoch":11}`),
	})
	if err != nil || duplicate {
		t.Fatalf("Accept() row=%#v duplicate=%t err=%v", row, duplicate, err)
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	executionStore, err := persistence.NewRuntimePlatformCommandExecutionStore(
		db,
		runtimeWorkerScheduledRunTransactionOwner{transactions},
		owner,
		scope,
		time.Second,
		2,
	)
	if err != nil {
		t.Fatal(err)
	}
	lease, found, err := executionStore.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("Claim() lease=%#v found=%t err=%v", lease, found, err)
	}
	const poisonMessage = "release executor rejected corrupt payload"
	if err := executionStore.Poison(t.Context(), lease, errors.New(poisonMessage)); err != nil {
		t.Fatal(err)
	}
	errorDigest := sha256.Sum256([]byte(poisonMessage))
	config := integrationMetadataWorkerConfig(owner, t.TempDir(), view)
	if err := repairProduction(
		t.Context(),
		integrationWorkerSourceRevision,
		config,
		RepairRequest{Poison: persistence.RuntimePoisonRepair{
			RowID: lease.Row.RowID, ExpectedRetryCount: lease.Attempt,
			ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
			RepairID:                "repair-platform-command", Reason: "operator corrected command payload authority",
		}},
		permissionStateFactory,
	); err != nil {
		t.Fatal(err)
	}
	var state, repairID string
	var attempt, repairCount int
	if err := db.QueryRow(`SELECT State, Attempt, RepairId, RepairCount
FROM _System_RuntimePlatformCommand WHERE Id = ?`, lease.Row.RowID).
		Scan(&state, &attempt, &repairID, &repairCount); err != nil {
		t.Fatal(err)
	}
	if state != persistence.RuntimePlatformCommandAccepted || attempt != 0 ||
		repairID != "repair-platform-command" || repairCount != 1 {
		t.Fatalf(
			"repaired platform command = state %q attempt %d repair %q count %d",
			state,
			attempt,
			repairID,
			repairCount,
		)
	}
}

func TestRuntimePlatformCommandWorkerRealTiDBProductionComposition(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID:         "activate-platform-command-audit",
			WorkFamily:          audit.AuditHistoryFanoutWorkFamily,
			DeploymentID:        "runtime-platform-command-audit",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-platform-command-events",
			WorkFamily:          eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID:        "runtime-platform-command-events",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-platform-command-rollups",
			WorkFamily:          records.RuntimeRollupDeltaWorkFamily,
			DeploymentID:        "runtime-platform-command-rollups",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-platform-command-records",
			WorkFamily:          persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID:        "runtime-platform-command-records",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		activateWorker(t, control, activation)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID:         "activate-platform-command-worker",
		WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "runtime-platform-command-worker",
		WorkerReleaseDigest: integrationWorkerDigest,
	})
	snapshotArtifact, viewArtifact, view := commandExecutionTestArtifacts(t)
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(),
		RuntimeCapabilityID: view.CapabilityID(),
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	storeRoot := t.TempDir()
	artifactStore, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{snapshotArtifact, viewArtifact} {
		if err := artifactStore.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`CREATE TABLE RuntimePlatformCommandEffect (
CommandId VARCHAR(191) PRIMARY KEY,
CommandKind VARCHAR(255) NOT NULL,
PayloadFormatVersion BIGINT NOT NULL,
Payload MEDIUMBLOB NOT NULL
)`); err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewRuntimePlatformCommandRepository(db)
	accept := func(rowID, commandID, kind string) persistence.RuntimePlatformCommandRow {
		t.Helper()
		row, duplicate, acceptErr := repository.Accept(
			t.Context(),
			persistence.RuntimePlatformCommandAcceptance{
				RowID: rowID, OrgID: scope.OrgID,
				AppInstallationID: scope.AppInstallationID, CallerWorkloadID: "integration-worker",
				RuntimeCapabilityID: scope.RuntimeCapabilityID, MetadataBinding: encodedBinding,
				ExecutionScopeDigest: executionScopeDigest,
				ScopeDigest:          strings.Repeat("7", 64),
				CommandID:            commandID, IdempotencyKey: commandID + "-request",
				CommandKind: kind, PayloadFormatVersion: 1,
				RequestDigest: "sha256:" + strings.Repeat("6", 64),
				Payload:       []byte(`{"value":1}`), Receipts: []byte(`{"CatalogRevision":7,"SecurityEpoch":11}`),
			},
		)
		if acceptErr != nil || duplicate {
			t.Fatalf("Accept(%s) row=%#v duplicate=%t err=%v", commandID, row, duplicate, acceptErr)
		}
		return row
	}
	row := accept("platform-command-worker-row", "platform-command-worker", "runtime.test")
	retryRow := accept("platform-command-retry-row", "platform-command-retry", "runtime.test")
	permanentRow := accept(
		"platform-command-permanent-row",
		"platform-command-permanent",
		"runtime.test",
	)
	oversizedRow := accept(
		"platform-command-oversized-row",
		"platform-command-oversized",
		"runtime.test",
	)
	unownedRow := accept(
		"platform-command-unowned-row",
		"platform-command-unowned",
		"runtime.unowned",
	)
	config := integrationMetadataWorkerConfig(owner, storeRoot, view)
	config.Role.PollInterval = "50ms"
	config.Role.LeaseDuration = "2s"
	executorAttempts := make(map[string]int)
	var executorAttemptsMu sync.Mutex
	dependencies := productionAssemblyDependencies{
		permissionStateFactory: permissionStateFactory,
		platformCommandExecutorRegistry: func(
			_ context.Context,
			_ *sql.DB,
			runtime *recordmutation.RecordsMutationRuntime,
			_ string,
		) (*platformCommandExecutorRegistry, error) {
			return newPlatformCommandExecutorRegistry(
				map[string]map[uint32]PlatformCommandExecutor{
					"runtime.test": {
						1: platformCommandIntegrationExecutor{
							transactions: runtime.Transactions,
							mu:           &executorAttemptsMu,
							attempts:     executorAttempts,
						},
					},
				},
				nil,
			)
		},
	}
	process, workerDB, err := assembleProductionWithDependencies(
		context.Background(),
		integrationWorkerSourceRevision,
		config,
		dependencies,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	waitForRuntimePlatformCommandState(t, db, row.RowID, persistence.RuntimePlatformCommandCompleted)
	waitForRuntimePlatformCommandState(
		t,
		db,
		retryRow.RowID,
		persistence.RuntimePlatformCommandCompleted,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		permanentRow.RowID,
		persistence.RuntimePlatformCommandFailed,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		oversizedRow.RowID,
		persistence.RuntimePlatformCommandFailed,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		unownedRow.RowID,
		persistence.RuntimePlatformCommandPoisoned,
	)
	stopWorker(t, process, workerDB)

	var effectCount, retryAttempt, permanentAttempt, oversizedAttempt, unownedAttempt int
	var result, retryResult []byte
	if err := db.QueryRow(
		`SELECT
(SELECT COUNT(*) FROM RuntimePlatformCommandEffect),
(SELECT ResultPayload FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT ResultPayload FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?)`,
		row.RowID,
		retryRow.RowID,
		retryRow.RowID,
		permanentRow.RowID,
		oversizedRow.RowID,
		unownedRow.RowID,
	).Scan(
		&effectCount,
		&result,
		&retryAttempt,
		&retryResult,
		&permanentAttempt,
		&oversizedAttempt,
		&unownedAttempt,
	); err != nil {
		t.Fatal(err)
	}
	decodeResult := func(value []byte) []byte {
		t.Helper()
		const prefix = "base64:"
		if !strings.HasPrefix(string(value), prefix) {
			t.Fatalf("Platform-command result encoding = %q, want %q prefix", value, prefix)
		}
		decoded, decodeErr := base64.StdEncoding.Strict().DecodeString(strings.TrimPrefix(string(value), prefix))
		if decodeErr != nil {
			t.Fatalf("decode Platform-command result: %v", decodeErr)
		}
		return decoded
	}
	result = decodeResult(result)
	retryResult = decodeResult(retryResult)
	if effectCount != 2 ||
		string(result) != `{"status":"applied"}` ||
		retryAttempt != 2 ||
		string(retryResult) != `{"status":"applied"}` ||
		permanentAttempt != 1 ||
		oversizedAttempt != 1 ||
		unownedAttempt != 1 {
		t.Fatalf(
			"Platform-command effects=%d result=%s retry=%d/%s permanent=%d oversized=%d unowned=%d",
			effectCount,
			result,
			retryAttempt,
			retryResult,
			permanentAttempt,
			oversizedAttempt,
			unownedAttempt,
		)
	}

	restartedProcess, restartedDB, err := assembleProductionWithDependencies(
		context.Background(),
		integrationWorkerSourceRevision,
		config,
		dependencies,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := restartedProcess.Start(); err != nil {
		_ = restartedDB.Close()
		t.Fatal(err)
	}
	assertWorkerOperationalSurface(t, restartedProcess, owner.WorkFamily)
	time.Sleep(250 * time.Millisecond)
	stopWorker(t, restartedProcess, restartedDB)
	var restartedOversizedAttempt int
	var restartedOversizedState string
	if err := db.QueryRow(
		`SELECT
(SELECT COUNT(*) FROM RuntimePlatformCommandEffect),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?)`,
		oversizedRow.RowID,
		oversizedRow.RowID,
	).Scan(
		&effectCount,
		&restartedOversizedAttempt,
		&restartedOversizedState,
	); err != nil {
		t.Fatal(err)
	}
	if effectCount != 2 ||
		restartedOversizedAttempt != oversizedAttempt ||
		restartedOversizedState != persistence.RuntimePlatformCommandFailed {
		t.Fatalf(
			"restarted Platform-command Worker effects=%d oversized=%d/%s",
			effectCount,
			restartedOversizedAttempt,
			restartedOversizedState,
		)
	}
}

func TestRuntimeFlowCommandWorkerRealTiDBExactAuthorityAndRestart(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	for _, statement := range []string{
		`CREATE TABLE Work (Id VARCHAR(255) PRIMARY KEY, Name VARCHAR(255) NULL)`,
		`INSERT INTO Work (Id, Name) VALUES ('a01000000000001AAA', 'before'), ('a01000000000002AAA', 'before')`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID:         "activate-runtime-flow-command-audit",
			WorkFamily:          audit.AuditHistoryFanoutWorkFamily,
			DeploymentID:        "runtime-flow-command-audit",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-runtime-flow-command-events",
			WorkFamily:          eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID:        "runtime-flow-command-events",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-runtime-flow-command-rollups",
			WorkFamily:          records.RuntimeRollupDeltaWorkFamily,
			DeploymentID:        "runtime-flow-command-rollups",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID:         "activate-runtime-flow-command-records",
			WorkFamily:          persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID:        "runtime-flow-command-records",
			WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		activateWorker(t, control, activation)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID:         "activate-runtime-flow-command-worker",
		WorkFamily:          persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "runtime-flow-command-worker",
		WorkerReleaseDigest: integrationWorkerDigest,
	})

	sourceStore, commandView, mutationView, flowView :=
		workerRuntimeFlowCommandMappingViews(t)
	commandBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(commandView)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(commandBinding)
	if err != nil {
		t.Fatal(err)
	}
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	artifactIdentities := []struct {
		kind   catalogsnapshot.ArtifactKind
		digest catalogsnapshot.Digest
	}{
		{
			kind:   catalogsnapshot.ArtifactCatalogSnapshot,
			digest: commandBinding.CatalogSnapshotDigest,
		},
		{
			kind:   catalogsnapshot.ArtifactRuntimeMetadataView,
			digest: commandView.Digest(),
		},
		{
			kind:   catalogsnapshot.ArtifactRuntimeMetadataView,
			digest: mutationView.Digest(),
		},
		{
			kind:   catalogsnapshot.ArtifactRuntimeMetadataView,
			digest: flowView.Digest(),
		},
	}
	for _, identity := range artifactIdentities {
		artifact, getErr := sourceStore.Get(
			t.Context(),
			identity.kind,
			identity.digest,
		)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	scope := persistence.RuntimeWorkScope{
		OrgID:               commandView.OrgID(),
		AppInstallationID:   commandView.AppInstallationID(),
		RuntimeCapabilityID: commandView.CapabilityID(),
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewRuntimePlatformCommandRepository(db)
	accept := func(
		rowID string,
		workloadID string,
		targetID string,
		definitionRef string,
		definitionVersion string,
	) persistence.RuntimePlatformCommandRow {
		t.Helper()
		canonicalInputs, marshalErr := json.Marshal(map[string]interface{}{
			"sourceRecord": map[string]interface{}{"remoteName": "after"},
			"targetId":     targetID,
		})
		if marshalErr != nil {
			t.Fatal(marshalErr)
		}
		payload, marshalErr := json.Marshal(map[string]interface{}{
			"canonicalInputs":   json.RawMessage(canonicalInputs),
			"definitionRef":     definitionRef,
			"definitionVersion": definitionVersion,
		})
		if marshalErr != nil {
			t.Fatal(marshalErr)
		}
		row, duplicate, acceptErr := repository.Accept(
			t.Context(),
			persistence.RuntimePlatformCommandAcceptance{
				RowID:                rowID,
				OrgID:                scope.OrgID,
				AppInstallationID:    scope.AppInstallationID,
				CallerWorkloadID:     workloadID,
				RuntimeCapabilityID:  scope.RuntimeCapabilityID,
				MetadataBinding:      encodedBinding,
				ExecutionScopeDigest: executionScopeDigest,
				ScopeDigest:          strings.Repeat("4", 64),
				CommandID:            rowID,
				IdempotencyKey:       rowID + "-request",
				CommandKind:          RuntimeFlowCommandKind,
				PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
				RequestDigest:        "sha256:" + strings.Repeat("5", 64),
				Payload:              payload,
				Receipts:             []byte(`{"CatalogRevision":11,"SecurityEpoch":1}`),
			},
		)
		if acceptErr != nil || duplicate {
			t.Fatalf(
				"Accept(%s) row=%#v duplicate=%t err=%v",
				rowID,
				row,
				duplicate,
				acceptErr,
			)
		}
		return row
	}
	authorized := accept(
		"runtime-flow-command-authorized",
		"integration-a",
		"a01000000000001AAA",
		"ExactChild",
		"version-child",
	)
	unauthorized := accept(
		"runtime-flow-command-unauthorized",
		"integration-other",
		"a01000000000002AAA",
		"ExactChild",
		"version-child",
	)
	retriable := accept(
		"runtime-flow-command-retriable",
		"integration-a",
		"a01000000000002AAA",
		"RetryChild",
		"version-retry-child",
	)
	dependencyInputs, err := semantic.CanonicalSourceJSON(`{"targetId":"lead-missing"}`)
	if err != nil {
		t.Fatal(err)
	}
	dependencyPayload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef:     "DependencyChild",
		DefinitionVersion: "version-dependency-child",
		CanonicalInputs:   []byte(dependencyInputs),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "contacts",
			SourceID:   "remote-contact-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	dependencyRow, duplicate, err := repository.Accept(
		t.Context(),
		persistence.RuntimePlatformCommandAcceptance{
			RowID: "runtime-flow-command-dependency", OrgID: scope.OrgID,
			AppInstallationID: scope.AppInstallationID, CallerWorkloadID: "integration-a",
			RuntimeCapabilityID: scope.RuntimeCapabilityID, MetadataBinding: encodedBinding,
			ExecutionScopeDigest: executionScopeDigest, ScopeDigest: strings.Repeat("4", 64),
			CommandID: "runtime-flow-command-dependency", IdempotencyKey: "dependency-request",
			CommandKind: RuntimeFlowCommandKind, PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV2,
			RequestDigest: "sha256:" + strings.Repeat("6", 64), Payload: dependencyPayload,
			Receipts: []byte(`{"CatalogRevision":11,"SecurityEpoch":1}`),
		},
	)
	if err != nil || duplicate {
		t.Fatalf("accept dependency command row=%#v duplicate=%t err=%v", dependencyRow, duplicate, err)
	}
	config := integrationMetadataWorkerConfig(owner, storeRoot, commandView)
	config.Role.PollInterval = "50ms"
	config.Role.LeaseDuration = "2s"
	config.Role.MaxAttempts = 2
	process, workerDB, err := assembleProduction(
		context.Background(),
		integrationWorkerSourceRevision,
		config,
		permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	waitForRuntimePlatformCommandState(
		t,
		db,
		authorized.RowID,
		persistence.RuntimePlatformCommandCompleted,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		unauthorized.RowID,
		persistence.RuntimePlatformCommandFailed,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		retriable.RowID,
		persistence.RuntimePlatformCommandPoisoned,
	)
	waitForRuntimePlatformCommandState(
		t,
		db,
		dependencyRow.RowID,
		persistence.RuntimePlatformCommandBlocked,
	)
	stopWorker(t, process, workerDB)

	var result []byte
	var authorizedName, unauthorizedName string
	var authorizedAttempt, unauthorizedAttempt, retriableAttempt int
	var dependencyAttempt int
	var dependencyKey string
	if err := db.QueryRow(
		`SELECT
(SELECT ResultPayload FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT DependencyKeyHash FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Name FROM Work WHERE Id = 'a01000000000001AAA'),
(SELECT Name FROM Work WHERE Id = 'a01000000000002AAA')`,
		authorized.RowID,
		authorized.RowID,
		unauthorized.RowID,
		retriable.RowID,
		dependencyRow.RowID,
		dependencyRow.RowID,
	).Scan(
		&result,
		&authorizedAttempt,
		&unauthorizedAttempt,
		&retriableAttempt,
		&dependencyAttempt,
		&dependencyKey,
		&authorizedName,
		&unauthorizedName,
	); err != nil {
		t.Fatal(err)
	}
	const prefix = "base64:"
	if !strings.HasPrefix(string(result), prefix) {
		t.Fatalf("Runtime Flow command result encoding=%q", result)
	}
	decoded, err := base64.StdEncoding.Strict().DecodeString(
		strings.TrimPrefix(string(result), prefix),
	)
	if err != nil {
		t.Fatal(err)
	}
	var outcome runtimeFlowCommandResult
	if err := json.Unmarshal(decoded, &outcome); err != nil {
		t.Fatal(err)
	}
	dependencyKeyBytes, dependencyKeyErr := hex.DecodeString(dependencyKey)
	if outcome.ExecutionID == "" ||
		string(outcome.CanonicalOutputs) != "{}" ||
		authorizedAttempt != 1 ||
		unauthorizedAttempt != 1 ||
		retriableAttempt != config.Role.MaxAttempts ||
		dependencyAttempt != 1 || dependencyKeyErr != nil || len(dependencyKeyBytes) != sha256.Size ||
		authorizedName != "after" ||
		unauthorizedName != "before" {
		t.Fatalf(
			"Runtime Flow command outcome=%#v authorized=%d/%q unauthorized=%d/%q retriable=%d dependency=%d/%q",
			outcome,
			authorizedAttempt,
			authorizedName,
			unauthorizedAttempt,
			unauthorizedName,
			retriableAttempt,
			dependencyAttempt,
			dependencyKey,
		)
	}

	restarted, restartedDB, err := assembleProduction(
		context.Background(),
		integrationWorkerSourceRevision,
		config,
		permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := restarted.Start(); err != nil {
		_ = restartedDB.Close()
		t.Fatal(err)
	}
	time.Sleep(250 * time.Millisecond)
	stopWorker(t, restarted, restartedDB)
	var restartedAuthorizedAttempt, restartedUnauthorizedAttempt, restartedRetriableAttempt, restartedDependencyAttempt int
	var restartedDependencyState string
	if err := db.QueryRow(
		`SELECT
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?)`,
		authorized.RowID,
		unauthorized.RowID,
		retriable.RowID,
		dependencyRow.RowID,
		dependencyRow.RowID,
	).Scan(
		&restartedAuthorizedAttempt,
		&restartedUnauthorizedAttempt,
		&restartedRetriableAttempt,
		&restartedDependencyAttempt,
		&restartedDependencyState,
	); err != nil {
		t.Fatal(err)
	}
	// The dependency retry is one minute, so a process restart cannot burn
	// the normal attempt budget or lose the durable BLOCKED state.
	if restartedDependencyAttempt != dependencyAttempt ||
		restartedDependencyState != persistence.RuntimePlatformCommandBlocked {
		t.Fatalf("restart changed dependency wait: attempt=%d state=%q", restartedDependencyAttempt, restartedDependencyState)
	}
	if restartedAuthorizedAttempt != authorizedAttempt ||
		restartedUnauthorizedAttempt != unauthorizedAttempt ||
		restartedRetriableAttempt != retriableAttempt {
		t.Fatalf(
			"restart replayed terminal commands: authorized=%d unauthorized=%d retriable=%d",
			restartedAuthorizedAttempt,
			restartedUnauthorizedAttempt,
			restartedRetriableAttempt,
		)
	}
}

func TestRuntimeFlowCommandWorkerRealTiDBFinalizesInboundIdentityInOwnedTransaction(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	if _, err := db.Exec(`CREATE TABLE Work (
Id VARCHAR(255) PRIMARY KEY,
Name VARCHAR(255) NULL,
RemoteKey VARCHAR(255) NULL
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(
		`INSERT INTO Work (Id, Name, RemoteKey) VALUES (?, ?, ?)`,
		"a01000000000003AAA",
		"before",
		"remote-123",
	); err != nil {
		t.Fatal(err)
	}
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID: "activate-runtime-finalization-audit", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
			DeploymentID: "runtime-finalization-audit", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-finalization-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID: "runtime-finalization-events", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-finalization-rollups", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
			DeploymentID: "runtime-finalization-rollups", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-finalization-records", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID: "runtime-finalization-records", WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		activateWorker(t, control, activation)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-runtime-finalization-worker", WorkFamily: persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID: "runtime-finalization-worker", WorkerReleaseDigest: integrationWorkerDigest,
	})

	sourceStore, commandView, mutationView, flowView :=
		workerRuntimeFlowCommandFinalizationViews(t)
	commandBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(commandView)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(commandBinding)
	if err != nil {
		t.Fatal(err)
	}
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, identity := range []struct {
		kind   catalogsnapshot.ArtifactKind
		digest catalogsnapshot.Digest
	}{
		{catalogsnapshot.ArtifactCatalogSnapshot, commandBinding.CatalogSnapshotDigest},
		{catalogsnapshot.ArtifactRuntimeMetadataView, commandView.Digest()},
		{catalogsnapshot.ArtifactRuntimeMetadataView, mutationView.Digest()},
		{catalogsnapshot.ArtifactRuntimeMetadataView, flowView.Digest()},
	} {
		artifact, getErr := sourceStore.Get(t.Context(), identity.kind, identity.digest)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	scope := persistence.RuntimeWorkScope{
		OrgID:               commandView.OrgID(),
		AppInstallationID:   commandView.AppInstallationID(),
		RuntimeCapabilityID: commandView.CapabilityID(),
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	canonicalInputs, err := json.Marshal(map[string]interface{}{
		"sourceRecord": map[string]interface{}{"remoteName": "after"},
		"targetId":     "a01000000000003AAA",
	})
	if err != nil {
		t.Fatal(err)
	}
	payload, err := json.Marshal(map[string]interface{}{
		"canonicalInputs":   json.RawMessage(canonicalInputs),
		"definitionRef":     "ExactChild",
		"definitionVersion": "version-child",
	})
	if err != nil {
		t.Fatal(err)
	}
	row, duplicate, err := persistence.NewRuntimePlatformCommandRepository(db).Accept(
		t.Context(),
		persistence.RuntimePlatformCommandAcceptance{
			RowID: "runtime-flow-command-finalization", OrgID: scope.OrgID,
			AppInstallationID: scope.AppInstallationID, CallerWorkloadID: "integration-a",
			RuntimeCapabilityID: scope.RuntimeCapabilityID, MetadataBinding: encodedBinding,
			ExecutionScopeDigest: executionScopeDigest, ScopeDigest: strings.Repeat("4", 64),
			CommandID: "runtime-flow-command-finalization", IdempotencyKey: "finalization-request",
			CommandKind: RuntimeFlowCommandKind, PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
			RequestDigest: "sha256:" + strings.Repeat("5", 64), Payload: payload,
			Receipts: []byte(`{"CatalogRevision":11,"SecurityEpoch":1}`),
		},
	)
	if err != nil || duplicate {
		t.Fatalf("accept finalization command row=%#v duplicate=%t err=%v", row, duplicate, err)
	}
	config := integrationMetadataWorkerConfig(owner, storeRoot, commandView)
	config.Role.PollInterval = "50ms"
	config.Role.LeaseDuration = "2s"
	config.Role.MaxAttempts = 2
	process, workerDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, config, permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	waitForRuntimePlatformCommandState(
		t, db, row.RowID, persistence.RuntimePlatformCommandCompleted,
	)
	stopWorker(t, process, workerDB)

	var name, linkStatus, remoteID, evidenceStatus, consumedBy string
	var attempt int
	if err := db.QueryRow(`SELECT
(SELECT Name FROM Work WHERE Id = 'a01000000000003AAA'),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Status FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000003AAA'),
(SELECT RemoteID FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000003AAA'),
(SELECT EvidenceStatus FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceRecordId = 'a01000000000003AAA'),
(SELECT COALESCE(ConsumedByLinkKey, '') FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceRecordId = 'a01000000000003AAA')`,
		row.RowID,
	).Scan(&name, &attempt, &linkStatus, &remoteID, &evidenceStatus, &consumedBy); err != nil {
		t.Fatal(err)
	}
	if name != "after" || attempt != 1 ||
		linkStatus != automation.ConnectorPushLinkStatusFinalized || remoteID != "remote-123" ||
		evidenceStatus != automation.ConnectorInboundEvidenceStatusConsumed || consumedBy == "" {
		t.Fatalf(
			"finalization state name=%q attempt=%d link=%q/%q evidence=%q/%q",
			name, attempt, linkStatus, remoteID, evidenceStatus, consumedBy,
		)
	}

	restarted, restartedDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, config, permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := restarted.Start(); err != nil {
		_ = restartedDB.Close()
		t.Fatal(err)
	}
	time.Sleep(250 * time.Millisecond)
	stopWorker(t, restarted, restartedDB)
	var restartedAttempt, linkCount, evidenceCount int
	if err := db.QueryRow(`SELECT
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000003AAA'),
(SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE SourceRecordId = 'a01000000000003AAA')`,
		row.RowID,
	).Scan(&restartedAttempt, &linkCount, &evidenceCount); err != nil {
		t.Fatal(err)
	}
	if restartedAttempt != attempt || linkCount != 1 || evidenceCount != 1 {
		t.Fatalf(
			"restart replayed finalization: attempt=%d links=%d evidence=%d",
			restartedAttempt, linkCount, evidenceCount,
		)
	}
}

func TestRuntimeFlowCommandWorkerRealTiDBRecoversIdentityAfterRollbackAndReplays(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	if _, err := db.Exec(`CREATE TABLE Work (
Id VARCHAR(255) PRIMARY KEY,
Name VARCHAR(255) NULL,
RemoteKey VARCHAR(255) NULL,
Archived TINYINT(1) NOT NULL DEFAULT 0,
IsDeleted TINYINT(1) NOT NULL DEFAULT 0
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(
		`INSERT INTO Work (Id, Name, RemoteKey, Archived) VALUES (?, ?, ?, 0)`,
		"a01000000000004AAA",
		"before",
		"remote-456",
	); err != nil {
		t.Fatal(err)
	}
	if err := automation.NewSQLConnectorPushDurableStore(db).InsertAuthoredEvidence(
		t.Context(),
		automation.ConnectorInboundRemoteIDEvidence{
			ID: "recovery-evidence-1", ConnectorOwnerKey: "companion-pack",
			ConnectorNamespace: "connector-a", SourceObject: "Work",
			SourceRecordId: "a01000000000004AAA", RemoteIDField: "RemoteKey",
			RemoteId: "remote-456",
		},
	); err != nil {
		t.Fatal(err)
	}

	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID: "activate-runtime-recovery-audit", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
			DeploymentID: "runtime-recovery-audit", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-recovery-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID: "runtime-recovery-events", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-recovery-rollups", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
			DeploymentID: "runtime-recovery-rollups", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-runtime-recovery-records", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID: "runtime-recovery-records", WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		activateWorker(t, control, activation)
	}
	owner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-runtime-recovery-worker", WorkFamily: persistence.RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID: "runtime-recovery-worker", WorkerReleaseDigest: integrationWorkerDigest,
	})

	sourceStore, commandView, mutationView, flowView := workerRuntimeFlowCommandRecoveryViews(t)
	commandBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(commandView)
	if err != nil {
		t.Fatal(err)
	}
	encodedBinding, err := catalogsnapshot.EncodeRuntimeWorkBinding(commandBinding)
	if err != nil {
		t.Fatal(err)
	}
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, identity := range []struct {
		kind   catalogsnapshot.ArtifactKind
		digest catalogsnapshot.Digest
	}{
		{catalogsnapshot.ArtifactCatalogSnapshot, commandBinding.CatalogSnapshotDigest},
		{catalogsnapshot.ArtifactRuntimeMetadataView, commandView.Digest()},
		{catalogsnapshot.ArtifactRuntimeMetadataView, mutationView.Digest()},
		{catalogsnapshot.ArtifactRuntimeMetadataView, flowView.Digest()},
	} {
		artifact, getErr := sourceStore.Get(t.Context(), identity.kind, identity.digest)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: commandView.OrgID(), AppInstallationID: commandView.AppInstallationID(),
		RuntimeCapabilityID: commandView.CapabilityID(),
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	canonicalInputs, err := semantic.CanonicalSourceJSON(`{
  "sourceRecord":{"remoteName":"after"},
  "targetId":"a01000000000004AAA",
  "remoteId":"remote-456"
}`)
	if err != nil {
		t.Fatal(err)
	}
	payload, err := flowcommandsemanticv2.Encode(flowcommandsemanticv2.Payload{
		DefinitionRef: "ExactChild", DefinitionVersion: "version-child",
		CanonicalInputs: []byte(canonicalInputs),
		DependencySource: flowcommandsemanticv2.DependencySourceScope{
			ObjectKind: "remoteWork", SourceID: "remote-456",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	row, duplicate, err := persistence.NewRuntimePlatformCommandRepository(db).Accept(
		t.Context(),
		persistence.RuntimePlatformCommandAcceptance{
			RowID: "runtime-flow-command-recovery", OrgID: scope.OrgID,
			AppInstallationID: scope.AppInstallationID, CallerWorkloadID: "integration-a",
			RuntimeCapabilityID: scope.RuntimeCapabilityID, MetadataBinding: encodedBinding,
			ExecutionScopeDigest: executionScopeDigest, ScopeDigest: strings.Repeat("6", 64),
			CommandID: "runtime-flow-command-recovery", IdempotencyKey: "recovery-request",
			CommandKind: RuntimeFlowCommandKind, PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV2,
			RequestDigest: "sha256:" + strings.Repeat("7", 64), Payload: payload,
			Receipts: []byte(`{"CatalogRevision":11,"SecurityEpoch":1}`),
		},
	)
	if err != nil || duplicate {
		t.Fatalf("accept recovery command row=%#v duplicate=%t err=%v", row, duplicate, err)
	}
	config := integrationMetadataWorkerConfig(owner, storeRoot, commandView)
	config.Role.PollInterval = "50ms"
	config.Role.LeaseDuration = "2s"
	config.Role.MaxAttempts = 3
	process, workerDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, config, permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	waitForRuntimePlatformCommandState(t, db, row.RowID, persistence.RuntimePlatformCommandCompleted)
	stopWorker(t, process, workerDB)
	authority, err := automation.NewSQLConnectorPushDurableStore(db).ResolveFinalizedRemoteLocator(
		t.Context(),
		automation.ConnectorRemoteLocatorQuery{
			ConnectorOwnerKey: "companion-pack", DestinationSlotID: "remote-work", RemoteID: "remote-456",
		},
	)
	if err != nil || authority == nil {
		t.Fatalf("recovered locator authority=%#v err=%v", authority, err)
	}

	var name, linkStatus, remoteID, evidenceStatus, consumedBy string
	var attempt, eventCount int
	if err := db.QueryRow(`SELECT
(SELECT Name FROM Work WHERE Id = 'a01000000000004AAA'),
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT Status FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000004AAA'),
(SELECT RemoteID FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000004AAA'),
(SELECT EvidenceStatus FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = 'recovery-evidence-1'),
(SELECT COALESCE(ConsumedByLinkKey, '') FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = 'recovery-evidence-1'),
(SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE RecordKey = 'Work/a01000000000004AAA')`,
		row.RowID,
	).Scan(&name, &attempt, &linkStatus, &remoteID, &evidenceStatus, &consumedBy, &eventCount); err != nil {
		t.Fatal(err)
	}
	if name != "after" || attempt != 2 || eventCount != 1 ||
		linkStatus != automation.ConnectorPushLinkStatusFinalized || remoteID != "remote-456" ||
		evidenceStatus != automation.ConnectorInboundEvidenceStatusConsumed || consumedBy == "" {
		t.Fatalf(
			"recovery state name=%q attempt=%d events=%d link=%q/%q evidence=%q/%q",
			name, attempt, eventCount, linkStatus, remoteID, evidenceStatus, consumedBy,
		)
	}

	restarted, restartedDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, config, permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := restarted.Start(); err != nil {
		_ = restartedDB.Close()
		t.Fatal(err)
	}
	time.Sleep(250 * time.Millisecond)
	stopWorker(t, restarted, restartedDB)
	var restartedAttempt, linkCount, evidenceCount, restartedEventCount int
	if err := db.QueryRow(`SELECT
(SELECT Attempt FROM _System_RuntimePlatformCommand WHERE Id = ?),
(SELECT COUNT(*) FROM _System_ConnectorRemoteLink WHERE SourceRecordID = 'a01000000000004AAA'),
(SELECT COUNT(*) FROM _System_ConnectorInboundRemoteIDEvidence WHERE Id = 'recovery-evidence-1'),
(SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE RecordKey = 'Work/a01000000000004AAA')`,
		row.RowID,
	).Scan(&restartedAttempt, &linkCount, &evidenceCount, &restartedEventCount); err != nil {
		t.Fatal(err)
	}
	if restartedAttempt != attempt || linkCount != 1 || evidenceCount != 1 || restartedEventCount != 1 {
		t.Fatalf(
			"restart replayed recovery: attempt=%d links=%d evidence=%d events=%d",
			restartedAttempt, linkCount, evidenceCount, restartedEventCount,
		)
	}
}

func waitForRuntimePlatformCommandState(
	t *testing.T,
	db *sql.DB,
	rowID string,
	want string,
) {
	t.Helper()
	deadline := time.Now().Add(10 * time.Second)
	for {
		var state string
		var lastError string
		err := db.QueryRow(
			`SELECT State, COALESCE(LastError, '') FROM _System_RuntimePlatformCommand WHERE Id = ?`,
			rowID,
		).Scan(&state, &lastError)
		if err == nil && state == want {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"Platform-command row %q state=%q error=%q query=%v, want %q",
				rowID,
				state,
				lastError,
				err,
				want,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func TestRuntimeRecordEventWorkerGlobalScheduledOwnerDoesNotActivateTimedFlow(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	var eventOwner persistence.RuntimeWorkOwnership
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID: "activate-disabled-timed-audit", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
			DeploymentID: "runtime-disabled-timed-audit", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-disabled-timed-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID: "runtime-disabled-timed-events", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-disabled-timed-rollups", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
			DeploymentID: "runtime-disabled-timed-rollups", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-disabled-timed-commands", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID: "runtime-disabled-timed-commands", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-unscoped-scheduled-owner", WorkFamily: automation.RuntimeScheduledFlowWorkFamily,
			DeploymentID: "runtime-unscoped-scheduled-owner", WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		owner := activateWorker(t, control, activation)
		if activation.WorkFamily == eventfeature.RuntimeRecordEventDispatchWorkFamily {
			eventOwner = owner
		}
	}
	if err := persistence.VerifyActiveRuntimeWorkFamily(
		t.Context(),
		db,
		automation.RuntimeScheduledFlowWorkFamily,
	); err != nil {
		t.Fatalf("global scheduled-flow owner = %v, want active but insufficient for timed activation", err)
	}

	snapshotArtifact, mutationArtifact, mutationView, flowArtifact, flowView :=
		compileWorkerMutationArtifacts(t)
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{
		snapshotArtifact,
		mutationArtifact,
		flowArtifact,
	} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`CREATE TABLE ScheduledTarget (
Id VARCHAR(191) PRIMARY KEY,
Status VARCHAR(191) NULL,
IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
LastModifiedById VARCHAR(191) NULL,
LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(
		`INSERT INTO ScheduledTarget (Id, Status) VALUES ('a02000000000001AAA', 'before')`,
	); err != nil {
		t.Fatal(err)
	}
	eventID := enqueueDualBoundTimedRecordEvent(
		t,
		db,
		mutationView,
		flowView,
		"a02000000000001AAA",
		0,
	)
	config := integrationMetadataWorkerConfig(eventOwner, storeRoot, mutationView)
	config.Role.MaxAttempts = 1
	process, workerDB, err := assembleProduction(
		t.Context(),
		integrationWorkerSourceRevision,
		config,
		permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	retryCount, lastError := waitForRecordEventStatus(t, db, eventID, "failed")
	stopWorker(t, process, workerDB)
	if retryCount != 1 ||
		!strings.Contains(lastError, "durable timed Flow handoff is not active") {
		t.Fatalf("disabled timed event retry=%d error=%q", retryCount, lastError)
	}
	var status string
	if err := db.QueryRow(
		`SELECT Status FROM ScheduledTarget WHERE Id = 'a02000000000001AAA'`,
	).Scan(&status); err != nil {
		t.Fatal(err)
	}
	if status != "before" {
		t.Fatalf("disabled timed Flow effect status = %q, want no early effect", status)
	}
	var reservations int
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM _System_ScheduledRunReservation
WHERE FlowDefinitionId = 'definition-timed-integration' AND InputKey = 'a02000000000001AAA'`,
	).Scan(&reservations); err != nil {
		t.Fatal(err)
	}
	if reservations != 0 {
		t.Fatalf("disabled timed Flow reservations = %d, want none", reservations)
	}
}

func TestRuntimeScheduledFlowWorkerRealTiDBProductionComposition(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("runtime_worker_integration requires the declared TiDB environment")
	}
	permissionStateFactory := integrationPermissionStateFactory(t)
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID: "activate-scheduled-audit", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
			DeploymentID: "runtime-scheduled-audit", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-scheduled-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID: "runtime-scheduled-events", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-scheduled-rollups", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
			DeploymentID: "runtime-scheduled-rollups", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "activate-scheduled-commands", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID: "runtime-scheduled-commands", WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		_ = activateWorker(t, control, activation)
	}
	scheduledOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "activate-scheduled-role", WorkFamily: automation.RuntimeScheduledFlowWorkFamily,
		DeploymentID: "runtime-scheduled-role", WorkerReleaseDigest: integrationWorkerDigest,
	})
	ageScheduledFlowOwnershipCutover(t, db, scheduledOwner)

	snapshotArtifact, mutationArtifact, mutationView, flowArtifact, flowView :=
		compileWorkerMutationArtifacts(t)
	startupSnapshotArtifact, startupMutationArtifact, _, startupFlowArtifact, startupFlowView :=
		compileWorkerMutationArtifactsForRevision(
			t,
			catalogsnapshot.CatalogRevision(8),
		)
	scheduledVersion, ok := flowView.FlowVersion("version-scheduled-integration")
	if !ok || scheduledVersion == nil ||
		len(scheduledVersion.Metadata.RecordUpdates) != 1 ||
		len(scheduledVersion.Metadata.RecordUpdates[0].InputAssignments) != 1 {
		t.Fatalf("scheduled Flow view lost its update metadata: %#v", scheduledVersion)
	}
	updateValue, ok := scheduledVersion.Metadata.RecordUpdates[0].InputAssignments[0].Value.(map[string]interface{})
	if !ok || updateValue["stringValue"] != "scheduled-complete" {
		t.Fatalf("scheduled Flow update value = %#v, want typed literal", updateValue)
	}
	appReleaseDigest := catalogsnapshot.ComputeDigest([]byte("runtime-scheduled-app-release"))
	serviceReleaseDigest := catalogsnapshot.ComputeDigest([]byte("runtime-scheduled-service-release"))
	receiptArtifact, err := catalogsnapshot.BuildServingCompatibilityReceipt(
		catalogsnapshot.ServingCompatibilityReceiptInput{
			OrgID: flowView.OrgID(), AppInstallationID: flowView.AppInstallationID(),
			AppReleaseDigest: appReleaseDigest, AppCapabilityID: integrationScheduledAppCapability,
			RuntimeCapabilityID: flowsemantic.CapabilityName,
			WorkloadID:          "runtime-scheduled-worker", ServiceReleaseDigest: serviceReleaseDigest,
			OwnershipEpoch:            scheduledOwner.OwnershipEpoch,
			CatalogSnapshotDigest:     startupSnapshotArtifact.Digest(),
			RuntimeMetadataViewDigest: startupFlowArtifact.Digest(),
			Lineage:                   startupFlowView.Lineage(), SecurityEpoch: 1,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	inactiveReceiptArtifact, err := catalogsnapshot.BuildServingCompatibilityReceipt(
		catalogsnapshot.ServingCompatibilityReceiptInput{
			OrgID: flowView.OrgID(), AppInstallationID: flowView.AppInstallationID(),
			AppReleaseDigest: appReleaseDigest, AppCapabilityID: integrationScheduledAppCapability,
			RuntimeCapabilityID: flowsemantic.CapabilityName,
			WorkloadID:          "runtime-scheduled-worker", ServiceReleaseDigest: serviceReleaseDigest,
			OwnershipEpoch:            scheduledOwner.OwnershipEpoch,
			CatalogSnapshotDigest:     startupSnapshotArtifact.Digest(),
			RuntimeMetadataViewDigest: startupFlowArtifact.Digest(),
			Lineage:                   startupFlowView.Lineage(), SecurityEpoch: 2,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{
		snapshotArtifact,
		mutationArtifact,
		flowArtifact,
		startupSnapshotArtifact,
		startupMutationArtifact,
		startupFlowArtifact,
		receiptArtifact,
		inactiveReceiptArtifact,
	} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	seedScheduledCatalogCapability(
		t,
		db,
		startupFlowView,
		appReleaseDigest,
		serviceReleaseDigest,
		receiptArtifact.Digest(),
		scheduledOwner.OwnershipEpoch,
	)
	if _, err := db.Exec(`CREATE TABLE ScheduledTarget (
Id VARCHAR(191) PRIMARY KEY,
Status VARCHAR(191) NULL,
IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
LastModifiedById VARCHAR(191) NULL,
LastModifiedDate DATETIME(6) NULL
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`INSERT INTO ScheduledTarget (Id, Status) VALUES
('scheduled-success', 'before'),
('scheduled-rollback', 'before')`); err != nil {
		t.Fatal(err)
	}

	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(flowView)
	if err != nil {
		t.Fatal(err)
	}
	transactions := persistence.NewTransactionManager(database.NewConnectionFromDB(db))
	reservations, err := automation.NewFencedScheduledRunReservationService(
		db, runtimeWorkerScheduledRunTransactionOwner{transactions},
		scheduledOwner, binding, 10*time.Second, 3,
	)
	if err != nil {
		t.Fatal(err)
	}
	successKey, err := automation.NewScheduledRunReservationKey(
		"definition-scheduled-integration",
		"version-scheduled-integration",
		"walking-slice-success",
		"scheduled-success",
	)
	if err != nil {
		t.Fatal(err)
	}
	rollbackKey, err := automation.NewScheduledRunReservationKey(
		"definition-scheduled-rollback",
		"version-scheduled-rollback",
		"walking-slice-rollback",
		"scheduled-rollback",
	)
	if err != nil {
		t.Fatal(err)
	}
	successLease, err := reservations.Claim(
		t.Context(), successKey,
		[]byte(`{"$Record":{"Id":"scheduled-success","Status":"before"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	rollbackLease, err := reservations.Claim(
		t.Context(), rollbackKey,
		[]byte(`{"$Record":{"Id":"scheduled-rollback","Status":"before"}}`),
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_ScheduledRunReservation
SET LeaseExpiresAt = UTC_TIMESTAMP(6) - INTERVAL 1 SECOND
WHERE Id IN (?, ?)`, successLease.ReservationID, rollbackLease.ReservationID); err != nil {
		t.Fatal(err)
	}

	config := integrationMetadataWorkerConfig(scheduledOwner, storeRoot, startupFlowView)
	config.Role.AppCapabilityID = integrationScheduledAppCapability
	config.Role.ExpectedCompatibilityReceiptDigest = string(receiptArtifact.Digest())
	config.Role.PollInterval = "50ms"
	config.Role.LeaseDuration = "10s"
	mispinned := config
	mispinned.Role.ExpectedCompatibilityReceiptDigest = string(inactiveReceiptArtifact.Digest())
	_, mispinnedDB, err := assembleProduction(
		context.Background(),
		integrationWorkerSourceRevision,
		mispinned,
		permissionStateFactory,
	)
	if mispinnedDB != nil {
		_ = mispinnedDB.Close()
	}
	if err == nil || !strings.Contains(err.Error(), "configured receipt is not the active Catalog binding") {
		t.Fatalf("inactive epoch-matching receipt error = %v, want fail closed", err)
	}
	process, workerDB, err := assembleProduction(
		context.Background(),
		integrationWorkerSourceRevision,
		config,
		permissionStateFactory,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		_ = workerDB.Close()
		t.Fatal(err)
	}
	processResult := make(chan error, 1)
	go func() { processResult <- process.Wait() }()
	assertWorkerOperationalSurface(t, process, scheduledOwner.WorkFamily)
	waitForScheduledWalkingSlice(
		t,
		db,
		processResult,
	)
	var effectEventID string
	var payloadJSON []byte
	if err := db.QueryRow(`SELECT Id, Payload
FROM _System_RecordEventOutbox
WHERE RecordKey = 'ScheduledTarget/scheduled-success'
ORDER BY CreatedDate
LIMIT 1`).Scan(&effectEventID, &payloadJSON); err != nil {
		t.Fatal(err)
	}
	var envelope eventfeature.InternalRecordEventEnvelope
	if err := json.Unmarshal(payloadJSON, &envelope); err != nil {
		t.Fatal(err)
	}
	effectBinding, err := catalogsnapshot.OpenRuntimeWorkBindingSet(
		envelope.InternalMetadata["runtime_work_metadata_binding"],
	)
	if err != nil {
		t.Fatal(err)
	}
	mutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		t.Fatal(err)
	}
	if effectBinding.Primary != mutationBinding ||
		effectBinding.FlowWork == nil ||
		*effectBinding.FlowWork != binding {
		t.Fatalf(
			"scheduled effect binding = %#v, want mutation %s plus durable Flow %s; startup Flow view was %s",
			effectBinding,
			mutationBinding.RuntimeMetadataViewDigest,
			binding.RuntimeMetadataViewDigest,
			startupFlowView.Digest(),
		)
	}
	nextOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "transfer-scheduled-role", WorkFamily: scheduledOwner.WorkFamily,
		DeploymentID: "runtime-scheduled-role-next", WorkerReleaseDigest: integrationWorkerDigest,
		ExpectedOwnershipEpoch: scheduledOwner.OwnershipEpoch,
	})
	ageScheduledFlowOwnershipCutover(t, db, nextOwner)
	select {
	case err := <-processResult:
		if !errors.Is(err, persistence.ErrRuntimeWorkOwnershipLost) {
			t.Fatalf("stale scheduled worker exit = %v", err)
		}
	case <-time.After(5 * time.Second):
		t.Fatal("scheduled worker did not stop after ownership transfer")
	}
	stopWorker(t, process, workerDB)

	nextConfig := integrationMetadataWorkerConfig(nextOwner, storeRoot, startupFlowView)
	nextConfig.Role.AppCapabilityID = integrationScheduledAppCapability
	nextConfig.Role.ExpectedCompatibilityReceiptDigest = string(receiptArtifact.Digest())
	nextConfig.Role.PollInterval = "50ms"
	nextConfig.Role.LeaseDuration = "10s"
	nextProcess, nextDB, err := assembleProduction(
		context.Background(),
		integrationWorkerSourceRevision,
		nextConfig,
		permissionStateFactory,
	)
	if err != nil {
		t.Fatalf("next Worker owner must retain independently fenced Catalog binding: %v", err)
	}
	stopWorker(t, nextProcess, nextDB)
}

func seedScheduledCatalogCapability(
	t *testing.T,
	db *sql.DB,
	view *catalogsnapshot.RuntimeMetadataView,
	appReleaseDigest, serviceReleaseDigest, receiptDigest catalogsnapshot.Digest,
	ownershipEpoch uint64,
) {
	t.Helper()
	for _, statement := range catalogapp.SchemaStatementsForTest() {
		if _, err := db.ExecContext(t.Context(), statement); err != nil {
			t.Fatal(err)
		}
	}
	now := time.Now().UTC()
	if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_CatalogAppInstallation
(AppInstallationId, OrgId, AppId, State, ActiveReleaseDigest, CatalogRevision,
 ActivationEpoch, LiveAppKey, CreatedDate, LastModifiedDate)
VALUES (?, ?, 'runtime-scheduled-test', 'ACTIVE', ?, ?, 1, ?, ?, ?)`,
		view.AppInstallationID(), view.OrgID(), string(appReleaseDigest),
		uint64(view.Lineage().CatalogRevision), strings.Repeat("1", 64), now, now,
	); err != nil {
		t.Fatal(err)
	}
	if _, err := db.ExecContext(t.Context(), `INSERT INTO _System_CatalogHandlerBinding
(HandlerBindingId, OrgId, AppInstallationId, AppReleaseDigest, CapabilityId, HandlerRef,
 WorkloadId, ServiceReleaseDigest, OwnershipEpoch, State, CompatibilityReceiptDigest,
 NaturalKey, ActiveKey, CreatedDate, LastModifiedDate)
VALUES ('00000000-0000-7000-8000-000000000080', ?, ?, ?, ?, 'runtime.scheduled-flow',
        'runtime-scheduled-worker', ?, ?, 'ACTIVE', ?, ?, ?, ?, ?)`,
		view.OrgID(), view.AppInstallationID(), string(appReleaseDigest),
		integrationScheduledAppCapability, string(serviceReleaseDigest), ownershipEpoch,
		string(receiptDigest), strings.Repeat("2", 64), strings.Repeat("3", 64), now, now,
	); err != nil {
		t.Fatal(err)
	}
}

func waitForScheduledWalkingSlice(
	t *testing.T,
	db *sql.DB,
	processResult <-chan error,
) {
	t.Helper()
	// Every scheduled effect deliberately holds the work-family ownership row
	// through its business transaction. The production walking slice exercises
	// several such effects, so race instrumentation and a cold TiDB serialize
	// them behind that fence. Keep a bounded integration budget without
	// weakening the production ownership lock.
	deadline := time.Now().Add(30 * time.Second)
	for {
		select {
		case err := <-processResult:
			t.Fatalf("scheduled Worker exited before walking slice converged: %v", err)
		default:
		}
		var successStatus, rollbackStatus string
		var successReservation, rollbackReservation string
		var catalogDiscoveredReservations int
		err := db.QueryRow(`SELECT
(SELECT Status FROM ScheduledTarget WHERE Id = 'scheduled-success'),
(SELECT Status FROM ScheduledTarget WHERE Id = 'scheduled-rollback'),
(SELECT Status FROM _System_ScheduledRunReservation
  WHERE FlowDefinitionId = 'definition-scheduled-integration'
    AND ScheduleWindowKey = 'walking-slice-success' AND InputKey = 'scheduled-success'),
(SELECT Status FROM _System_ScheduledRunReservation
  WHERE FlowDefinitionId = 'definition-scheduled-rollback'
    AND ScheduleWindowKey = 'walking-slice-rollback' AND InputKey = 'scheduled-rollback'),
(SELECT COUNT(*) FROM _System_ScheduledRunReservation
  WHERE FlowDefinitionId = 'definition-scheduled-rollback'
    AND ScheduleWindowKey <> 'walking-slice-rollback'
    AND Status = 'failed')`).
			Scan(
				&successStatus, &rollbackStatus,
				&successReservation, &rollbackReservation,
				&catalogDiscoveredReservations,
			)
		if err == nil &&
			successStatus == "scheduled-complete" &&
			rollbackStatus == "before" &&
			successReservation == "completed" &&
			rollbackReservation == "failed" &&
			catalogDiscoveredReservations > 0 {
			return
		}
		if time.Now().After(deadline) {
			rows, rowsErr := db.Query(`SELECT FlowDefinitionId, InputKey, Status, Attempt,
       COALESCE(LastError, ''), LeaseExpiresAt
FROM _System_ScheduledRunReservation
ORDER BY FlowDefinitionId, InputKey`)
			durableRows := make([]string, 0)
			if rowsErr == nil {
				for rows.Next() {
					var definitionID, inputKey, status, lastError string
					var attempt int
					var leaseExpiresAt sql.NullTime
					if scanErr := rows.Scan(
						&definitionID,
						&inputKey,
						&status,
						&attempt,
						&lastError,
						&leaseExpiresAt,
					); scanErr != nil {
						durableRows = append(durableRows, "scan-error="+scanErr.Error())
						break
					}
					lease := "<none>"
					if leaseExpiresAt.Valid {
						lease = leaseExpiresAt.Time.UTC().Format(time.RFC3339Nano)
					}
					durableRows = append(durableRows, fmt.Sprintf(
						"%s/%s=%s attempt=%d lease=%s error=%q",
						definitionID,
						inputKey,
						status,
						attempt,
						lease,
						lastError,
					))
				}
				_ = rows.Close()
			}
			t.Fatalf(
				"scheduled walking slice did not converge: records=%q/%q reservations=%q/%q Catalog-discovered=%d err=%v rowsErr=%v rows=%v",
				successStatus, rollbackStatus,
				successReservation, rollbackReservation,
				catalogDiscoveredReservations, err, rowsErr, durableRows,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func integrationPermissionStateFactory(t *testing.T) permissionStateGateFactory {
	t.Helper()
	info := &debug.BuildInfo{Deps: []*debug.Module{
		{Path: "github.com/pingcap-inc/nexuscrm-contract", Version: "v0.2.9"},
		{Path: "github.com/pingcap-inc/nexuscrm-api-host", Version: "v0.18.0"},
	}}
	return func(
		ctx context.Context,
		db *sql.DB,
		config runtimepermissionstate.Config,
	) (*runtimepermissionstate.Gate, error) {
		return runtimepermissionstate.NewIntegrationGate(ctx, db, config, info)
	}
}

// assertMetadataWorkerAssemblyRequiresOwner proves the producer closure in the
// same order Runtime verifies it. A metadata worker is not ready merely because
// its own queue owner exists: every durable output family produced by the shared
// mutation graph must already be independently fenced.
func assertMetadataWorkerAssemblyRequiresOwner(
	t *testing.T,
	config runtimeconfig.Worker,
	family string,
	permissionStateFactory permissionStateGateFactory,
) {
	t.Helper()
	_, workerDB, err := assembleProduction(
		context.Background(), integrationWorkerSourceRevision, config, permissionStateFactory,
	)
	if workerDB != nil {
		_ = workerDB.Close()
	}
	if err == nil {
		t.Fatalf("metadata worker started without required durable producer owner %q", family)
	}
	if !strings.Contains(err.Error(), family) {
		t.Fatalf("missing durable producer owner %q error = %v", family, err)
	}
}

func activateWorker(t *testing.T, control *persistence.RuntimeWorkOwnershipControl, request persistence.RuntimeWorkOwnershipActivation) persistence.RuntimeWorkOwnership {
	t.Helper()
	owner, err := control.Activate(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	return owner
}

// ageScheduledFlowOwnershipCutover places a composition fixture after the
// stopped-upgrade isolation window. The focused Automation integration tests
// separately prove that a fresh ownership row must fail closed until this
// legacy-producer lookback has elapsed.
func ageScheduledFlowOwnershipCutover(
	t *testing.T,
	db *sql.DB,
	owner persistence.RuntimeWorkOwnership,
) {
	t.Helper()
	result, err := db.Exec(`UPDATE _System_RuntimeWorkOwnership
SET LastModifiedDate = UTC_TIMESTAMP(6) - INTERVAL 2 MINUTE
WHERE WorkFamily = ? AND DeploymentId = ? AND WorkerReleaseDigest = ? AND OwnershipEpoch = ?`,
		owner.WorkFamily,
		owner.DeploymentID,
		owner.WorkerReleaseDigest,
		owner.EpochDecimal(),
	)
	if err != nil {
		t.Fatal(err)
	}
	rows, err := result.RowsAffected()
	if err != nil {
		t.Fatal(err)
	}
	if rows != 1 {
		t.Fatalf("aged scheduled Flow ownership rows = %d, want exact active tuple", rows)
	}
}

func integrationWorkerConfig(owner persistence.RuntimeWorkOwnership) runtimeconfig.Worker {
	return runtimeconfig.Worker{AdminAddress: "127.0.0.1:0", Role: runtimeconfig.WorkerRole{
		Schema: runtimeconfig.WorkerConfigSchemaV3, PermissionStateProtocol: runtimeconfig.PermissionStateProtocolV1,
		WorkFamily:   owner.WorkFamily,
		DeploymentID: owner.DeploymentID, WorkerReleaseDigest: owner.WorkerReleaseDigest, OwnershipEpoch: owner.OwnershipEpoch,
		WorkerSourceRevision: integrationWorkerSourceRevision,
		BatchSize:            10, PollInterval: "50ms", OwnershipHealthInterval: "1s",
		LeaseDuration: "2s", RetryJitterMax: "5ms", MaxAttempts: 5,
		DBMaxOpenConns: 4, DBMaxIdleConns: 2, ShutdownTimeout: "5s",
	}}
}

func integrationMetadataWorkerConfig(owner persistence.RuntimeWorkOwnership, storeRoot string, view *catalogsnapshot.RuntimeMetadataView) runtimeconfig.Worker {
	config := integrationWorkerConfig(owner)
	lineage := view.Lineage()
	config.Role.OrgID = view.OrgID()
	config.Role.AppInstallationID = view.AppInstallationID()
	config.Role.RuntimeCapabilityID = view.CapabilityID()
	config.Role.CatalogSnapshotStoreRoot = storeRoot
	config.Role.SchemaGeneration = uint64(lineage.SchemaGeneration)
	config.Role.EngineGeneration = uint64(lineage.EngineGeneration)
	if owner.WorkFamily == eventfeature.RuntimeRecordEventDispatchWorkFamily {
		config.Role.ProcessedRetention = "1s"
		config.Role.ProcessedCleanupInterval = "1s"
	}
	return config
}

func insertExpiredProcessedRecordEvent(t *testing.T, db *sql.DB, binding catalogsnapshot.RuntimeWorkBinding) string {
	t.Helper()
	const id = "record-event-processed-expired"
	scopeDigest := mustRuntimeWorkScopeDigest(t, persistence.RuntimeWorkScope{
		OrgID: binding.OrgID, AppInstallationID: binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	})
	_, err := db.Exec(`INSERT INTO _System_RecordEventOutbox
(Id, OrgId, AppInstallationId, RuntimeCapabilityId, ExecutionScopeDigest, EventType, Payload, Status, RetryCount,
 ProcessedDate, CreatedDate, LastModifiedDate)
VALUES (?, ?, ?, ?, ?, 'RecordUpdated', '{}', 'processed', 0,
        UTC_TIMESTAMP(6) - INTERVAL 2 SECOND, UTC_TIMESTAMP(6) - INTERVAL 2 SECOND, UTC_TIMESTAMP(6) - INTERVAL 2 SECOND)`,
		id, binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID, scopeDigest)
	if err != nil {
		t.Fatal(err)
	}
	return id
}

func waitForRecordEventDeletion(t *testing.T, db *sql.DB, id string) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		var count int
		err := db.QueryRow(`SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE Id = ?`, id).Scan(&count)
		if err == nil && count == 0 {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("processed record event %s was not cleaned up: count=%d err=%v", id, count, err)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func enqueueBoundRecordEvent(t *testing.T, db *sql.DB, view *catalogsnapshot.RuntimeMetadataView) string {
	t.Helper()
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	ctx := eventfeature.WithDurableExecutionBinding(context.Background(), encoded)
	ctx, err = persistence.WithRuntimeWorkScope(ctx, persistence.RuntimeWorkScope{
		OrgID: binding.OrgID, AppInstallationID: binding.AppInstallationID, RuntimeCapabilityID: binding.RuntimeCapabilityID,
	})
	if err != nil {
		t.Fatal(err)
	}
	service := eventfeature.NewOutboxService(
		persistence.NewRecordEventOutboxRepository(db), eventfeature.NewEventBus(),
		persistence.NewTransactionManager(database.NewConnectionFromDB(db)),
	)
	if err := service.EnqueueEvent(ctx, sharedevents.RecordUpdated, eventfeature.RecordEventPayload{
		ObjectAPIName: "Child", Record: models.SObject{constants.FieldID: "a01000000000001AAA", "ParentId": "parent-1"},
	}); err != nil {
		t.Fatal(err)
	}
	var id string
	if err := db.QueryRow(`SELECT Id FROM _System_RecordEventOutbox
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?`,
		binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID).Scan(&id); err != nil {
		t.Fatal(err)
	}
	return id
}

func enqueueDualBoundTimedRecordEvent(
	t *testing.T,
	db *sql.DB,
	mutationView *catalogsnapshot.RuntimeMetadataView,
	flowView *catalogsnapshot.RuntimeMetadataView,
	recordID string,
	eventAge time.Duration,
) string {
	t.Helper()
	mutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		t.Fatal(err)
	}
	flowBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(flowView)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBindingSet(
		catalogsnapshot.RuntimeWorkBindingSet{
			Primary:  mutationBinding,
			FlowWork: &flowBinding,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx := eventfeature.WithDurableExecutionBinding(t.Context(), encoded)
	ctx, err = persistence.WithRuntimeWorkScope(ctx, persistence.RuntimeWorkScope{
		OrgID:               mutationBinding.OrgID,
		AppInstallationID:   mutationBinding.AppInstallationID,
		RuntimeCapabilityID: mutationBinding.RuntimeCapabilityID,
	})
	if err != nil {
		t.Fatal(err)
	}
	service := eventfeature.NewOutboxService(
		persistence.NewRecordEventOutboxRepository(db),
		eventfeature.NewEventBus(),
		persistence.NewTransactionManager(database.NewConnectionFromDB(db)),
	)
	if err := service.EnqueueEvent(
		ctx,
		sharedevents.RecordUpdated,
		eventfeature.RecordEventPayload{
			ObjectAPIName: "ScheduledTarget",
			Record: models.SObject{
				constants.FieldID: recordID,
				"Status":          "before",
			},
		},
	); err != nil {
		t.Fatal(err)
	}
	var id string
	if err := db.QueryRow(
		`SELECT Id FROM _System_RecordEventOutbox
WHERE OrgId = ? AND AppInstallationId = ? AND RuntimeCapabilityId = ?
  AND RecordKey = ?`,
		mutationBinding.OrgID,
		mutationBinding.AppInstallationID,
		mutationBinding.RuntimeCapabilityID,
		"ScheduledTarget/"+recordID,
	).Scan(&id); err != nil {
		t.Fatal(err)
	}
	if eventAge > 0 {
		if _, err := db.Exec(
			`UPDATE _System_RecordEventOutbox
SET CreatedDate = UTC_TIMESTAMP(6) - INTERVAL ? MICROSECOND,
    LastModifiedDate = UTC_TIMESTAMP(6) - INTERVAL ? MICROSECOND
WHERE Id = ?`,
			eventAge.Microseconds(),
			eventAge.Microseconds(),
			id,
		); err != nil {
			t.Fatal(err)
		}
	}
	return id
}

func waitForRecordEventStatus(t *testing.T, db *sql.DB, id, status string) (int, string) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		var current string
		var retries int
		var lastError sql.NullString
		err := db.QueryRow(`SELECT Status, RetryCount, ErrorMessage FROM _System_RecordEventOutbox WHERE Id = ?`, id).
			Scan(&current, &retries, &lastError)
		if err == nil && current == status {
			return retries, lastError.String
		}
		if time.Now().After(deadline) {
			t.Fatalf("record event %s status=%q retries=%d error=%q query=%v, want %q", id, current, retries, lastError.String, err, status)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func prepareRollupRecords(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, statement := range []string{
		`CREATE TABLE Parent (Id VARCHAR(255) PRIMARY KEY, ChildCount BIGINT NULL, LastModifiedById VARCHAR(255) NULL, LastModifiedDate DATETIME(6) NULL)`,
		`CREATE TABLE Child (Id VARCHAR(255) PRIMARY KEY, ParentId VARCHAR(255) NULL, IsDeleted TINYINT(1) NOT NULL DEFAULT 0)`,
		`INSERT INTO Parent (Id, ChildCount) VALUES ('parent-1', 0)`,
		`INSERT INTO Child (Id, ParentId, IsDeleted) VALUES ('child-1', 'parent-1', 0), ('child-2', 'parent-1', 0), ('child-deleted', 'parent-1', 1)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
}

func insertBoundRollupDelta(t *testing.T, db *sql.DB, binding catalogsnapshot.RuntimeWorkBinding) {
	t.Helper()
	scopeDigest := mustRuntimeWorkScopeDigest(t, persistence.RuntimeWorkScope{
		OrgID:               binding.OrgID,
		AppInstallationID:   binding.AppInstallationID,
		RuntimeCapabilityID: binding.RuntimeCapabilityID,
	})
	_, err := db.Exec(`INSERT INTO _System_RollupDelta
(Id, SourceObject, SourceRecordId, ParentObject, ParentRecordId, RollupField, Delta, DedupeKey,
 ScopeDigest, OrgId, AppInstallationId, RuntimeCapabilityId, CatalogRevision, SchemaGeneration, EngineGeneration,
 CatalogSnapshotDigest, RuntimeMetadataViewDigest)
VALUES ('rollup-1', 'Child', 'child-1', 'Parent', 'parent-1', 'ChildCount', 1, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		strings.Repeat("d", 64), scopeDigest, binding.OrgID, binding.AppInstallationID, binding.RuntimeCapabilityID,
		binding.CatalogRevision, binding.SchemaGeneration, binding.EngineGeneration,
		binding.CatalogSnapshotDigest, binding.RuntimeMetadataViewDigest)
	if err != nil {
		t.Fatal(err)
	}
}

func waitForRollupEffect(t *testing.T, db *sql.DB, want int64) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		var queueRows int
		var count sql.NullInt64
		err := db.QueryRow(`SELECT
(SELECT COUNT(*) FROM _System_RollupDelta WHERE Id = 'rollup-1'),
(SELECT ChildCount FROM Parent WHERE Id = 'parent-1')`).Scan(&queueRows, &count)
		if err == nil && queueRows == 0 && count.Valid && count.Int64 == want {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("rollup queue=%d parent count=%v query=%v, want count=%d", queueRows, count, err, want)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func insertAuditOutbox(t *testing.T, db *sql.DB, recordID string) int64 {
	t.Helper()
	sequence := int64(1)
	payload, err := json.Marshal(audit.AuditEvent{
		ObjectApiName: "Issue", RecordId: recordID, Action: audit.ActionUpdate,
		ChangedById: "user-1", Entrypoint: audit.EntrypointRestAPI, Outcome: audit.OutcomeSuccess,
		FieldDiffs: []audit.FieldDiff{{FieldName: "Subject", OldValue: "before", NewValue: "after", HistorySequence: &sequence}},
	})
	if err != nil {
		t.Fatal(err)
	}
	result, err := db.Exec(`INSERT INTO _System_AuditOutbox (Payload, EnqueuedAt, Attempts) VALUES (?, CURRENT_TIMESTAMP(6), 0)`, payload)
	if err != nil {
		t.Fatal(err)
	}
	id, err := result.LastInsertId()
	if err != nil {
		t.Fatal(err)
	}
	return id
}

func assertAuditCounts(t *testing.T, db *sql.DB, outboxID int64, wantOutbox, wantEffects int) {
	t.Helper()
	var outbox, effects int
	if err := db.QueryRow(`SELECT
(SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?),
(SELECT COUNT(*) FROM _System_AuditLog WHERE OutboxRowId = ?)`, outboxID, outboxID).Scan(&outbox, &effects); err != nil {
		t.Fatal(err)
	}
	if outbox != wantOutbox || effects != wantEffects {
		t.Fatalf("audit counts outbox=%d effects=%d, want %d/%d", outbox, effects, wantOutbox, wantEffects)
	}
}

func waitForAuditCounts(t *testing.T, db *sql.DB, outboxID int64, wantOutbox, wantEffects int) {
	t.Helper()
	deadline := time.Now().Add(5 * time.Second)
	for {
		var outbox, effects int
		err := db.QueryRow(`SELECT
(SELECT COUNT(*) FROM _System_AuditOutbox WHERE Id = ?),
(SELECT COUNT(*) FROM _System_AuditLog WHERE OutboxRowId = ?)`, outboxID, outboxID).Scan(&outbox, &effects)
		if err == nil && outbox == wantOutbox && effects == wantEffects {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf("audit did not converge outbox=%d effects=%d err=%v", outbox, effects, err)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

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

func waitProcess(process interface{ Wait() error }) <-chan error {
	done := make(chan error, 1)
	go func() { done <- process.Wait() }()
	return done
}

func assertWorkerOperationalSurface(t *testing.T, process interface{ AdminAddress() string }, workFamily string) {
	t.Helper()
	base := "http://" + process.AdminAddress()
	response, err := http.Get(base + processhttp.LivePath)
	if err != nil {
		t.Fatalf("GET %s: %v", processhttp.LivePath, err)
	}
	_ = response.Body.Close()
	if response.StatusCode != http.StatusOK {
		t.Fatalf("GET %s status=%d, want 200", processhttp.LivePath, response.StatusCode)
	}

	// Start establishes the process surface before the asynchronous ownership
	// controller verifies the durable family tuple. Readiness must therefore be
	// observed as a state transition, not assumed to be synchronous with Start.
	deadline := time.Now().Add(5 * time.Second)
	for {
		response, err = http.Get(base + processhttp.ReadyPath)
		if err == nil {
			_ = response.Body.Close()
			if response.StatusCode == http.StatusOK {
				break
			}
		}
		if time.Now().After(deadline) {
			if err != nil {
				t.Fatalf("GET %s before deadline: %v", processhttp.ReadyPath, err)
			}
			t.Fatalf("GET %s status=%d before deadline, want 200", processhttp.ReadyPath, response.StatusCode)
		}
		time.Sleep(20 * time.Millisecond)
	}
	executionClass := workerExecutionClassMetadata
	if _, ok := supportedPayloadWorkFamilies[workFamily]; ok {
		executionClass = workerExecutionClassPayload
	}
	for path, facts := range map[string][]string{
		processhttp.VersionPath: {`"role":"runtime-worker"`, `"version":"` + integrationWorkerSourceRevision + `"`},
		processhttp.CapabilitiesPath: {
			`"role":"runtime-worker"`,
			`"capabilities":[`,
			`"` + workFamily + `"`,
			`"permission_state.config=` + runtimeconfig.WorkerConfigSchemaV3 + `"`,
			`"permission_state.role=reader-writer"`,
			`"` + runtimeconfig.PermissionStateProtocolV1 + `"`,
		},
		runtimeprocess.MetricsPath: {
			`nexuscrm_runtime_process_ready{role="runtime-worker"} 1`,
			`nexuscrm_runtime_process_readiness_check_known{role="runtime-worker"} 1`,
			`nexuscrm_runtime_worker_info{execution_class="` + executionClass + `",work_family="` + workFamily + `"} 1`,
			`nexuscrm_runtime_worker_authority_checks_total{authority="permission_state"}`,
			`nexuscrm_runtime_worker_authority_checks_total{authority="work_ownership"}`,
			`nexuscrm_runtime_worker_queue_depth`,
			`nexuscrm_runtime_worker_db_pool_max_open_connections 4`,
		},
	} {
		response, err := http.Get(base + path)
		if err != nil {
			t.Fatalf("GET %s: %v", path, err)
		}
		body, readErr := io.ReadAll(response.Body)
		_ = response.Body.Close()
		if readErr != nil || response.StatusCode != http.StatusOK {
			t.Fatalf("GET %s status=%d read=%v", path, response.StatusCode, readErr)
		}
		for _, fact := range facts {
			if !strings.Contains(string(body), fact) {
				t.Fatalf("GET %s missing %q: %s", path, fact, body)
			}
		}
	}
	response, err = http.Get(base + "/debug/pprof/goroutine")
	if err != nil {
		t.Fatalf("GET loopback diagnostics: %v", err)
	}
	_ = response.Body.Close()
	if response.StatusCode != http.StatusOK {
		t.Fatalf("GET loopback diagnostics status=%d, want 200", response.StatusCode)
	}
}

func openWorkerScratchDB(t *testing.T) (*sql.DB, func()) {
	t.Helper()
	host, port := os.Getenv("TIDB_HOST"), os.Getenv("TIDB_PORT")
	if port == "" {
		port = "4000"
	}
	user, password := os.Getenv("TIDB_USER"), os.Getenv("TIDB_PASSWORD")
	admin, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/?parseTime=true&loc=UTC", user, password, host, port))
	if err != nil {
		t.Fatal(err)
	}
	name := fmt.Sprintf("nexuscrm_runtime_worker_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE DATABASE `" + name + "`"); err != nil {
		_ = admin.Close()
		t.Fatal(err)
	}
	t.Setenv("TIDB_DATABASE", name)
	t.Setenv("TIDB_TLS_DISABLE", "1")
	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true&loc=UTC&collation=utf8mb4_unicode_ci", user, password, host, port, name))
	if err != nil {
		t.Fatal(err)
	}
	prepareWorkerSchema(t, db)
	return db, func() {
		_ = db.Close()
		_, _ = admin.Exec("DROP DATABASE IF EXISTS `" + name + "`")
		_ = admin.Close()
	}
}

func prepareWorkerSchema(t *testing.T, db *sql.DB) {
	t.Helper()
	wanted := workerScratchSystemTables()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	repository := persistence.NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if !wanted[definition.TableName] {
			continue
		}
		if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
			t.Fatalf("create canonical worker table %s: %v", definition.TableName, err)
		}
		delete(wanted, definition.TableName)
	}
	if len(wanted) != 0 {
		t.Fatalf("canonical worker schema is missing tables: %v", wanted)
	}
	permissionState, err := persistence.NewPermissionStateRepository(db)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := permissionState.EnsureInitialized(context.Background()); err != nil {
		t.Fatalf("initialize Permission State: %v", err)
	}
}