package persistence

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"testing"
	"time"

	database "github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
)

type runtimePlatformCommandIntegrationTransactionOwner struct {
	*TransactionManager
}

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

func TestRuntimePlatformCommandExecutionTiDBFencesEffectsRestartAndRepair(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	// Command availability is a UTC authority, independent of the TiDB session
	// timezone. Keep one session at a deliberately non-UTC offset so this test
	// catches writers that fall back to CURRENT_TIMESTAMP defaults while Claim
	// compares against UTC_TIMESTAMP.
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	if _, err := db.Exec(`SET time_zone = '+08:00'`); err != nil {
		t.Fatal(err)
	}
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	createRuntimePlatformCommandOwnershipTable(t, db)
	if _, err := db.Exec(`CREATE TABLE CommandEffect (
		CommandId VARCHAR(255) PRIMARY KEY,
		Value VARCHAR(255) NOT NULL
	)`); err != nil {
		t.Fatal(err)
	}
	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:         "activate-platform-command-one",
		WorkFamily:          RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "platform-command-one",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := RuntimeWorkScope{
		OrgID: "org-1", AppInstallationID: "installation-1",
		RuntimeCapabilityID: "runtime.commands.accept.v1",
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	transactionOwner := runtimePlatformCommandIntegrationTransactionOwner{transactions}
	store, err := NewRuntimePlatformCommandExecutionStore(
		db, transactionOwner, owner, scope, time.Second, 2,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	repository := NewRuntimePlatformCommandRepository(db)
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-1", scope)
	otherScope := RuntimeWorkScope{
		OrgID: "org-1", AppInstallationID: "installation-2",
		RuntimeCapabilityID: scope.RuntimeCapabilityID,
	}
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "other-command", otherScope)
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-mixed-authority", scope)
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET ExecutionScopeDigest = ?
WHERE CommandId = ?`,
		"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
		"command-mixed-authority",
	); err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err == nil {
		t.Fatal("CheckReady() accepted mixed explicit and derived execution authority")
	}
	if _, err := db.Exec(`DELETE FROM _System_RuntimePlatformCommand WHERE CommandId = ?`,
		"command-mixed-authority",
	); err != nil {
		t.Fatal(err)
	}
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-partial-dependency", scope)
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET DependencyRecordId = 'orphaned-record' WHERE CommandId = 'command-partial-dependency'`); err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err == nil {
		t.Fatal("CheckReady() accepted partial dependency residue on an ACCEPTED row")
	}
	if _, err := db.Exec(`DELETE FROM _System_RuntimePlatformCommand WHERE CommandId = ?`,
		"command-partial-dependency",
	); err != nil {
		t.Fatal(err)
	}

	lease, found, err := store.Claim(t.Context())
	if err != nil || !found || lease.Row.CommandID != "command-1" || lease.Attempt != 1 {
		t.Fatalf("Claim() = %#v found=%t err=%v", lease, found, err)
	}
	if err := store.Execute(t.Context(), lease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("test transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandEffect (CommandId, Value) VALUES (?, ?)`,
			lease.Row.CommandID, "committed"); err != nil {
			return nil, err
		}
		return []byte(`{"status":"completed"}`), nil
	}); err != nil {
		t.Fatal(err)
	}
	var effect, state string
	var count int
	if err := db.QueryRow(`SELECT Value FROM CommandEffect WHERE CommandId = ?`, lease.Row.CommandID).Scan(&effect); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?`, lease.Row.RowID).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if effect != "committed" || state != RuntimePlatformCommandCompleted {
		t.Fatalf("effect/state = %q/%q", effect, state)
	}

	// Restart sees the terminal row and cannot claim another AppInstallation.
	restarted, err := NewRuntimePlatformCommandExecutionStore(
		db, transactionOwner, owner, scope, time.Second, 2,
	)
	if err != nil {
		t.Fatal(err)
	}
	if lease, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("restart Claim() = %#v found=%t err=%v", lease, found, err)
	}

	// A V2 dependency is persisted only after the effect transaction rolls
	// back. The due replay retains the command's execution-attempt budget,
	// reopens the exact row, and clears dependency state on the same-transaction
	// terminal commit.
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-dependency", scope)
	dependencyLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found || dependencyLease.Attempt != 1 {
		t.Fatalf("dependency Claim() = %#v found=%t err=%v", dependencyLease, found, err)
	}
	rollbackCause := errors.New("dependency published")
	err = restarted.Execute(t.Context(), dependencyLease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("test transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandEffect (CommandId, Value) VALUES (?, ?)`,
			dependencyLease.Row.CommandID, "must-rollback-before-block"); err != nil {
			return nil, err
		}
		return nil, rollbackCause
	})
	if !errors.Is(err, rollbackCause) {
		t.Fatalf("dependency Execute() error = %v", err)
	}
	dependency := runtimePlatformCommandDependencyForTest(t, "Lead", "lead-1")
	if err := restarted.Block(t.Context(), dependencyLease, dependency, time.Second); err != nil {
		t.Fatal(err)
	}
	var blockedAttempt int
	var dependencyKey string
	if err := db.QueryRow(`SELECT State, Attempt, DependencyKeyHash
FROM _System_RuntimePlatformCommand WHERE Id = ?`, dependencyLease.Row.RowID).
		Scan(&state, &blockedAttempt, &dependencyKey); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandBlocked || blockedAttempt != 1 || dependencyKey != dependency.KeyHash {
		t.Fatalf("blocked state/attempt/dependency = %q/%d/%q", state, blockedAttempt, dependencyKey)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET DependencyKeyHash = ? WHERE Id = ?`, strings.Repeat("f", 64), dependencyLease.Row.RowID); err != nil {
		t.Fatal(err)
	}
	if err := restarted.CheckReady(t.Context()); err == nil {
		t.Fatal("CheckReady() accepted a dependency whose persisted key hash was tampered")
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET DependencyKeyHash = ? WHERE Id = ?`, dependency.KeyHash, dependencyLease.Row.RowID); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM CommandEffect WHERE CommandId = ?`,
		dependencyLease.Row.CommandID).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatal("business effect survived dependency suspension")
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET AvailableAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6)) WHERE Id = ?`,
		dependencyLease.Row.RowID); err != nil {
		t.Fatal(err)
	}
	woken, found, err := restarted.Claim(t.Context())
	if err != nil || !found || woken.Row.RowID != dependencyLease.Row.RowID ||
		woken.Attempt != 1 || woken.Row.Dependency == nil ||
		woken.Row.Dependency.KeyHash != dependency.KeyHash {
		t.Fatalf("dependency wake Claim() = %#v found=%t err=%v", woken, found, err)
	}
	if err := restarted.Replay(t.Context(), woken, errors.New("dependency recovered; replay exact command")); err != nil {
		t.Fatal(err)
	}
	var replayState string
	var replayDependencyCount int
	if err := db.QueryRow(`SELECT State,
       (DependencyKind IS NOT NULL OR DependencyBlockedAt IS NOT NULL)
FROM _System_RuntimePlatformCommand WHERE Id = ?`, woken.Row.RowID).
		Scan(&replayState, &replayDependencyCount); err != nil {
		t.Fatal(err)
	}
	if replayState != RuntimePlatformCommandAccepted || replayDependencyCount != 0 {
		t.Fatalf("replay disposition state/dependency = %q/%d", replayState, replayDependencyCount)
	}
	replayLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found || replayLease.Row.RowID != dependencyLease.Row.RowID ||
		replayLease.Attempt != 2 || replayLease.AttemptsRemaining != 0 ||
		replayLease.Row.Dependency != nil {
		t.Fatalf("exact replay Claim() = %#v found=%t err=%v", replayLease, found, err)
	}
	if err := restarted.Execute(t.Context(), replayLease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("test transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandEffect (CommandId, Value) VALUES (?, ?)`,
			replayLease.Row.CommandID, "committed-after-wake"); err != nil {
			return nil, err
		}
		return []byte(`{"status":"completed-after-wake"}`), nil
	}); err != nil {
		t.Fatal(err)
	}
	var dependencyColumns int
	if err := db.QueryRow(`SELECT State, Attempt,
       (DependencyKind IS NOT NULL OR DependencyBlockedAt IS NOT NULL)
FROM _System_RuntimePlatformCommand WHERE Id = ?`, woken.Row.RowID).
		Scan(&state, &blockedAttempt, &dependencyColumns); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandCompleted || blockedAttempt != 2 || dependencyColumns != 0 {
		t.Fatalf("completed wake state/attempt/dependency = %q/%d/%d", state, blockedAttempt, dependencyColumns)
	}

	// Exact replay must not silently replace the durable prerequisite. A changed
	// dependency is deterministic corruption and is terminalized through the
	// normal fenced disposition while preserving the original rollback proof.
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-dependency-drift", scope)
	driftLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("dependency drift Claim() = %#v found=%t err=%v", driftLease, found, err)
	}
	if err := restarted.Execute(t.Context(), driftLease, func(context.Context) ([]byte, error) {
		return nil, rollbackCause
	}); !errors.Is(err, rollbackCause) {
		t.Fatalf("dependency drift first Execute() error = %v", err)
	}
	if err := restarted.Block(t.Context(), driftLease, dependency, time.Second); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET AvailableAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6)) WHERE Id = ?`, driftLease.Row.RowID); err != nil {
		t.Fatal(err)
	}
	driftReplay, found, err := restarted.Claim(t.Context())
	if err != nil || !found || driftReplay.Attempt != driftLease.Attempt {
		t.Fatalf("dependency drift replay Claim() = %#v found=%t err=%v", driftReplay, found, err)
	}
	if err := restarted.Execute(t.Context(), driftReplay, func(context.Context) ([]byte, error) {
		return nil, rollbackCause
	}); !errors.Is(err, rollbackCause) {
		t.Fatalf("dependency drift replay Execute() error = %v", err)
	}
	changedDependency := runtimePlatformCommandDependencyForTest(t, "Contact", "contact-2")
	if err := restarted.Block(t.Context(), driftReplay, changedDependency, time.Second); !errors.Is(err, ErrRuntimePlatformCommandDependencyChanged) {
		t.Fatalf("changed dependency Block() error = %v", err)
	}
	if err := restarted.Fail(t.Context(), driftReplay, ErrRuntimePlatformCommandDependencyChanged); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT State, (DependencyKind IS NOT NULL OR DependencyBlockedAt IS NOT NULL)
FROM _System_RuntimePlatformCommand WHERE Id = ?`, driftReplay.Row.RowID).
		Scan(&state, &dependencyColumns); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandFailed || dependencyColumns != 0 {
		t.Fatalf("dependency drift terminal state/dependency = %q/%d", state, dependencyColumns)
	}

	// A dependency wait has a fixed release-owned bound; it cannot remain an
	// immortal queue row or consume ordinary execution attempts while waiting.
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-dependency-expired", scope)
	expiredLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("dependency expiry Claim() = %#v found=%t err=%v", expiredLease, found, err)
	}
	if err := restarted.Execute(t.Context(), expiredLease, func(context.Context) ([]byte, error) {
		return nil, rollbackCause
	}); !errors.Is(err, rollbackCause) {
		t.Fatalf("dependency expiry Execute() error = %v", err)
	}
	if err := restarted.Block(t.Context(), expiredLease, dependency, time.Second); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET AvailableAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6)),
    DependencyBlockedAt = TIMESTAMPADD(MINUTE, -31, UTC_TIMESTAMP(6))
