package runtimeworker

import (
	"bytes"
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	contractmodels "github.com/pingcap-inc/nexuscrm-contract/pkg/models"
	applicationresultsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
)

type platformCommandStoreProbe struct {
	lease                  persistence.RuntimePlatformCommandLease
	found                  bool
	hasWork                bool
	hintErr                error
	hintCalls              int
	claimCalls             int
	completed              int
	applicationCompleted   int
	publicationPrepared    int
	applicationCompletion  persistence.RuntimePlatformCommandApplicationCompletion
	retried                int
	replayed               int
	failed                 int
	poisoned               int
	blocked                int
	dependency             *contractmodels.FlowExecutionDependency
	completeErr            error
	commitErr              error
	retryErr               error
	retryCause             error
	failureResult          []byte
	publicationLease       persistence.RuntimePlatformCommandPublicationLease
	publicationFound       bool
	publicationHint        bool
	publicationClaimed     int
	publicationCompleted   int
	publicationRetried     int
	publicationResult      []byte
	publicationReceipts    []byte
	publicationCompleteErr error
}

func (s *platformCommandStoreProbe) Block(
	_ context.Context,
	_ persistence.RuntimePlatformCommandLease,
	dependency contractmodels.FlowExecutionDependency,
	_ time.Duration,
) error {
	s.blocked++
	s.dependency = &dependency
	return nil
}

func (*platformCommandStoreProbe) CheckReady(context.Context) error { return nil }
func (s *platformCommandStoreProbe) HasWork(context.Context) (bool, error) {
	s.hintCalls++
	return s.hasWork, s.hintErr
}
func (s *platformCommandStoreProbe) HasPublicationWork(context.Context) (bool, error) {
	return s.publicationHint, nil
}
func (s *platformCommandStoreProbe) ClaimPublication(context.Context) (persistence.RuntimePlatformCommandPublicationLease, bool, error) {
	s.publicationClaimed++
	return s.publicationLease, s.publicationFound, nil
}
func (s *platformCommandStoreProbe) CompletePublication(
	_ context.Context,
	_ persistence.RuntimePlatformCommandPublicationLease,
	result []byte,
	receipts []byte,
) error {
	s.publicationCompleted++
	s.publicationResult = append([]byte(nil), result...)
	s.publicationReceipts = append([]byte(nil), receipts...)
	return s.publicationCompleteErr
}
func (s *platformCommandStoreProbe) RetryPublication(
	_ context.Context,
	_ persistence.RuntimePlatformCommandPublicationLease,
	cause error,
	_ time.Duration,
) error {
	s.publicationRetried++
	s.retryCause = cause
	return s.retryErr
}
func (s *platformCommandStoreProbe) Claim(context.Context) (persistence.RuntimePlatformCommandLease, bool, error) {
	s.claimCalls++
	return s.lease, s.found, nil
}
func (s *platformCommandStoreProbe) Execute(
	ctx context.Context,
	_ persistence.RuntimePlatformCommandLease,
	effect func(context.Context) ([]byte, error),
) error {
	if _, err := effect(ctx); err != nil {
		return err
	}
	s.completed++
	if s.completeErr != nil {
		return s.completeErr
	}
	return s.commitErr
}
func (s *platformCommandStoreProbe) ExecuteApplication(
	ctx context.Context,
	_ persistence.RuntimePlatformCommandLease,
	effect func(context.Context) (persistence.RuntimePlatformCommandApplicationCompletion, error),
) error {
	completion, err := effect(ctx)
	if err != nil {
		return err
	}
	s.applicationCompleted++
	s.applicationCompletion = completion
	if s.completeErr != nil {
		return s.completeErr
	}
	return s.commitErr
}
func (s *platformCommandStoreProbe) PreparePublication(
	ctx context.Context,
	_ persistence.RuntimePlatformCommandLease,
	effect func(context.Context) ([]byte, error),
) error {
	if _, err := effect(ctx); err != nil {
		return err
	}
	s.publicationPrepared++
	if s.completeErr != nil {
		return s.completeErr
	}
	return s.commitErr
}
func (s *platformCommandStoreProbe) Retry(_ context.Context, _ persistence.RuntimePlatformCommandLease, cause error, _ time.Duration) error {
	s.retried++
	s.retryCause = cause
	return s.retryErr
}
func (s *platformCommandStoreProbe) Replay(_ context.Context, _ persistence.RuntimePlatformCommandLease, cause error) error {
	s.replayed++
	s.retryCause = cause
	return s.retryErr
}
func (s *platformCommandStoreProbe) Fail(context.Context, persistence.RuntimePlatformCommandLease, error) error {
	s.failed++
	return nil
}
func (s *platformCommandStoreProbe) FailWithResult(_ context.Context, _ persistence.RuntimePlatformCommandLease, encoded []byte) error {
	s.failed++
	s.failureResult = append([]byte(nil), encoded...)
	return nil
}
func (s *platformCommandStoreProbe) Poison(context.Context, persistence.RuntimePlatformCommandLease, error) error {
	s.poisoned++
	return nil
}

