package run

import (
	"encoding/json"
	"fmt"
	"reflect"
	"strconv"
	"strings"
	"time"

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

func (kernel *Kernel) resolveFanOut(transition *Transition, scopeID string) error {
	scope, err := transition.State.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("fan-out scope %q is invalid", scopeID)
	}
	if scope.Status != ScopeActive {
		return nil
	}
	if !fanOutItemsTerminal(scope.FanOut) {
		return nil
	}
	if err := collapseScopeTaskWatchFrontiers(
		transition.State,
		scope.ID,
		scope.FanOut.JoinActivationID,
		scope.FanOut.OwnerActivationID,
	); err != nil {
		return err
	}
	scope, _ = transition.State.scope(scopeID)
	join, err := transition.State.activation(scope.FanOut.JoinActivationID)
	if err != nil {
		return err
	}
	if scope.FanOut.ResourceFailure {
		scope.Status = ScopeFailed
		scope.FanOut.Outcome = "failed"
		join.Failure = &Failure{
			Code:    workItemResourceLimitFailureCode,
			Message: workItemResourceLimitMessage,
			Stage:   "work_item",
			Problem: FailureProblem(workItemResourceLimitFailureCode),
			Hint:    FailureHint(workItemResourceLimitFailureCode),
		}
	} else {
		scope.Status = ScopeSucceeded
		scope.FanOut.Outcome = "completed"
	}
	if join.Status == ActivationWaiting {
		join.Status = ActivationReady
	}
	transition.Events = append(transition.Events, Event{
		Kind: "fan_out_resolved", ScopeID: scope.ID, Port: scope.FanOut.Outcome,
	})
	return nil
}

func fanOutItemByID(scope *Scope, id string) *WorkItemState {
	if scope == nil || scope.FanOut == nil {
		return nil
	}
	for index := range scope.FanOut.Items {
		if scope.FanOut.Items[index].ID == id {
			return &scope.FanOut.Items[index]
		}
	}
	return nil
}

func fanOutItemsTerminal(fanOut *FanOutState) bool {
	for index := range fanOut.Items {
		if !fanOut.Items[index].Status.Terminal() {
			return false
		}
	}
	return true
}

func fanOutItemByKey(fanOut *FanOutState, key string) *WorkItemState {
	if fanOut == nil {
		return nil
	}
	for index := range fanOut.Items {
		if fanOut.Items[index].Key == key {
			return &fanOut.Items[index]
		}
	}
	return nil
}

func fanOutItemValues(fanOut *FanOutState, item WorkItemState) Values {
	responder := any(item.Responder)
	if len(item.Responders) > 0 {
		responderObject := map[string]any{"selectOne": stringsToAny(item.Responders)}
		if strategy := item.SelectOneStrategy; strategy != "" {
			// Render the additive strategy back so the item body's
			// input.responder carries it losslessly to the send_message intent.
			responderObject["strategy"] = string(strategy)
		}
		responder = responderObject
	}
	return Values{
		plan.WorkItemFieldKey:            item.Key,
		plan.WorkItemFieldDependsOn:      stringsToAny(item.DependsOn),
		plan.WorkItemFieldWorkScope:      messageScopeObject(item.WorkScope),
		plan.WorkItemFieldRelatedScopes:  messageScopesToAny(item.RelatedScopes),
		plan.WorkItemFieldResponder:      responder,
		plan.WorkItemFieldInstruction:    item.Instruction,
		plan.FanOutItemFieldDependencies: fanOutItemDependencyView(fanOut, &item),
	}
}

func fanOutItemEnvelope(fanOut *FanOutState, item WorkItemState) map[string]any {
	envelope := fanOutTerminalEnvelope(item)
	envelope[plan.FanOutItemFieldDependencies] = map[string]any(fanOutItemDependencyView(fanOut, &item))
	return envelope
}

func fanOutItemDependencyView(fanOut *FanOutState, item *WorkItemState) Values {
	if item == nil || item.Status == WorkItemPending || item.Status == WorkItemBlocked ||
		item.Failure != nil && item.Failure.Code == workItemResourceLimitFailureCode {
		return Values{}
	}
	return fanOutDependencyView(fanOut, item)
}