WHERE Id = ?`, expiredLease.Row.RowID); err != nil {
		t.Fatal(err)
	}
	if expired, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("expired dependency Claim() = %#v found=%t err=%v", expired, found, err)
	}
	if err := db.QueryRow(`SELECT State, Attempt,
       (DependencyKind IS NOT NULL OR DependencyBlockedAt IS NOT NULL)
FROM _System_RuntimePlatformCommand WHERE Id = ?`, expiredLease.Row.RowID).
		Scan(&state, &blockedAttempt, &dependencyColumns); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandFailed || blockedAttempt != 1 || dependencyColumns != 0 {
		t.Fatalf("expired dependency state/attempt/dependency = %q/%d/%d", state, blockedAttempt, dependencyColumns)
	}

	// A process stop after Claim but before any Runtime effect leaves only a
	// lease. A restarted worker may safely reclaim it with a new attempt.
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-reclaim", scope)
	abandoned, found, err := restarted.Claim(t.Context())
	if err != nil || !found || abandoned.Row.CommandID != "command-reclaim" {
		t.Fatalf("abandoned Claim() = %#v found=%t err=%v", abandoned, found, err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6))
WHERE Id = ?`, abandoned.Row.RowID); err != nil {
		t.Fatal(err)
	}
	afterCrash, err := NewRuntimePlatformCommandExecutionStore(
		db, transactionOwner, owner, scope, time.Second, 2,
	)
	if err != nil {
		t.Fatal(err)
	}
	reclaimed, found, err := afterCrash.Claim(t.Context())
	if err != nil || !found || reclaimed.Row.RowID != abandoned.Row.RowID || reclaimed.Attempt != 2 {
		t.Fatalf("reclaim after restart = %#v found=%t err=%v", reclaimed, found, err)
	}
	if err := afterCrash.Fail(t.Context(), reclaimed, errors.New("test terminal after reclaim")); err != nil {
		t.Fatal(err)
	}

	// Repeated process loss without an executor disposition must also converge
	// to poison at the attempt bound. This exercises Claim's expired-lease
	// terminal transition rather than the explicit Retry path below.
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-crash-exhaustion", scope)
	firstCrash, found, err := restarted.Claim(t.Context())
	if err != nil || !found || firstCrash.Row.CommandID != "command-crash-exhaustion" ||
		firstCrash.Attempt != 1 {
		t.Fatalf("first crash Claim() = %#v found=%t err=%v", firstCrash, found, err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6))
