package orchestrator

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
)

func TestCallbackServiceGetCurrentUsesFrozenWakeReceipt(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
	})
	require.NoError(t, err)
	require.Equal(t, "wri_1", result.RunID)
	require.Equal(t, "wfx_1", result.CallbackID)
	require.Equal(t, "a000001", result.ActivationID)
	require.Equal(t, "say", result.NodeID)
	require.Equal(t, run.EffectStatusAwaiting, result.Status)
	require.JSONEq(t, `{"word":"一心一意"}`, string(result.Input))
	require.JSONEq(t, `[{"name":"word","schema":{"type":"string","required":true,"default":null,"values":null,"items":null}}]`, string(result.OutputContract))
	require.Equal(t, &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, result.DeliveryScope)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_other", Action: CallbackActionGetCurrent,
	})
	require.ErrorIs(t, err, ErrCallbackUnauthorized)
}

func TestCallbackServiceAuthorizesLegacyUniqueWakeReceiptForAgentNameRef(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackResponder(t, repository, "calculator-a")
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.WakeAgentIDs = []string{"agt_other"}
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now: func() time.Time { return now },
	})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1",
		AgentID: "agt_other", Action: CallbackActionGetCurrent,
	})
	require.NoError(t, err)
	require.Equal(t, "wfx_1", result.CallbackID)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1",
		AgentID: "agt_exact", Action: CallbackActionGetCurrent,
	})
	require.ErrorIs(t, err, ErrCallbackUnauthorized)
}

func TestCallbackServiceUsesExplicitAuthorityWhenTaskWatchWakesOtherAgents(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.CallbackResponderAgentID = "agt_assigned"
	receipt.WakeAgentIDs = []string{"agt_assigned", "agt_task_watcher", "agt_assignee"}
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1",
		AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
	})
	require.NoError(t, err)
	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1",
		AgentID: "agt_task_watcher", Action: CallbackActionGetCurrent,
	})
	require.ErrorIs(t, err, ErrCallbackUnauthorized)
}

func TestCallbackServiceGetCurrentRejectsMalformedDeliveryScopeReceipt(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.DeliveryScope = &domain.MessageScopeValue{
		Kind:    domain.MessageScopeKindChannel,
		ScopeID: " chn_game",
	}
	repository.effect.ResultJSON, err = json.Marshal(receipt)
	require.NoError(t, err)
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
	})
	require.ErrorContains(t, err, "delivery scope is invalid")
}

func TestCallbackServiceGetCurrentAllowsLegacyReceiptWithoutDeliveryScope(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.DeliveryScope = nil
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
	})
	require.NoError(t, err)
	require.Nil(t, result.DeliveryScope)
}

func TestCallbackServiceGetCurrentRejectsNoLongerWaitingCallback(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	for _, tc := range []struct {
		name   string
		mutate func(*callbackRepositoryStub)
	}{
		{name: "run canceled", mutate: func(repository *callbackRepositoryStub) {
			repository.run.Status = run.RunStatusCanceled
		}},
		{name: "effect completed", mutate: func(repository *callbackRepositoryStub) {
			repository.effect.Status = run.EffectStatusCompleted
		}},
		{name: "activation no longer waiting", mutate: func(repository *callbackRepositoryStub) {
			repository.activation.Status = run.ActivationStatusSucceeded
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			repository := callbackServiceFixture(t, now)
			tc.mutate(repository)
			service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
			require.NoError(t, err)
			_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
				WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
			})
			require.ErrorIs(t, err, ErrCallbackNotWaiting)
		})
	}
}