// fanOutTerminalEnvelope deliberately omits dependencies. It is used as the
// value of item.dependencies so dependency envelopes never recursively embed
// their own dependency graphs.
func fanOutTerminalEnvelope(item WorkItemState) map[string]any {
	var failure any
	if item.Failure != nil {
		failure = map[string]any{"code": item.Failure.Code}
		if strings.TrimSpace(item.Failure.Problem) != "" {
			failure.(map[string]any)["problem"] = item.Failure.Problem
		} else if strings.TrimSpace(FailureProblem(item.Failure.Code)) != "" {
			failure.(map[string]any)["problem"] = FailureProblem(item.Failure.Code)
		}
		if strings.TrimSpace(item.Failure.Hint) != "" {
			failure.(map[string]any)["hint"] = item.Failure.Hint
		} else if strings.TrimSpace(FailureHint(item.Failure.Code)) != "" {
			failure.(map[string]any)["hint"] = FailureHint(item.Failure.Code)
		}
		if strings.TrimSpace(item.Failure.Stage) != "" {
			failure.(map[string]any)["stage"] = item.Failure.Stage
		}
		if strings.TrimSpace(item.Failure.Path) != "" {
			failure.(map[string]any)["path"] = item.Failure.Path
		}
		if item.Failure.Retryable {
			failure.(map[string]any)["retryable"] = true
		}
		if item.Failure.WorkItem != nil {
			workItem := map[string]any{}
			if strings.TrimSpace(item.Failure.WorkItem.Key) != "" {
				workItem["key"] = item.Failure.WorkItem.Key
			}
			if item.Failure.WorkItem.Index != nil {
				workItem["index"] = json.Number(strconv.Itoa(*item.Failure.WorkItem.Index))
			}
			if strings.TrimSpace(item.Failure.WorkItem.DependencyPath) != "" {
				workItem["dependencyPath"] = item.Failure.WorkItem.DependencyPath
			}
			failure.(map[string]any)["workItem"] = workItem
		}
	}
	var value any
	if item.Value != nil {
		// Terminal values are immutable. Projection callers either evaluate them
		// read-only or normalize into the one persistence destination.
		value = map[string]any(item.Value)
	}
	return map[string]any{
		"id":            item.ID,
		"key":           item.Key,
		"index":         json.Number(strconv.Itoa(item.Index)),
		"status":        string(item.Status),
		"dependsOn":     stringsToAny(item.DependsOn),
		"blockedBy":     stringsToAny(item.BlockedBy),
		"responder":     item.Responder,
		"workScope":     messageScopeObject(item.WorkScope),
		"relatedScopes": messageScopesToAny(item.RelatedScopes),
		"instruction":   item.Instruction,
		"value":         value,
		"failure":       failure,
	}
}

func fanOutResultEnvelopes(scope *Scope) []any {
	if scope == nil || scope.FanOut == nil {
		return nil
	}
	result := make([]any, len(scope.FanOut.Items))
	for index, item := range scope.FanOut.Items {
		result[index] = fanOutItemEnvelope(scope.FanOut, item)
	}
	return result
}

