package run

import (
	"testing"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
)

func TestFanOutTaskWatchFrontierReleasesSourceAfterRunningSuccessorAndLaterBlockedSibling(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRuntimePlan)
	sharedTask := taskWatchScopeObject("tsk_shared")
	items := []any{
		workItemRaw("source", nil, "@dev1"),
		workItemRaw("blocker", nil, "@dev2"),
		workItemRaw("viable", []string{"source"}, "@dev1"),
		workItemRaw("doomed", []string{"source", "blocker"}, "@dev2"),
	}
	for _, raw := range items {
		raw.(map[string]any)[plan.WorkItemFieldWorkScope] = sharedTask
	}
	state := startFanOut(t, kernel, items)

	sourceIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "source", activationItemKey(t, state, sourceIntent.ActivationID))
	state = observeTaskWatchResponder(t, kernel, state, sourceIntent, "agt_source")
	blockerIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "blocker", activationItemKey(t, state, blockerIntent.ActivationID))

	state = completeTaskWatchCallback(
		t, kernel, state, sourceIntent, "completed", "agt_source", Values{"answer": "source"},
	)
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemRunning, fanOutItemByKey(scope.FanOut, "viable").Status)
	require.Equal(t, WorkItemPending, fanOutItemByKey(scope.FanOut, "doomed").Status)
	require.NotEmpty(
		t,
		fanOutItemByKey(scope.FanOut, "source").TaskWatchFrontier,
		"source remains a carrier until its other successor becomes terminal",
	)

	state = completeIntent(
		t,
		kernel,
		state,
		blockerIntent,
		"failed",
		Values{},
		&Failure{Code: "blocker.failed"},
	)
	scope = onlyFanOutScope(t, state)
	require.Equal(t, WorkItemBlocked, fanOutItemByKey(scope.FanOut, "doomed").Status)
	require.Empty(
		t,
		fanOutItemByKey(scope.FanOut, "source").TaskWatchFrontier,
		"the block transition must immediately release a carrier already copied to every viable successor",
	)

	viableIntent, _ := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "viable", activationItemKey(t, state, viableIntent.ActivationID))
	requireTaskWatchHandoff(t, viableIntent, "tsk_shared", []string{"agt_source"})
}

func TestFanOutTaskWatchFrontierReleasesSourceAfterNeverStartedSuccessorIsCanceled(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRuntimePlan)
	sharedTask := taskWatchScopeObject("tsk_shared")
	items := []any{
		workItemRaw("source", nil, "@dev1"),
		workItemRaw("slot_holder", nil, "@dev2"),
		workItemRaw("viable", []string{"source"}, "@dev1"),
		workItemRaw("canceled", []string{"source"}, "@dev2"),
	}
	for _, raw := range items {
		raw.(map[string]any)[plan.WorkItemFieldWorkScope] = sharedTask
	}
	state := startFanOut(t, kernel, items)

	sourceIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	state = observeTaskWatchResponder(t, kernel, state, sourceIntent, "agt_source")
	slotHolderIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	state = observeTaskWatchResponder(t, kernel, state, slotHolderIntent, "agt_holder")
	state = completeTaskWatchCallback(
		t, kernel, state, sourceIntent, "completed", "agt_source", Values{"answer": "source"},
	)

	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemRunning, fanOutItemByKey(scope.FanOut, "viable").Status)
	canceled := fanOutItemByKey(scope.FanOut, "canceled")
	require.Equal(t, WorkItemPending, canceled.Status)
	canceled.Status = WorkItemCanceled
	canceled.Failure = &Failure{Code: "test.canceled"}
	require.NoError(t, releaseFanOutInputTaskWatchFrontiers(state, scope.ID))
	scope = onlyFanOutScope(t, state)
	require.Empty(t, fanOutItemByKey(scope.FanOut, "source").TaskWatchFrontier)

	viableIntent, _ := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "viable", activationItemKey(t, state, viableIntent.ActivationID))
	requireTaskWatchHandoff(t, viableIntent, "tsk_shared", []string{"agt_source"})
}

