package run

import (
	"fmt"
	"reflect"
	"sort"
	"strings"

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

func (kernel *Kernel) executeFanOut(
	transition *Transition,
	activationID string,
	context EvalContext,
) error {
	state := transition.State
	activation, err := state.activation(activationID)
	if err != nil {
		return err
	}
	node, ok := kernel.graph.Node(activation.NodeID)
	if !ok || node.Config.FanOut == nil {
		return fmt.Errorf("fan-out activation %q has no compiled config", activationID)
	}
	raw, err := EvalExpression(node.Config.FanOut.Items, context)
	if err != nil {
		return kernel.failActivation(transition, activationID, "fan_out.items_invalid", err)
	}
	items, err := NormalizeWorkItems("fan_out.items", raw, nil)
	if err != nil {
		return kernel.failActivation(transition, activationID, "fan_out.items_invalid", err)
	}
	return kernel.commitPreparedRoutes(transition, activationID, func(prepared *Transition) error {
		owner, err := prepared.State.activation(activationID)
		if err != nil {
			return err
		}
		owner.Status = ActivationSucceeded
		owner.SelectedPorts = []string{"item"}
		// Fan-out itself has no user-visible output; keep nil normalized so a
		// persistence JSON round-trip is semantically identical.
		owner.Output = nil
		return kernel.openFanOut(prepared, activationID, items)
	})
}

func (kernel *Kernel) openFanOut(
	transition *Transition,
	ownerActivationID string,
	items []WorkItem,
) error {
	state := transition.State
	owner, err := state.activation(ownerActivationID)
	if err != nil {
		return err
	}
	node, ok := kernel.graph.Node(owner.NodeID)
	if !ok || node.Config.FanOut == nil {
		return fmt.Errorf("fan-out owner %q has no config", owner.NodeID)
	}
	artifactScope, ok := kernel.graph.Scope(plan.ArtifactScopeFanOut, owner.NodeID)
	if !ok || artifactScope.FanOut == nil {
		return fmt.Errorf("fan-out owner %q has no scope", owner.NodeID)
	}
	config := node.Config.FanOut
	if config.FailureMode != plan.FanOutFailureCollect {
		return fmt.Errorf("fan-out owner %q has unsupported failure mode %q", owner.NodeID, config.FailureMode)
	}
	if config.MaxConcurrency < 1 || config.MaxConcurrency > plan.MaxFanOutConcurrency {
		return fmt.Errorf(
			"fan-out owner %q maxConcurrency %d is outside 1..%d",
			owner.NodeID,
			config.MaxConcurrency,
			plan.MaxFanOutConcurrency,
		)
	}
	if len(items) > plan.MaxWorkItems {
		return fmt.Errorf("fan-out owner %q exceeds the %d WorkItem limit", owner.NodeID, plan.MaxWorkItems)
	}
	scope := state.appendScope(plan.ArtifactScopeFanOut, owner.NodeID, owner.Lineage)
	join, err := state.appendActivation(artifactScope.FanOut.JoinID, "", Values{}, owner.Lineage, ActivationWaiting)
	if err != nil {
		return err
	}
	join.ParentActivationID = ownerActivationID
	join.TriggerScopeID = scope.ID
	join.FanOutResourceCausal = owner.FanOutResourceCausal
	frozen := make([]WorkItemState, len(items))
	for index, item := range items {
		frozen[index] = WorkItemState{
			ID:                         WorkItemSemanticID(ownerActivationID, item.Key),
			Key:                        item.Key,
			Index:                      index,
			Status:                     WorkItemPending,
			DependsOn:                  append([]string{}, item.DependsOn...),
			BlockedBy:                  []string{},
			Responder:                  item.Responder,
			Responders:                 append([]string(nil), item.Responders...),
			SelectOneStrategy:          item.SelectOneStrategy,
			WorkScope:                  item.WorkScope,
			RelatedScopes:              append([]domain.MessageScopeValue{}, item.RelatedScopes...),
			Instruction:                item.Instruction,
			ActivatedResponderAgentIDs: []string{},
		}
	}
	// appendActivation may reallocate the activation slice; refresh both
	// aggregate pointers before recording ownership.
	scope, _ = state.scope(scope.ID)
	scope.FanOut = &FanOutState{
		OwnerActivationID: ownerActivationID,
		JoinNodeID:        artifactScope.FanOut.JoinID,
		JoinActivationID:  join.ID,
		MaxConcurrency:    config.MaxConcurrency,
		FailureMode:       config.FailureMode,
		Items:             frozen,
	}
	owner, _ = state.activation(ownerActivationID)
	owner.OwnedScopeIDs = append(owner.OwnedScopeIDs, scope.ID)
	transition.Events = append(transition.Events, Event{
		Kind: "fan_out_opened", ActivationID: owner.ID, NodeID: owner.NodeID, ScopeID: scope.ID,
	})
	return kernel.scheduleFanOut(transition, scope.ID)
}

func (kernel *Kernel) scheduleFanOut(transition *Transition, scopeID string) (returnErr error) {
	defer func() {
		if returnErr == nil {
			returnErr = validateFanOutHardStateBudget(
				"fan_out "+scopeID+" settled schedule",
				transition.State,
			)
		}
	}()
	for {
		scope, err := transition.State.scope(scopeID)
		if err != nil {
			return err
		}
		if scope.FanOut == nil {
			return fmt.Errorf("scope %q is not a fan-out", scopeID)
		}
		if scope.Status != ScopeActive {
			return nil
		}
		changed, err := kernel.blockUnrunnableItems(transition, scope)
		if err != nil {
			return err
		}
		if changed {
			// A newly blocked successor will never receive a causal frontier
			// copy. Re-evaluate dependency carriers immediately so another
			// already-running viable successor can become the sole owner rather
			// than waiting until the entire fan-out resolves.
			if err := releaseFanOutInputTaskWatchFrontiers(transition.State, scopeID); err != nil {
				return err
			}
		}
		scope, _ = transition.State.scope(scopeID)
		running := 0
		pending := 0
		allTerminal := true
		for index := range scope.FanOut.Items {
			switch scope.FanOut.Items[index].Status {
			case WorkItemRunning:
				running++
				allTerminal = false
			case WorkItemPending:
				pending++
				allTerminal = false
			default:
				allTerminal = allTerminal && scope.FanOut.Items[index].Status.Terminal()
			}
		}
		if allTerminal {
			return kernel.resolveFanOut(transition, scopeID)
		}
		launched := false
		for index := range scope.FanOut.Items {
			if running >= scope.FanOut.MaxConcurrency {
				break
			}
			item := &scope.FanOut.Items[index]
			if item.Status != WorkItemPending || !fanOutDependenciesSucceeded(scope.FanOut, item) {
				continue
			}
			item.Status = WorkItemRunning
			if err := kernel.launchFanOutItem(transition, scopeID, item.ID); err != nil {
				return err
			}
			launched = true
			scope, _ = transition.State.scope(scopeID)
			item = fanOutItemByID(scope, item.ID)
			if item != nil && item.Status == WorkItemRunning {
				running++
			}
		}
		if changed || launched {
			continue
		}
		if running == 0 && pending != 0 {
			return fmt.Errorf("fan-out scope %q has pending WorkItems but no runnable or running item", scopeID)
		}
		return nil
	}
}

func (kernel *Kernel) blockUnrunnableItems(transition *Transition, scope *Scope) (bool, error) {
	if scope == nil || scope.FanOut == nil {
		return false, nil
	}
	changed := false
	for index := range scope.FanOut.Items {
		item := &scope.FanOut.Items[index]
		if item.Status != WorkItemPending {
			continue
		}
		blockedBy := fanOutBlockers(scope.FanOut, item)
		if len(blockedBy) == 0 {
			continue
		}
		item.Status = WorkItemBlocked
		item.BlockedBy = blockedBy
		blockedDiagnostics := make([]FailureDiagnostic, 0, len(blockedBy))
		for _, dependencyKey := range blockedBy {
			blockedDiagnostics = append(blockedDiagnostics, FailureDiagnostic{
				Code:    "work_item.dependency_failed",
				Path:    fmt.Sprintf("%s.dependsOn", item.Key),
				Field:   "dependsOn",
				Message: fmt.Sprintf("blocked by unsuccessful dependency %q", dependencyKey),
			})
		}
		item.Failure = &Failure{
			Code:    "work_item.dependency_blocked",
			Message: "blocked by unsuccessful dependencies: " + strings.Join(blockedBy, ", "),
			Stage:   "work_item",
			Problem: FailureProblem("work_item.dependency_blocked"),
			Hint:    FailureHint("work_item.dependency_blocked"),
			Path:    workItemFailurePath(item, "dependency_blocked"),
			WorkItem: &FailureWorkItem{
				Key: item.Key, Index: &item.Index, DependencyPath: item.Key + ".dependsOn",
			},
			Diagnostics: blockedDiagnostics,
		}
		budgetErr := kernel.validateFanOutItemCandidate(
			"fan_out blocked item "+item.Key,
			transition.State,
			scope.ID,
			item.ID,
			true,
		)
		if budgetErr != nil {
			if !isFanOutResourceLimit(budgetErr) {
				return false, budgetErr
			}
			kernel.compactFanOutItemResourceState(transition.State, scope.ID, item)
			if err := kernel.validateFanOutItemCandidate(
				"fan_out resource-blocked item "+item.Key,
				transition.State,
				scope.ID,
				item.ID,
				true,
			); err != nil {
				return false, err
			}
			transition.Events = append(transition.Events, FailureEvent("fan_out_item_failed", item.Failure, scope.ID, "", ""))
		} else {
			transition.Events = append(transition.Events, FailureEvent("fan_out_item_blocked", item.Failure, scope.ID, "", ""))
		}
		changed = true
	}
	return changed, nil
}

func fanOutBlockers(fanOut *FanOutState, item *WorkItemState) []string {
	if fanOut == nil || item == nil {
		return nil
	}
	indexes := make(map[string]int, len(fanOut.Items))
	for index := range fanOut.Items {
		indexes[fanOut.Items[index].Key] = index
	}
	blocked := make(map[string]struct{})
	for _, dependencyKey := range item.DependsOn {
		dependency := fanOutItemByKey(fanOut, dependencyKey)
		if dependency == nil {
			continue
		}
		switch dependency.Status {
		case WorkItemFailed, WorkItemCanceled:
			blocked[dependency.Key] = struct{}{}
		case WorkItemBlocked:
			if len(dependency.BlockedBy) == 0 {
				blocked[dependency.Key] = struct{}{}
				continue
			}
			for _, key := range dependency.BlockedBy {
				blocked[key] = struct{}{}
			}
		}
	}
	result := make([]string, 0, len(blocked))
	for key := range blocked {
		result = append(result, key)
	}
	sort.Slice(result, func(left, right int) bool {
		return indexes[result[left]] < indexes[result[right]]
	})
	return result
}

func fanOutDependenciesSucceeded(fanOut *FanOutState, item *WorkItemState) bool {
	if fanOut == nil || item == nil {
		return false
	}
	for _, dependencyKey := range item.DependsOn {
		dependency := fanOutItemByKey(fanOut, dependencyKey)
		if dependency == nil || dependency.Status != WorkItemSucceeded {
			return false
		}
	}
	return true
}

// fanOutDependencyView projects direct predecessor results from their single
// durable WorkItem Value. The returned envelopes are read-only: value points
// at immutable terminal state instead of cloning the same payload once per
// dependency edge.
func fanOutDependencyView(fanOut *FanOutState, item *WorkItemState) Values {
	result := Values{}
	if fanOut == nil || item == nil {
		return result
	}
	for _, dependencyKey := range item.DependsOn {
		dependency := fanOutItemByKey(fanOut, dependencyKey)
		if dependency != nil && dependency.Status == WorkItemSucceeded {
			result[dependency.Key] = fanOutTerminalEnvelope(*dependency)
		}
	}
	return result
}

// evalFanOutEntryBindings evaluates direct item loads as borrowed read-only
// views. The complete expanded object is budgeted before NormalizeContract
// takes the single owned copy for the child activation input.
func evalFanOutEntryBindings(
	path string,
	bindings []plan.ArtifactBinding,
	contract plan.ArtifactContract,
	context EvalContext,
) (Values, error) {
	values := make(Values, len(bindings))
	for _, binding := range bindings {
		value, err := evalFanOutBindingExpression(binding.Value, context)
		if err != nil {
			return nil, fmt.Errorf("%s.%s: %w", path, binding.Target, err)
		}
		if value == nil && binding.Value.MayBeAbsent() {
			continue
		}
		values[binding.Target] = value
		if err := validateFanOutPayloadBudget(path, values); err != nil {
			return nil, err
		}
	}
	normalized, err := NormalizeContract(path, contract, values)
	if err != nil {
		return nil, err
	}
	if err := validateFanOutPayloadBudget(path, normalized); err != nil {
		return nil, err
	}
	return normalized, nil
}

func evalFanOutBindingExpression(
	expression plan.ArtifactExpression,
	context EvalContext,
) (any, error) {
	return evalExpressionBorrowed(expression, context)
}

func activationBelongsToFanOutItem(activation *Activation, scopeID, itemID string) bool {
	if activation == nil {
		return false
	}
	for _, reference := range activation.Lineage {
		if reference.Kind == plan.ArtifactScopeFanOut &&
			reference.ScopeID == scopeID && reference.ItemID == itemID {
			return true
		}
	}
	return false
}

func scopeLineageContainsFanOutItem(lineage []ScopeRef, scopeID, itemID string) bool {
	for _, reference := range lineage {
		if reference.Kind == plan.ArtifactScopeFanOut &&
			reference.ScopeID == scopeID && reference.ItemID == itemID {
			return true
		}
	}
	return false
}

// compactFanOutItemDurableHistory removes payload copies that can no longer
// participate in execution after an item is deterministically converted to a
// resource failure. Keeping every intermediate Input/Output (and nested scope
// value) would let a chain of individually-admitted steps consume the entire
// Run budget before the closing join. The graph/audit skeleton and the compact
// rejected-completion receipt remain durable; only unusable business payload
// copies owned by this failed item are retired.
func (kernel *Kernel) compactFanOutItemDurableHistory(state *State, scopeID, itemID string) {
	if state == nil || strings.TrimSpace(scopeID) == "" || strings.TrimSpace(itemID) == "" {
		return
	}
	for index := range state.Activations {
		activation := &state.Activations[index]
		if !activationBelongsToFanOutItem(activation, scopeID, itemID) {
			continue
		}
		compact := activation.Status != ActivationSucceeded
		if activation.Status == ActivationSucceeded {
			node, ok := kernel.graph.Node(activation.NodeID)
			if ok {
				switch node.Opcode {
				case plan.OpcodeSetVariable,
					plan.OpcodeSwitchFirstMatch,
					plan.OpcodeSwitchAllMatch,
					plan.OpcodeFork,
					plan.OpcodeFanOut,
					plan.OpcodeLoop,
					plan.OpcodeFailPlan,
					plan.OpcodeEndSucceeded,
					plan.OpcodeEndFailed:
					compact = true
				}
			}
		}
		if compact {
			activation.Input = Values{}
			activation.Output = nil
			activation.TaskWatchFrontier = nil
		}
	}
}

func (kernel *Kernel) compactFanOutItemResourceState(state *State, scopeID string, item *WorkItemState) {
	if state == nil || item == nil {
		return
	}
	itemID := item.ID
	projectCompactFanOutResourceFailure(item)
	cancelFanOutItemWork(state, scopeID, itemID, workItemResourceLimitFailureCode)
	kernel.compactFanOutItemDurableHistory(state, scopeID, itemID)
}

// fanOutContextForActivation identifies both ordinary body activations (via
// lineage) and the closing join (via TriggerScopeID). The latter matters for
// merge expressions and downstream bindings: they are part of the same
// WorkItem aggregate even though the join itself intentionally carries the
// owner's parent lineage.
func fanOutContextForActivation(
	state *State,
	activation *Activation,
) (*Scope, *WorkItemState, bool) {
	if state == nil || activation == nil {
		return nil, nil, false
	}
	for index := len(activation.Lineage) - 1; index >= 0; index-- {
		reference := activation.Lineage[index]
		if reference.Kind != plan.ArtifactScopeFanOut {
			continue
		}
		scope, err := state.scope(reference.ScopeID)
		if err != nil || scope.FanOut == nil {
			return nil, nil, false
		}
		return scope, fanOutItemByID(scope, reference.ItemID), true
	}
	if activation.TriggerScopeID != "" {
		if scope, err := state.scope(activation.TriggerScopeID); err == nil && scope.FanOut != nil {
			return scope, nil, true
		}
	}
	return nil, nil, false
}

func (kernel *Kernel) failFanOutActivationResource(
	transition *Transition,
	activationID string,
	cause error,
) error {
	if transition == nil || transition.State == nil {
		return fmt.Errorf("fan-out activation resource failure has invalid state")
	}
	activation, err := transition.State.activation(activationID)
	if err != nil {
		return err
	}
	scope, item, ok := fanOutContextForActivation(transition.State, activation)
	if (!ok || scope == nil || scope.FanOut == nil) && !activation.FanOutResourceCausal {
		return cause
	}
	if item == nil {
		// The WorkItems are already terminal truth at the closing join (or this
		// activation is a later causal descendant). Never rewrite an unrelated
		// successful item to make a route fit: take this node's recoverable failed
		// port atomically instead.
		alreadyRoutingResourceFailure := activation.Failure != nil &&
			activation.Failure.Code == workItemResourceLimitFailureCode
		node, nodeOK := kernel.graph.Node(activation.NodeID)
		hasFailedPort := false
		if nodeOK {
			_, hasFailedPort = outputPort(node, "failed")
		}
		if !nodeOK || !hasFailedPort || alreadyRoutingResourceFailure {
			return kernel.failFanOutCausalRunResource(transition, activation.ID)
		}
		activation.Failure = &Failure{
			Code:    workItemResourceLimitFailureCode,
			Message: workItemResourceLimitMessage,
			Stage:   "work_item",
			Problem: FailureProblem(workItemResourceLimitFailureCode),
			Hint:    FailureHint(workItemResourceLimitFailureCode),
		}
		return kernel.finishPorts(
			transition,
			activation.ID,
			[]portValue{{Port: "failed", Value: Values{}}},
		)
	}
	activation.Status = ActivationFailed
	activation.Output = nil
	activation.SelectedPorts = nil
	activation.Failure = &Failure{
		Code:    workItemResourceLimitFailureCode,
		Message: workItemResourceLimitMessage,
		Stage:   "work_item",
		Problem: FailureProblem(workItemResourceLimitFailureCode),
		Hint:    FailureHint(workItemResourceLimitFailureCode),
	}
	if err := kernel.failFanOutItemResource(
		transition,
		scope,
		item,
		activation.TaskWatchFrontier,
	); err != nil {
		return err
	}
	if scope.Status == ScopeActive {
		if err := kernel.scheduleFanOut(transition, scope.ID); err != nil {
			return err
		}
	}
	return validateFanOutHardStateBudget("fan_out activation resource settlement", transition.State)
}

func (kernel *Kernel) failFanOutCausalRunResource(transition *Transition, activationID string) error {
	if transition == nil || transition.State == nil {
		return fmt.Errorf("fan-out causal Run resource failure has invalid state")
	}
	activation, err := transition.State.activation(activationID)
	if err != nil {
		return err
	}
	activation.Status = ActivationFailed
	activation.Input = Values{}
	activation.Output = nil
	activation.SelectedPorts = nil
	activation.TaskWatchFrontier = nil
	activation.Failure = &Failure{
		Code:    workItemResourceLimitFailureCode,
		Message: workItemResourceLimitMessage,
		Stage:   "work_item",
		Problem: FailureProblem(workItemResourceLimitFailureCode),
		Hint:    FailureHint(workItemResourceLimitFailureCode),
	}
	transition.State.Output = nil
	transition.State.Status = StatusFailed
	transition.State.Failure = cloneFailure(activation.Failure)
	cancelOpenWork(transition.State, activation.ID)
	transition.Events = append(transition.Events, Event{
		Kind: "run_failed", ActivationID: activation.ID, NodeID: activation.NodeID,
		Code: workItemResourceLimitFailureCode,
	})
	return validateFanOutHardStateBudget("fan_out causal Run resource failure", transition.State)
}

func freezeFanOutActivatedResponder(
	state *State,
	activation *Activation,
	agentID string,
) (string, bool, error) {
	agentID = strings.TrimSpace(agentID)
	if state == nil || activation == nil || agentID == "" {
		return "", false, nil
	}
	for lineageIndex := len(activation.Lineage) - 1; lineageIndex >= 0; lineageIndex-- {
		reference := activation.Lineage[lineageIndex]
		if reference.Kind != plan.ArtifactScopeFanOut || strings.TrimSpace(reference.ItemID) == "" {
			continue
		}
		scope, err := state.scope(reference.ScopeID)
		if err != nil || scope.FanOut == nil {
			return "", false, fmt.Errorf("callback activation %q has invalid fan-out lineage", activation.ID)
		}
		item := fanOutItemByID(scope, reference.ItemID)
		if item == nil {
			return "", false, fmt.Errorf("callback activation %q has unknown WorkItem lineage", activation.ID)
		}
		for _, existing := range item.ActivatedResponderAgentIDs {
			if existing == agentID {
				return scope.ID, false, nil
			}
		}
		item.ActivatedResponderAgentIDs = append(item.ActivatedResponderAgentIDs, agentID)
		return scope.ID, true, nil
	}
	return "", false, nil
}

func (kernel *Kernel) launchFanOutItem(transition *Transition, scopeID, itemID string) error {
	state := transition.State
	scope, err := state.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("fan-out scope %q is invalid", scopeID)
	}
	item := fanOutItemByID(scope, itemID)
	if item == nil || item.Status != WorkItemRunning {
		return fmt.Errorf("fan-out scope %q item %q is not running", scopeID, itemID)
	}
	owner, err := state.activation(scope.FanOut.OwnerActivationID)
	if err != nil {
		return err
	}
	incomingTaskWatchFrontier, err := fanOutItemIncomingTaskWatchFrontier(state, scope, item)
	if err != nil {
		return err
	}
	dependencies := fanOutDependencyView(scope.FanOut, item)
	materializedInstruction, err := materializeWorkItemInstruction(
		"fan_out item "+item.Key+" instruction",
		item.Instruction,
		item.DependsOn,
		dependencies,
	)
	if err != nil {
		return kernel.failFanOutItemEvaluation(
			transition, scope, item, incomingTaskWatchFrontier,
			workItemInstructionReferenceFailureCode, err, "invalid instruction",
		)
	}
	edges := kernel.graph.Outgoing(owner.NodeID, "item")
	if len(edges) != 1 {
		return fmt.Errorf("fan-out owner %q item port has %d edges, want one", owner.NodeID, len(edges))
	}
	edge := edges[0]
	targetNode, ok := kernel.graph.Node(edge.Target.NodeID)
	if !ok {
		return fmt.Errorf("fan-out entry edge %q target is missing", edge.ID)
	}
	contract, ok := inputPort(targetNode, edge.Target.Port)
	if !ok {
		return fmt.Errorf("fan-out entry edge %q target port is missing", edge.ID)
	}
	context, err := kernel.evalContext(state, owner)
	if err != nil {
		return err
	}
	itemValues := fanOutItemValues(scope.FanOut, *item)
	// Freeze the single-pass expansion in the child activation input. A
	// recovered Run therefore renders the exact same effect request without
	// consulting dependency outputs again.
	itemValues[plan.WorkItemFieldInstruction] = materializedInstruction
	context.FanOutItems[scope.OwnerID] = itemValues
	input, err := evalFanOutEntryBindings("edge "+edge.ID, edge.Bindings, contract, context)
	if err != nil {
		if isFanOutResourceLimit(err) {
			return kernel.failFanOutItemResource(
				transition, scope, item, incomingTaskWatchFrontier,
			)
		}
		return kernel.failFanOutItemEvaluation(
			transition, scope, item, incomingTaskWatchFrontier,
			"edge.binding_failed", err, "binding failure",
		)
	}
	hasEntry, hasArrival := false, false
	for _, scopeTransition := range edge.ScopeTransitions {
		if scopeTransition.ScopeKind != plan.ArtifactScopeFanOut || scopeTransition.OwnerID != scope.OwnerID {
			continue
		}
		switch scopeTransition.Action {
		case "item":
			hasEntry = true
		case "arrive":
			hasArrival = true
		}
	}
	if !hasEntry {
		return fmt.Errorf("fan-out entry edge %q has no item transition", edge.ID)
	}
	if hasArrival {
		return kernel.completeDirectFanOutArrival(
			transition, scope, item, input, incomingTaskWatchFrontier,
		)
	}
	index := item.Index
	lineage := cloneScopeRefs(owner.Lineage)
	lineage = append(lineage, ScopeRef{
		ScopeID: scope.ID, Kind: plan.ArtifactScopeFanOut, OwnerID: scope.OwnerID,
		ItemID: item.ID, ItemKey: item.Key, ItemIndex: &index,
	})
	activationCount := len(state.Activations)
	nextActivation := state.NextActivation
	activation, err := state.appendActivation(edge.Target.NodeID, edge.Target.Port, input, lineage, ActivationReady)
	if err != nil {
		return err
	}
	activation.ParentActivationID = owner.ID
	activation.FanOutResourceCausal = owner.FanOutResourceCausal
	activation.TaskWatchFrontier = incomingTaskWatchFrontier
	item.RootActivationID = activation.ID
	if err := kernel.validateFanOutStateAdmission("fan_out item launch", state, item.ID); err != nil {
		state.Activations = state.Activations[:activationCount]
		state.NextActivation = nextActivation
		item.RootActivationID = ""
		if isFanOutResourceLimit(err) {
			return kernel.failFanOutItemResource(
				transition, scope, item, incomingTaskWatchFrontier,
			)
		}
		return err
	}
	transition.Events = append(transition.Events, Event{Kind: "fan_out_item_started", ScopeID: scope.ID})
	transition.Events = append(transition.Events, Event{
		Kind: "activation_created", ActivationID: activation.ID, NodeID: activation.NodeID, ScopeID: scope.ID,
	})
	return releaseFanOutInputTaskWatchFrontiers(state, scope.ID)
}