// validateFanOutItemCandidate admits the complete durable candidate, not just
// the raw items array. The authored closing join may wrap the array in an
// object, select multiple fields, or bind the same dependency view more than
// once; evaluating it with borrowed loads accounts for that exact expanded
// output before normalization performs one owned materialization. A projected
// finish also includes the join output and downstream activation input in the
// candidate Run state, so ordinary routing cannot defer aggregate rejection to
// ProjectTurn/Store.
func (kernel *Kernel) validateFanOutItemCandidate(
	path string,
	state *State,
	scopeID string,
	forceItemID string,
	admitState bool,
) error {
	scope, err := state.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("%s has invalid fan-out scope %q", path, scopeID)
	}
	join, err := state.activation(scope.FanOut.JoinActivationID)
	if err != nil {
		return err
	}
	node, ok := kernel.graph.Node(join.NodeID)
	if !ok || node.Config.Join == nil {
		return fmt.Errorf("%s has invalid fan-out join %q", path, join.NodeID)
	}
	context, err := kernel.evalContext(state, join)
	if err != nil {
		return err
	}
	result, err := evalFanOutEntryBindings(
		path+" join.merge",
		node.Config.Join.Merge,
		node.Config.Join.Result,
		context,
	)
	var mergeResourceErr error
	if err != nil {
		if isFanOutResourceLimit(err) {
			mergeResourceErr = err
		} else {
			// An authored merge may legitimately require a terminal-only field.
			// Its semantic error is handled by the real join once the scope
			// resolves; this projection is exclusively a resource boundary.
			if admitState {
				if stateErr := kernel.validateFanOutStateAdmission(path, state, forceItemID); stateErr != nil {
					return stateErr
				}
			}
		}
	} else if payloadErr := validateFanOutPayloadBudget(path+" join output", result); payloadErr != nil {
		mergeResourceErr = payloadErr
	} else if admitState {
		if stateErr := kernel.validateFanOutStateAdmission(path, state, forceItemID); stateErr != nil {
			return stateErr
		}
	}
	allTerminal := fanOutItemsTerminal(scope.FanOut)
	if mergeResourceErr == nil && err == nil && allTerminal {
		outcome, projectionErr := kernel.projectFanOutClosingSettlement(
			path+" closing candidate",
			state,
			scope.ID,
			"",
			false,
			false,
		)
		if projectionErr != nil {
			if !isFanOutResourceLimit(projectionErr) {
				// Dynamic binding/semantic failures are handled by the real node's
				// compiled failed path. They do not justify changing WorkItem truth.
				return nil
			}
			compactOutcome, compactErr := kernel.projectFanOutClosingSettlement(
				path+" current-item compact settlement",
				state,
				scope.ID,
				forceItemID,
				true,
				false,
			)
			if compactErr == nil && compactOutcome == fanOutClosingProjectionCompleted {
				return projectionErr
			}
			if compactErr != nil && !isFanOutResourceLimit(compactErr) {
				return compactErr
			}
			return kernel.markFanOutClosingResourceFailure(path, state, scope.ID)
		}
		if outcome == fanOutClosingProjectionFailed {
			// Semantic join failures already route through the compiled failed
			// port. They are recoverable reducer outcomes, not resource leaks.
			return nil
		}
	}

	// A candidate is admitted only when its deterministic all-resource-failed
	// settlement can also be persisted after the real resolveFanOut mutation,
	// closing merge, and downstream routes. This exact projection replaces the
	// old fixed reserve and includes status/outcome plus Task Watch collapse.
	fallbackOutcome, fallbackErr := kernel.projectFanOutClosingSettlement(
		path+" compact settlement",
		state,
		scope.ID,
		forceItemID,
		true,
		false,
	)
	if fallbackErr != nil {
		return fallbackErr
	}
	if mergeResourceErr != nil {
		// If compacting the candidate makes the authored merge fit, the
		// oversized contribution is attributable to this WorkItem and the
		// caller must fail it locally. If even the compact terminal envelopes
		// cannot satisfy the static merge, no WorkItem value can repair it;
		// let the real join take its deterministic recoverable failed port.
		if fallbackOutcome == fanOutClosingProjectionCompleted {
			return mergeResourceErr
		}
		if allTerminal {
			return kernel.markFanOutClosingResourceFailure(path, state, scope.ID)
		}
		return nil
	}
	return nil
}

func (kernel *Kernel) markFanOutClosingResourceFailure(path string, state *State, scopeID string) error {
	if _, projectionErr := kernel.projectFanOutClosingSettlement(
		path+" recoverable resource-failed settlement",
		state,
		scopeID,
		"",
		false,
		true,
	); projectionErr != nil {
		return projectionErr
	}
	scope, err := state.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("%s has invalid fan-out scope %q", path, scopeID)
	}
	scope.FanOut.ResourceFailure = true
	return validateFanOutHardStateBudget(path+" resource-failure marker", state)
}

type fanOutClosingProjectionOutcome uint8

const (
	fanOutClosingProjectionCompleted fanOutClosingProjectionOutcome = iota
	fanOutClosingProjectionFailed
)