type platformCommandObservationSourceProbe struct {
	state persistence.RuntimePlatformCommandExecutionState
	err   error
}

type platformCommandPublicationOwnerProbe struct {
	completion platformCommandPublicationCompletion
	err        error
	calls      int
}

func (p *platformCommandPublicationOwnerProbe) Publish(
	context.Context,
	persistence.RuntimePlatformCommandPublicationLease,
) (platformCommandPublicationCompletion, error) {
	p.calls++
	return p.completion, p.err
}

func (s *platformCommandObservationSourceProbe) Observe(context.Context) (persistence.RuntimePlatformCommandExecutionState, error) {
	return s.state, s.err
}

func platformCommandTestObservationFactory() platformCommandObservationFactory {
	return func() (platformCommandObservationSource, func() error, error) {
		return &platformCommandObservationSourceProbe{}, func() error { return nil }, nil
	}
}

type platformCommandBinderProbe struct {
	view    *catalogsnapshot.RuntimeMetadataView
	bindErr error
	err     error
}

func (b platformCommandBinderProbe) OpenCommand(
	context.Context,
	string,
	string,
	uint32,
) (platformCommandExecutionAuthority, error) {
	if b.err != nil {
		return nil, b.err
	}
	return platformCommandAuthorityProbe{view: b.view, bindErr: b.bindErr}, nil
}

type platformCommandAuthorityProbe struct {
	view    *catalogsnapshot.RuntimeMetadataView
	bindErr error
}

func (a platformCommandAuthorityProbe) AllowsCommand(kind string, version uint32) bool {
	return a.view != nil && a.view.AllowsCommand(kind, version)
}

func (a platformCommandAuthorityProbe) CommandExecutionContext(
	kind string,
	version uint32,
	authorizedWorkloadID string,
	definitionRef string,
	definitionVersion string,
) (string, bool) {
	if a.view == nil {
		return "", false
	}
	return a.view.CommandExecutionContext(
		kind,
		version,
		authorizedWorkloadID,
		definitionRef,
		definitionVersion,
	)
}

func (a platformCommandAuthorityProbe) Bind(ctx context.Context) (context.Context, error) {
	if a.bindErr != nil {
		return nil, a.bindErr
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: a.view.OrgID(), AppInstallationID: a.view.AppInstallationID(),
		RuntimeCapabilityID: a.view.CapabilityID(),
	}
	return commandExecutionAuthority{
		authorizationView: a.view,
		executionView:     a.view,
		scope:             scope,
	}.Bind(ctx)
}

type platformCommandExecutorProbe struct {
	err       error
	calls     int
	boundView bool
}

type platformCommandApplicationExecutorProbe struct {
	completion persistence.RuntimePlatformCommandApplicationCompletion
	err        error
	calls      int
	boundView  bool
}

type platformCommandPublicationExecutorProbe struct {
	result    []byte
	err       error
	calls     int
	boundView bool
}

func (e *platformCommandPublicationExecutorProbe) ExecutePlatformPublicationCommand(
	ctx context.Context,
	_ PlatformCommandExecution,
) ([]byte, error) {
	e.calls++
	_, e.boundView = metadata.RuntimeMetadataViewFromContext(ctx)
	return append([]byte(nil), e.result...), e.err
}

func (e *platformCommandApplicationExecutorProbe) ExecutePlatformApplicationCommand(
	ctx context.Context,
	_ PlatformCommandExecution,
) (persistence.RuntimePlatformCommandApplicationCompletion, error) {
	e.calls++
	_, e.boundView = metadata.RuntimeMetadataViewFromContext(ctx)
	return e.completion, e.err
}

type platformCommandRecoveryProbe struct {
	calls     int
	boundView bool
	request   automation.ConnectorInboundRemoteIdentityRecoveryRequest
	err       error
	cancel    context.CancelFunc
}

func (p *platformCommandRecoveryProbe) Recover(
	ctx context.Context,
	request automation.ConnectorInboundRemoteIdentityRecoveryRequest,
) (*automation.ConnectorInboundRemoteIdentityRecoveryResult, error) {
	p.calls++
	p.request = request
	_, p.boundView = metadata.RuntimeMetadataViewFromContext(ctx)
	if p.cancel != nil {
		p.cancel()
	}
	if p.err != nil {
		return nil, p.err
	}
	return &automation.ConnectorInboundRemoteIdentityRecoveryResult{
		EvidenceID: "evidence-1", LinkKey: "link-1", SourceObject: "GenericSource",
		SourceRecordID: "source-1", RemoteIDField: "RemoteKey",
	}, nil
}

