package run

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

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
)

var reducerNow = time.Date(2026, 7, 21, 12, 0, 0, 0, time.UTC)

func TestActionOpcodesProduceTypedIntentsAndResume(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name       string
		golden     string
		input      Values
		intentKind IntentKind
		port       string
		output     Values
	}{
		{"sent", "send_message_sent.json", Values{}, IntentSendMessage, "completed", Values{}},
		{"callback", "send_message_callback.json", Values{"channel": "#workflow"}, IntentSendMessage, "completed", Values{"reply": "done"}},
		{"wait", "wait.json", Values{}, IntentWait, "completed", Values{}},
		{"approval", "request_approval.json", Values{}, IntentApproval, "approved", Values{}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			kernel, _ := goldenKernel(t, test.golden)
			state, err := kernel.Start(test.input)
			require.NoError(t, err)
			step, err := kernel.Step(state, reducerNow)
			require.NoError(t, err)
			require.Len(t, step.Intents, 1)
			require.Equal(t, test.intentKind, step.Intents[0].Kind)
			if test.name == "callback" {
				require.True(t, step.Intents[0].Callback)
				require.Equal(t, "#workflow", step.Intents[0].Target)
			}
			completion, err := kernel.Complete(step.State, Completion{ActivationID: step.Intents[0].ActivationID, Port: test.port, Output: test.output}, reducerNow)
			require.NoError(t, err)
			terminal := stepUntilTerminal(t, kernel, completion.State, 5)
			require.Equal(t, StatusSucceeded, terminal.Status)
		})
	}
}

func TestDynamicRequiredActionFieldsFailThroughCompiledFailedPort(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name   string
		golden string
		nodeID string
		code   string
		mutate func(*plan.ArtifactNode)
	}{
		{
			name: "message target", golden: "send_message_sent.json", nodeID: "action", code: "message.target_invalid",
			mutate: func(node *plan.ArtifactNode) {
				target := dynamicStringTemplate(t, " \n\t ")
				node.Config.SendMessage.Target = &target
			},
		},
		{
			name: "message text", golden: "send_message_sent.json", nodeID: "action", code: "message.text_invalid",
			mutate: func(node *plan.ArtifactNode) { node.Config.SendMessage.Text = dynamicStringTemplate(t, " \n\t ") },
		},
		{
			name: "approval assignee", golden: "request_approval.json", nodeID: "action", code: "approval.assignee_invalid",
			mutate: func(node *plan.ArtifactNode) {
				node.Config.RequestApproval.Assignee = dynamicStringTemplate(t, " \n\t ")
			},
		},
		{
			name: "approval summary", golden: "request_approval.json", nodeID: "action", code: "approval.summary_invalid",
			mutate: func(node *plan.ArtifactNode) {
				node.Config.RequestApproval.Summary = dynamicStringTemplate(t, " \n\t ")
			},
		},
		{
			name: "approval summary protocol limit", golden: "request_approval.json", nodeID: "action", code: "approval.summary_invalid",
			mutate: func(node *plan.ArtifactNode) {
				node.Config.RequestApproval.Summary = dynamicStringTemplate(t, strings.Repeat("批", maxApprovalSummaryRunes+1))
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			kernel, graph := goldenKernel(t, test.golden)
			node, ok := graph.Node(test.nodeID)
			require.True(t, ok)
			test.mutate(node)

			state, err := kernel.Start(Values{})
			require.NoError(t, err)
			transition, err := kernel.Step(state, reducerNow)
			require.NoError(t, err)
			require.Empty(t, transition.Intents, "invalid rendered input must not create an effect")
			require.NotContains(t, eventKinds(transition.Events), "effect_requested")
			require.NotContains(t, eventKinds(transition.Events), "approval_requested")
			for _, activation := range transition.State.Activations {
				require.NotEqual(t, ActivationWaiting, activation.Status)
			}
			action, err := transition.State.activation(state.Activations[0].ID)
			require.NoError(t, err)
			require.NotNil(t, action.Failure)
			require.Equal(t, test.code, action.Failure.Code)
			require.Equal(t, []string{"failed"}, action.SelectedPorts)

			terminal := stepUntilTerminal(t, kernel, transition.State, 2)
			require.Equal(t, StatusFailed, terminal.Status)
		})
	}
}

