package orchestrator

import (
	"context"
	"errors"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
)

type effectRepositoryStub struct {
	effect            run.Effect
	effects           []run.Effect
	candidates        []run.EffectCandidate
	claims            []run.ClaimEffectParams
	claimedEffects    []run.Effect
	run               run.Run
	resolutions       []run.ResolveEffectParams
	frozenAssignee    string
	freezeLeaseToken  string
	freezeCallCounter int

	selectParams   []run.SelectCallbackResponderCapacityParams
	selectResult   run.SelectCallbackResponderCapacityResult
	selectErr      error
	promoteParams  []run.PromoteCallbackResponderWaitersParams
	promoteResults []run.SelectCallbackResponderCapacityResult
	promoteErr     error
}

func (stub *effectRepositoryStub) ListPendingEffects(context.Context, run.ListPendingEffectsParams) ([]run.EffectCandidate, error) {
	if stub.candidates != nil {
		return append([]run.EffectCandidate(nil), stub.candidates...), nil
	}
	return []run.EffectCandidate{{EffectID: stub.effect.ID, WorkspaceID: stub.effect.WorkspaceID}}, nil
}

func (stub *effectRepositoryStub) ClaimEffect(_ context.Context, params run.ClaimEffectParams) (run.Effect, bool, error) {
	stub.claims = append(stub.claims, params)
	effect := stub.effect
	if len(stub.effects) >= len(stub.claims) {
		effect = stub.effects[len(stub.claims)-1]
	}
	expires := params.Now.Add(params.LeaseTTL).UTC().Truncate(time.Microsecond)
	effect.LeaseExpiresAt = &expires
	stub.claimedEffects = append(stub.claimedEffects, effect)
	return effect, true, nil
}

func (stub *effectRepositoryStub) ClaimCallbackEffectCapacity(_ context.Context, params run.ClaimCallbackEffectCapacityParams) (run.ClaimCallbackEffectCapacityResult, error) {
	stub.effect.CallbackResponderAgentID = params.ResponderAgentID
	return run.ClaimCallbackEffectCapacityResult{Effect: stub.effect, Claimed: true}, nil
}

func (stub *effectRepositoryStub) ReconcileAgentExecutionLeases(context.Context, string, int) (int, error) {
	return 0, nil
}

func (stub *effectRepositoryStub) SelectCallbackResponderCapacity(_ context.Context, params run.SelectCallbackResponderCapacityParams) (run.SelectCallbackResponderCapacityResult, error) {
	stub.selectParams = append(stub.selectParams, params)
	if stub.selectErr != nil {
		return run.SelectCallbackResponderCapacityResult{}, stub.selectErr
	}
	result := stub.selectResult
	if result.SelectedResponderAgentID != "" {
		stub.effect.CallbackResponderAgentID = result.SelectedResponderAgentID
		result.Effect = stub.effect
	}
	return result, nil
}

func (stub *effectRepositoryStub) PromoteCallbackResponderWaiters(_ context.Context, _ string, params run.PromoteCallbackResponderWaitersParams, _ int) ([]run.SelectCallbackResponderCapacityResult, error) {
	stub.promoteParams = append(stub.promoteParams, params)
	if stub.promoteErr != nil {
		return nil, stub.promoteErr
	}
	return append([]run.SelectCallbackResponderCapacityResult(nil), stub.promoteResults...), nil
}

type responderResolverStub struct {
	agentID string
	err     error
	calls   int
}

func (stub *responderResolverStub) ResolveCallbackResponder(_ context.Context, request MessageRequest) (string, error) {
	stub.calls++
	if stub.err != nil {
		return "", stub.err
	}
	return stub.agentID, nil
}

func (stub *effectRepositoryStub) FreezeApprovalAssignee(_ context.Context, workspaceID, runID, effectID, leaseToken, memberID string) error {
	stub.freezeCallCounter++
	stub.frozenAssignee = memberID
	stub.freezeLeaseToken = leaseToken
	stub.effect.ApprovalAssigneeMemberID = memberID
	return nil
}