// projectFanOutClosingSettlement mirrors the real terminal path on an isolated
// State: optional compact WorkItem fallback, resolveFanOut (including scope
// outcome and Task Watch collapse), authored join.merge, and every downstream
// route. A resource-invalid static merge is projected through the join's
// compiled failed port, which is recoverable even when no individual item
// value can make the authored completed output fit.
func (kernel *Kernel) projectFanOutClosingSettlement(
	path string,
	state *State,
	scopeID string,
	forceItemID string,
	compactLive bool,
	forceResourceFailure bool,
) (fanOutClosingProjectionOutcome, error) {
	projected := &Transition{State: state.Clone(), Intents: []Intent{}, Events: []Event{}}
	scope, err := projected.State.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fanOutClosingProjectionFailed, fmt.Errorf("%s has invalid fan-out scope %q", path, scopeID)
	}
	if forceResourceFailure {
		scope.FanOut.ResourceFailure = true
	}
	if compactLive {
		kernel.compactFanOutClosingProjection(projected.State, scopeID, forceItemID)
	}
	for index := range scope.FanOut.Items {
		if !scope.FanOut.Items[index].Status.Terminal() {
			return fanOutClosingProjectionFailed, fmt.Errorf(
				"%s leaves WorkItem %q non-terminal",
				path,
				scope.FanOut.Items[index].Key,
			)
		}
	}
	if err := kernel.resolveFanOut(projected, scopeID); err != nil {
		return fanOutClosingProjectionFailed, err
	}
	scope, _ = projected.State.scope(scopeID)
	join, err := projected.State.activation(scope.FanOut.JoinActivationID)
	if err != nil {
		return fanOutClosingProjectionFailed, err
	}
	if scope.FanOut.Outcome != "completed" {
		firstDescendant := len(projected.State.Activations)
		if finishErr := kernel.finishPortsInPlace(
			projected,
			join.ID,
			[]portValue{{Port: scope.FanOut.Outcome, Value: Values{}}},
		); finishErr != nil {
			return fanOutClosingProjectionFailed, finishErr
		}
		if closureErr := kernel.projectFanOutPureDescendantClosure(
			path+" resource-failed descendants",
			projected,
			firstDescendant,
		); closureErr != nil {
			return fanOutClosingProjectionFailed, closureErr
		}
		return fanOutClosingProjectionFailed, nil
	}
	node, ok := kernel.graph.Node(join.NodeID)
	if !ok || node.Config.Join == nil {
		return fanOutClosingProjectionFailed, fmt.Errorf("%s has invalid fan-out join %q", path, join.NodeID)
	}
	context, err := kernel.evalContext(projected.State, join)
	if err != nil {
		return fanOutClosingProjectionFailed, err
	}
	result, mergeErr := evalFanOutEntryBindings(
		path+" join.merge",
		node.Config.Join.Merge,
		node.Config.Join.Result,
		context,
	)
	if mergeErr != nil {
		if isFanOutResourceLimit(mergeErr) {
			join.Failure = &Failure{
				Code:    workItemResourceLimitFailureCode,
				Message: workItemResourceLimitMessage,
				Stage:   "work_item",
				Problem: FailureProblem(workItemResourceLimitFailureCode),
				Hint:    FailureHint(workItemResourceLimitFailureCode),
			}
		} else {
			join.Failure = failureFromCause("join.merge_failed", "work_item", mergeErr)
		}
		firstDescendant := len(projected.State.Activations)
		if finishErr := kernel.finishPortsInPlace(
			projected,
			join.ID,
			[]portValue{{Port: "failed", Value: Values{}}},
		); finishErr != nil {
			return fanOutClosingProjectionFailed, finishErr
		}
		if closureErr := kernel.projectFanOutPureDescendantClosure(
			path+" failed descendants",
			projected,
			firstDescendant,
		); closureErr != nil {
			return fanOutClosingProjectionFailed, closureErr
		}
		return fanOutClosingProjectionFailed, nil
	}
	firstDescendant := len(projected.State.Activations)
	if finishErr := kernel.finishPortsInPlace(
		projected,
		join.ID,
		[]portValue{{Port: "completed", Value: result}},
	); finishErr != nil {
		return fanOutClosingProjectionCompleted, finishErr
	}
	if closureErr := kernel.projectFanOutPureDescendantClosure(
		path+" completed descendants",
		projected,
		firstDescendant,
	); closureErr != nil {
		return fanOutClosingProjectionCompleted, closureErr
	}
	return fanOutClosingProjectionCompleted, nil
}