WHERE Id = ?`, firstCrash.Row.RowID); err != nil {
		t.Fatal(err)
	}
	secondCrash, found, err := restarted.Claim(t.Context())
	if err != nil || !found || secondCrash.Row.RowID != firstCrash.Row.RowID ||
		secondCrash.Attempt != 2 {
		t.Fatalf("second crash Claim() = %#v found=%t err=%v", secondCrash, found, err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6))
WHERE Id = ?`, secondCrash.Row.RowID); err != nil {
		t.Fatal(err)
	}
	if exhausted, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("exhausted crash Claim() = %#v found=%t err=%v, want poison disposition", exhausted, found, err)
	}
	if err := db.QueryRow(`SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?`,
		secondCrash.Row.RowID,
	).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandPoisoned {
		t.Fatalf("crash-exhausted command state = %q, want poison", state)
	}

	acceptRuntimePlatformCommandForExecutionTest(t, repository, "command-rollback", scope)
	rollbackLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("rollback Claim() found=%t err=%v", found, err)
	}
	err = restarted.Execute(t.Context(), rollbackLease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("test transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandEffect (CommandId, Value) VALUES (?, ?)`,
			rollbackLease.Row.CommandID, "must-rollback"); err != nil {
			return nil, err
		}
		// The store owns the terminal transition. An invalid empty terminal
		// result must roll the effect back before commit.
		return nil, nil
	})
	if err == nil {
		t.Fatal("transaction without terminal command transition committed")
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM CommandEffect WHERE CommandId = ?`,
		rollbackLease.Row.CommandID,
	).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatal("business effect survived a missing terminal transition")
	}
	if err := restarted.Retry(t.Context(), rollbackLease, errors.New("retryable"), 0); err != nil {
		t.Fatal(err)
	}
	retryLease, found, err := restarted.Claim(t.Context())
	if err != nil || !found || retryLease.Attempt != 2 || retryLease.AttemptsRemaining != 0 {
		t.Fatalf("retry Claim() = %#v found=%t err=%v", retryLease, found, err)
	}
	if err := restarted.Retry(t.Context(), retryLease, errors.New("still retryable"), 0); err != nil {
		t.Fatal(err)
	}
	if err := db.QueryRow(`SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?`,
		retryLease.Row.RowID,
	).Scan(&state); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandPoisoned {
		t.Fatalf("exhausted command state = %q", state)
	}
	errorDigest := sha256.Sum256([]byte("still retryable"))
	if err := restarted.RepairPoison(t.Context(), RuntimePoisonRepair{
		RowID: retryLease.Row.RowID, ExpectedRetryCount: retryLease.Attempt,
		ExpectedLastErrorSHA256: hex.EncodeToString(errorDigest[:]),
		RepairID:                "repair-platform-command-one", Reason: "operator confirmed replay",
	}); err != nil {
		t.Fatal(err)
	}
	repaired, found, err := restarted.Claim(t.Context())
	if err != nil || !found || repaired.Attempt != 1 || repaired.AttemptsRemaining != 1 {
		t.Fatalf("repaired Claim() = %#v found=%t err=%v", repaired, found, err)
	}

	// Ownership transfer makes the old lease incapable of committing effects.
	nextOwner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:            "activate-platform-command-two",
		WorkFamily:             RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:           "platform-command-two",
		WorkerReleaseDigest:    "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		ExpectedOwnershipEpoch: owner.OwnershipEpoch,
	})
	if err != nil {
		t.Fatal(err)
	}
	err = restarted.Execute(t.Context(), repaired, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("test transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandEffect (CommandId, Value) VALUES (?, ?)`,
			repaired.Row.CommandID, "stale-owner"); err != nil {
			return nil, err
		}
		return []byte(`{"status":"stale"}`), nil
	})
	if err == nil || !errors.Is(err, ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("stale owner transaction err = %v", err)
	}
	if err := db.QueryRow(`SELECT COUNT(*) FROM CommandEffect WHERE CommandId = ?`,
		repaired.Row.CommandID,
	).Scan(&count); err != nil {
		t.Fatal(err)
	}
	if count != 0 {
		t.Fatal("stale owner effect committed")
	}
	if nextOwner.OwnershipEpoch != owner.OwnershipEpoch+1 {
		t.Fatalf("transferred epoch = %d", nextOwner.OwnershipEpoch)
	}
}