func (stub *effectRepositoryStub) ResolveEffect(_ context.Context, params run.ResolveEffectParams) (run.Effect, []run.Event, error) {
	stub.resolutions = append(stub.resolutions, params)
	return stub.effect, nil, nil
}

func (stub *effectRepositoryStub) LoadRun(context.Context, string, string) (run.Run, error) {
	return stub.run, nil
}

type messageWriterStub struct {
	request        MessageRequest
	result         MessageResult
	err            error
	calls          int
	recoverRequest MessageRequest
	recoverID      string
	recoverResult  MessageResult
	recoverErr     error
	recoverCalls   int
	after          func()
}

func (stub *messageWriterStub) WriteRunMessage(_ context.Context, request MessageRequest) (MessageResult, error) {
	stub.request = request
	stub.calls++
	if stub.after != nil {
		stub.after()
	}
	return stub.result, stub.err
}

func (stub *messageWriterStub) RecoverRunMessageReceipt(_ context.Context, request MessageRequest, messageID string) (MessageResult, error) {
	stub.recoverRequest = request
	stub.recoverID = messageID
	stub.recoverCalls++
	if stub.after != nil {
		stub.after()
	}
	return stub.recoverResult, stub.recoverErr
}

type approvalWriterStub struct {
	request ApprovalRequest
	receipt ApprovalReceipt
	err     error
}

func (stub *approvalWriterStub) WriteApproval(_ context.Context, request ApprovalRequest) (ApprovalReceipt, error) {
	stub.request = request
	return stub.receipt, stub.err
}

type assigneeResolverStub struct {
	workspaceID string
	reference   string
	memberID    string
	err         error
}

func (stub *assigneeResolverStub) ResolveWorkflowApprovalAssignee(_ context.Context, workspaceID, reference string) (string, error) {
	stub.workspaceID, stub.reference = workspaceID, reference
	return stub.memberID, stub.err
}

func TestEffectDispatcherDeliversMessageAndAwaitsFrozenResponderCallback(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, true)
	repository := &effectRepositoryStub{effect: effect}
	deliveryScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg", WakeAgentIDs: []string{"agt_a", "agt_watcher", "agt_a"},
		CallbackResponderAgentID: "agt_a", DeliveryScope: &deliveryScope,
	}}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}})
	require.NoError(t, err)
	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Equal(t, effect.ID, messages.request.EffectID)
	require.Len(t, repository.resolutions, 1)
	resolved := repository.resolutions[0]
	require.Equal(t, run.EffectStatusAwaiting, resolved.Status)
	require.Equal(t, "agt_a", resolved.CallbackResponderAgentID)
	require.Equal(t, now.Add(parkedRunDelay).UTC(), resolved.RunnableAt.UTC())
	result, err := DecodeEffectResult(resolved.ResultJSON)
	require.NoError(t, err)
	require.Empty(t, result.Port)
	require.Equal(t, "msg", result.MessageID)
	require.Equal(t, []string{"agt_a", "agt_watcher"}, result.WakeAgentIDs)
	require.Equal(t, "agt_a", result.CallbackResponderAgentID)
	require.Equal(t, &deliveryScope, result.DeliveryScope)
}