func TestFanOutTaskWatchFrontierKeepsDependencyWhenEverySuccessorTerminatesBeforeCopy(t *testing.T) {
	t.Parallel()
	state := terminalFanOutTaskWatchState([]WorkItemState{
		{
			ID:                "wfi_dependency",
			Key:               "dependency",
			Status:            WorkItemSucceeded,
			TaskWatchFrontier: []string{"act_last_handler"},
		},
		{
			ID:        "wfi_blocked",
			Key:       "blocked",
			Status:    WorkItemBlocked,
			DependsOn: []string{"dependency"},
			BlockedBy: []string{"missing_gate"},
		},
		{
			ID:        "wfi_canceled",
			Key:       "canceled",
			Status:    WorkItemCanceled,
			DependsOn: []string{"dependency"},
			Failure:   &Failure{Code: "test.canceled"},
		},
	}, nil)

	require.NoError(t, releaseFanOutInputTaskWatchFrontiers(state, "scp_fan_out"))
	scope, err := state.scope("scp_fan_out")
	require.NoError(t, err)
	require.Equal(
		t,
		[]string{"act_last_handler"},
		fanOutItemByKey(scope.FanOut, "dependency").TaskWatchFrontier,
		"terminal successors that never received a frontier copy cannot retire the final handler",
	)

	transition := &Transition{State: state}
	require.NoError(t, (&Kernel{}).resolveFanOut(transition, "scp_fan_out"))
	join, err := transition.State.activation("act_join")
	require.NoError(t, err)
	require.Equal(t, ActivationReady, join.Status)
	require.Equal(t, []string{"act_last_handler"}, join.TaskWatchFrontier)
	requireTaskWatchFrontierAgents(t, transition.State, []string{"agt_last_handler"})
}

func TestFanOutTaskWatchFrontierKeepsOwnerWhenEveryRootTerminatesBeforeCopy(t *testing.T) {
	t.Parallel()
	state := terminalFanOutTaskWatchState([]WorkItemState{
		{
			ID:        "wfi_blocked_root",
			Key:       "blocked_root",
			Status:    WorkItemBlocked,
			BlockedBy: []string{"missing_gate"},
		},
		{
			ID:      "wfi_canceled_root",
			Key:     "canceled_root",
			Status:  WorkItemCanceled,
			Failure: &Failure{Code: "test.canceled"},
		},
	}, []string{"act_last_handler"})

	require.NoError(t, releaseFanOutInputTaskWatchFrontiers(state, "scp_fan_out"))
	owner, err := state.activation("act_owner")
	require.NoError(t, err)
	require.Equal(
		t,
		[]string{"act_last_handler"},
		owner.TaskWatchFrontier,
		"all never-started terminal roots leave the owner as the only live carrier",
	)

	transition := &Transition{State: state}
	require.NoError(t, (&Kernel{}).resolveFanOut(transition, "scp_fan_out"))
	join, err := transition.State.activation("act_join")
	require.NoError(t, err)
	require.Equal(t, ActivationReady, join.Status)
	require.Equal(t, []string{"act_last_handler"}, join.TaskWatchFrontier)
	requireTaskWatchFrontierAgents(t, transition.State, []string{"agt_last_handler"})
}

func terminalFanOutTaskWatchState(items []WorkItemState, ownerFrontier []string) *State {
	taskScope := domain.MessageScopeValue{
		Kind:    domain.MessageScopeKindTask,
		ScopeID: "tsk_shared",
	}
	return &State{
		Status: StatusRunning,
		Activations: []Activation{
			{
				ID:       "act_last_handler",
				Sequence: 1,
				Status:   ActivationSucceeded,
				CallbackTaskWatch: &CallbackTaskWatchState{
					TaskScope:        taskScope,
					ResponderRef:     "@last-handler",
					ResponderAgentID: "agt_last_handler",
				},
			},
			{
				ID:                "act_owner",
				Sequence:          2,
				Status:            ActivationSucceeded,
				TaskWatchFrontier: cloneTaskWatchFrontier(ownerFrontier),
			},
			{
				ID:       "act_join",
				Sequence: 3,
				Status:   ActivationWaiting,
			},
		},
		Scopes: []Scope{{
			ID:     "scp_fan_out",
			Status: ScopeActive,
			FanOut: &FanOutState{
				OwnerActivationID: "act_owner",
				JoinActivationID:  "act_join",
				Items:             items,
			},
		}},
	}
}