func (e *platformCommandExecutorProbe) ExecutePlatformCommand(
	ctx context.Context,
	_ PlatformCommandExecution,
) ([]byte, error) {
	e.calls++
	_, e.boundView = metadata.RuntimeMetadataViewFromContext(ctx)
	if e.err != nil {
		return nil, e.err
	}
	return []byte(`{"status":"completed"}`), nil
}

func TestPlatformCommandProcessorOwnsTerminalAndRetryClassification(t *testing.T) {
	view := commandExecutionTestView(t)
	for _, test := range []struct {
		name        string
		executeErr  error
		completeErr error
		commitErr   error
		wantDone    int
		wantRetry   int
		wantFailed  int
		wantErr     bool
	}{
		{name: "success", wantDone: 1},
		{name: "retryable executor", executeErr: errors.New("retryable"), wantRetry: 1},
		{
			name: "permanent executor",
			executeErr: &PlatformCommandExecutionError{
				Cause: errors.New("invalid payload"), Permanent: true,
			},
			wantFailed: 1,
		},
		{
			name:      "unknown commit",
			commitErr: errors.New("commit outcome unknown"),
			wantDone:  1, wantErr: true,
		},
		{
			name:        "terminal transition failure",
			completeErr: persistence.ErrRuntimePlatformCommandLeaseLost,
			wantDone:    1, wantErr: true,
		},
		{
			name:        "deterministic invalid terminal result",
			completeErr: persistence.ErrRuntimePlatformCommandResultInvalid,
			wantDone:    1, wantFailed: 1,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := &platformCommandStoreProbe{
				found:       true,
				completeErr: test.completeErr,
				commitErr:   test.commitErr,
				lease: persistence.RuntimePlatformCommandLease{
					Row: persistence.RuntimePlatformCommandRow{
						CommandID: "command-1", CommandKind: "runtime.test",
						PayloadFormatVersion: 1, Payload: []byte(`{"value":1}`),
					},
					ImmutableBinding: "binding", CallerWorkloadID: "integration-1",
				},
			}
			executor := &platformCommandExecutorProbe{err: test.executeErr}
			registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
				"runtime.test": {1: executor},
			}, nil)
			if err != nil {
				t.Fatal(err)
			}
			processor, err := newPlatformCommandProcessor(
				store,
				platformCommandTestObservationFactory(),
				platformCommandBinderProbe{view: view},
				registry,
				0,
			)
			if err != nil {
				t.Fatal(err)
			}
			processed, found, err := processor.ProcessOne(t.Context(), 1)
			if found != true || (err != nil) != test.wantErr {
				t.Fatalf("ProcessOne() processed=%d found=%t err=%v", processed, found, err)
			}
			if store.completed != test.wantDone || store.retried != test.wantRetry ||
				store.failed != test.wantFailed || !executor.boundView {
				t.Fatalf("dispositions complete/retry/fail = %d/%d/%d",
					store.completed, store.retried, store.failed)
			}
		})
	}
}

func TestPlatformCommandProcessorCommitsPublicationPendingResultWithoutTerminalSuccess(t *testing.T) {
	view := commandExecutionTestView(t)
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 1, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-1",
		},
	}
	executor := &platformCommandPublicationExecutorProbe{result: []byte(`{"pending":true}`)}
	registry, err := newPlatformCommandExecutorRegistry(nil, nil, map[string]map[uint32]platformCommandPublicationExecutor{
		"runtime.test": {1: executor},
	})
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: view}, registry, 0,
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 1 {
		t.Fatalf("ProcessOne() processed=%d found=%t err=%v", processed, found, err)
	}
	if store.publicationPrepared != 1 || store.completed != 0 || store.failed != 0 ||
		executor.calls != 1 || !executor.boundView {
		t.Fatalf("publication prepare/legacy complete/fail/executor = %d/%d/%d/%d bound=%t",
			store.publicationPrepared, store.completed, store.failed, executor.calls, executor.boundView)
	}
}