func TestRuntimePlatformCommandExecutionTiDBRunsIndependentPartitionsConcurrently(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	db.SetMaxOpenConns(8)
	db.SetMaxIdleConns(8)
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	createRuntimePlatformCommandOwnershipTable(t, db)

	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:         "activate-platform-command-parallel",
		WorkFamily:          RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "platform-command-parallel",
		WorkerReleaseDigest: "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := RuntimeWorkScope{
		OrgID: "org-parallel", AppInstallationID: "installation-parallel",
		RuntimeCapabilityID: "runtime.commands.accept.v1",
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	store, err := NewRuntimePlatformCommandExecutionStore(
		db,
		runtimePlatformCommandIntegrationTransactionOwner{transactions},
		owner,
		scope,
		30*time.Second,
		2,
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	repository := NewRuntimePlatformCommandRepository(db)
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "parallel-command-a", scope, "partition-a", "")
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "parallel-command-b", scope, "partition-b", "")

	leaseA, found, err := store.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("claim A found=%t err=%v", found, err)
	}
	leaseB, found, err := store.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("claim B found=%t err=%v", found, err)
	}
	if leaseA.Row.SerializationKeyDigest != "" || leaseB.Row.SerializationKeyDigest != "" {
		t.Fatalf("Integration-shaped claims unexpectedly have trusted aggregate keys: %q/%q", leaseA.Row.SerializationKeyDigest, leaseB.Row.SerializationKeyDigest)
	}
	if RuntimeWorkFenceSlot(runtimePlatformCommandEffectFenceKey(leaseA)) ==
		RuntimeWorkFenceSlot(runtimePlatformCommandEffectFenceKey(leaseB)) {
		t.Fatal("test command identities unexpectedly share an effect slot")
	}

	entered := make(chan string, 2)
	release := make(chan struct{})
	results := make(chan error, 2)
	run := func(lease RuntimePlatformCommandLease) {
		results <- store.Execute(t.Context(), lease, func(context.Context) ([]byte, error) {
			entered <- lease.Row.CommandID
			<-release
			return []byte(`{"status":"completed"}`), nil
		})
	}
	go run(leaseA)
	go run(leaseB)

	seen := map[string]bool{}
	timer := time.NewTimer(5 * time.Second)
	for len(seen) < 2 {
		select {
		case commandID := <-entered:
			seen[commandID] = true
		case <-timer.C:
			close(release)
			t.Fatalf("independent effects did not enter concurrently; entered=%v", seen)
		}
	}
	if !timer.Stop() {
		<-timer.C
	}
	close(release)
	for range 2 {
		if err := <-results; err != nil {
			t.Fatalf("parallel effect execution: %v", err)
		}
	}

	acceptRuntimePlatformCommandForExecutionTest(t, repository, "ordered-command-a", scope, "partition-shared", "")
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "ordered-command-b", scope, "partition-shared", "")
	orderedA, found, err := store.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("ordered claim A found=%t err=%v", found, err)
	}
	if orderedB, found, err := store.Claim(t.Context()); err != nil || found {
		t.Fatalf("same-partition concurrent claim = %#v found=%t err=%v", orderedB, found, err)
	}
	if err := store.Execute(t.Context(), orderedA, func(context.Context) ([]byte, error) {
		return []byte(`{"status":"completed"}`), nil
	}); err != nil {
		t.Fatal(err)
	}
	orderedB, found, err := store.Claim(t.Context())
	if err != nil || !found || orderedB.Row.PartitionKey != orderedA.Row.PartitionKey {
		t.Fatalf("same-partition successor claim = %#v found=%t err=%v", orderedB, found, err)
	}
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "ordered-command-c", scope, "partition-shared", "")
	dependency := runtimePlatformCommandDependencyForTest(t, "Contact", "contact-parallel")
	if err := store.Block(t.Context(), orderedB, dependency, time.Minute); err != nil {
		t.Fatalf("block same-partition owner: %v", err)
	}
	if orderedC, found, err := store.Claim(t.Context()); err != nil || found {
		t.Fatalf("blocked partition admitted successor = %#v found=%t err=%v", orderedC, found, err)
	}
}