func TestCallbackServiceReturnsNotReadyWhileDeliveryReceiptMayConverge(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	for _, status := range []string{run.EffectStatusPending, run.EffectStatusDispatching} {
		t.Run(status, func(t *testing.T) {
			for _, command := range []CallbackCommand{
				{Action: CallbackActionGetCurrent},
				{Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"意气风发"}`), IdempotencyKey: "decision-1"},
				{Action: CallbackActionFail, FailureCode: "agent.failed", IdempotencyKey: "decision-1"},
			} {
				t.Run(command.Action, func(t *testing.T) {
					repository := callbackServiceFixture(t, now)
					repository.effect.Status = status
					repository.effect.ResultJSON = nil
					service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
					require.NoError(t, err)

					command.WorkspaceID = "wsp_1"
					command.RunID = "wri_1"
					command.CallbackID = "wfx_1"
					command.AgentID = "agt_assigned"
					_, err = service.ExecuteCallback(context.Background(), command)
					require.ErrorIs(t, err, ErrCallbackNotReady)
					require.NotErrorIs(t, err, ErrCallbackNotWaiting)
				})
			}
		})
	}
}

func TestCallbackServiceDoesNotReportNotReadyOutsideLiveDispatchWindow(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	for _, tc := range []struct {
		name   string
		mutate func(*callbackRepositoryStub)
	}{
		{name: "pending effect with terminal run", mutate: func(repository *callbackRepositoryStub) {
			repository.effect.Status = run.EffectStatusPending
			repository.run.Status = run.RunStatusCanceled
		}},
		{name: "dispatching effect with terminal activation", mutate: func(repository *callbackRepositoryStub) {
			repository.activation.Status = run.ActivationStatusCanceled
		}},
		{name: "awaiting effect without receipt", mutate: func(repository *callbackRepositoryStub) {
			repository.effect.Status = run.EffectStatusAwaiting
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			repository := callbackServiceFixture(t, now)
			repository.effect.Status = run.EffectStatusDispatching
			repository.effect.ResultJSON = json.RawMessage(" \n ")
			tc.mutate(repository)
			service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
			require.NoError(t, err)

			_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
				WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned", Action: CallbackActionGetCurrent,
			})
			require.ErrorIs(t, err, ErrCallbackNotWaiting)
			require.NotErrorIs(t, err, ErrCallbackNotReady)
		})
	}
}

func TestCallbackServiceNonAwaitingFailureResultIsAuthoritativelyNotWaiting(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	nonReceipt, err := EncodeEffectResult(EffectResult{
		Port: "failed", Output: runkernel.Values{}, FailureCode: "message.delivery_failed", FailureMessage: "receipt lookup exhausted",
	})
	require.NoError(t, err)
	for _, status := range []string{
		run.EffectStatusDelivered,
		run.EffectStatusCompleted,
		run.EffectStatusFailed,
		run.EffectStatusCanceled,
	} {
		t.Run(status, func(t *testing.T) {
			for _, command := range []CallbackCommand{
				{Action: CallbackActionGetCurrent},
				{Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"意气风发"}`), IdempotencyKey: "decision-1"},
				{Action: CallbackActionFail, FailureCode: "agent.failed", IdempotencyKey: "decision-1"},
			} {
				t.Run(command.Action, func(t *testing.T) {
					repository := callbackServiceFixture(t, now)
					repository.effect.Status = status
					repository.effect.ResultJSON = nonReceipt
					service, serviceErr := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
					require.NoError(t, serviceErr)
					command.WorkspaceID, command.RunID, command.CallbackID, command.AgentID = "wsp_1", "wri_1", "wfx_1", "agt_assigned"

					_, serviceErr = service.ExecuteCallback(context.Background(), command)
					require.ErrorIs(t, serviceErr, ErrCallbackNotWaiting)
					require.NotErrorIs(t, serviceErr, ErrCallbackNotReady)
				})
			}
		})
	}
}