func TestPlatformCommandProcessorClaimsPublicationBeforeBusinessAndCompletesExactResult(t *testing.T) {
	store := &platformCommandStoreProbe{
		found: true, publicationFound: true, publicationHint: true,
		publicationLease: persistence.RuntimePlatformCommandPublicationLease{
			Row: persistence.RuntimePlatformCommandRow{State: persistence.RuntimePlatformCommandPublishing},
		},
	}
	owner := &platformCommandPublicationOwnerProbe{completion: platformCommandPublicationCompletion{
		ResultPayload: []byte(`{"terminal":true}`), Receipts: []byte(`{"CatalogRevision":8}`),
	}}
	registry, err := newPlatformCommandExecutorRegistry(
		map[string]map[uint32]PlatformCommandExecutor{"runtime.test": {1: &platformCommandExecutorProbe{}}}, nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{}, registry, 0,
		withPlatformCommandPublicationOwner(owner),
	)
	if err != nil {
		t.Fatal(err)
	}
	if pending, err := processor.HasWork(t.Context()); err != nil || !pending {
		t.Fatalf("HasWork() = %t, %v", pending, err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 1 || owner.calls != 1 ||
		store.publicationClaimed != 1 || store.publicationCompleted != 1 ||
		store.publicationRetried != 0 || store.claimCalls != 0 ||
		string(store.publicationResult) != `{"terminal":true}` {
		t.Fatalf("publication processed/found/error=%d/%t/%v owner=%d claim=%d complete=%d retry=%d business=%d result=%s",
			processed, found, err, owner.calls, store.publicationClaimed, store.publicationCompleted,
			store.publicationRetried, store.claimCalls, store.publicationResult)
	}
}

func TestPlatformCommandProcessorPublicationFailureNeverReexecutesBusinessMutation(t *testing.T) {
	for _, test := range []struct {
		name       string
		publishErr error
		commitErr  error
		wantRetry  int
		wantErr    bool
	}{
		{name: "publisher retry", publishErr: errors.New("Catalog unavailable"), wantRetry: 1},
		{name: "terminal commit uncertain", commitErr: persistence.ErrRuntimePlatformCommandOutcomeUncertain, wantErr: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := &platformCommandStoreProbe{
				found: true, publicationFound: true,
				publicationLease: persistence.RuntimePlatformCommandPublicationLease{
					Row: persistence.RuntimePlatformCommandRow{State: persistence.RuntimePlatformCommandPublishing},
				},
				publicationCompleteErr: test.commitErr,
			}
			owner := &platformCommandPublicationOwnerProbe{
				completion: platformCommandPublicationCompletion{
					ResultPayload: []byte(`{"terminal":true}`), Receipts: []byte(`{}`),
				},
				err: test.publishErr,
			}
			registry, err := newPlatformCommandExecutorRegistry(
				map[string]map[uint32]PlatformCommandExecutor{"runtime.test": {1: &platformCommandExecutorProbe{}}}, nil,
			)
			if err != nil {
				t.Fatal(err)
			}
			processor, err := newPlatformCommandProcessor(
				store, platformCommandTestObservationFactory(), platformCommandBinderProbe{}, registry, 0,
				withPlatformCommandPublicationOwner(owner),
			)
			if err != nil {
				t.Fatal(err)
			}
			_, found, processErr := processor.ProcessOne(t.Context(), 1)
			if !found || (processErr != nil) != test.wantErr ||
				store.publicationRetried != test.wantRetry || store.claimCalls != 0 || store.publicationPrepared != 0 {
				t.Fatalf("ProcessOne() found=%t err=%v retry=%d businessClaim=%d mutation=%d",
					found, processErr, store.publicationRetried, store.claimCalls, store.publicationPrepared)
			}
		})
	}
}

func TestPlatformCommandProcessorWritesTypedFailureOnlyAfterPublicationEffectRollback(t *testing.T) {
	view := commandExecutionTestView(t)
	encoded := []byte(`{"schema":"typed-business-failure"}`)
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 1, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-1",
		},
	}
	executor := &platformCommandPublicationExecutorProbe{err: &PlatformCommandBusinessFailure{
		Cause: errors.New("stale Layout version"), Encoded: encoded,
	}}
	registry, err := newPlatformCommandExecutorRegistry(nil, nil, map[string]map[uint32]platformCommandPublicationExecutor{
		"runtime.test": {1: executor},
	})
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: view}, registry, 0,
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 1 {
		t.Fatalf("ProcessOne() processed=%d found=%t err=%v", processed, found, err)
	}
	if store.publicationPrepared != 0 || store.failed != 1 || !bytes.Equal(store.failureResult, encoded) ||
		store.retried != 0 || store.poisoned != 0 {
		t.Fatalf("publication/fail/retry/poison = %d/%d/%d/%d result=%q",
			store.publicationPrepared, store.failed, store.retried, store.poisoned, store.failureResult)
	}
}