func TestEffectDispatcherTaskWatchCallbackWakesRunAtCommittedDeliveryReceipt(t *testing.T) {
	startedAt := time.Date(2026, 7, 31, 8, 0, 0, 0, time.UTC)
	completedAt := startedAt.Add(5 * time.Second)
	effect := claimedMessageEffect(t, startedAt, true)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	taskScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_callback_work",
	}
	handoff := &runkernel.TaskWatchHandoff{
		TaskScope: taskScope, WorkItemID: "wfi_callback",
		CausalPredecessorAgentIDs: []string{},
	}
	request.Intent.Target = ""
	request.Intent.Scope = &taskScope
	request.Intent.WorkScopeMode = runkernel.WorkScopeModeFixed
	request.Intent.WorkScope = &taskScope
	request.Intent.TaskWatchHandoff = handoff
	request.Message.TargetRef = ""
	request.Message.Scope = &taskScope
	request.Message.WorkScopeMode = runkernel.WorkScopeModeFixed
	request.Message.WorkScope = &taskScope
	request.Message.TaskWatchHandoff = &runkernel.TaskWatchHandoff{
		TaskScope: taskScope, WorkItemID: "wfi_callback",
		CausalPredecessorAgentIDs: []string{},
	}
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)

	current := startedAt
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{
		result: MessageResult{
			MessageID: "msg", WakeAgentIDs: []string{"agt_a"},
			CallbackResponderAgentID: "agt_a", DeliveryScope: &taskScope,
		},
		after: func() { current = completedAt },
	}
	dispatcher, err := NewEffectDispatcher(
		repository,
		messages,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return current }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Len(t, repository.resolutions, 1)
	resolved := repository.resolutions[0]
	require.Equal(t, run.EffectStatusAwaiting, resolved.Status)
	require.Equal(t, completedAt, resolved.Now)
	require.NotNil(t, resolved.DeliveredAt)
	require.Equal(t, completedAt, *resolved.DeliveredAt)
	require.Equal(t, completedAt, resolved.RunnableAt,
		"a Task Watch callback receipt must wake the Run immediately so it can freeze the responder")
}

func TestEffectDispatcherPersistsNormalizedReceiptForAgentNameRef(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 5, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, true)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.Responder = "calculator-a"
	request.Message.ResponderAgentRef = "calculator-a"
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)
	effect.AttemptCount = 3
	repository := &effectRepositoryStub{effect: effect}
	deliveryScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg", WakeAgentIDs: []string{"agt_other"}, DeliveryScope: &deliveryScope,
	}}
	dispatcher, err := NewEffectDispatcher(
		repository,
		messages,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_other"}},
	)
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Len(t, repository.resolutions, 1)
	resolution := repository.resolutions[0]
	require.Equal(t, run.EffectStatusAwaiting, resolution.Status)
	require.Equal(t, "agt_other", resolution.CallbackResponderAgentID)
	result, err := DecodeEffectResult(resolution.ResultJSON)
	require.NoError(t, err)
	require.Equal(t, []string{"agt_other"}, result.WakeAgentIDs)
	require.Equal(t, "agt_other", result.CallbackResponderAgentID, "legacy unique wake receipt freezes explicit authority")
}

func TestEffectDispatcherPublishesTrustedStartThreadOutput(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, false)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.StartThread = true
	request.Message.StartThread = true
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)

	parentScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	threadScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindThread, ScopeID: "chn_bcdefghijklmno",
	}
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg_root", DeliveryScope: &parentScope, ThreadScope: &threadScope,
	}}
	dispatcher, err := NewEffectDispatcher(
		repository,
		messages,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	result, err := DecodeEffectResult(repository.resolutions[0].ResultJSON)
	require.NoError(t, err)
	require.Equal(t, runkernel.Values{
		"messageId": "msg_root",
		"threadScope": map[string]any{
			"kind": string(threadScope.Kind), "scopeId": threadScope.ScopeID,
		},
	}, result.Output)
	require.Equal(t, &parentScope, result.DeliveryScope)
	require.Equal(t, &threadScope, result.ThreadScope)
}

func TestEffectDispatcherRejectsUntrustedStartThreadReceipt(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, false)
	effect.AttemptCount = 3
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.StartThread = true
	request.Message.StartThread = true
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)

	parentScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg_root", DeliveryScope: &parentScope,
	}}
	dispatcher, err := NewEffectDispatcher(
		repository,
		messages,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)

	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	result, err := DecodeEffectResult(repository.resolutions[0].ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "failed", result.Port)
	require.Equal(t, "message.delivery_failed", result.FailureCode)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureMessage)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureHint)
	require.NotContains(t, result.FailureMessage, "threadScope is required")
	require.Empty(t, result.Output)
}