func TestCallbackServiceCompletesTypedOutputAndPreservesDeliveryReceipt(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 456789000, time.UTC)
	repository := callbackServiceFixture(t, now)
	var captured run.ResolveAwaitingEffectParams
	repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
		captured = params
		completedEffect := repository.effect
		completedEffect.Status = run.EffectStatusCompleted
		completedRun := repository.run
		return run.ResolveAwaitingEffectResult{Run: completedRun, Effect: completedEffect}, nil
	}
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"意气风发"}`), IdempotencyKey: "decision-1",
	})
	require.NoError(t, err)
	require.False(t, result.Replayed)
	require.Equal(t, run.EffectStatusCompleted, result.Status)
	require.Equal(t, "wra_1", captured.ActivationID)
	require.Equal(t, "wfx_1", captured.EffectID)
	require.Equal(t, callbackCompletedPort, captured.SelectedPort)
	require.Equal(t, now.UTC().Truncate(time.Microsecond), captured.Now)
	require.JSONEq(t, `{"word":"意气风发"}`, string(captured.ActivationOutputJSON))
	receipt, err := DecodeEffectResult(captured.EffectResultJSON)
	require.NoError(t, err)
	require.Equal(t, callbackCompletedPort, receipt.Port)
	require.Equal(t, "msg_1", receipt.MessageID)
	require.Equal(t, []string{"agt_assigned"}, receipt.WakeAgentIDs)
	require.Equal(t, &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, receipt.DeliveryScope)
	require.Equal(t, "decision-1", receipt.IdempotencyKey)
	require.Equal(t, "意气风发", receipt.Output["word"])
	require.Len(t, captured.Events, 1)
	require.Equal(t, "message_callback_completed", captured.Events[0].Type)
}

func TestCallbackServiceAuthorizesAndNormalizesScopeOutput(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 456789000, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackOutputContract(t, repository, plan.ArtifactContract{{
		Name: "work_scope",
		Schema: plan.ArtifactSchema{
			Type:     plan.TypeScope,
			Required: true,
		},
	}})
	var captured run.ResolveAwaitingEffectParams
	repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
		captured = params
		completed := repository.effect
		completed.Status = run.EffectStatusCompleted
		return run.ResolveAwaitingEffectResult{Run: repository.run, Effect: completed}, nil
	}
	authorizer := &callbackScopeAuthorizerStub{}
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now:             func() time.Time { return now },
		ScopeAuthorizer: authorizer,
	})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action:         CallbackActionComplete,
		Output:         json.RawMessage(`{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`),
		IdempotencyKey: "decision-scope",
	})
	require.NoError(t, err)
	require.Equal(t, &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"}, result.DeliveryScope)
	require.Equal(t, []callbackScopeAuthorizationCall{{
		WorkspaceID:   "wsp_1",
		AgentID:       "agt_assigned",
		OwnerUserID:   "usr_owner",
		OwnerMemberID: "mem_owner",
		Scope: domain.MessageScopeValue{
			Kind:    domain.MessageScopeKindChannel,
			ScopeID: "chn_work",
		},
	}}, authorizer.calls)
	require.JSONEq(t, `{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`, string(captured.ActivationOutputJSON))
	receipt, err := DecodeEffectResult(captured.EffectResultJSON)
	require.NoError(t, err)
	require.Equal(t, map[string]any{
		"kind":    "channel",
		"scopeId": "chn_work",
	}, receipt.Output["work_scope"])
}

func TestCallbackServiceDeduplicatesScopeAuthorizationInsideCommitGuard(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 456789000, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackOutputContract(t, repository, plan.ArtifactContract{{
		Name: "work_scopes",
		Schema: plan.ArtifactSchema{
			Type:     plan.TypeArray,
			Required: true,
			Items: &plan.ArtifactSchema{
				Type:     plan.TypeScope,
				Required: true,
			},
		},
	}})
	repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
		completed := repository.effect
		completed.Status = run.EffectStatusCompleted
		return run.ResolveAwaitingEffectResult{Run: repository.run, Effect: completed}, nil
	}
	authorizer := &callbackScopeAuthorizerStub{}
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now:             func() time.Time { return now },
		ScopeAuthorizer: authorizer,
	})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete,
		Output: json.RawMessage(`{"work_scopes":[
			{"kind":"channel","scopeId":"chn_work"},
			{"kind":"channel","scopeId":"chn_work"}
		]}`),
		IdempotencyKey: "decision-scope-array",
	})
	require.NoError(t, err)
	require.Len(t, authorizer.calls, 1)
	require.Equal(t, domain.MessageScopeValue{
		Kind:    domain.MessageScopeKindChannel,
		ScopeID: "chn_work",
	}, authorizer.calls[0].Scope)
}

func TestCallbackServiceAuthorizesScopeOutputsInStableNormalizedOrder(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 456789000, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackOutputContract(t, repository, plan.ArtifactContract{{
		Name: "work_scopes",
		Schema: plan.ArtifactSchema{
			Type:     plan.TypeArray,
			Required: true,
			Items: &plan.ArtifactSchema{
				Type:     plan.TypeScope,
				Required: true,
			},
		},
	}})
	repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
		completed := repository.effect
		completed.Status = run.EffectStatusCompleted
		return run.ResolveAwaitingEffectResult{Run: repository.run, Effect: completed}, nil
	}
	authorizer := &callbackScopeAuthorizerStub{}
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now:             func() time.Time { return now },
		ScopeAuthorizer: authorizer,
	})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete,
		Output: json.RawMessage(`{"work_scopes":[
			{"kind":"task","scopeId":"tsk_z"},
			{"kind":"thread","scopeId":"thr_b"},
			{"kind":"channel","scopeId":"chn_z"},
			{"kind":"task","scopeId":"tsk_a"},
			{"kind":"dm","scopeId":"dm_m"},
			{"kind":"channel","scopeId":"chn_a"}
		]}`),
		IdempotencyKey: "decision-scope-order",
	})
	require.NoError(t, err)

	authorizedScopes := make([]domain.MessageScopeValue, 0, len(authorizer.calls))
	for _, call := range authorizer.calls {
		authorizedScopes = append(authorizedScopes, call.Scope)
	}
	require.Equal(t, []domain.MessageScopeValue{
		{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_a"},
		{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_z"},
		{Kind: domain.MessageScopeKindDM, ScopeID: "dm_m"},
		{Kind: domain.MessageScopeKindTask, ScopeID: "tsk_a"},
		{Kind: domain.MessageScopeKindTask, ScopeID: "tsk_z"},
		{Kind: domain.MessageScopeKindThread, ScopeID: "thr_b"},
	}, authorizedScopes)
}

func TestCallbackServiceRejectsEffectIdentityThatDiffersFromLockedRun(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackOutputContract(t, repository, plan.ArtifactContract{{
		Name:   "work_scope",
		Schema: plan.ArtifactSchema{Type: plan.TypeScope, Required: true},
	}})
	repository.run.Identity.AuthorMemberID = "mem_other"
	resolved := false
	repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
		resolved = true
		return run.ResolveAwaitingEffectResult{}, nil
	}
	authorizer := &callbackScopeAuthorizerStub{}
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now:             func() time.Time { return now },
		ScopeAuthorizer: authorizer,
	})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action:         CallbackActionComplete,
		Output:         json.RawMessage(`{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`),
		IdempotencyKey: "decision-scope",
	})
	require.ErrorIs(t, err, ErrCallbackUnauthorized)
	require.False(t, resolved)
	require.Empty(t, authorizer.calls)
}

func TestCallbackServiceRejectsUnauthorizedScopeOutputBeforeCompletion(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	for _, tc := range []struct {
		name string
		err  error
	}{
		{name: "forbidden", err: storeerr.Forbidden("scope denied")},
		{name: "not found", err: storeerr.NotFound("scope missing")},
		{name: "kind mismatch", err: storeerr.BadRequest("scope kind mismatch")},
	} {
		t.Run(tc.name, func(t *testing.T) {
			repository := callbackServiceFixture(t, now)
			setCallbackOutputContract(t, repository, plan.ArtifactContract{{
				Name:   "work_scope",
				Schema: plan.ArtifactSchema{Type: plan.TypeScope, Required: true},
			}})
			resolved := false
			repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
				resolved = true
				return run.ResolveAwaitingEffectResult{}, nil
			}
			service, err := NewCallbackService(repository, CallbackServiceOptions{
				Now:             func() time.Time { return now },
				ScopeAuthorizer: &callbackScopeAuthorizerStub{err: tc.err},
			})
			require.NoError(t, err)

			_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
				WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
				Action:         CallbackActionComplete,
				Output:         json.RawMessage(`{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`),
				IdempotencyKey: "decision-scope",
			})
			require.ErrorIs(t, err, ErrCallbackOutputInvalid)
			require.False(t, resolved)
		})
	}
}

func TestCallbackServiceSameScopeDecisionReplayDoesNotReauthorize(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	setCallbackOutputContract(t, repository, plan.ArtifactContract{{
		Name:   "work_scope",
		Schema: plan.ArtifactSchema{Type: plan.TypeScope, Required: true},
	}})
	repository.effect.Status = run.EffectStatusCompleted
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.Port = callbackCompletedPort
	receipt.Output = runkernel.Values{"work_scope": map[string]any{
		"kind":    "channel",
		"scopeId": "chn_work",
	}}
	receipt.IdempotencyKey = "decision-scope"
	receipt.CompletedAt = &now
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	repository.activation.Status = run.ActivationStatusSucceeded
	repository.activation.OutputJSON = json.RawMessage(`{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`)
	repository.activation.SelectedPort = callbackCompletedPort
	service, err := NewCallbackService(repository, CallbackServiceOptions{
		Now: func() time.Time { return now.Add(time.Minute) },
		// Deliberately no mutable scope authorizer: committed same-result replay
		// is derived exclusively from the frozen receipt.
	})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action:         CallbackActionComplete,
		Output:         json.RawMessage(`{"work_scope":{"kind":"channel","scopeId":"chn_work"}}`),
		IdempotencyKey: "decision-scope",
	})
	require.NoError(t, err)
	require.True(t, result.Replayed)
}

func TestCallbackServiceValidatesContractAndFailure(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)

	t.Run("contract", func(t *testing.T) {
		repository := callbackServiceFixture(t, now)
		service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
		require.NoError(t, err)
		_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
			WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
			Action: CallbackActionComplete, Output: json.RawMessage(`{"extra":true}`), IdempotencyKey: "decision-1",
		})
		require.ErrorIs(t, err, ErrCallbackOutputInvalid)
	})

	t.Run("failure", func(t *testing.T) {
		repository := callbackServiceFixture(t, now)
		var captured run.ResolveAwaitingEffectParams
		repository.resolve = func(params run.ResolveAwaitingEffectParams) (run.ResolveAwaitingEffectResult, error) {
			captured = params
			effect := repository.effect
			effect.Status = run.EffectStatusCompleted
			return run.ResolveAwaitingEffectResult{Run: repository.run, Effect: effect}, nil
		}
		service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now }})
		require.NoError(t, err)
		_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
			WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
			Action: CallbackActionFail, FailureCode: "agent.failed", FailureMessage: "no answer", IdempotencyKey: "decision-1",
		})
		require.NoError(t, err)
		require.Equal(t, callbackFailedPort, captured.SelectedPort)
		require.JSONEq(t, `{}`, string(captured.ActivationOutputJSON))
		result, err := DecodeEffectResult(captured.EffectResultJSON)
		require.NoError(t, err)
		require.Equal(t, "agent.failed", result.FailureCode)
		// The responder's arbitrary FailureMessage is never persisted: the durable
		// contract carries the mapped hint/problem only.
		require.Equal(t, runkernel.FailureHint("agent.failed"), result.FailureMessage)
		require.Equal(t, runkernel.FailureHint("agent.failed"), result.FailureHint)
		require.Equal(t, runkernel.FailureProblem("agent.failed"), result.FailureProblem)
		require.Empty(t, result.FailureDiagnostics)
	})
}

func TestCallbackServiceSameDecisionReplaysAndDifferentDecisionConflicts(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	repository.effect.Status = run.EffectStatusCompleted
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.Port = callbackCompletedPort
	receipt.Output = runkernel.Values{"word": "意气风发"}
	receipt.IdempotencyKey = "decision-1"
	receipt.CompletedAt = &now
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	repository.activation.Status = run.ActivationStatusSucceeded
	repository.activation.OutputJSON = json.RawMessage(`{"word":"意气风发"}`)
	repository.activation.SelectedPort = callbackCompletedPort
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now.Add(time.Minute) }})
	require.NoError(t, err)

	result, err := service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"意气风发"}`), IdempotencyKey: "decision-1",
	})
	require.NoError(t, err)
	require.True(t, result.Replayed)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"发扬光大"}`), IdempotencyKey: "decision-1",
	})
	require.ErrorIs(t, err, ErrCallbackAlreadyResolved)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_assigned",
		Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"意气风发"}`), IdempotencyKey: "decision-2",
	})
	require.ErrorIs(t, err, ErrCallbackAlreadyResolved)
}