func TestPlatformCommandProcessorRoutesV3OnlyThroughApplicationResultOwner(t *testing.T) {
	view := commandExecutionTestView(t)
	for _, test := range []struct {
		name          string
		completion    persistence.RuntimePlatformCommandApplicationCompletion
		executeErr    error
		completeErr   error
		commitErr     error
		wantProcessed int
		wantApplied   int
		wantRetry     int
		wantPoison    int
		wantErr       bool
	}{
		{
			name: "applied terminal",
			completion: persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome: applicationresultsemantic.OutcomeApplied,
				Body:    []byte(`{"executionId":"flow-1"}`),
			},
			wantProcessed: 1, wantApplied: 1,
		},
		{
			name: "rejected terminal",
			completion: persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome:     applicationresultsemantic.OutcomeRejected,
				FailureCode: applicationresultsemantic.FailureApplicationRejected,
				Body:        []byte(`{"category":"validation","code":"VALIDATION"}`),
			},
			wantProcessed: 1, wantApplied: 1,
		},
		{
			name: "failed terminal",
			completion: persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome:     applicationresultsemantic.OutcomeFailed,
				FailureCode: applicationresultsemantic.FailureRuntimeExecutionFailed,
				Body:        []byte(`{"category":"internal","code":"INTERNAL_ERROR"}`),
			},
			wantProcessed: 1, wantApplied: 1,
		},
		{name: "ordinary retry", executeErr: errors.New("temporarily unavailable"), wantRetry: 1},
		{
			name: "permanent infrastructure error is poison not failed",
			executeErr: &PlatformCommandExecutionError{
				Cause: errors.New("unsupported policy"), Permanent: true,
			},
			wantProcessed: 1, wantPoison: 1,
		},
		{
			name: "invalid typed completion is poison not failed",
			completion: persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome: applicationresultsemantic.OutcomeApplied,
				Body:    []byte(`{"executionId":"flow-1"}`),
			},
			completeErr:   persistence.ErrRuntimePlatformCommandResultInvalid,
			wantProcessed: 1, wantApplied: 1, wantPoison: 1,
		},
		{
			name: "uncertain commit has no second disposition",
			completion: persistence.RuntimePlatformCommandApplicationCompletion{
				Outcome: applicationresultsemantic.OutcomeApplied,
				Body:    []byte(`{"executionId":"flow-1"}`),
			},
			commitErr:   errors.New("commit outcome unknown"),
			wantApplied: 1, wantErr: true,
		},
		{
			name:          "unexpected dependency is poison not blocked",
			executeErr:    &PlatformCommandDependencyError{},
			wantProcessed: 1, wantPoison: 1,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := &platformCommandStoreProbe{
				found: true, completeErr: test.completeErr, commitErr: test.commitErr,
				lease: persistence.RuntimePlatformCommandLease{
					Row: persistence.RuntimePlatformCommandRow{
						CommandID: "command-v3", CommandKind: RuntimeFlowCommandKind,
						PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV3,
						Payload:              []byte(`{"value":1}`),
					},
					ImmutableBinding: "binding", CallerWorkloadID: "integration-a",
				},
			}
			executor := &platformCommandApplicationExecutorProbe{
				completion: test.completion,
				err:        test.executeErr,
			}
			registry, err := newPlatformCommandExecutorRegistry(
				nil,
				map[string]map[uint32]platformCommandApplicationExecutor{
					RuntimeFlowCommandKind: {
						RuntimeFlowCommandPayloadFormatVersionV3: executor,
					},
				},
			)
			if err != nil {
				t.Fatal(err)
			}
			processor, err := newPlatformCommandProcessor(
				store,
				platformCommandTestObservationFactory(),
				platformCommandBinderProbe{view: view},
				registry,
				0,
			)
			if err != nil {
				t.Fatal(err)
			}
			processed, found, err := processor.ProcessOne(t.Context(), 1)
			if !found || processed != test.wantProcessed || (err != nil) != test.wantErr ||
				store.applicationCompleted != test.wantApplied || store.completed != 0 ||
				store.retried != test.wantRetry || store.poisoned != test.wantPoison ||
				store.failed != 0 || store.blocked != 0 || executor.calls != 1 ||
				!executor.boundView {
				t.Fatalf(
					"ProcessOne() processed=%d found=%t err=%v application/legacy/retry/poison/fail/block=%d/%d/%d/%d/%d/%d executor=%#v",
					processed, found, err, store.applicationCompleted, store.completed,
					store.retried, store.poisoned, store.failed, store.blocked, executor,
				)
			}
		})
	}
}

func TestPlatformCommandRegistryRejectsLegacyV3Executor(t *testing.T) {
	registry, err := newPlatformCommandExecutorRegistry(
		map[string]map[uint32]PlatformCommandExecutor{
			RuntimeFlowCommandKind: {
				RuntimeFlowCommandPayloadFormatVersionV3: &platformCommandExecutorProbe{},
			},
		},
		nil,
	)
	if err == nil || registry != nil {
		t.Fatalf("newPlatformCommandExecutorRegistry() = %#v, %v, want V3 legacy rejection", registry, err)
	}
}

