package orchestrator

import (
	"encoding/json"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	runstore "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 TestProjectTurnRejectsOversizedStateAsDeterministicResourceFailure(t *testing.T) {
	_, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 3, 55, 0, 0, time.UTC)
	state := &runkernel.State{
		Status: runkernel.StatusRunning, Input: runkernel.Values{},
		Variables:   runkernel.Values{"oversized": strings.Repeat("x", runstore.MaxRunStateBytes)},
		Activations: []runkernel.Activation{}, Scopes: []runkernel.Scope{},
	}
	run := runstore.Run{ID: "wri_run", WorkspaceID: "wsp", Revision: 2}
	_, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease"},
		Graph: graph, Transition: runkernel.Transition{State: state}, Now: now,
	})
	require.ErrorIs(t, err, errRuntimeStateLimit)
}

func TestProjectTurnRejectsOversizedSinglePayloadAsDeterministicResourceFailure(t *testing.T) {
	_, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 3, 56, 0, 0, time.UTC)
	state := &runkernel.State{
		Status: runkernel.StatusSucceeded, Input: runkernel.Values{}, Variables: runkernel.Values{},
		Output:      runkernel.Values{"oversized": strings.Repeat("x", runstore.MaxRunPayloadBytes)},
		Activations: []runkernel.Activation{}, Scopes: []runkernel.Scope{},
	}
	run := runstore.Run{ID: "wri_run", WorkspaceID: "wsp", Revision: 2}
	_, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease"},
		Graph: graph, Transition: runkernel.Transition{State: state}, Now: now,
	})
	require.ErrorIs(t, err, errRuntimePayloadLimit)
}