func TestRuntimePlatformCommandClaimPlanScopesUnrelatedBacklogOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")

	unrelatedScope := strings.Repeat("e", 64)
	tx, err := db.BeginTx(t.Context(), nil)
	if err != nil {
		t.Fatal(err)
	}
	insert, err := tx.PrepareContext(t.Context(), `INSERT INTO _System_RuntimePlatformCommand
(Id, OrgId, AppInstallationId, CallerWorkloadId, RuntimeCapabilityId, MetadataBinding,
 ExecutionScopeDigest, ScopeDigest, CommandId, IdempotencyKey, CommandKind, PartitionKey,
 SerializationKeyDigest, PayloadFormatVersion, RequestDigest, Payload, Receipts, State,
 AvailableAt, CreatedDate, LastModifiedDate)
VALUES (?, 'other-org', 'other-installation', 'other-workload', 'runtime.commands.accept.v1',
        '{"binding":"other"}', ?, ?, ?, ?, 'runtime.test', ?, ?, 1,
        'sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb',
        '{"value":1}', '{"CatalogRevision":7,"SecurityEpoch":11}', 'ACCEPTED',
        UTC_TIMESTAMP(6), UTC_TIMESTAMP(6), UTC_TIMESTAMP(6))`)
	if err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	for index := 0; index < 20_000; index++ {
		id := fmt.Sprintf("unrelated-%05d", index)
		serializationKey := fmt.Sprintf("%064x", index+1)
		if _, err := insert.ExecContext(
			t.Context(), id, unrelatedScope, strings.Repeat("f", 64), id, id, "partition-"+id, serializationKey,
		); err != nil {
			_ = insert.Close()
			_ = tx.Rollback()
			t.Fatal(err)
		}
	}
	if err := insert.Close(); err != nil {
		_ = tx.Rollback()
		t.Fatal(err)
	}
	if err := tx.Commit(); err != nil {
		t.Fatal(err)
	}

	scope := RuntimeWorkScope{
		OrgID: "org-plan", AppInstallationID: "installation-plan",
		RuntimeCapabilityID: "runtime.commands.accept.v1",
	}
	repository := NewRuntimePlatformCommandRepository(db)
	acceptRuntimePlatformCommandForExecutionTest(
		t, repository, "plan-command", scope, "plan-partition", strings.Repeat("a", 64),
	)
	scopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	var now time.Time
	if err := db.QueryRowContext(t.Context(), `SELECT UTC_TIMESTAMP(6)`).Scan(&now); err != nil {
		t.Fatal(err)
	}
	plan := runtimePlatformCommandExplainAnalyzeText(
		t,
		db,
		runtimePlatformCommandClaimSQL,
		scopeDigest,
		scope.OrgID,
		scope.AppInstallationID,
		scope.RuntimeCapabilityID,
		now,
		now,
		now,
		now,
		now,
	)
	if strings.Contains(plan, "TableFullScan") {
		t.Fatalf("Platform-command claim scans unrelated backlog:\n%s", plan)
	}
	for _, indexName := range []string{
		"idx_runtime_platform_command_claim",
		"idx_runtime_platform_command_partition",
		"idx_runtime_platform_command_serialization",
	} {
		if !strings.Contains(plan, indexName) {
			t.Fatalf("Platform-command claim plan does not use %s:\n%s", indexName, plan)
		}
	}
}

func runtimePlatformCommandExplainAnalyzeText(
	t *testing.T,
	db *sql.DB,
	queryText string,
	args ...interface{},
) string {
	t.Helper()
	rows, err := db.QueryContext(t.Context(), "EXPLAIN ANALYZE "+queryText, args...)
	if err != nil {
		t.Fatal(err)
	}
	defer rows.Close()
	columns, err := rows.Columns()
	if err != nil {
		t.Fatal(err)
	}
	var plan strings.Builder
	for rows.Next() {
		values := make([]sql.RawBytes, len(columns))
		destinations := make([]interface{}, len(columns))
		for index := range values {
			destinations[index] = &values[index]
		}
		if err := rows.Scan(destinations...); err != nil {
			t.Fatal(err)
		}
		for _, value := range values {
			plan.Write(value)
			plan.WriteByte('\t')
		}
		plan.WriteByte('\n')
	}
	if err := rows.Err(); err != nil {
		t.Fatal(err)
	}
	return plan.String()
}