func TestPlatformCommandProcessorKeepsV3RecoveryReplayNonterminal(t *testing.T) {
	request := automation.ConnectorInboundRemoteIdentityRecoveryRequest{
		Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
			ConnectorOwnerKey: "owner-1", ConnectorNamespace: "connector-1",
			DestinationSlotID: "destination-1", RemoteID: "remote-1",
		},
	}
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-v3-recovery", CommandKind: RuntimeFlowCommandKind,
				PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV3,
				Payload:              []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-a",
			AttemptsRemaining: 1,
		},
	}
	executor := &platformCommandApplicationExecutorProbe{
		err: &PlatformCommandRecoveryError{Request: request},
	}
	registry, err := newPlatformCommandExecutorRegistry(
		nil,
		map[string]map[uint32]platformCommandApplicationExecutor{
			RuntimeFlowCommandKind: {
				RuntimeFlowCommandPayloadFormatVersionV3: executor,
			},
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	recovery := &platformCommandRecoveryProbe{}
	processor, err := newPlatformCommandProcessor(
		store,
		platformCommandTestObservationFactory(),
		platformCommandBinderProbe{view: commandExecutionTestView(t)},
		registry,
		0,
		withPlatformCommandRecoveryOwner(recovery),
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 0 || store.applicationCompleted != 0 ||
		store.replayed != 1 || store.retried != 0 || store.failed != 0 ||
		store.poisoned != 0 || store.blocked != 0 || recovery.calls != 1 ||
		!recovery.boundView || recovery.request.Selector != request.Selector {
		t.Fatalf(
			"processed=%d found=%t err=%v application/replay/retry/fail/poison/block=%d/%d/%d/%d/%d/%d recovery=%#v",
			processed, found, err, store.applicationCompleted, store.replayed,
			store.retried, store.failed, store.poisoned, store.blocked, recovery,
		)
	}
}

func TestPlatformCommandProcessorPersistsDependencyOnlyAfterExecutorRollback(t *testing.T) {
	request, err := contractmodels.NewFlowIdentityConvergenceDependencyRequest(
		"Lead", "lead-1", "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)
	}
	view := commandExecutionTestView(t)
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 2, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-1",
		},
	}
	executor := &platformCommandExecutorProbe{
		err: &PlatformCommandDependencyError{Dependency: dependency},
	}
	registry, err := newPlatformCommandExecutorRegistry(
		map[string]map[uint32]PlatformCommandExecutor{
			"runtime.test": {2: executor},
		},
		nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: view}, registry, 0,
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 0 || store.blocked != 1 ||
		store.retried != 0 || store.failed != 0 || store.poisoned != 0 ||
		store.dependency == nil || store.dependency.KeyHash != dependency.KeyHash {
		t.Fatalf(
			"ProcessOne dependency processed=%d found=%t err=%v block/retry/fail/poison=%d/%d/%d/%d dependency=%#v",
			processed, found, err, store.blocked, store.retried, store.failed,
			store.poisoned, store.dependency,
		)
	}
}

func TestPlatformCommandProcessorDoesNotCommitRecoveryWithoutReplayAttempt(t *testing.T) {
	view := commandExecutionTestView(t)
	request := automation.ConnectorInboundRemoteIdentityRecoveryRequest{
		Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
			ConnectorOwnerKey: "owner-1", ConnectorNamespace: "connector-1",
			DestinationSlotID: "destination-1", RemoteID: "remote-1",
		},
	}
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 2, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-1",
			AttemptsRemaining: 0,
		},
	}
	executor := &platformCommandExecutorProbe{err: &PlatformCommandRecoveryError{Request: request}}
	registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
		"runtime.test": {2: executor},
	}, nil)
	if err != nil {
		t.Fatal(err)
	}
	recovery := &platformCommandRecoveryProbe{}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: view}, registry, 0,
		withPlatformCommandRecoveryOwner(recovery),
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 0 || store.retried != 1 ||
		recovery.calls != 0 || store.retryCause == nil ||
		!strings.Contains(store.retryCause.Error(), "requires one remaining replay attempt") {
		t.Fatalf("processed=%d found=%t err=%v retried=%d recovery=%d cause=%v",
			processed, found, err, store.retried, recovery.calls, store.retryCause)
	}
}

func TestPlatformCommandProcessorRecoversOnlyAfterEffectRollbackThenReplays(t *testing.T) {
	view := commandExecutionTestView(t)
	request := automation.ConnectorInboundRemoteIdentityRecoveryRequest{
		Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
			ConnectorOwnerKey: "owner-1", ConnectorNamespace: "connector-1",
			DestinationSlotID: "destination-1", RemoteID: "remote-1",
		},
	}
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 2, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding", CallerWorkloadID: "integration-1", AttemptsRemaining: 1,
		},
	}
	executor := &platformCommandExecutorProbe{err: &PlatformCommandRecoveryError{Request: request}}
	registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
		"runtime.test": {2: executor},
	}, nil)
	if err != nil {
		t.Fatal(err)
	}
	recovery := &platformCommandRecoveryProbe{}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: view}, registry, 0,
		withPlatformCommandRecoveryOwner(recovery),
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(t.Context(), 1)
	if err != nil || !found || processed != 0 || store.completed != 0 ||
		store.replayed != 1 || store.retried != 0 || store.failed != 0 || recovery.calls != 1 ||
		!recovery.boundView || recovery.request.Selector != request.Selector {
		t.Fatalf("processed=%d found=%t err=%v complete/replay/retry/fail=%d/%d/%d/%d recovery=%#v",
			processed, found, err, store.completed, store.replayed, store.retried, store.failed, recovery)
	}
}