func TestEffectDispatcherStartThreadCrashReplayProducesByteEquivalentOutput(t *testing.T) {
	t.Parallel()

	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, false)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.StartThread = true
	request.Message.StartThread = true
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)

	parentScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	threadScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindThread, ScopeID: "chn_bcdefghijklmno",
	}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg_root", DeliveryScope: &parentScope, ThreadScope: &threadScope,
	}}

	dispatch := func() []byte {
		repository := &effectRepositoryStub{effect: effect}
		dispatcher, createErr := NewEffectDispatcher(
			repository,
			messages,
			nil,
			nil,
			EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}},
		)
		require.NoError(t, createErr)
		_, dispatchErr := dispatcher.DispatchOnce(context.Background())
		require.NoError(t, dispatchErr)
		require.Len(t, repository.resolutions, 1)
		return append([]byte(nil), repository.resolutions[0].ResultJSON...)
	}

	first := dispatch()
	// Model the crash window after the atomic message/thread transaction but
	// before effect resolution: the reclaimed attempt gets the same committed
	// receipt from the stable clientMessageId.
	messages.result.IdempotentReplay = true
	replayed := dispatch()
	require.JSONEq(t, string(first), string(replayed))
	require.Equal(t, string(first), string(replayed), "authoritative effect output must be byte-equivalent")
	require.Equal(t, 2, messages.calls)
	require.Equal(t, request.Message.ClientMessageID, messages.request.ClientMessageID)
}

func TestEffectDispatcherRetriesThenFailsMessageDelivery(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, false)
	effect.AttemptCount = 1
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{err: errors.New("network")}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}})
	require.NoError(t, err)
	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, run.EffectStatusPending, repository.resolutions[0].Status)
	require.Equal(t, now.Add(time.Second), *repository.resolutions[0].RetryAt)

	effect.AttemptCount = 3
	repository = &effectRepositoryStub{effect: effect}
	dispatcher, err = NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}})
	require.NoError(t, err)
	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, run.EffectStatusFailed, repository.resolutions[0].Status)
	result, err := DecodeEffectResult(repository.resolutions[0].ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "failed", result.Port)
	require.Equal(t, "message.delivery_failed", result.FailureCode)
}