func (kernel *Kernel) compactFanOutClosingProjection(state *State, scopeID, forceItemID string) {
	scope, _ := state.scope(scopeID)
	compactIDs := make([]string, 0, len(scope.FanOut.Items))
	for index := range scope.FanOut.Items {
		item := &scope.FanOut.Items[index]
		if item.ID != forceItemID && item.Status.Terminal() {
			continue
		}
		compactIDs = append(compactIDs, item.ID)
	}
	for _, itemID := range compactIDs {
		scope, _ = state.scope(scopeID)
		item := fanOutItemByID(scope, itemID)
		kernel.compactFanOutItemResourceState(state, scope.ID, item)
	}
}

// projectFanOutPureDescendantClosure follows deterministic successors after
// the closing join until no activation is immediately reducible. External
// effects are reduced to their durable Waiting state (their intents are kept
// only on this isolated projection), and nested fan-out owners are reduced
// through opening/scheduling. This accounts for status, owned-scope, callback
// Task Watch, and other state deltas that otherwise appear only on the next
// Step. Parallel joins can become ready after their lower-index activation was
// first scanned, so the traversal repeats until no new activation runs.
func (kernel *Kernel) projectFanOutPureDescendantClosure(
	path string,
	transition *Transition,
	firstActivation int,
) error {
	if transition == nil || transition.State == nil {
		return fmt.Errorf("%s has invalid projected transition", path)
	}
	for {
		progressed := false
		for index := firstActivation; index < len(transition.State.Activations); index++ {
			activation := &transition.State.Activations[index]
			if activation.Status != ActivationReady {
				continue
			}
			if _, ok := kernel.graph.Node(activation.NodeID); !ok {
				return fmt.Errorf("%s activation %q has no node", path, activation.ID)
			}
			activationID := activation.ID
			firstIntent := len(transition.Intents)
			if err := kernel.execute(transition, activationID, time.Unix(0, 0).UTC()); err != nil {
				return err
			}
			if err := validateFanOutIntentRange(path, transition.Intents, firstIntent); err != nil {
				return err
			}
			progressed = true
			if err := validateFanOutHardStateBudget(path, transition.State); err != nil {
				return err
			}
			if transition.State.Status.Terminal() {
				return nil
			}
		}
		if !progressed {
			return validateFanOutHardStateBudget(path, transition.State)
		}
	}
}

func cancelFanOutItemWork(state *State, scopeID, itemID, code string) {
	for index := range state.Activations {
		activation := &state.Activations[index]
		if activation.Status != ActivationReady && activation.Status != ActivationWaiting {
			continue
		}
		for _, reference := range activation.Lineage {
			if reference.ScopeID == scopeID && reference.ItemID == itemID {
				activation.Status = ActivationCanceled
				activation.Failure = &Failure{Code: code}
				break
			}
		}
	}
	for index := range state.Scopes {
		nested := &state.Scopes[index]
		if nested.ID == scopeID || nested.Status != ScopeActive {
			continue
		}
		for _, reference := range nested.ParentLineage {
			if reference.ScopeID == scopeID && reference.ItemID == itemID {
				nested.Status = ScopeCanceled
				cancelFanOutItems(nested.FanOut, code)
				break
			}
		}
	}
}

// validateFanOutState fails closed on corrupt or partially persisted dynamic
// state before the reducer can schedule duplicate work after recovery.
func (kernel *Kernel) validateFanOutState(state *State) error {
	if state == nil {
		return fmt.Errorf("Run state is nil")
	}
	for scopeIndex := range state.Scopes {
		scope := &state.Scopes[scopeIndex]
		if scope.FanOut == nil {
			if scope.Kind == plan.ArtifactScopeFanOut {
				return fmt.Errorf("scope %q is fan_out without fanOut state", scope.ID)
			}
			continue
		}
		if err := kernel.validateFanOutScopeState(state, scope); err != nil {
			return err
		}
	}
	return validateFanOutActivationLineage(state)
}