func TestPlatformCommandRecoveryLeavesCancellationAndLostDispositionForDurableReclaim(t *testing.T) {
	request := automation.ConnectorInboundRemoteIdentityRecoveryRequest{Selector: automation.ConnectorInboundRemoteIdentityRecoverySelector{
		ConnectorOwnerKey: "owner-1", ConnectorNamespace: "connector-1",
		DestinationSlotID: "destination-1", RemoteID: "remote-1",
	}}
	for _, tc := range []struct {
		name         string
		cancel       bool
		retryErr     error
		wantErr      error
		wantReplayed int
	}{
		{name: "owner cancellation leaves lease for reclaim", cancel: true, wantErr: context.Canceled},
		{name: "recovery commit before lost replay disposition remains reclaimable", retryErr: persistence.ErrRuntimePlatformCommandLeaseLost, wantErr: persistence.ErrRuntimePlatformCommandLeaseLost, wantReplayed: 1},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ctx, cancel := context.WithCancel(t.Context())
			defer cancel()
			store := &platformCommandStoreProbe{
				found: true, retryErr: tc.retryErr,
				lease: persistence.RuntimePlatformCommandLease{Row: persistence.RuntimePlatformCommandRow{
					CommandID: "command-1", CommandKind: "runtime.test",
					PayloadFormatVersion: 2, Payload: []byte(`{"value":1}`),
				}, ImmutableBinding: "binding", CallerWorkloadID: "integration-1", AttemptsRemaining: 1},
			}
			executor := &platformCommandExecutorProbe{err: &PlatformCommandRecoveryError{Request: request}}
			registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
				"runtime.test": {2: executor},
			}, nil)
			if err != nil {
				t.Fatal(err)
			}
			recovery := &platformCommandRecoveryProbe{}
			if tc.cancel {
				recovery.cancel = cancel
			}
			processor, err := newPlatformCommandProcessor(
				store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: commandExecutionTestView(t)}, registry, 0,
				withPlatformCommandRecoveryOwner(recovery),
			)
			if err != nil {
				t.Fatal(err)
			}
			processed, found, err := processor.ProcessOne(ctx, 1)
			if !found || processed != 0 || !errors.Is(err, tc.wantErr) || store.replayed != tc.wantReplayed || store.retried != 0 ||
				store.completed != 0 || store.failed != 0 || store.poisoned != 0 {
				t.Fatalf("processed=%d found=%t err=%v replayed/retried=%d/%d complete/fail/poison=%d/%d/%d",
					processed, found, err, store.replayed, store.retried, store.completed, store.failed, store.poisoned)
			}
		})
	}
}

func TestPlatformCommandProcessorFailsClosedBeforeExecutor(t *testing.T) {
	view := commandExecutionTestView(t)
	for _, test := range []struct {
		name          string
		binder        platformCommandBinderProbe
		kind          string
		registryKind  string
		wantRetry     int
		wantPoison    int
		wantProcessed int
		wantErr       bool
	}{
		{
			name: "temporarily missing immutable view",
			binder: platformCommandBinderProbe{
				err: retryableMetadataBindingError(errors.New("missing")),
			},
			kind: "runtime.test", registryKind: "runtime.test", wantRetry: 1,
		},
		{
			name: "invalid immutable binding",
			binder: platformCommandBinderProbe{
				err: permanentMetadataBindingError(errors.New("invalid")),
			},
			kind: "runtime.test", registryKind: "runtime.test",
			wantPoison: 1, wantProcessed: 1,
		},
		{
			name: "immutable view rejects command", binder: platformCommandBinderProbe{view: view},
			kind: "runtime.other", registryKind: "runtime.other",
			wantPoison: 1, wantProcessed: 1,
		},
		{
			name: "unregistered release executor", binder: platformCommandBinderProbe{view: view},
			kind: "runtime.test", registryKind: "runtime.other",
			wantPoison: 1, wantProcessed: 1,
		},
		{
			name: "execution envelope installation fails closed",
			binder: platformCommandBinderProbe{
				view: view, bindErr: errors.New("scope installation failed"),
			},
			kind: "runtime.test", registryKind: "runtime.test", wantErr: true,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			store := &platformCommandStoreProbe{
				found: true,
				lease: persistence.RuntimePlatformCommandLease{
					Row: persistence.RuntimePlatformCommandRow{
						CommandID: "command-1", CommandKind: test.kind,
						PayloadFormatVersion: 1, Payload: []byte(`{"value":1}`),
					},
					ImmutableBinding: "binding",
				},
			}
			executor := &platformCommandExecutorProbe{}
			registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
				test.registryKind: {1: executor},
			}, nil)
			if err != nil {
				t.Fatal(err)
			}
			processor, err := newPlatformCommandProcessor(
				store, platformCommandTestObservationFactory(), test.binder, registry, 0,
			)
			if err != nil {
				t.Fatal(err)
			}
			processed, found, err := processor.ProcessOne(t.Context(), 1)
			if (err != nil) != test.wantErr || !found || processed != test.wantProcessed ||
				store.retried != test.wantRetry ||
				store.poisoned != test.wantPoison ||
				executor.calls != 0 {
				t.Fatalf(
					"ProcessOne() processed=%d found=%t err=%v retried=%d poisoned=%d calls=%d",
					processed, found, err, store.retried, store.poisoned, executor.calls,
				)
			}
		})
	}
}