func TestEffectDispatcherReconstructsBoundMessageReceiptAfterLostWriterResponse(t *testing.T) {
	startedAt := time.Date(2026, 7, 31, 9, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, startedAt, true)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	effect.AttemptCount = 1

	// The writer atomically commits the visible message and Effect binding, but
	// its response is lost. The dispatcher's old Effect snapshot is unbound, so
	// the ordinary retry path returns it to pending.
	firstRepository := &effectRepositoryStub{effect: effect}
	firstWriter := &messageWriterStub{err: errors.New("response lost after commit")}
	firstWriter.after = func() { firstRepository.effect.MessageID = "msg_committed" }
	firstDispatcher, err := NewEffectDispatcher(
		firstRepository,
		firstWriter,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return startedAt }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)
	_, err = firstDispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, 1, firstWriter.calls)
	require.Equal(t, "msg_committed", firstRepository.effect.MessageID, "the external transaction committed its binding")
	require.Len(t, firstRepository.resolutions, 1)
	require.Equal(t, run.EffectStatusPending, firstRepository.resolutions[0].Status)

	// By the time the lease is reclaimed, the retry budget is exhausted. The
	// durable binding changes the obligation: call the idempotent recoverer
	// again and keep the Effect pending on another transport error instead of
	// resolving a false failure without a receipt. send_message carries no
	// runtime deadline, so exhausted delivery must never resolve timed_out.
	reclaimAt := startedAt.Add(2 * time.Second)
	reclaimed := effect
	reclaimed.MessageID = "msg_committed"
	reclaimed.AttemptCount = int64(request.Intent.Retry.MaxAttempts)
	boundRepository := &effectRepositoryStub{effect: reclaimed}
	boundWriter := &messageWriterStub{recoverErr: errors.New("receipt reconstruction temporarily unavailable")}
	boundDispatcher, err := NewEffectDispatcher(
		boundRepository,
		boundWriter,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return reclaimAt }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)
	_, err = boundDispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Zero(t, boundWriter.calls, "a bound Effect must never replay the ordinary message writer")
	require.Equal(t, 1, boundWriter.recoverCalls, "a bound Effect must recover its receipt even after its deadline")
	require.Equal(t, "msg_committed", boundWriter.recoverID)
	require.Equal(t, request.Message.ClientMessageID, boundWriter.recoverRequest.ClientMessageID)
	require.Len(t, boundRepository.resolutions, 1)
	recovery := boundRepository.resolutions[0]
	require.Equal(t, run.EffectStatusPending, recovery.Status)
	require.Equal(t, reclaimAt.Add(time.Second), *recovery.RetryAt)
	require.Equal(t, *recovery.RetryAt, recovery.RunnableAt)

	// A subsequent idempotent replay reconstructs the normalized callback
	// receipt. It is persisted even though the original delivery deadline is in
	// the past; the engine may then apply the callback's normal timeout rules
	// without losing the committed responder/watch identity.
	deliveryScope := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	replayAt := *recovery.RetryAt
	replayRepository := &effectRepositoryStub{effect: reclaimed}
	replayWriter := &messageWriterStub{recoverResult: MessageResult{
		MessageID: "msg_committed", WakeAgentIDs: []string{"agt_a"}, CallbackResponderAgentID: "agt_a",
		DeliveryScope: &deliveryScope, IdempotentReplay: true,
	}}
	replayDispatcher, err := NewEffectDispatcher(
		replayRepository,
		replayWriter,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return replayAt }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)
	_, err = replayDispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Zero(t, replayWriter.calls)
	require.Equal(t, 1, replayWriter.recoverCalls)
	require.Len(t, replayRepository.resolutions, 1)
	receiptResolution := replayRepository.resolutions[0]
	require.Equal(t, run.EffectStatusAwaiting, receiptResolution.Status)
	receipt, err := DecodeEffectResult(receiptResolution.ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "msg_committed", receipt.MessageID)
	require.Equal(t, "agt_a", receipt.CallbackResponderAgentID)
}

func TestEffectDispatcherClampsBoundReceiptRecoveryBackoffAboveClockPrecision(t *testing.T) {
	now := time.Date(2026, 7, 31, 9, 30, 0, 123456000, time.UTC)
	effect := claimedMessageEffect(t, now, true)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.Retry.BackoffNanos = 1
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)
	effect.MessageID = "msg_committed"
	effect.AttemptCount = int64(request.Intent.Retry.MaxAttempts)
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{recoverErr: errors.New("receipt not visible yet")}
	dispatcher, err := NewEffectDispatcher(
		repository,
		messages,
		nil,
		nil,
		EffectDispatcherOptions{WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"}},
	)
	require.NoError(t, err)

	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Zero(t, messages.calls)
	require.Equal(t, 1, messages.recoverCalls)
	require.Len(t, repository.resolutions, 1)
	resolution := repository.resolutions[0]
	require.Equal(t, run.EffectStatusPending, resolution.Status)
	require.NotNil(t, resolution.RetryAt)
	require.Equal(t, now.Add(time.Second), *resolution.RetryAt)
	require.True(t, resolution.RetryAt.After(now), "sub-microsecond authored backoff must not hot-loop")
	require.Equal(t, *resolution.RetryAt, resolution.RunnableAt)
}