func TestRenderedActionIdentifiersAreTrimmedWithoutChangingMessageContent(t *testing.T) {
	t.Parallel()
	kernel, graph := goldenKernel(t, "send_message_sent.json")
	node, ok := graph.Node("action")
	require.True(t, ok)
	target := dynamicStringTemplate(t, "  #workflow \n")
	node.Config.SendMessage.Target = &target
	node.Config.SendMessage.Text = dynamicStringTemplate(t, "\t hello agent  ")
	state, err := kernel.Start(Values{})
	require.NoError(t, err)
	transition, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Len(t, transition.Intents, 1)
	require.Equal(t, "#workflow", transition.Intents[0].Target)
	require.Equal(t, "\t hello agent  ", transition.Intents[0].Text)

	kernel, graph = goldenKernel(t, "request_approval.json")
	node, ok = graph.Node("action")
	require.True(t, ok)
	node.Config.RequestApproval.Assignee = dynamicStringTemplate(t, "  @alice \n")
	node.Config.RequestApproval.Summary = dynamicStringTemplate(t, "\t Approve this run  ")
	state, err = kernel.Start(Values{})
	require.NoError(t, err)
	transition, err = kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Len(t, transition.Intents, 1)
	require.Equal(t, "@alice", transition.Intents[0].Assignee)
	require.Equal(t, "Approve this run", transition.Intents[0].Summary)
}

func TestSetVariableIsAtomicAndUsesOldStateSnapshot(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "set_variable.json")
	state, err := kernel.Start(Values{})
	require.NoError(t, err)
	step, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Empty(t, step.Intents)
	require.Equal(t, json.Number("2"), step.State.Variables["count"])
	terminal := stepUntilTerminal(t, kernel, step.State, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
}

func TestForkJoinAllAndScopeTransitionsAsSet(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "fork_join_all.json")
	state, err := kernel.Start(Values{"value": "answer"})
	require.NoError(t, err)
	fork, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Len(t, fork.State.Scopes, 1)
	parallel := fork.State.Scopes[0].Parallel
	require.Equal(t, "completed", parallel.Outcome)
	require.Equal(t, ScopeSucceeded, fork.State.Scopes[0].Status)
	require.Len(t, StableReadyIDs(fork.State), 1) // join, not one join activation per arrival
	require.Empty(t, fork.State.Activations[0].ParentActivationID)
	require.Equal(t, fork.State.Activations[0].ID, fork.State.Activations[1].ParentActivationID)
	for _, branch := range parallel.Branches {
		require.Equal(t, BranchSucceeded, branch.Status)
	}
	terminal := stepUntilTerminal(t, kernel, fork.State, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.Equal(t, "answer", terminal.Output["value"])
}

func TestJoinAnyChoosesFirstSuccessfulNotFirstTerminal(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "fork_join_any.json")
	state, err := kernel.Start(Values{"value": "answer"})
	require.NoError(t, err)
	transition := Transition{State: state, Intents: []Intent{}, Events: []Event{}}
	ownerID := state.Activations[0].ID
	require.NoError(t, kernel.openParallel(&transition, ownerID, []string{"left", "right"}))
	scope := &state.Scopes[0]
	require.NoError(t, kernel.failParallelBranch(&transition, ScopeRef{ScopeID: scope.ID, Kind: plan.ArtifactScopeParallel, OwnerID: "fanout", Branch: "left"}, "left.failed"))
	require.Equal(t, ScopeActive, scope.Status)
	join, err := state.activation(scope.Parallel.JoinActivationID)
	require.NoError(t, err)
	require.Equal(t, ActivationWaiting, join.Status)

	require.NoError(t, kernel.recordParallelArrival(&transition, scope.ID, "right", Values{"value": "winner"}))
	require.Equal(t, ScopeSucceeded, scope.Status)
	require.Equal(t, "right", scope.Parallel.WinnerBranch)
	require.Equal(t, ActivationReady, join.Status)
	terminal := stepUntilTerminal(t, kernel, state, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.Equal(t, "winner", terminal.Output["value"])
}