func TestCallbackServiceRejectsAmbiguousLegacyCallbackAuthority(t *testing.T) {
	now := time.Date(2026, 7, 21, 1, 2, 3, 0, time.UTC)
	repository := callbackServiceFixture(t, now)
	repository.effect.Status = run.EffectStatusCompleted
	receipt, err := DecodeEffectResult(repository.effect.ResultJSON)
	require.NoError(t, err)
	receipt.WakeAgentIDs = append(receipt.WakeAgentIDs, "agt_auto_wake")
	receipt.Port = callbackCompletedPort
	receipt.Output = runkernel.Values{"word": "意气风发"}
	receipt.IdempotencyKey = "winner-decision"
	receipt.CompletedAt = &now
	repository.effect.ResultJSON, err = EncodeEffectResult(receipt)
	require.NoError(t, err)
	repository.activation.Status = run.ActivationStatusSucceeded
	repository.activation.OutputJSON = json.RawMessage(`{"word":"意气风发"}`)
	repository.activation.SelectedPort = callbackCompletedPort
	service, err := NewCallbackService(repository, CallbackServiceOptions{Now: func() time.Time { return now.Add(time.Second) }})
	require.NoError(t, err)

	_, err = service.ExecuteCallback(context.Background(), CallbackCommand{
		WorkspaceID: "wsp_1", RunID: "wri_1", CallbackID: "wfx_1", AgentID: "agt_auto_wake",
		Action: CallbackActionComplete, Output: json.RawMessage(`{"word":"发扬光大"}`), IdempotencyKey: "loser-decision",
	})
	require.ErrorIs(t, err, ErrCallbackUnauthorized)
}