func TestEffectDispatcherUsesFreshClaimAndCompletionTimesAcrossSlowBatch(t *testing.T) {
	start := time.Date(2026, 7, 21, 8, 30, 0, 0, time.UTC)
	current := start
	first := claimedMessageEffect(t, start, false)
	first.ID = "effect_first"
	second := claimedMessageEffect(t, start, false)
	second.ID = "effect_second"
	repository := &effectRepositoryStub{
		effects: []run.Effect{first, second},
		candidates: []run.EffectCandidate{
			{EffectID: first.ID, WorkspaceID: first.WorkspaceID},
			{EffectID: second.ID, WorkspaceID: second.WorkspaceID},
		},
	}
	messages := &messageWriterStub{
		result: MessageResult{
			MessageID: "msg",
			DeliveryScope: &domain.MessageScopeValue{
				Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
			},
		},
		after: func() {
			if current.Equal(start) {
				current = current.Add(45 * time.Second)
			}
		},
	}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", LeaseTTL: 30 * time.Second, Now: func() time.Time { return current }, Responders: &responderResolverStub{agentID: "agt_a"},
	})
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 2, Claimed: 2, Resolved: 2}, outcome)
	require.Equal(t, 2, messages.calls, "send_message carries no runtime deadline, so both effects are delivered")
	require.Len(t, repository.claims, 2)
	require.Equal(t, start, repository.claims[0].Now)
	require.Equal(t, start.Add(45*time.Second), repository.claims[1].Now)
	require.Equal(t, start.Add(75*time.Second), *repository.claimedEffects[1].LeaseExpiresAt,
		"the second lease must start when the second effect is actually claimed")
	require.Len(t, repository.resolutions, 2)
	require.Equal(t, start.Add(45*time.Second), repository.resolutions[0].Now,
		"the first receipt must use the external write completion time")
	require.Equal(t, run.EffectStatusCompleted, repository.resolutions[0].Status)
	require.Equal(t, start.Add(45*time.Second), repository.resolutions[1].Now)
	require.Equal(t, run.EffectStatusCompleted, repository.resolutions[1].Status)
	completed, err := DecodeEffectResult(repository.resolutions[1].ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "completed", completed.Port)
}

func TestEffectDispatcherRejectsMessageReceiptWithoutTrustedDeliveryScope(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 40, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, true)
	effect.AttemptCount = 3
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{result: MessageResult{MessageID: "msg"}}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"},
	})
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Len(t, repository.resolutions, 1)
	resolution := repository.resolutions[0]
	require.Equal(t, run.EffectStatusFailed, resolution.Status)
	require.Nil(t, resolution.DeliveredAt, "an invalid writer receipt must not become a delivered callback capability")
	result, err := DecodeEffectResult(resolution.ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "failed", result.Port)
	require.Equal(t, "message.delivery_failed", result.FailureCode)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureMessage)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureHint)
	require.NotContains(t, result.FailureMessage, "deliveryScope is required")
	require.Empty(t, result.MessageID)
	require.Nil(t, result.DeliveryScope)
}

func TestEffectDispatcherRejectsMessageReceiptForDifferentRequestedScope(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 42, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, false)
	effect.AttemptCount = 3
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	requested := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_abcdefghijklmn",
	}
	request.Intent.Target = ""
	request.Intent.Scope = &requested
	request.Message.TargetRef = ""
	request.Message.Scope = &requested
	effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)

	delivered := domain.MessageScopeValue{
		Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn",
	}
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg", DeliveryScope: &delivered,
	}}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", Now: func() time.Time { return now }, Responders: &responderResolverStub{agentID: "agt_a"},
	})
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Len(t, repository.resolutions, 1)
	resolution := repository.resolutions[0]
	require.Equal(t, run.EffectStatusFailed, resolution.Status)
	require.Nil(t, resolution.DeliveredAt)
	result, err := DecodeEffectResult(resolution.ResultJSON)
	require.NoError(t, err)
	require.Equal(t, "failed", result.Port)
	require.Equal(t, "message.delivery_failed", result.FailureCode)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureMessage)
	require.Equal(t, runkernel.FailureHint("message.delivery_failed"), result.FailureHint)
	require.NotContains(t, result.FailureMessage, "deliveryScope does not match requested scope")
}