func (kernel *Kernel) completeDirectFanOutArrival(
	transition *Transition,
	scope *Scope,
	item *WorkItemState,
	input Values,
	incomingTaskWatchFrontier []string,
) error {
	item.TaskWatchFrontier = incomingTaskWatchFrontier
	item.Status = WorkItemSucceeded
	item.Value = input
	budgetErr := kernel.validateFanOutItemCandidate(
		"fan_out direct arrival", transition.State, scope.ID, item.ID, true,
	)
	if budgetErr == nil {
		transition.Events = append(transition.Events,
			Event{Kind: "fan_out_item_started", ScopeID: scope.ID},
			Event{Kind: "fan_out_item_arrived", ScopeID: scope.ID},
		)
		return releaseFanOutInputTaskWatchFrontiers(transition.State, scope.ID)
	}
	item.Status = WorkItemRunning
	item.Value = nil
	if isFanOutResourceLimit(budgetErr) {
		return kernel.failFanOutItemResource(
			transition, scope, item, incomingTaskWatchFrontier,
		)
	}
	return budgetErr
}

// workItemFailurePath produces a stable, bounded path label identifying the
// failing dynamic WorkItem and the dependency/materialization step.
func (kernel *Kernel) failFanOutItemEvaluation(
	transition *Transition,
	scope *Scope,
	item *WorkItemState,
	incomingTaskWatchFrontier []string,
	failureCode string,
	cause error,
	candidateLabel string,
) error {
	state := transition.State
	item.TaskWatchFrontier = incomingTaskWatchFrontier
	item.Status = WorkItemFailed
	item.Failure = failureFromCause(failureCode, "work_item", cause)
	item.Failure.Path = workItemFailurePath(item, candidateLabel)
	if item.Failure.WorkItem == nil {
		item.Failure.WorkItem = &FailureWorkItem{
			Key: item.Key, Index: &item.Index, DependencyPath: candidateLabel,
		}
	}
	transition.Events = append(transition.Events, FailureEvent("fan_out_item_failed", item.Failure, scope.ID, "", ""))
	if budgetErr := kernel.validateFanOutItemCandidate(
		"fan_out "+candidateLabel+" "+item.Key,
		state,
		scope.ID,
		item.ID,
		true,
	); budgetErr != nil {
		if !isFanOutResourceLimit(budgetErr) {
			return budgetErr
		}
		kernel.compactFanOutItemResourceState(state, scope.ID, item)
		if err := kernel.validateFanOutStateAdmission(
			"fan_out compact "+candidateLabel+" "+item.Key,
			state,
			item.ID,
		); err != nil {
			return err
		}
	}
	return releaseFanOutInputTaskWatchFrontiers(state, scope.ID)
}