func TestProjectTurnPersistsCallbackIntentAndFrozenIdentity(t *testing.T) {
	kernel, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 4, 0, 0, 0, time.UTC)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	transition, err := kernel.Step(state, now)
	require.NoError(t, err)
	require.Equal(t, runkernel.StatusWaiting, transition.State.Status)
	require.Len(t, transition.Intents, 1)

	run := runstore.Run{
		ID: "wri_run", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		WorkflowNameSnapshot: "成语接龙",
		Status:               runstore.RunStatusRunning, Revision: 2, StateJSON: []byte(`{}`),
		Identity: invocationIdentity(), StartedAt: ptrTime(now), RunnableAt: now,
	}
	fence := runstore.TurnFence{WorkspaceID: "wsp", RunID: "wri_run", LeaseToken: "lease"}
	projected, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: fence, Graph: graph, Transition: transition, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	require.Equal(t, runstore.RunStatusWaiting, projected.Status)
	require.True(t, projected.ReleaseLease)
	require.Equal(t, now.Add(parkedRunDelay), projected.RunnableAt)
	require.Len(t, projected.Activations, 1)
	require.Equal(t, runstore.MutationInsert, projected.Activations[0].Operation)
	require.Equal(t, "a00000000000000000001", projected.Activations[0].Value.LogicalKey)
	require.Equal(t, runstore.ActivationStatusWaiting, projected.Activations[0].Value.Status)
	require.Equal(t, int64(1), projected.Activations[0].Value.AttemptCount)
	require.Len(t, projected.Effects, 1)
	effect := projected.Effects[0].Value
	require.Equal(t, runstore.EffectStatusPending, effect.Status)
	require.Equal(t, EffectKindMessage, effect.Kind)
	require.NotEmpty(t, effect.IdempotencyKey)
	require.NotEqual(t, transition.Intents[0].ActivationID, effect.ActivationID)
	request, err := DecodeEffectRequest(effect.RequestJSON)
	require.NoError(t, err)
	require.True(t, request.Message.Callback)
	require.Equal(t, "owner-user", request.Message.OwnerUserID)
	require.Equal(t, "author-member", request.Message.AuthorMemberID)
	require.Equal(t, "成语接龙", request.Message.WorkflowName)
	require.Equal(t, effect.ID, request.Message.EffectID)
	require.Equal(t, "workflow-run:"+effect.ID, request.Message.ClientMessageID)
	require.Equal(t, "@agent answer in 9 characters or more", request.Message.Content)
}

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

	kernel, graph := callbackKernel(t)
	node, ok := graph.Node("notify")
	require.True(t, ok)
	node.Config.SendMessage.Target = nil
	node.Config.SendMessage.Scope = &plan.ArtifactExpression{
		Op:         plan.ArtifactOpConst,
		ResultType: plan.TypeScope,
		Value:      plan.ArtifactLiteral(`{"kind":"task","scopeId":"tsk_abcdefghijklmn"}`),
	}
	now := time.Date(2026, 7, 28, 19, 45, 0, 0, time.UTC)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	transition, err := kernel.Step(state, now)
	require.NoError(t, err)
	require.Len(t, transition.Intents, 1)

	run := runstore.Run{
		ID: "wri_scope", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		Status: runstore.RunStatusRunning, Revision: 2, StateJSON: []byte(`{}`),
		Identity: invocationIdentity(), StartedAt: ptrTime(now), RunnableAt: now,
	}
	projected, err := ProjectTurn(ProjectTurnInput{
		Run: run,
		Fence: runstore.TurnFence{
			WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease",
		},
		Graph: graph, Transition: transition, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	require.Len(t, projected.Effects, 1)
	request, err := DecodeEffectRequest(projected.Effects[0].Value.RequestJSON)
	require.NoError(t, err)
	require.Empty(t, request.Message.TargetRef)
	require.Equal(t, &domain.MessageScopeValue{
		Kind: domain.MessageScopeKindTask, ScopeID: "tsk_abcdefghijklmn",
	}, request.Message.Scope)
	require.Equal(t, request.Intent.Scope, request.Message.Scope)
}

func TestProjectTurnUpdatesExistingActivationWithoutDuplicatingEffect(t *testing.T) {
	kernel, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 4, 0, 0, 0, time.UTC)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	run := runstore.Run{
		ID: "wri_run", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		Status: runstore.RunStatusRunning, Revision: 2, StateJSON: []byte(`{}`),
		Identity: invocationIdentity(), StartedAt: ptrTime(now), RunnableAt: now,
	}
	fence := runstore.TurnFence{WorkspaceID: "wsp", RunID: "wri_run", LeaseToken: "lease"}
	first, err := ProjectTurn(ProjectTurnInput{Run: run, Fence: fence, Graph: graph, Transition: waiting, SourceKind: "manual", Now: now})
	require.NoError(t, err)
	existingActivation := first.Activations[0].Value
	existingActivation.CreatedAt = now
	existingEffect := first.Effects[0].Value
	existingEffect.Status = runstore.EffectStatusCompleted
	existingEffect.CompletedAt = ptrTime(now)

	completed, err := kernel.Complete(waiting.State, runkernel.Completion{
		ActivationID: waiting.Intents[0].ActivationID, Port: "completed", Output: runkernel.Values{"reply": "done"},
	}, now.Add(time.Second))
	require.NoError(t, err)
	projected, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: fence, Graph: graph, Transition: completed,
		ExistingActivations: []runstore.Activation{existingActivation},
		ExistingEffects:     []runstore.Effect{existingEffect}, SourceKind: "manual", Now: now.Add(time.Second),
	})
	require.NoError(t, err)
	require.Empty(t, projected.Effects)
	require.Len(t, projected.Activations, 2)
	require.Equal(t, runstore.MutationUpdate, projected.Activations[0].Operation)
	require.Equal(t, runstore.ActivationStatusWaiting, projected.Activations[0].ExpectedStatus)
	require.Equal(t, runstore.ActivationStatusSucceeded, projected.Activations[0].Value.Status)
	require.Equal(t, int64(1), projected.Activations[0].Value.AttemptCount)
	require.Equal(t, runstore.MutationInsert, projected.Activations[1].Operation)
	require.Equal(t, runstore.ActivationStatusReady, projected.Activations[1].Value.Status)
	require.Zero(t, projected.Activations[1].Value.AttemptCount)
	require.Equal(t, existingActivation.ID, projected.Activations[1].Value.ParentActivationID)
	require.Equal(t, now.Add(time.Second), projected.RunnableAt)
}

func TestProjectTurnRetainsWaitingActivationScheduleWithoutRepeatedIntent(t *testing.T) {
	kernel, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 4, 15, 0, 0, time.UTC)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	run := runstore.Run{
		ID: "wri_run", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		Status: runstore.RunStatusWaiting, Revision: 2, StateJSON: []byte(`{}`),
		Identity: invocationIdentity(), StartedAt: ptrTime(now), RunnableAt: now,
	}
	fence := runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease"}
	first, err := ProjectTurn(ProjectTurnInput{Run: run, Fence: fence, Graph: graph, Transition: waiting, SourceKind: "manual", Now: now})
	require.NoError(t, err)
	prior := first.Activations[0].Value
	prior.CreatedAt = now
	prior.UpdatedAt = now
	require.NotNil(t, prior.RetryAt)
	require.Nil(t, prior.DeadlineAt, "send_message carries no runtime deadline, so no DeadlineAt is projected")

	quietTurn := runkernel.Transition{State: waiting.State, Intents: []runkernel.Intent{}, Events: []runkernel.Event{}}
	projected, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: fence, Graph: graph, Transition: quietTurn,
		ExistingActivations: []runstore.Activation{prior}, ExistingEffects: []runstore.Effect{first.Effects[0].Value},
		SourceKind: "manual", Now: now.Add(time.Minute),
	})
	require.NoError(t, err)
	require.Empty(t, projected.Activations, "unchanged waiting metadata must not be cleared or rewritten")
}