func callbackServiceFixture(t *testing.T, now time.Time) *callbackRepositoryStub {
	t.Helper()
	request, err := EncodeEffectRequest(EffectRequest{
		Intent: runkernel.Intent{
			Kind: runkernel.IntentSendMessage, ActivationID: "a000001", NodeID: "say", Callback: true,
			CallbackOutput: plan.ArtifactContract{{Name: "word", Schema: plan.ArtifactSchema{Type: plan.TypeString, Required: true}}},
			Retry:          plan.ArtifactRetryPolicy{MaxAttempts: 1, BackoffNanos: int64(time.Second)},
		},
		Message: &MessageRequest{
			WorkspaceID: "wsp_1", PlanID: "wfp_1", PlanVersionID: "wfv_1", RunID: "wri_1",
			ActivationID: "a000001", EffectID: "wfx_1", NodeID: "say", SourceKind: "manual",
			OwnerUserID: "usr_owner", AuthorMemberID: "mem_owner", TargetRef: "#game",
			Content: "@agent 一心一意，成语接龙输出一个成语", ClientMessageID: "workflow-run:test", Callback: true,
		},
	})
	require.NoError(t, err)
	receipt, err := EncodeEffectResult(EffectResult{
		MessageID:     "msg_1",
		WakeAgentIDs:  []string{"agt_assigned"},
		DeliveryScope: &domain.MessageScopeValue{Kind: domain.MessageScopeKindChannel, ScopeID: "chn_game"},
	})
	require.NoError(t, err)
	return &callbackRepositoryStub{
		run: run.Run{
			ID: "wri_1", WorkspaceID: "wsp_1", Status: run.RunStatusWaiting,
			Identity: invocation.ExecutionIdentitySnapshot{
				OwnerUserID:       "usr_owner",
				OwnerMemberID:     "mem_owner",
				PrincipalMemberID: "mem_owner",
				AuthorMemberID:    "mem_owner",
			},
		},
		activation: run.Activation{
			ID: "wra_1", WorkspaceID: "wsp_1", RunID: "wri_1", LogicalKey: "a000001", NodeID: "say",
			Status: run.ActivationStatusWaiting, InputJSON: json.RawMessage(`{"word":"一心一意"}`),
		},
		effect: run.Effect{
			ID: "wfx_1", WorkspaceID: "wsp_1", RunID: "wri_1", ActivationID: "wra_1", Kind: EffectKindMessage,
			Status: run.EffectStatusAwaiting, CallbackResponderAgentID: "agt_assigned",
			RequestJSON: request, ResultJSON: receipt,
		},
	}
}

func setCallbackOutputContract(t *testing.T, repository *callbackRepositoryStub, contract plan.ArtifactContract) {
	t.Helper()
	request, err := DecodeEffectRequest(repository.effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.CallbackOutput = contract
	repository.effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)
}

func setCallbackResponder(t *testing.T, repository *callbackRepositoryStub, responderAgentRef string) {
	t.Helper()
	request, err := DecodeEffectRequest(repository.effect.RequestJSON)
	require.NoError(t, err)
	request.Intent.Responder = responderAgentRef
	request.Message.ResponderAgentRef = responderAgentRef
	repository.effect.RequestJSON, err = EncodeEffectRequest(request)
	require.NoError(t, err)
}