func (kernel *Kernel) validateFanOutScopeState(state *State, scope *Scope) error {
	if scope == nil || scope.FanOut == nil {
		return fmt.Errorf("fan-out scope state is nil")
	}
	if scope.Kind != plan.ArtifactScopeFanOut {
		return fmt.Errorf("scope %q has fanOut state with kind %q", scope.ID, scope.Kind)
	}
	artifactScope, ok := kernel.graph.Scope(plan.ArtifactScopeFanOut, scope.OwnerID)
	if !ok || artifactScope.FanOut == nil || artifactScope.FanOut.JoinID != scope.FanOut.JoinNodeID {
		return fmt.Errorf("scope %q does not match its compiled fan-out scope", scope.ID)
	}
	owner, err := state.activation(scope.FanOut.OwnerActivationID)
	if err != nil || owner.NodeID != scope.OwnerID {
		return fmt.Errorf("scope %q has invalid owner activation", scope.ID)
	}
	if owner.Status != ActivationSucceeded ||
		len(owner.SelectedPorts) != 1 || owner.SelectedPorts[0] != "item" ||
		!containsString(owner.OwnedScopeIDs, scope.ID) {
		return fmt.Errorf("scope %q owner activation does not own an opened fan-out", scope.ID)
	}
	join, err := state.activation(scope.FanOut.JoinActivationID)
	if err != nil || join.NodeID != scope.FanOut.JoinNodeID ||
		join.TriggerScopeID != scope.ID || join.ParentActivationID != owner.ID ||
		!reflect.DeepEqual(join.Lineage, owner.Lineage) {
		return fmt.Errorf("scope %q has invalid join activation", scope.ID)
	}
	if scope.FanOut.MaxConcurrency < 1 || scope.FanOut.MaxConcurrency > plan.MaxFanOutConcurrency {
		return fmt.Errorf("scope %q has invalid maxConcurrency %d", scope.ID, scope.FanOut.MaxConcurrency)
	}
	if scope.FanOut.FailureMode != plan.FanOutFailureCollect {
		return fmt.Errorf("scope %q has unsupported failureMode %q", scope.ID, scope.FanOut.FailureMode)
	}
	if len(scope.FanOut.Items) > plan.MaxWorkItems {
		return fmt.Errorf("scope %q exceeds the %d WorkItem limit", scope.ID, plan.MaxWorkItems)
	}
	itemKeys, running, allTerminal, frozen, err := validateFanOutItems(scope)
	if err != nil {
		return err
	}
	if running > scope.FanOut.MaxConcurrency {
		return fmt.Errorf(
			"scope %q has %d running WorkItems above maxConcurrency %d",
			scope.ID,
			running,
			scope.FanOut.MaxConcurrency,
		)
	}
	if _, err := NormalizeWorkItems("fan_out.state.items", WorkItemsValue(frozen), nil); err != nil {
		return fmt.Errorf("scope %q has invalid frozen WorkItems: %w", scope.ID, err)
	}
	if err := validateFanOutItemExecutions(state, scope, owner, itemKeys); err != nil {
		return err
	}
	return validateFanOutScopeOutcome(state, scope, join, allTerminal)
}