func (kernel *Kernel) recordFanOutArrival(
	transition *Transition,
	scopeID, itemID string,
	value Values,
	taskWatchFrontier []string,
) error {
	scope, err := transition.State.scope(scopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("fan-out arrival scope %q is invalid", scopeID)
	}
	if scope.Status != ScopeActive {
		return nil
	}
	item := fanOutItemByID(scope, itemID)
	if item == nil {
		return fmt.Errorf("fan-out scope %q has no item %q", scopeID, itemID)
	}
	if item.Status == WorkItemSucceeded {
		if reflect.DeepEqual(item.Value, value) {
			return nil
		}
		return fmt.Errorf("fan-out item %q replay has conflicting value", item.Key)
	}
	if item.Status != WorkItemRunning {
		return nil
	}
	item.TaskWatchFrontier, err = normalizedTaskWatchFrontier(
		transition.State,
		item.TaskWatchFrontier,
		taskWatchFrontier,
	)
	if err != nil {
		return err
	}
	if budgetErr := validateFanOutPayloadBudget("fan_out item arrival", value); budgetErr != nil {
		return budgetErr
	}
	item.Status = WorkItemSucceeded
	item.Value = value
	// Once the item has reached its single durable result, pure reducer
	// intermediates are no longer causal inputs. Retire their duplicate payload
	// copies while preserving externally-completable activations for exact
	// callback/timer replay.
	kernel.compactFanOutItemDurableHistory(transition.State, scope.ID, item.ID)
	budgetErr := kernel.validateFanOutItemCandidate(
		"fan_out item arrival", transition.State, scope.ID, item.ID, false,
	)
	if budgetErr == nil {
		transition.Events = append(transition.Events, Event{
			Kind: "fan_out_item_arrived", ScopeID: scope.ID,
		})
		return kernel.scheduleFanOut(transition, scope.ID)
	}
	item.Status = WorkItemRunning
	item.Value = nil
	return budgetErr
}