func TestEffectDispatcherSchedulesRetryFromExternalFailureCompletionTime(t *testing.T) {
	start := time.Date(2026, 7, 21, 8, 45, 0, 0, time.UTC)
	current := start
	effect := claimedMessageEffect(t, start, false)
	effect.AttemptCount = 1
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{
		err: errors.New("network"),
		after: func() {
			current = current.Add(7 * time.Second)
		},
	}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", Now: func() time.Time { return current }, Responders: &responderResolverStub{agentID: "agt_a"},
	})
	require.NoError(t, err)

	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Len(t, repository.resolutions, 1)
	resolution := repository.resolutions[0]
	require.Equal(t, start.Add(7*time.Second), resolution.Now)
	require.Equal(t, start.Add(8*time.Second), *resolution.RetryAt)
}

// TestEffectDispatcherFrozenResponderReusesIDWithoutResolver proves P0 at the
// dispatcher boundary: once effect.CallbackResponderAgentID is frozen, an unbound
// retry/reclaim never calls the mutable resolver; the writer receives the frozen
// ID as the internal runtime fence, and the effect/queue/lease identity stays A.
func TestEffectDispatcherFrozenResponderReusesIDWithoutResolver(t *testing.T) {
	now := time.Date(2026, 7, 21, 8, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, now, true)
	effect.CallbackResponderAgentID = "agt_a" // already frozen
	repository := &effectRepositoryStub{effect: effect}
	deliveryScope := domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_abcdefghijklmn"}
	messages := &messageWriterStub{result: MessageResult{
		MessageID: "msg", WakeAgentIDs: []string{"agt_a"},
		CallbackResponderAgentID: "agt_a", DeliveryScope: &deliveryScope,
	}}
	// The resolver, if called, would drift to B. It must never be called.
	responders := &responderResolverStub{agentID: "agt_b"}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", Now: func() time.Time { return now }, Responders: responders,
	})
	require.NoError(t, err)

	outcome, err := dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Equal(t, DispatchOutcome{Candidates: 1, Claimed: 1, Resolved: 1}, outcome)
	require.Zero(t, responders.calls, "the mutable resolver must never be called once the responder is frozen")
	require.Equal(t, "agt_a", messages.request.ExpectedCallbackResponderAgentID, "the writer must receive the frozen ID as the runtime fence")
	require.Len(t, repository.resolutions, 1)
	require.Equal(t, "agt_a", repository.resolutions[0].CallbackResponderAgentID, "the receipt must keep the frozen responder")
}

// TestEffectDispatcherFrozenResponderSurvivesRetryableWriterFailure proves that a
// retryable unbound writer failure keeps the same frozen responder: the retry
// path never re-resolves and the second attempt still targets A.
func TestEffectDispatcherFrozenResponderSurvivesRetryableWriterFailure(t *testing.T) {
	startedAt := time.Date(2026, 7, 21, 9, 0, 0, 0, time.UTC)
	effect := claimedMessageEffect(t, startedAt, true)
	effect.CallbackResponderAgentID = "agt_a"
	repository := &effectRepositoryStub{effect: effect}
	messages := &messageWriterStub{err: errors.New("network")}
	responders := &responderResolverStub{agentID: "agt_b"}
	dispatcher, err := NewEffectDispatcher(repository, messages, nil, nil, EffectDispatcherOptions{
		WorkerID: "effects", Now: func() time.Time { return startedAt }, Responders: responders,
	})
	require.NoError(t, err)

	_, err = dispatcher.DispatchOnce(context.Background())
	require.NoError(t, err)
	require.Zero(t, responders.calls, "a retryable unbound failure must not re-resolve a frozen responder")
	require.Len(t, repository.resolutions, 1)
	require.Equal(t, run.EffectStatusPending, repository.resolutions[0].Status, "retryable failure schedules a retry")
}