func validateFanOutItems(scope *Scope) (map[string]*WorkItemState, int, bool, []WorkItem, error) {
	frozen := make([]WorkItem, len(scope.FanOut.Items))
	itemIDs := make(map[string]struct{}, len(scope.FanOut.Items))
	itemKeys := make(map[string]*WorkItemState, len(scope.FanOut.Items))
	running := 0
	allTerminal := true
	for itemIndex := range scope.FanOut.Items {
		item := &scope.FanOut.Items[itemIndex]
		if item.DependsOn == nil || item.BlockedBy == nil ||
			item.RelatedScopes == nil ||
			item.ActivatedResponderAgentIDs == nil {
			return nil, 0, false, nil, fmt.Errorf(
				"scope %q item %q has non-canonical null collection state",
				scope.ID,
				item.Key,
			)
		}
		if item.Index != itemIndex {
			return nil, 0, false, nil, fmt.Errorf(
				"scope %q item %q has index %d, want %d",
				scope.ID,
				item.Key,
				item.Index,
				itemIndex,
			)
		}
		if item.ID != WorkItemSemanticID(scope.FanOut.OwnerActivationID, item.Key) {
			return nil, 0, false, nil, fmt.Errorf("scope %q item %q has invalid semantic ID", scope.ID, item.Key)
		}
		if _, duplicate := itemIDs[item.ID]; duplicate {
			return nil, 0, false, nil, fmt.Errorf("scope %q duplicates item ID %q", scope.ID, item.ID)
		}
		if _, duplicate := itemKeys[item.Key]; duplicate {
			return nil, 0, false, nil, fmt.Errorf("scope %q duplicates item key %q", scope.ID, item.Key)
		}
		itemIDs[item.ID] = struct{}{}
		itemKeys[item.Key] = item
		seenResponderAgentIDs := make(map[string]struct{}, len(item.ActivatedResponderAgentIDs))
		for _, agentID := range item.ActivatedResponderAgentIDs {
			if agentID != strings.TrimSpace(agentID) || agentID == "" {
				return nil, 0, false, nil, fmt.Errorf("scope %q item %q has non-canonical activated responder", scope.ID, item.Key)
			}
			if _, duplicate := seenResponderAgentIDs[agentID]; duplicate {
				return nil, 0, false, nil, fmt.Errorf("scope %q item %q duplicates activated responder %q", scope.ID, item.Key, agentID)
			}
			seenResponderAgentIDs[agentID] = struct{}{}
		}
		switch item.Status {
		case WorkItemPending:
			allTerminal = false
			if item.Failure != nil || item.Value != nil || item.RootActivationID != "" ||
				len(item.BlockedBy) != 0 ||
				len(item.ActivatedResponderAgentIDs) != 0 {
				return nil, 0, false, nil, fmt.Errorf("scope %q pending item %q has outcome data", scope.ID, item.Key)
			}
		case WorkItemRunning:
			running++
			allTerminal = false
			if item.Failure != nil || item.Value != nil ||
				len(item.BlockedBy) != 0 || item.RootActivationID == "" {
				return nil, 0, false, nil, fmt.Errorf(
					"scope %q running item %q has invalid execution state",
					scope.ID,
					item.Key,
				)
			}
		case WorkItemSucceeded:
			if item.Failure != nil || item.Value == nil || len(item.BlockedBy) != 0 {
				return nil, 0, false, nil, fmt.Errorf(
					"scope %q succeeded item %q has invalid outcome",
					scope.ID,
					item.Key,
				)
			}
		case WorkItemFailed:
			if item.Failure == nil || item.Value != nil || len(item.BlockedBy) != 0 {
				return nil, 0, false, nil, fmt.Errorf("scope %q unsuccessful item %q has no failure", scope.ID, item.Key)
			}
		case WorkItemBlocked:
			if item.Failure == nil || item.Value != nil ||
				len(item.BlockedBy) == 0 || item.RootActivationID != "" ||
				len(item.ActivatedResponderAgentIDs) != 0 {
				return nil, 0, false, nil, fmt.Errorf(
					"scope %q blocked item %q has invalid blocked state",
					scope.ID,
					item.Key,
				)
			}
		case WorkItemCanceled:
			if item.Failure == nil || item.Value != nil || len(item.BlockedBy) != 0 {
				return nil, 0, false, nil, fmt.Errorf("scope %q canceled item %q has no failure", scope.ID, item.Key)
			}
		default:
			return nil, 0, false, nil, fmt.Errorf(
				"scope %q item %q has unknown status %q",
				scope.ID,
				item.Key,
				item.Status,
			)
		}
		frozen[itemIndex] = WorkItem{
			Key:               item.Key,
			DependsOn:         append([]string(nil), item.DependsOn...),
			WorkScope:         item.WorkScope,
			RelatedScopes:     append([]domain.MessageScopeValue(nil), item.RelatedScopes...),
			Responder:         item.Responder,
			Responders:        append([]string(nil), item.Responders...),
			SelectOneStrategy: item.SelectOneStrategy,
			Instruction:       item.Instruction,
		}
	}
	return itemKeys, running, allTerminal, frozen, nil
}