func TestFanOutTaskWatchFrontierFailedPreDeliveryProjectionPreservesUncoveredBranch(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRuntimePlan)
	sharedTask := taskWatchScopeObject("tsk_shared")
	items := []any{
		workItemRaw("root", nil, "@dev1"),
		workItemRaw("fails_before_delivery", []string{"root"}, "@dev2"),
		workItemRaw("sibling", []string{"root"}, "@dev1"),
	}
	for _, raw := range items {
		raw.(map[string]any)[plan.WorkItemFieldWorkScope] = sharedTask
	}
	state := startFanOut(t, kernel, items)

	rootIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	state = observeTaskWatchResponder(t, kernel, state, rootIntent, "agt_root")
	state = completeTaskWatchCallback(
		t, kernel, state, rootIntent, "completed", "agt_root", Values{"answer": "root"},
	)

	failedIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "fails_before_delivery", activationItemKey(t, state, failedIntent.ActivationID))
	state = completeIntent(
		t,
		kernel,
		state,
		failedIntent,
		"failed",
		Values{},
		&Failure{Code: "delivery.failed"},
	)
	scope := onlyFanOutScope(t, state)
	require.Equal(t, WorkItemFailed, fanOutItemByKey(scope.FanOut, "fails_before_delivery").Status)

	siblingIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "sibling", activationItemKey(t, state, siblingIntent.ActivationID))
	requireTaskWatchHandoff(
		t,
		siblingIntent,
		"tsk_shared",
		[]string{},
	)

	// The failed branch never committed a successor Task message, so its
	// restored root handler remains the final handler for that causal leaf. A
	// sibling callback replaces only its own copy of the root frontier and must
	// not unwatch the uncovered branch's Agent.
	state = observeTaskWatchResponder(t, kernel, state, siblingIntent, "agt_sibling")
	state = completeTaskWatchCallback(
		t, kernel, state, siblingIntent, "completed", "agt_sibling", Values{"answer": "sibling"},
	)
	requireTaskWatchFrontierAgents(t, state, []string{"agt_root", "agt_sibling"})
}

func TestFanOutTaskWatchFrontierDeliveredSiblingThenPreDeliveryFailurePreservesUncoveredBranch(t *testing.T) {
	t.Parallel()
	kernel := compilePlanKernel(t, fanOutRuntimePlan)
	sharedTask := taskWatchScopeObject("tsk_shared")
	items := []any{
		workItemRaw("root", nil, "@dev1"),
		workItemRaw("delivered", []string{"root"}, "@dev2"),
		workItemRaw("fails_before_delivery", []string{"root"}, "@dev1"),
	}
	for _, raw := range items {
		raw.(map[string]any)[plan.WorkItemFieldWorkScope] = sharedTask
	}
	state := startFanOut(t, kernel, items)

	rootIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	state = observeTaskWatchResponder(t, kernel, state, rootIntent, "agt_root")
	state = completeTaskWatchCallback(
		t, kernel, state, rootIntent, "completed", "agt_root", Values{"answer": "root"},
	)

	deliveredIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "delivered", activationItemKey(t, state, deliveredIntent.ActivationID))
	requireTaskWatchHandoff(t, deliveredIntent, "tsk_shared", []string{})
	state = observeTaskWatchResponder(t, kernel, state, deliveredIntent, "agt_delivered")

	failedIntent, state := stepToNextTaskWatchCallback(t, kernel, state)
	require.Equal(t, "fails_before_delivery", activationItemKey(t, state, failedIntent.ActivationID))
	state = completeIntent(
		t,
		kernel,
		state,
		failedIntent,
		"failed",
		Values{},
		&Failure{Code: "delivery.failed"},
	)
	state = completeTaskWatchCallback(
		t, kernel, state, deliveredIntent, "completed", "agt_delivered", Values{"answer": "delivered"},
	)

	// Reversing the delivery/failure order produces the same joined frontier:
	// no Task transaction ever installed a successor for the failed branch, so
	// the root Agent remains watched alongside the delivered sibling Agent.
	requireTaskWatchFrontierAgents(t, state, []string{"agt_root", "agt_delivered"})
}