func TestRuntimePlatformCommandPublicationDurabilityOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	db.SetMaxOpenConns(4)
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	createRuntimePlatformCommandOwnershipTable(t, db)
	if _, err := db.Exec(`CREATE TABLE CommandPublicationEffect (
		CommandId VARCHAR(255) PRIMARY KEY,
		Value VARCHAR(255) NOT NULL
	)`); err != nil {
		t.Fatal(err)
	}
	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	owner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:         "activate-platform-publication-one",
		WorkFamily:          RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "platform-publication-one",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := RuntimeWorkScope{
		OrgID: "org-publication", AppInstallationID: "installation-publication",
		RuntimeCapabilityID: "runtime.commands.accept.v1",
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	transactionOwner := runtimePlatformCommandIntegrationTransactionOwner{transactions}
	newStore := func() *RuntimePlatformCommandExecutionStore {
		store, storeErr := NewRuntimePlatformCommandExecutionStore(
			db, transactionOwner, owner, scope, time.Second, 1,
			WithRuntimePlatformCommandPublicationAttempts(2),
		)
		if storeErr != nil {
			t.Fatal(storeErr)
		}
		return store
	}
	store := newStore()
	if err := store.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	repository := NewRuntimePlatformCommandRepository(db)
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	serializationKey := strings.Repeat("c", 64)
	accept := func(commandID, partitionKey, serializationKeyDigest string) RuntimePlatformCommandRow {
		row, duplicate, acceptErr := repository.Accept(t.Context(), RuntimePlatformCommandAcceptance{
			RowID: commandID + "-row", OrgID: scope.OrgID, AppInstallationID: scope.AppInstallationID,
			CallerWorkloadID: "publication-integration", RuntimeCapabilityID: scope.RuntimeCapabilityID,
			MetadataBinding: `{"binding":"exact"}`, ExecutionScopeDigest: executionScopeDigest,
			ScopeDigest: strings.Repeat("a", 64), CommandID: commandID,
			IdempotencyKey: commandID + "-idempotency", CommandKind: "runtime.test",
			PartitionKey: partitionKey, SerializationKeyDigest: serializationKeyDigest,
			PayloadFormatVersion: 1,
			RequestDigest:        "sha256:" + strings.Repeat("b", 64),
			Payload:              []byte(`{"value":1}`),
			Receipts:             []byte(`{"CatalogRevision":7,"SecurityEpoch":11}`),
		})
		if acceptErr != nil || duplicate {
			t.Fatalf("Accept(%s) duplicate=%t err=%v", commandID, duplicate, acceptErr)
		}
		return row
	}
	first := accept("publication-a", "caller-partition", serializationKey)
	second := accept("publication-b", "different-caller-partition", serializationKey)
	lease, found, err := store.Claim(t.Context())
	if err != nil || !found || lease.Row.RowID != first.RowID {
		t.Fatalf("first Claim() = %#v found=%t err=%v", lease, found, err)
	}
	pending := []byte(`{"post_image":"pending"}`)
	if err := store.PreparePublication(t.Context(), lease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("publication effect transaction is absent")
		}
		if _, err := tx.ExecContext(txCtx, `INSERT INTO CommandPublicationEffect (CommandId, Value) VALUES (?, ?)`,
			lease.Row.CommandID, "committed"); err != nil {
			return nil, err
		}
		return pending, nil
	}); err != nil {
		t.Fatal(err)
	}
	var state string
	var businessAttempt, publishAttempt int
	if err := db.QueryRow(`SELECT State, Attempt, PublishAttempt
	FROM _System_RuntimePlatformCommand WHERE Id = ?`, first.RowID).
		Scan(&state, &businessAttempt, &publishAttempt); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandPublishing || businessAttempt != 1 || publishAttempt != 0 {
		t.Fatalf("prepared state/attempts = %q/%d/%d", state, businessAttempt, publishAttempt)
	}
	restarted := newStore()
	if later, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("serialized successor Claim() = %#v found=%t err=%v", later, found, err)
	}
	publication, found, err := restarted.ClaimPublication(t.Context())
	if err != nil || !found || publication.Row.RowID != first.RowID || publication.Attempt != 1 {
		t.Fatalf("ClaimPublication() = %#v found=%t err=%v", publication, found, err)
	}
	// Simulate a process crash after the first publication claim. The restarted
	// worker resumes only publication; the committed effect and business attempt
	// remain untouched.
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6)) WHERE Id = ?`, first.RowID); err != nil {
		t.Fatal(err)
	}
	restarted = newStore()
	publication, found, err = restarted.ClaimPublication(t.Context())
	if err != nil || !found || publication.Attempt != 2 || publication.AttemptsRemaining != 0 {
		t.Fatalf("publication retry = %#v found=%t err=%v", publication, found, err)
	}
	finalFailure := errors.New("catalog unavailable after final attempt")
	if err := restarted.RetryPublication(t.Context(), publication, finalFailure, 0); err != nil {
		t.Fatal(err)
	}
	var recoveryState string
	if err := db.QueryRow(`SELECT State, Attempt, PublishAttempt, COALESCE(RecoveryState, '')
	FROM _System_RuntimePlatformCommand WHERE Id = ?`, first.RowID).
		Scan(&state, &businessAttempt, &publishAttempt, &recoveryState); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandPoisoned || recoveryState != RuntimePlatformCommandPublishing ||
		businessAttempt != 1 || publishAttempt != 2 {
		t.Fatalf("poison state/recovery/attempts = %q/%q/%d/%d", state, recoveryState, businessAttempt, publishAttempt)
	}
	if successor, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("publication-poisoned successor Claim() = %#v found=%t err=%v", successor, found, err)
	}
	var successorState string
	if err := db.QueryRow(`SELECT State FROM _System_RuntimePlatformCommand WHERE Id = ?`, second.RowID).
		Scan(&successorState); err != nil || successorState != RuntimePlatformCommandAccepted {
		t.Fatalf("serialized successor state = %q err=%v", successorState, err)
	}
	lastErrorDigest := sha256.Sum256([]byte(finalFailure.Error()))
	if err := restarted.RepairPoison(t.Context(), RuntimePoisonRepair{
		RowID: first.RowID, ExpectedRetryCount: 2,
		ExpectedLastErrorSHA256: hex.EncodeToString(lastErrorDigest[:]),
		RepairID:                "repair-publication-integration", Reason: "publisher authority restored",
	}); err != nil {
		t.Fatal(err)
	}
	publication, found, err = restarted.ClaimPublication(t.Context())
	if err != nil || !found || publication.Row.RowID != first.RowID || publication.Attempt != 1 {
		t.Fatalf("repaired ClaimPublication() = %#v found=%t err=%v", publication, found, err)
	}

	// A declared partition owns source ordering even when the operation has no
	// trusted aggregate key. It does not become aggregate identity authority.
	accept("public-partition-a", "same-public-key", "")
	accept("public-partition-b", "same-public-key", "")
	publicA, found, err := restarted.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("public partition first Claim() = %#v found=%t err=%v", publicA, found, err)
	}
	if publicB, found, err := restarted.Claim(t.Context()); err != nil || found {
		t.Fatalf("public partition concurrent Claim() = %#v found=%t err=%v", publicB, found, err)
	}
	if err := restarted.Execute(t.Context(), publicA, func(context.Context) ([]byte, error) {
		return []byte(`{"status":"completed"}`), nil
	}); err != nil {
		t.Fatalf("complete first public partition command: %v", err)
	}
	publicB, found, err := restarted.Claim(t.Context())
	if err != nil || !found || publicB.Row.PartitionKey != publicA.Row.PartitionKey {
		t.Fatalf("public partition successor Claim() = %#v found=%t err=%v", publicB, found, err)
	}

	// Readiness fails closed if a stopped upgrade discovers multiple active
	// owners for a trusted aggregate key. Partition ordering and aggregate
	// authority remain distinct, as proved above.
	duplicateKey := strings.Repeat("d", 64)
	duplicateA := accept("duplicate-owner-a", "", duplicateKey)
	duplicateB := accept("duplicate-owner-b", "", duplicateKey)
	for _, row := range []RuntimePlatformCommandRow{duplicateA, duplicateB} {
		identity, identityErr := runtimePlatformCommandPublicationIdentity(
			row.ScopeDigest, row.CommandID, row.RequestDigest,
		)
		if identityErr != nil {
			t.Fatal(identityErr)
		}
		if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
SET State = 'PUBLISHING', Attempt = 1, ResultPayload = ?, PublicationIdentity = ?
WHERE Id = ?`, encodeRuntimePlatformCommandResult([]byte(`{"pending":true}`)), identity, row.RowID); err != nil {
			t.Fatal(err)
		}
	}
	if err := restarted.CheckReady(t.Context()); err == nil ||
		!strings.Contains(err.Error(), "multiple active owners share a serialization key") {
		t.Fatalf("CheckReady() duplicate serialization owners error = %v", err)
	}
}