func validateFanOutItemExecutions(
	state *State,
	scope *Scope,
	owner *Activation,
	itemKeys map[string]*WorkItemState,
) error {
	for itemIndex := range scope.FanOut.Items {
		item := &scope.FanOut.Items[itemIndex]
		expectedBlockers := fanOutBlockers(scope.FanOut, item)
		if item.Status == WorkItemBlocked && !reflect.DeepEqual(item.BlockedBy, expectedBlockers) {
			return fmt.Errorf("scope %q item %q has invalid blockedBy", scope.ID, item.Key)
		}
		if item.Status == WorkItemPending && len(expectedBlockers) != 0 {
			return fmt.Errorf("scope %q pending item %q already has unsuccessful dependencies", scope.ID, item.Key)
		}
		started := item.Status == WorkItemRunning || item.Status == WorkItemSucceeded ||
			item.Status == WorkItemFailed ||
			item.Status == WorkItemCanceled && item.RootActivationID != ""
		if started {
			for _, dependencyKey := range item.DependsOn {
				if itemKeys[dependencyKey].Status != WorkItemSucceeded {
					return fmt.Errorf(
						"scope %q started item %q before dependency %q succeeded",
						scope.ID,
						item.Key,
						dependencyKey,
					)
				}
			}
		}
		if item.RootActivationID != "" {
			root, err := state.activation(item.RootActivationID)
			if err != nil || root.ParentActivationID != owner.ID ||
				!lineageContainsFanOutItem(root.Lineage, scope.ID, item) {
				return fmt.Errorf("scope %q item %q has invalid root activation", scope.ID, item.Key)
			}
		}
	}
	return nil
}

func validateFanOutScopeOutcome(state *State, scope *Scope, join *Activation, allTerminal bool) error {
	switch scope.Status {
	case ScopeActive:
		if scope.FanOut.Outcome != "" || allTerminal || join.Status != ActivationWaiting {
			return fmt.Errorf("scope %q active fan-out has resolved outcome", scope.ID)
		}
	case ScopeSucceeded:
		if scope.FanOut.ResourceFailure || scope.FanOut.Outcome != "completed" || !allTerminal ||
			(join.Status != ActivationReady && join.Status != ActivationSucceeded &&
				!(join.Status == ActivationCanceled && state.Status.Terminal())) {
			return fmt.Errorf("scope %q succeeded fan-out is not completely settled", scope.ID)
		}
	case ScopeFailed:
		resourceFailed := scope.FanOut.ResourceFailure && scope.FanOut.Outcome == "failed" &&
			(join.Status == ActivationReady || join.Status == ActivationSucceeded ||
				join.Status == ActivationFailed ||
				(join.Status == ActivationCanceled && state.Status.Terminal()))
		if !allTerminal || !resourceFailed {
			return fmt.Errorf("scope %q failed fan-out has outcome %q", scope.ID, scope.FanOut.Outcome)
		}
	case ScopeCanceled:
		if !allTerminal || join.Status != ActivationCanceled ||
			scope.FanOut.Outcome != "" && scope.FanOut.Outcome != "canceled" {
			return fmt.Errorf("scope %q canceled fan-out has outcome %q", scope.ID, scope.FanOut.Outcome)
		}
	default:
		return fmt.Errorf("scope %q fan-out has unsupported status %q", scope.ID, scope.Status)
	}
	return nil
}

func validateFanOutActivationLineage(state *State) error {
	for activationIndex := range state.Activations {
		activation := &state.Activations[activationIndex]
		for _, reference := range activation.Lineage {
			if reference.Kind != plan.ArtifactScopeFanOut {
				continue
			}
			scope, err := state.scope(reference.ScopeID)
			if err != nil || scope.FanOut == nil {
				return fmt.Errorf("activation %q references invalid fan-out scope %q", activation.ID, reference.ScopeID)
			}
			item := fanOutItemByID(scope, reference.ItemID)
			if item == nil || reference.ItemKey != item.Key || reference.ItemIndex == nil || *reference.ItemIndex != item.Index {
				return fmt.Errorf("activation %q has invalid fan-out item lineage", activation.ID)
			}
		}
	}
	return nil
}

func containsString(values []string, wanted string) bool {
	for _, value := range values {
		if value == wanted {
			return true
		}
	}
	return false
}

func lineageContainsFanOutItem(lineage []ScopeRef, scopeID string, item *WorkItemState) bool {
	if item == nil {
		return false
	}
	for _, reference := range lineage {
		if reference.ScopeID == scopeID && reference.Kind == plan.ArtifactScopeFanOut &&
			reference.ItemID == item.ID && reference.ItemKey == item.Key &&
			reference.ItemIndex != nil && *reference.ItemIndex == item.Index {
			return true
		}
	}
	return false
}