func TestPlatformCommandProcessorLeavesCancellationForDurableReconciliation(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel()
	store := &platformCommandStoreProbe{
		found: true,
		lease: persistence.RuntimePlatformCommandLease{
			Row: persistence.RuntimePlatformCommandRow{
				CommandID: "command-1", CommandKind: "runtime.test",
				PayloadFormatVersion: 1, Payload: []byte(`{"value":1}`),
			},
			ImmutableBinding: "binding",
		},
	}
	executor := &platformCommandExecutorProbe{err: context.Canceled}
	registry, err := newPlatformCommandExecutorRegistry(map[string]map[uint32]PlatformCommandExecutor{
		"runtime.test": {1: executor},
	}, nil)
	if err != nil {
		t.Fatal(err)
	}
	processor, err := newPlatformCommandProcessor(
		store, platformCommandTestObservationFactory(), platformCommandBinderProbe{view: commandExecutionTestView(t)},
		registry, 0,
	)
	if err != nil {
		t.Fatal(err)
	}
	processed, found, err := processor.ProcessOne(ctx, 1)
	if !errors.Is(err, context.Canceled) || !found || processed != 0 ||
		store.retried != 0 || store.failed != 0 || store.poisoned != 0 {
		t.Fatalf("canceled ProcessOne() processed=%d found=%t err=%v dispositions=%d/%d/%d",
			processed, found, err, store.retried, store.failed, store.poisoned)
	}
}

func commandExecutionTestView(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	_, _, view := commandExecutionTestArtifacts(t)
	return view
}

func commandExecutionTestArtifacts(
	t *testing.T,
) (
	catalogsnapshot.Artifact,
	catalogsnapshot.Artifact,
	*catalogsnapshot.RuntimeMetadataView,
) {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(7)
	componentID := "component.command.runtime-test"
	componentV2ID := "component.command.runtime-test-v2"
	componentV3ID := "component.command.runtime-flow-v3"
	unownedComponentID := "component.command.runtime-unowned"
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: "org-test",
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: revision, SchemaGeneration: 1, EngineGeneration: 1,
		},
		Objects: []catalogsnapshot.ObjectInput{{
			CatalogRevision: revision,
			Metadata: &models.ObjectMetadata{
				ID: "object-work", APIName: "Work", Label: "Work", PluralLabel: "Work",
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{{
					ID: "field-work-id", APIName: constants.FieldID, Label: "ID",
					Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true,
				}},
			},
		}},
		CommandRegistrations: []catalogsnapshot.CommandRegistrationInput{
			{
				CatalogRevision: revision, ComponentID: componentID, DefinitionVersion: "1",
				ContentDigest: catalogsnapshot.ComputeDigest([]byte("runtime.test/v1")),
				CommandKind:   "runtime.test", PayloadFormatVersion: 1,
			},
			{
				CatalogRevision: revision, ComponentID: componentV2ID, DefinitionVersion: "2",
				ContentDigest: catalogsnapshot.ComputeDigest([]byte("runtime.test/v2")),
				CommandKind:   "runtime.test", PayloadFormatVersion: 2,
			},
			{
				CatalogRevision: revision, ComponentID: componentV3ID, DefinitionVersion: "3",
				ContentDigest: catalogsnapshot.ComputeDigest([]byte(RuntimeFlowCommandKind + "/v3")),
				CommandKind:   RuntimeFlowCommandKind, PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV3,
			},
			{
				CatalogRevision: revision, ComponentID: unownedComponentID, DefinitionVersion: "1",
				ContentDigest: catalogsnapshot.ComputeDigest([]byte("runtime.unowned/v1")),
				CommandKind:   "runtime.unowned", PayloadFormatVersion: 1,
			},
		},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision: revision, AppInstallationID: "installation-test",
			CapabilityID: "runtime.commands.accept.v1", Profile: catalogsnapshot.ProjectionCommandAcceptanceV1,
			DefinitionComponentIDs: []string{componentID, componentV2ID, componentV3ID, unownedComponentID},
		}},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: "installation-test", CapabilityID: "runtime.commands.accept.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact, view
}