func TestEndFailedInsideBranchFailsBranchNotRun(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "fork_join_any.json")
	state, err := kernel.Start(Values{"value": "answer"})
	require.NoError(t, err)
	transition := Transition{State: state, Intents: []Intent{}, Events: []Event{}}
	ownerID := state.Activations[0].ID
	require.NoError(t, kernel.openParallel(&transition, ownerID, []string{"left", "right"}))
	scope := &state.Scopes[0]
	failed, err := state.appendActivation("failure", "in", Values{}, []ScopeRef{{ScopeID: scope.ID, Kind: plan.ArtifactScopeParallel, OwnerID: "fanout", Branch: "left"}}, ActivationReady)
	require.NoError(t, err)
	require.NoError(t, kernel.execute(&transition, failed.ID, reducerNow))
	require.NotEqual(t, StatusFailed, state.Status)
	require.Equal(t, BranchFailed, branchByPort(t, scope.Parallel.Branches, "left").Status)
	join, err := state.activation(scope.Parallel.JoinActivationID)
	require.NoError(t, err)
	require.Equal(t, ActivationWaiting, join.Status)
	require.NoError(t, kernel.recordParallelArrival(&transition, scope.ID, "right", Values{"value": "winner"}))
	require.Equal(t, ActivationReady, join.Status)
}

func TestSwitchAllMatchActivatesOnlyMatchedBranches(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "switch_all_match.json")
	state, err := kernel.Start(Values{"value": "left right"})
	require.NoError(t, err)
	step, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Len(t, step.State.Scopes, 1)
	branches := step.State.Scopes[0].Parallel.Branches
	require.Equal(t, BranchInactive, branchByPort(t, branches, "fallback").Status)
	require.Equal(t, BranchSucceeded, branchByPort(t, branches, "left").Status)
	require.Equal(t, BranchSucceeded, branchByPort(t, branches, "right").Status)
	terminal := stepUntilTerminal(t, kernel, step.State, 4)
	require.Equal(t, StatusSucceeded, terminal.Status)
}

func TestCountedLoopBreakAndAutomaticCompletion(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "bounded_loop_first_match.json")
	state, err := kernel.Start(Values{"answer": "stop"})
	require.NoError(t, err)
	terminal := stepUntilTerminal(t, kernel, state, 12)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.Equal(t, "stop", terminal.Output["answer"])
	require.Len(t, terminal.Scopes, 1)
	require.Equal(t, 1, terminal.Scopes[0].Loop.Iteration)

	state, err = kernel.Start(Values{"answer": "again"})
	require.NoError(t, err)
	terminal = stepUntilTerminal(t, kernel, state, 30)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.Equal(t, "again", terminal.Output["answer"])
	require.Equal(t, 3, terminal.Scopes[0].Loop.Iteration)
}

func TestNestedParallelAndParallelInLoop(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "nested_parallel.json")
	state, err := kernel.Start(Values{"value": "nested"})
	require.NoError(t, err)
	terminal := stepUntilTerminal(t, kernel, state, 20)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.Len(t, terminal.Scopes, 2)

	kernel, _ = goldenKernel(t, "parallel_in_loop.json")
	state, err = kernel.Start(Values{"value": "stop"})
	if err != nil {
		// Keep the failure useful if the compiler golden's field has a
		// different name after a DSL revision.
		t.Fatalf("parallel_in_loop Start: %v", err)
	}
	terminal = stepUntilTerminal(t, kernel, state, 30)
	require.Equal(t, StatusSucceeded, terminal.Status)
	require.GreaterOrEqual(t, len(terminal.Scopes), 2)
}

func TestCompletionContractReplayCancelAndJSONRoundTrip(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "send_message_callback.json")
	state, err := kernel.Start(Values{"channel": "#workflow"})
	require.NoError(t, err)
	step, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	id := step.Intents[0].ActivationID
	_, err = kernel.Complete(step.State, Completion{ActivationID: id, Port: "completed", Output: Values{}}, reducerNow)
	require.ErrorContains(t, err, "required")

	completed, err := kernel.Complete(step.State, Completion{ActivationID: id, Port: "completed", Output: Values{"reply": "ok"}}, reducerNow)
	require.NoError(t, err)
	require.Len(t, completed.State.Activations, 2)
	require.Equal(t, id, completed.State.Activations[1].ParentActivationID)
	activation, err := completed.State.activation(id)
	require.NoError(t, err)
	replay, err := kernel.Complete(completed.State, Completion{ActivationID: id, Port: "completed", Output: activation.Output}, reducerNow)
	require.NoError(t, err)
	require.Equal(t, "completion_replayed", replay.Events[0].Kind)
	_, err = kernel.Complete(completed.State, Completion{ActivationID: id, Port: "completed", Output: Values{"reply": "different"}}, reducerNow)
	require.ErrorContains(t, err, "different result")
	terminal := stepUntilTerminal(t, kernel, completed.State, 4)
	replay, err = kernel.Complete(terminal, Completion{ActivationID: id, Port: "completed", Output: Values{"reply": "ok"}}, reducerNow)
	require.NoError(t, err)
	require.Equal(t, "completion_replayed", replay.Events[0].Kind)

	restored, err := JSONRoundTrip(completed.State)
	require.NoError(t, err)
	require.Equal(t, completed.State, restored)
	canceled, err := kernel.Cancel(restored)
	require.NoError(t, err)
	require.Equal(t, StatusCanceled, canceled.State.Status)
	_, err = kernel.Complete(canceled.State, Completion{ActivationID: id, Port: "completed", Output: Values{"reply": "ok"}}, reducerNow)
	require.ErrorContains(t, err, "terminal")
}