func TestRuntimePlatformCommandPublicationCompletionSurvivesOwnershipTransferOnTiDB(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	createRuntimePlatformCommandTestTable(t, db, "MEDIUMBLOB")
	createRuntimePlatformCommandOwnershipTable(t, db)
	if _, err := db.Exec(`CREATE TABLE CommandPublicationEffect (
		CommandId VARCHAR(255) PRIMARY KEY,
		Value VARCHAR(255) NOT NULL
	)`); err != nil {
		t.Fatal(err)
	}
	control, err := NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	firstOwner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:         "activate-publication-completion-one",
		WorkFamily:          RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:        "publication-completion-one",
		WorkerReleaseDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := RuntimeWorkScope{
		OrgID: "org-publication-completion", AppInstallationID: "installation-publication-completion",
		RuntimeCapabilityID: "runtime.commands.accept.v1",
	}
	transactions := NewTransactionManager(database.NewConnectionFromDB(db))
	transactionOwner := runtimePlatformCommandIntegrationTransactionOwner{transactions}
	newStore := func(owner RuntimeWorkOwnership) *RuntimePlatformCommandExecutionStore {
		store, storeErr := NewRuntimePlatformCommandExecutionStore(
			db, transactionOwner, owner, scope, time.Second, 1,
			WithRuntimePlatformCommandPublicationAttempts(3),
			WithRuntimePlatformCommandPublicationTerminalResultMaxBytes(
				runtimePlatformCommandPublicationPendingResultMaxBytes+1024,
			),
		)
		if storeErr != nil {
			t.Fatal(storeErr)
		}
		return store
	}
	firstStore := newStore(firstOwner)
	if err := firstStore.CheckReady(t.Context()); err != nil {
		t.Fatal(err)
	}
	repository := NewRuntimePlatformCommandRepository(db)
	acceptRuntimePlatformCommandForExecutionTest(t, repository, "publication-completion", scope)
	businessLease, found, err := firstStore.Claim(t.Context())
	if err != nil || !found {
		t.Fatalf("Claim() = %#v found=%t err=%v", businessLease, found, err)
	}
	pending := []byte(`{"schema":"pending","post_image":"committed-once"}`)
	if err := firstStore.PreparePublication(t.Context(), businessLease, func(txCtx context.Context) ([]byte, error) {
		tx, ok := transactions.ExtractTx(txCtx).(*sql.Tx)
		if !ok {
			return nil, errors.New("publication effect transaction is absent")
		}
		if _, insertErr := tx.ExecContext(txCtx,
			`INSERT INTO CommandPublicationEffect (CommandId, Value) VALUES (?, ?)`,
			businessLease.Row.CommandID, "committed-once",
		); insertErr != nil {
			return nil, insertErr
		}
		return pending, nil
	}); err != nil {
		t.Fatal(err)
	}
	firstPublication, found, err := firstStore.ClaimPublication(t.Context())
	if err != nil || !found || firstPublication.Attempt != 1 {
		t.Fatalf("first ClaimPublication() = %#v found=%t err=%v", firstPublication, found, err)
	}

	secondOwner, err := control.Activate(t.Context(), RuntimeWorkOwnershipActivation{
		OperationID:            "activate-publication-completion-two",
		WorkFamily:             RuntimePlatformCommandExecutionWorkFamily,
		DeploymentID:           "publication-completion-two",
		WorkerReleaseDigest:    "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		ExpectedOwnershipEpoch: firstOwner.OwnershipEpoch,
	})
	if err != nil {
		t.Fatal(err)
	}
	// A contract-valid pending envelope may consume the complete pending budget;
	// its terminal form adds immutable publication evidence. Exercise that
	// representation through the real TiDB completion path.
	terminal := bytes.Repeat([]byte("x"), runtimePlatformCommandPublicationPendingResultMaxBytes+1024)
	terminalReceipts := []byte(`{"CatalogRevision":8,"SecurityEpoch":12}`)
	if err := firstStore.CompletePublication(
		t.Context(), firstPublication, terminal, terminalReceipts,
	); !errors.Is(err, ErrRuntimeWorkOwnershipLost) {
		t.Fatalf("stale owner CompletePublication() = %v, want ownership lost", err)
	}
	if _, err := db.Exec(`UPDATE _System_RuntimePlatformCommand
		SET LeaseExpiresAt = TIMESTAMPADD(SECOND, -1, UTC_TIMESTAMP(6)) WHERE Id = ?`,
		businessLease.Row.RowID,
	); err != nil {
		t.Fatal(err)
	}

	secondStore := newStore(secondOwner)
	secondPublication, found, err := secondStore.ClaimPublication(t.Context())
	if err != nil || !found || secondPublication.Attempt != 2 ||
		secondPublication.Row.PublicationIdentity != firstPublication.Row.PublicationIdentity ||
		!bytes.Equal(secondPublication.Row.ResultPayload, pending) {
		t.Fatalf("transferred ClaimPublication() = %#v found=%t err=%v", secondPublication, found, err)
	}
	if err := secondStore.CompletePublication(
		t.Context(), secondPublication, terminal, terminalReceipts,
	); err != nil {
		t.Fatal(err)
	}

	var state string
	var businessAttempt, publishAttempt, effects int
	var storedResult sql.NullString
	var storedReceipts []byte
	if err := db.QueryRow(`SELECT State, Attempt, PublishAttempt, ResultPayload, Receipts,
		(SELECT COUNT(*) FROM CommandPublicationEffect WHERE CommandId = ?)
		FROM _System_RuntimePlatformCommand WHERE Id = ?`,
		businessLease.Row.CommandID, businessLease.Row.RowID,
	).Scan(&state, &businessAttempt, &publishAttempt, &storedResult, &storedReceipts, &effects); err != nil {
		t.Fatal(err)
	}
	decodedResult, err := decodeNullableRuntimePlatformCommandResult(storedResult)
	if err != nil {
		t.Fatal(err)
	}
	var compactReceipts bytes.Buffer
	if err := json.Compact(&compactReceipts, storedReceipts); err != nil {
		t.Fatal(err)
	}
	if state != RuntimePlatformCommandCompleted || businessAttempt != 1 || publishAttempt != 2 ||
		effects != 1 || !bytes.Equal(decodedResult, terminal) || !bytes.Equal(compactReceipts.Bytes(), terminalReceipts) {
		t.Fatalf("completed state/attempts/effects/result/receipts = %q/%d/%d/%d/%s/%s",
			state, businessAttempt, publishAttempt, effects, decodedResult, storedReceipts)
	}
	if lease, found, err := secondStore.Claim(t.Context()); err != nil || found {
		t.Fatalf("completed command was reclaimed for business execution: %#v found=%t err=%v", lease, found, err)
	}
}