func (kernel *Kernel) failFanOutItemResource(
	transition *Transition,
	scope *Scope,
	item *WorkItemState,
	taskWatchFrontier []string,
) error {
	if transition == nil || transition.State == nil || scope == nil || scope.FanOut == nil || item == nil {
		return fmt.Errorf("fan-out resource failure has invalid state")
	}
	item.TaskWatchFrontier = append([]string(nil), taskWatchFrontier...)
	item.Status = WorkItemFailed
	item.Value = nil
	item.Failure = &Failure{
		Code:    workItemResourceLimitFailureCode,
		Message: workItemResourceLimitMessage,
		Stage:   "work_item",
		Problem: FailureProblem(workItemResourceLimitFailureCode),
		Hint:    FailureHint(workItemResourceLimitFailureCode),
	}
	cancelFanOutItemWork(transition.State, scope.ID, item.ID, item.Failure.Code)
	kernel.compactFanOutItemDurableHistory(transition.State, scope.ID, item.ID)
	transition.Events = append(transition.Events, FailureEvent("fan_out_item_failed", item.Failure, scope.ID, "", ""))
	if err := releaseFanOutInputTaskWatchFrontiers(transition.State, scope.ID); err != nil {
		return err
	}
	return kernel.validateFanOutItemCandidate(
		"fan_out resource failure",
		transition.State,
		scope.ID,
		item.ID,
		true,
	)
}