func TestReducerRefusesActivationBeyondSharedRuntimeLimit(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "send_message_callback.json")
	state, err := kernel.Start(Values{"channel": "#workflow"})
	require.NoError(t, err)
	waiting, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Len(t, waiting.Intents, 1)
	waiting.State.NextActivation = uint64(plan.MaxRunActivations)

	_, err = kernel.Complete(waiting.State, Completion{
		ActivationID: waiting.Intents[0].ActivationID,
		Port:         "completed",
		Output:       Values{"reply": "done"},
	}, reducerNow)
	require.ErrorContains(t, err, "cannot exceed 10000 activations")
	require.Equal(t, ActivationWaiting, waiting.State.Activations[0].Status, "failed reduction must not mutate its input state")
}

func TestForkOpensParallelJoinWithoutTimeoutIntent(t *testing.T) {
	t.Parallel()
	kernel, _ := goldenKernel(t, "fork_join_all.json")
	state, err := kernel.Start(Values{"value": "ok"})
	require.NoError(t, err)
	transition, err := kernel.Step(state, reducerNow)
	require.NoError(t, err)
	require.Empty(t, transition.Intents, "fork must not arm a join timeout intent")
	require.Len(t, transition.State.Scopes, 1)
	parallel := transition.State.Scopes[0].Parallel
	require.NotNil(t, parallel)
	join, err := transition.State.activation(parallel.JoinActivationID)
	require.NoError(t, err)
	require.NotEmpty(t, join.ID)
}

func stepUntilTerminal(t *testing.T, kernel *Kernel, state *State, limit int) *State {
	t.Helper()
	for iteration := 0; iteration < limit && !state.Status.Terminal(); iteration++ {
		transition, err := kernel.Step(state, reducerNow.Add(time.Duration(iteration)*time.Second))
		require.NoError(t, err)
		state = transition.State
		for _, intent := range transition.Intents {
			var port string
			var output Values
			switch intent.Kind {
			case IntentSendMessage:
				port, output = "completed", Values{}
				if intent.Callback {
					for _, field := range intent.CallbackOutput {
						switch field.Schema.Type {
						case plan.TypeString, plan.TypeChannel, plan.TypeMember, plan.TypeAgent, plan.TypeEnum:
							output[field.Name] = "ok"
						}
					}
				}
			case IntentWait:
				port, output = "completed", Values{}
			case IntentApproval:
				port, output = "approved", Values{}
			}
			if port != "" {
				completion, completeErr := kernel.Complete(state, Completion{ActivationID: intent.ActivationID, Port: port, Output: output}, reducerNow)
				require.NoError(t, completeErr)
				state = completion.State
			}
		}
	}
	require.True(t, state.Status.Terminal(), "Run did not terminate: %#v", state)
	return state
}

func branchByPort(t *testing.T, branches []BranchState, port string) *BranchState {
	t.Helper()
	for index := range branches {
		if branches[index].Port == port {
			return &branches[index]
		}
	}
	t.Fatalf("missing branch %s", port)
	return nil
}

func dynamicStringTemplate(t *testing.T, value string) plan.ArtifactTemplate {
	t.Helper()
	literal, err := json.Marshal(value)
	require.NoError(t, err)
	expression := plan.ArtifactExpression{
		Op:         plan.ArtifactOpConst,
		ResultType: plan.TypeString,
		Value:      plan.ArtifactLiteral(literal),
	}
	return plan.ArtifactTemplate{Segments: []plan.ArtifactTemplateSegment{{
		Kind:       plan.ArtifactTemplateSegmentExpr,
		Expression: &expression,
	}}}
}

func eventKinds(events []Event) []string {
	kinds := make([]string, len(events))
	for index := range events {
		kinds[index] = events[index].Kind
	}
	return kinds
}