func TestProjectTurnDefersTerminalRunWhileEffectDispatchIsCommitted(t *testing.T) {
	kernel, graph := callbackKernel(t)
	now := time.Date(2026, 7, 21, 4, 45, 0, 0, time.UTC)
	state, err := kernel.Start(runkernel.Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, now)
	require.NoError(t, err)
	completed, err := kernel.Complete(waiting.State, runkernel.Completion{
		ActivationID: waiting.Intents[0].ActivationID, Port: "completed", Output: runkernel.Values{"reply": "done"},
	}, now.Add(time.Second))
	require.NoError(t, err)
	for !completed.State.Status.Terminal() {
		next, stepErr := kernel.Step(completed.State, now.Add(time.Second))
		require.NoError(t, stepErr)
		mergeTransition(&completed, next)
	}
	run := runstore.Run{
		ID: "wri_run", WorkspaceID: "wsp", InvocationID: "wiv", PlanID: "pln", PlanVersionID: "plv",
		Status: runstore.RunStatusRunning, Revision: 2, StateJSON: []byte(`{}`),
		Identity: invocationIdentity(), StartedAt: ptrTime(now), RunnableAt: now,
	}
	seed, err := ProjectTurn(ProjectTurnInput{
		Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease"},
		Graph: graph, Transition: waiting, SourceKind: "manual", Now: now,
	})
	require.NoError(t, err)
	committed := seed.Effects[0].Value
	committed.Status = runstore.EffectStatusDispatching
	committed.AttemptCount = 1
	committed.LeaseOwner = "effect-worker"
	committed.LeaseToken = "effect-lease"
	leaseExpiry := now.Add(20 * time.Second)
	committed.LeaseExpiresAt = &leaseExpiry
	retryAt := now.Add(30 * time.Second)
	retrying := committed
	retrying.Status = runstore.EffectStatusPending
	retrying.RetryAt = &retryAt
	retrying.LeaseOwner, retrying.LeaseToken, retrying.LeaseExpiresAt = "", "", nil

	for _, test := range []struct {
		name       string
		effect     runstore.Effect
		runnableAt time.Time
	}{
		{name: "dispatching", effect: committed, runnableAt: leaseExpiry},
		{name: "crossed dispatch pending retry", effect: retrying, runnableAt: retryAt},
	} {
		t.Run(test.name, func(t *testing.T) {
			projected, err := ProjectTurn(ProjectTurnInput{
				Run: run, Fence: runstore.TurnFence{WorkspaceID: run.WorkspaceID, RunID: run.ID, LeaseToken: "lease"},
				Graph: graph, Transition: completed, ExistingEffects: []runstore.Effect{test.effect},
				SourceKind: "manual", Now: now.Add(2 * time.Second),
			})
			require.NoError(t, err)
			require.Equal(t, runstore.RunStatusWaiting, projected.Status)
			require.Nil(t, projected.CompletedAt)
			require.Nil(t, projected.CanceledAt)
			require.Empty(t, projected.Effects, "projection must not revoke a crossed-dispatch commitment")
			require.Equal(t, test.runnableAt, projected.RunnableAt)
			var persisted runkernel.State
			require.NoError(t, json.Unmarshal(projected.StateJSON, &persisted))
			require.Equal(t, runkernel.StatusSucceeded, persisted.Status)
		})
	}
}

func callbackKernel(t *testing.T) (*runkernel.Kernel, *runkernel.Graph) {
	t.Helper()
	raw, err := os.ReadFile("../plan/testdata/ir/send_message_callback.json")
	require.NoError(t, err)
	graph, err := runkernel.DecodeArtifact(raw)
	require.NoError(t, err)
	kernel, err := runkernel.NewKernel(graph)
	require.NoError(t, err)
	return kernel, graph
}

func invocationIdentity() invocation.ExecutionIdentitySnapshot {
	return invocation.ExecutionIdentitySnapshot{
		OwnerUserID: "owner-user", OwnerMemberID: "owner-member",
		PrincipalMemberID: "principal-member", AuthorMemberID: "author-member",
	}
}