func runtimePlatformCommandDependencyForTest(
	t *testing.T,
	objectAPIName string,
	recordID string,
) contractmodels.FlowExecutionDependency {
	t.Helper()
	request, err := contractmodels.NewFlowIdentityConvergenceDependencyRequest(
		objectAPIName, recordID, "owner-1", "destination-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	dependency, err := contractmodels.BindFlowIdentityConvergenceDependency(
		request, "connector-1", "contacts", "remote-1",
	)
	if err != nil {
		t.Fatal(err)
	}
	return dependency
}

func acceptRuntimePlatformCommandForExecutionTest(
	t *testing.T,
	repository *RuntimePlatformCommandRepository,
	commandID string,
	scope RuntimeWorkScope,
	partitionKeys ...string,
) {
	t.Helper()
	partitionKey := ""
	serializationKeyDigest := ""
	if len(partitionKeys) > 0 {
		partitionKey = partitionKeys[0]
	}
	if len(partitionKeys) > 1 {
		serializationKeyDigest = partitionKeys[1]
	}
	executionScopeDigest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	_, duplicate, err := repository.Accept(t.Context(), RuntimePlatformCommandAcceptance{
		RowID: commandID + "-row", OrgID: scope.OrgID, AppInstallationID: scope.AppInstallationID,
		CallerWorkloadID: "integration-1", RuntimeCapabilityID: scope.RuntimeCapabilityID,
		MetadataBinding:      `{"binding":"exact"}`,
		ExecutionScopeDigest: executionScopeDigest,
		ScopeDigest:          "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		CommandID:            commandID, IdempotencyKey: commandID + "-idempotency",
		CommandKind: "runtime.test", PartitionKey: partitionKey,
		SerializationKeyDigest: serializationKeyDigest, PayloadFormatVersion: 1,
		RequestDigest: "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		Payload:       []byte(`{"value":1}`), Receipts: []byte(`{"CatalogRevision":7,"SecurityEpoch":11}`),
	})
	if err != nil || duplicate {
		t.Fatalf("Accept(%s) duplicate=%t err=%v", commandID, duplicate, err)
	}
}

func createRuntimePlatformCommandOwnershipTable(t *testing.T, db *sql.DB) {
	t.Helper()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatalf("load Runtime Worker system table catalog: %v", err)
	}
	wanted := map[string]bool{
		RuntimeWorkOwnershipTable: false,
		RuntimeWorkFenceTable:     false,
	}
	repository := NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		if err := repository.CreatePhysicalTable(t.Context(), definition); err != nil {
			t.Fatalf("create %s from system table catalog: %v", definition.TableName, err)
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("system table catalog does not declare %s", table)
		}
	}
}