func (kernel *Kernel) failFanOutItem(
	transition *Transition,
	reference ScopeRef,
	status WorkItemStatus,
	failure *Failure,
) error {
	if status != WorkItemFailed {
		return fmt.Errorf("unsupported failed WorkItem status %q", status)
	}
	scope, err := transition.State.scope(reference.ScopeID)
	if err != nil || scope.FanOut == nil {
		return fmt.Errorf("fan-out failure scope %q is invalid", reference.ScopeID)
	}
	if scope.Status != ScopeActive {
		return nil
	}
	item := fanOutItemByID(scope, reference.ItemID)
	if item == nil {
		return fmt.Errorf("fan-out scope %q has no item %q", scope.ID, reference.ItemID)
	}
	if item.Status.Terminal() {
		if item.Status == status && reflect.DeepEqual(item.Failure, failure) {
			return nil
		}
		return nil // a fenced late result after dependency/cancel resolution
	}
	if item.Status != WorkItemRunning {
		return fmt.Errorf("fan-out item %q is %s, want running", item.Key, item.Status)
	}
	if err := collapseFanOutItemTaskWatchFrontier(
		transition.State,
		reference.ScopeID,
		reference.ItemID,
	); err != nil {
		return err
	}
	// collapse may reallocate neither aggregate, but refresh the item pointer so
	// the ownership handoff remains explicit beside later state mutations.
	scope, _ = transition.State.scope(reference.ScopeID)
	item = fanOutItemByID(scope, reference.ItemID)
	item.Status = status
	item.Failure = cloneFailure(failure)
	if item.Failure == nil {
		item.Failure = &Failure{Code: "work_item.failed", Stage: "work_item", Problem: FailureProblem("work_item.failed"), Hint: FailureHint("work_item.failed")}
	}
	// Every fan-out item failure carries its own current WorkItem context so
	// persisted events and Run show/effect/activation can pinpoint the exact
	// item (key/index/dependency path), including inherited/generic failures
	// routed through this settling path.
	if item.Failure.WorkItem == nil {
		item.Failure.WorkItem = &FailureWorkItem{Key: item.Key, Index: &item.Index}
	}
	if strings.TrimSpace(item.Failure.Path) == "" {
		item.Failure.Path = workItemFailurePath(item, "settlement")
	}
	kernel.compactFanOutItemDurableHistory(transition.State, scope.ID, item.ID)
	if budgetErr := kernel.validateFanOutItemCandidate(
		"fan_out failed item "+item.Key,
		transition.State,
		scope.ID,
		item.ID,
		true,
	); budgetErr != nil {
		if !isFanOutResourceLimit(budgetErr) {
			return budgetErr
		}
		status = WorkItemFailed
		kernel.compactFanOutItemResourceState(transition.State, scope.ID, item)
		if err := kernel.validateFanOutItemCandidate(
			"fan_out compact failed item "+item.Key,
			transition.State,
			scope.ID,
			item.ID,
			true,
		); err != nil {
			return err
		}
	}
	cancelFanOutItemWork(transition.State, scope.ID, item.ID, item.Failure.Code)
	transition.Events = append(transition.Events, FailureEvent("fan_out_item_failed", item.Failure, scope.ID, "", ""))
	return kernel.scheduleFanOut(transition, scope.ID)
}