package run

import (
	"bytes"
	"crypto/sha256"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"

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

// The durable Workflow App approval protocol accepts at most 4000 Unicode
// characters. Enforce that boundary while the reducer can still take the
// compiled failed port instead of emitting an effect that can never dispatch.
const maxApprovalSummaryRunes = 4000

// Kernel is a deterministic reducer over one immutable artifact.
type Kernel struct {
	graph *Graph
}

func NewKernel(graph *Graph) (*Kernel, error) {
	if graph == nil || graph.Artifact == nil {
		return nil, fmt.Errorf("Run graph is nil")
	}
	return &Kernel{graph: graph}, nil
}

func (kernel *Kernel) Graph() *Graph { return kernel.graph }

// Start validates Plan input, applies input/state defaults, evaluates entry
// bindings, and creates exactly one ready entry activation.
func (kernel *Kernel) Start(input Values) (*State, error) {
	normalizedInput, err := NormalizeContract("plan.input", kernel.graph.Artifact.Contracts.Input, input)
	if err != nil {
		return nil, err
	}
	variables, err := NormalizeContract("run.state", kernel.graph.Artifact.Contracts.State, Values{})
	if err != nil {
		return nil, err
	}
	state := &State{
		Status:      StatusQueued,
		Input:       normalizedInput,
		Variables:   variables,
		Activations: []Activation{},
		Scopes:      []Scope{},
	}
	node, ok := kernel.graph.Node(kernel.graph.Artifact.Entry.NodeID)
	if !ok {
		return nil, fmt.Errorf("entry node %q is missing", kernel.graph.Artifact.Entry.NodeID)
	}
	contract, ok := inputPort(node, kernel.graph.Artifact.Entry.Port)
	if !ok {
		return nil, fmt.Errorf("entry port is missing")
	}
	entry, err := evalBindings("entry", kernel.graph.Artifact.Entry.Bindings, contract, EvalContext{
		PlanInput: normalizedInput,
		RunState:  variables,
	})
	if err != nil {
		return nil, err
	}
	if _, err := state.appendActivation(node.ID, kernel.graph.Artifact.Entry.Port, entry, nil, ActivationReady); err != nil {
		return nil, err
	}
	return state, nil
}

// Step executes the oldest ready activation.  now is explicit so replay is
// deterministic.  At most one activation is reduced per call.
func (kernel *Kernel) Step(current *State, now time.Time) (Transition, error) {
	if current == nil {
		return Transition{}, fmt.Errorf("Run state is nil")
	}
	if current.Status.Terminal() {
		return Transition{}, fmt.Errorf("Run is terminal (%s)", current.Status)
	}
	if err := kernel.validateFanOutState(current); err != nil {
		return Transition{}, fmt.Errorf("validate fan-out Run state: %w", err)
	}
	if err := validateTaskWatchFrontierState(kernel.graph, current); err != nil {
		return Transition{}, fmt.Errorf("validate Task Watch Run state: %w", err)
	}
	state := current.Clone()
	activationID := ""
	for index := range state.Activations {
		if state.Activations[index].Status == ActivationReady {
			activationID = state.Activations[index].ID
			break
		}
	}
	if activationID == "" {
		refreshStatus(state)
		return Transition{State: state, Intents: []Intent{}, Events: []Event{}}, nil
	}
	if state.Status == StatusQueued || state.Status == StatusWaiting {
		state.Status = StatusRunning
	}
	transition := Transition{State: state, Intents: []Intent{}, Events: []Event{{Kind: "activation_started", ActivationID: activationID}}}
	activation, _ := state.activation(activationID)
	resourceCausal := activationRequiresFanOutResourceAdmission(state, activation)
	if err := kernel.execute(&transition, activationID, now.UTC()); err != nil {
		if !resourceCausal || !isFanOutResourceLimit(err) {
			return Transition{}, err
		}
		state = current.Clone()
		if state.Status == StatusQueued || state.Status == StatusWaiting {
			state.Status = StatusRunning
		}
		transition = Transition{State: state, Intents: []Intent{}, Events: []Event{{Kind: "activation_started", ActivationID: activationID}}}
		if failErr := kernel.failFanOutActivationResource(&transition, activationID, err); failErr != nil {
			return Transition{}, failErr
		}
	} else if resourceCausal {
		admissionErr := validateFanOutIntentRange(
			"fan_out causal Step "+activationID,
			transition.Intents,
			0,
		)
		if admissionErr == nil {
			admissionErr = validateFanOutHardStateBudget(
				"fan_out causal Step "+activationID,
				state,
			)
		}
		if admissionErr != nil {
			state = current.Clone()
			if state.Status == StatusQueued || state.Status == StatusWaiting {
				state.Status = StatusRunning
			}
			transition = Transition{State: state, Intents: []Intent{}, Events: []Event{{Kind: "activation_started", ActivationID: activationID}}}
			if failErr := kernel.failFanOutActivationResource(&transition, activationID, admissionErr); failErr != nil {
				return Transition{}, failErr
			}
		}
	}
	refreshStatus(state)
	return transition, nil
}

func activationRequiresFanOutResourceAdmission(state *State, activation *Activation) bool {
	if activation == nil {
		return false
	}
	if activation.FanOutResourceCausal {
		return true
	}
	_, _, aware := fanOutContextForActivation(state, activation)
	return aware
}

// Complete feeds an effect, callback, approval decision, or timer result back
// into the graph.  Same-result replay is idempotent; conflicting replay fails.
func (kernel *Kernel) Complete(current *State, completion Completion, now time.Time) (Transition, error) {
	if current == nil {
		return Transition{}, fmt.Errorf("Run state is nil")
	}
	if err := kernel.validateFanOutState(current); err != nil {
		return Transition{}, fmt.Errorf("validate fan-out Run state: %w", err)
	}
	if err := validateTaskWatchFrontierState(kernel.graph, current); err != nil {
		return Transition{}, fmt.Errorf("validate Task Watch Run state: %w", err)
	}
	state := current.Clone()
	activation, err := state.activation(completion.ActivationID)
	if err != nil {
		return Transition{}, err
	}
	node, ok := kernel.graph.Node(activation.NodeID)
	if !ok {
		return Transition{}, fmt.Errorf("activation node %q is missing", activation.NodeID)
	}
	contract, ok := outputPort(node, completion.Port)
	if !ok {
		return Transition{}, fmt.Errorf("node %q has no output port %q", node.ID, completion.Port)
	}
	output, err := NormalizeContract("completion.output", contract, completion.Output)
	if err != nil {
		return Transition{}, err
	}
	rejectedCompletion, err := resourceRejectedCompletionReceipt(completion, output)
	if err != nil {
		return Transition{}, err
	}
	if state.Status == StatusCanceled {
		return Transition{}, fmt.Errorf("Run is terminal (%s)", state.Status)
	}
	if replay, handled, replayErr := completionReplay(
		state, activation, completion, output, rejectedCompletion,
	); handled || replayErr != nil {
		return replay, replayErr
	}
	if state.Status.Terminal() {
		return Transition{}, fmt.Errorf("Run is terminal (%s)", state.Status)
	}
	if activation.Status != ActivationWaiting {
		return Transition{}, fmt.Errorf("activation %q is %s, want waiting", activation.ID, activation.Status)
	}
	activation.Failure = cloneFailure(completion.Failure)
	responderAgentID, responderActivatedScopeID, responderActivated, err := kernel.freezeCompletionResponder(
		state, activation, node.Opcode, completion.CallbackResponderAgentID,
	)
	if err != nil {
		return Transition{}, err
	}
	// Carry the compact receipt through admission and route preparation. A
	// successful completion clears it before returning; a resource-local
	// failure retains it so replay never needs the rejected output bytes.
	activation.ResourceRejectedCompletion = rejectedCompletion
	if inFanOut, budgetErr := kernel.validateFanOutCompletionBudget(
		state,
		activation,
		completion.Port,
		output,
	); inFanOut && budgetErr != nil {
		if !isFanOutResourceLimit(budgetErr) {
			return Transition{}, budgetErr
		}
		transition := Transition{
			State:   state,
			Intents: []Intent{},
			Events: []Event{{
				Kind: "activation_failed", ActivationID: activation.ID,
				NodeID: node.ID, Code: workItemResourceLimitFailureCode,
			}},
		}
		if err := kernel.failFanOutActivationResource(
			&transition,
			activation.ID,
			budgetErr,
		); err != nil {
			return Transition{}, err
		}
		if err := validateFanOutHardStateBudget("fan_out completion resource failure", state); err != nil {
			return Transition{}, err
		}
		refreshStatus(state)
		return transition, nil
	}
	events := []Event{{Kind: "activation_completed", ActivationID: activation.ID, NodeID: node.ID, Port: completion.Port}}
	if responderActivated {
		events = append(events, Event{
			Kind: "fan_out_item_responder_activated", ActivationID: activation.ID,
			NodeID: node.ID, AgentID: responderAgentID,
			ScopeID: responderActivatedScopeID,
		})
	}
	transition := Transition{State: state, Intents: []Intent{}, Events: events}
	if err := kernel.finishPorts(&transition, activation.ID, []portValue{{Port: completion.Port, Value: output}}); err != nil {
		return Transition{}, err
	}
	activation, _ = state.activation(completion.ActivationID)
	acceptedCompletion := activation.Status == ActivationSucceeded &&
		len(activation.SelectedPorts) == 1 && activation.SelectedPorts[0] == completion.Port &&
		reflect.DeepEqual(activation.Output, output) &&
		reflect.DeepEqual(activation.Failure, completion.Failure)
	if acceptedCompletion {
		activation.ResourceRejectedCompletion = nil
	} else if activation.Failure == nil || activation.Failure.Code != workItemResourceLimitFailureCode {
		activation.ResourceRejectedCompletion = nil
	}
	if err := validateFanOutHardStateBudget("fan_out completion settled", state); err != nil {
		return Transition{}, err
	}
	refreshStatus(state)
	_ = now // completion time is persisted by the store/event layer
	return transition, nil
}

func resourceRejectedCompletionReceipt(
	completion Completion,
	output Values,
) (*ResourceRejectedCompletionState, error) {
	payload, err := json.Marshal(output)
	if err != nil {
		return nil, fmt.Errorf("encode completion replay digest: %w", err)
	}
	return &ResourceRejectedCompletionState{
		Port:                     completion.Port,
		OutputDigest:             deterministicCompletionOutputDigest(payload),
		FailureDigest:            completionFailureDigest(completion.Failure),
		CallbackResponderAgentID: strings.TrimSpace(completion.CallbackResponderAgentID),
	}, nil
}

func completionFailureDigest(failure *Failure) string {
	if failure == nil {
		return ""
	}
	hash := sha256.New()
	_, _ = hash.Write([]byte("loop.workflow.rejected-completion/failure/v1\x00"))
	for _, field := range []string{failure.Code, failure.Message} {
		_, _ = hash.Write([]byte(strconv.Itoa(len(field))))
		_, _ = hash.Write([]byte{0})
		_, _ = hash.Write([]byte(field))
	}
	// Include the deterministic truncation receipt so a replayed completion with a
	// different prior cut (omittedCount/droppedCount/truncatedFields/
	// serializedBytes/minimalFallback) produces a different digest instead of
	// silently matching the original.
	if failure.Truncation != nil {
		receipt := failure.Truncation
		_, _ = hash.Write([]byte{0x01})
		_, _ = hash.Write([]byte(strconv.Itoa(receipt.OmittedCount)))
		_, _ = hash.Write([]byte{0})
		_, _ = hash.Write([]byte(strconv.Itoa(receipt.DroppedCount)))
		_, _ = hash.Write([]byte{0})
		_, _ = hash.Write([]byte(strconv.Itoa(receipt.SerializedBytes)))
		_, _ = hash.Write([]byte{0})
		if receipt.DiagnosticsTruncated {
			_, _ = hash.Write([]byte("t"))
		} else {
			_, _ = hash.Write([]byte("f"))
		}
		_, _ = hash.Write([]byte{0})
		if receipt.MinimalFallback {
			_, _ = hash.Write([]byte("t"))
		} else {
			_, _ = hash.Write([]byte("f"))
		}
		for _, field := range receipt.TruncatedFields {
			_, _ = hash.Write([]byte(strconv.Itoa(len(field))))
			_, _ = hash.Write([]byte{0})
			_, _ = hash.Write([]byte(field))
		}
	}
	return fmt.Sprintf("sha256:%x", hash.Sum(nil))
}

// deterministicCompletionOutputDigest follows the durable Run Store's finite
// float64 number semantics without applying an ordinary payload byte or
// structural budget. The receipt is created precisely when a completion may
// exceed those budgets, so digesting it must remain total for every payload
// that already passed NormalizeContract.
//
// json.Marshal has already produced valid compact JSON with deterministic map
// key ordering and deterministic string escaping. Its only Store-visible spelling
// difference is json.Number, which NormalizeContract intentionally preserves.
// Rewrite those number tokens while streaming into SHA-256, avoiding another
// potentially multi-megabyte deterministic payload allocation. A syntactically
// valid number outside the finite runtime domain keeps its original spelling;
// the existing completion/budget path remains responsible for rejecting it.
func deterministicCompletionOutputDigest(payload []byte) string {
	hash := sha256.New()
	normalizedValid := true
	lastWritten := 0
	for index := 0; index < len(payload); {
		switch payload[index] {
		case '"':
			index++
			for index < len(payload) {
				switch payload[index] {
				case '\\':
					index += 2
				case '"':
					index++
					goto stringDone
				default:
					index++
				}
			}
		stringDone:
			continue
		case '-', '0', '1', '2', '3', '4', '5', '6', '7', '8', '9':
			start := index
			for index < len(payload) {
				switch payload[index] {
				case ',', '}', ']', ' ', '\t', '\n', '\r':
					goto numberDone
				default:
					index++
				}
			}
		numberDone:
			_, _ = hash.Write(payload[lastWritten:start])
			normalizedNum, err := runlimit.NormalizedNumberString(string(payload[start:index]))
			if err != nil {
				normalizedValid = false
				_, _ = hash.Write(payload[start:index])
			} else {
				_, _ = hash.Write([]byte(normalizedNum))
			}
			lastWritten = index
		default:
			index++
		}
	}
	_, _ = hash.Write(payload[lastWritten:])
	if !normalizedValid {
		// Keep non-normalized payload identity deterministic without allowing raw
		// bytes to collide with the deterministic digest domain. The fan-out budget
		// admission that follows converts the invalid finite-domain number into
		// an item-local resource failure rather than a whole-Run reducer error.
		invalidHash := sha256.New()
		_, _ = invalidHash.Write([]byte("loop.workflow.rejected-completion/invalid-canonical/v1\x00")) // frozen digest domain separator; renaming changes persisted rejection digests
		_, _ = invalidHash.Write(payload)
		return fmt.Sprintf("sha256:%x", invalidHash.Sum(nil))
	}
	return fmt.Sprintf("sha256:%x", hash.Sum(nil))
}

// Cancel is a pure terminal fence.  Late Complete calls are rejected.
func (kernel *Kernel) Cancel(current *State) (Transition, error) {
	if current == nil {
		return Transition{}, fmt.Errorf("Run state is nil")
	}
	if err := kernel.validateFanOutState(current); err != nil {
		return Transition{}, fmt.Errorf("validate fan-out Run state: %w", err)
	}
	if err := validateTaskWatchFrontierState(kernel.graph, current); err != nil {
		return Transition{}, fmt.Errorf("validate Task Watch Run state: %w", err)
	}
	state := current.Clone()
	if state.Status == StatusCanceled {
		return Transition{State: state, Intents: []Intent{}, Events: []Event{{Kind: "run_cancel_replayed"}}}, nil
	}
	if state.Status.Terminal() {
		return Transition{}, fmt.Errorf("Run is terminal (%s)", state.Status)
	}
	state.Status = StatusCanceled
	for index := range state.Activations {
		if state.Activations[index].Status == ActivationReady || state.Activations[index].Status == ActivationWaiting {
			state.Activations[index].Status = ActivationCanceled
		}
	}
	for index := range state.Scopes {
		if state.Scopes[index].Status == ScopeActive {
			state.Scopes[index].Status = ScopeCanceled
			cancelFanOutItems(state.Scopes[index].FanOut, "run.canceled")
		}
	}
	return Transition{State: state, Intents: []Intent{}, Events: []Event{{Kind: "run_canceled"}}}, nil
}

type portValue struct {
	Port  string
	Value Values
}

// routeBindingFailure is returned while preparing a completion's routes. It
// is deliberately not applied to State by routeEdge: finishPorts owns the
// transaction boundary and can therefore discard every partially prepared
// sibling route before failing the source activation once.
type routeBindingFailure struct {
	edgeID string
	cause  error
}

func (failure *routeBindingFailure) Error() string {
	return fmt.Sprintf("edge %q binding failed: %v", failure.edgeID, failure.cause)
}

func (failure *routeBindingFailure) Unwrap() error { return failure.cause }

func (kernel *Kernel) execute(transition *Transition, activationID string, now time.Time) error {
	state := transition.State
	activation, err := state.activation(activationID)
	if err != nil {
		return err
	}
	node, ok := kernel.graph.Node(activation.NodeID)
	if !ok {
		return fmt.Errorf("activation node %q is missing", activation.NodeID)
	}
	context, err := kernel.evalContext(state, activation)
	if err != nil {
		return err
	}
	switch node.Opcode {
	case plan.OpcodeSendMessageSent, plan.OpcodeSendMessageCallback:
		config := node.Config.SendMessage
		if config == nil || (config.Target == nil) == (config.Scope == nil) {
			return kernel.routeRecoverableFailure(transition, activation.ID, "message.destination_invalid", fmt.Errorf("compiled message requires exactly one target or scope"))
		}
		target := ""
		var scope *domain.MessageScopeValue
		if config.Scope != nil {
			value, evalErr := EvalExpression(*config.Scope, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.scope_invalid", evalErr)
			}
			normalized, normalizeErr := domain.NormalizeMessageScopeValue(value)
			if normalizeErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.scope_invalid", normalizeErr)
			}
			scope = &normalized
		} else {
			rendered, renderErr := EvalTemplate(*config.Target, context)
			if renderErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.target_invalid", renderErr)
			}
			target = strings.TrimSpace(rendered)
			if target == "" {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.target_invalid", fmt.Errorf("rendered message target is blank"))
			}
			parsedTarget, targetErr := domain.ParseMessageTargetReference(target)
			if targetErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.target_invalid", targetErr)
			}
			if parsedTarget.Kind == domain.MessageTargetKindWorkflow {
				return kernel.routeRecoverableFailure(
					transition,
					activation.ID,
					"message.target_invalid",
					fmt.Errorf("Workflow Builder edit scopes are not Workflow Run message destinations"),
				)
			}
		}
		responder := ""
		var responderCandidates []string
		responderStrategy := plan.SelectOneStrategy("")
		if config.Responder != nil {
			value, evalErr := EvalExpression(*config.Responder, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.responder_invalid", evalErr)
			}
			// A WorkItem responder may resolve to a selectOne pool + strategy.
			if object, ok := value.(map[string]any); ok {
				candidates, strategy, normalizeErr := selectOneCandidatesFromObject(object)
				if normalizeErr != nil {
					return kernel.routeRecoverableFailure(transition, activation.ID, "message.responder_select_one_invalid", normalizeErr)
				}
				responderCandidates = candidates
				responderStrategy = strategy
			} else {
				var ok bool
				responder, ok = value.(string)
				if !ok {
					return kernel.routeRecoverableFailure(transition, activation.ID, "message.responder_invalid", fmt.Errorf("responder evaluated to %T, want agent string or selectOne pool", value))
				}
				responder = strings.TrimSpace(responder)
				if responder == "" {
					return kernel.routeRecoverableFailure(transition, activation.ID, "message.responder_invalid", fmt.Errorf("responder is blank"))
				}
			}
		} else if config.SelectOne != nil {
			value, evalErr := EvalExpression(*config.SelectOne, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.responder_select_one_invalid", evalErr)
			}
			candidates, normalizeErr := normalizeSelectOneCandidates(value)
			if normalizeErr != nil {
				return kernel.routeRecoverableFailure(
					transition,
					activation.ID,
					"message.responder_select_one_invalid",
					normalizeErr,
				)
			}
			responderCandidates = candidates
		}
		var workItemsAllowedResponders []string
		var workItemsDependencyValueContract *plan.ArtifactContract
		if config.WorkItems != nil {
			value, evalErr := EvalExpression(config.WorkItems.AllowedResponders, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.work_items_invalid", evalErr)
			}
			workItemsAllowedResponders, evalErr = normalizeResponderArray(value)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.work_items_invalid", evalErr)
			}
			dependencyValueContract, contractErr := kernel.graph.WorkItemsDependencyValueContract(node.ID)
			if contractErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "message.work_items_invalid", contractErr)
			}
			workItemsDependencyValueContract = &dependencyValueContract
		}
		workScopeMode, workScope, relatedScopes, scopeErr := evalMessageCollaborationScopes(config, context)
		if scopeErr != nil {
			return kernel.routeRecoverableFailure(
				transition,
				activation.ID,
				"message.collaboration_scope_invalid",
				scopeErr,
			)
		}
		text, renderErr := EvalTemplate(config.Text, context)
		if renderErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "message.text_invalid", renderErr)
		}
		if strings.TrimSpace(text) == "" {
			return kernel.routeRecoverableFailure(transition, activation.ID, "message.text_invalid", fmt.Errorf("rendered message text is blank"))
		}
		var taskWatchHandoff *TaskWatchHandoff
		if node.Opcode == plan.OpcodeSendMessageCallback && (responder != "" || len(responderCandidates) > 0) {
			taskWatchHandoff, scopeErr = taskWatchHandoffForActivation(
				state,
				activation,
				responder,
				responderCandidates,
				scope,
				workScopeMode,
				workScope,
			)
			if scopeErr != nil {
				return kernel.routeRecoverableFailure(
					transition,
					activation.ID,
					"message.task_watch_handoff_invalid",
					scopeErr,
				)
			}
		}
		activation.Status = ActivationWaiting
		transition.Intents = append(transition.Intents, Intent{
			Kind:                             IntentSendMessage,
			ActivationID:                     activation.ID,
			NodeID:                           node.ID,
			Target:                           target,
			Scope:                            scope,
			Responder:                        responder,
			ResponderCandidates:              append([]string(nil), responderCandidates...),
			SelectOneStrategy:                effectiveSelectOneStrategy(responderStrategy, selectOneStrategyForConfig(config)),
			Text:                             text,
			Callback:                         node.Opcode == plan.OpcodeSendMessageCallback,
			StartThread:                      config.StartThread,
			WorkScopeMode:                    workScopeMode,
			WorkScope:                        workScope,
			RelatedScopes:                    relatedScopes,
			TaskWatchHandoff:                 taskWatchHandoff,
			CallbackOutput:                   append(plan.ArtifactContract(nil), config.CallbackOutput...),
			WorkItemsAllowedResponders:       append([]string(nil), workItemsAllowedResponders...),
			WorkItemsDependencyValueContract: workItemsDependencyValueContract,
			Retry:                            config.Retry,
		})
		transition.Events = append(transition.Events, Event{Kind: "effect_requested", ActivationID: activation.ID, NodeID: node.ID})
		return nil
	case plan.OpcodeWait:
		until, evalErr := EvalExpression(node.Config.Wait.Until, context)
		if evalErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "wait.until_invalid", evalErr)
		}
		wakeAt, evalErr := waitInstant(until, node.Config.Wait.Until.ResultType, now)
		if evalErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "wait.until_invalid", evalErr)
		}
		activation.Status = ActivationWaiting
		deadline := now.Add(time.Duration(node.Config.Wait.TimeoutNanos))
		transition.Intents = append(transition.Intents, Intent{Kind: IntentWait, ActivationID: activation.ID, NodeID: node.ID, WakeAt: &wakeAt, Deadline: &deadline, Retry: node.Config.Wait.Retry})
		transition.Events = append(transition.Events, Event{Kind: "wait_requested", ActivationID: activation.ID, NodeID: node.ID})
		return nil
	case plan.OpcodeRequestApproval:
		assignee, renderErr := EvalTemplate(node.Config.RequestApproval.Assignee, context)
		if renderErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "approval.assignee_invalid", renderErr)
		}
		assignee = strings.TrimSpace(assignee)
		if assignee == "" {
			return kernel.routeRecoverableFailure(transition, activation.ID, "approval.assignee_invalid", fmt.Errorf("rendered approval assignee is blank"))
		}
		summary, renderErr := EvalTemplate(node.Config.RequestApproval.Summary, context)
		if renderErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "approval.summary_invalid", renderErr)
		}
		summary = strings.TrimSpace(summary)
		if summary == "" {
			return kernel.routeRecoverableFailure(transition, activation.ID, "approval.summary_invalid", fmt.Errorf("rendered approval summary is blank"))
		}
		if utf8.RuneCountInString(summary) > maxApprovalSummaryRunes {
			return kernel.routeRecoverableFailure(transition, activation.ID, "approval.summary_invalid", fmt.Errorf("rendered approval summary exceeds %d characters", maxApprovalSummaryRunes))
		}
		var feedbackScope *domain.MessageScopeValue
		if node.Config.RequestApproval.FeedbackScope != nil {
			value, evalErr := EvalExpression(*node.Config.RequestApproval.FeedbackScope, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "approval.feedback_scope_invalid", evalErr)
			}
			normalized, normalizeErr := domain.NormalizeMessageScopeValue(value)
			if normalizeErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "approval.feedback_scope_invalid", normalizeErr)
			}
			feedbackScope = &normalized
		}
		activation.Status = ActivationWaiting
		transition.Intents = append(transition.Intents, Intent{Kind: IntentApproval, ActivationID: activation.ID, NodeID: node.ID, Assignee: assignee, Summary: summary, FeedbackScope: feedbackScope, Deadline: nil, Retry: node.Config.RequestApproval.Retry})
		transition.Events = append(transition.Events, Event{Kind: "approval_requested", ActivationID: activation.ID, NodeID: node.ID})
		return nil
	case plan.OpcodeSetVariable:
		candidate := cloneValues(state.Variables)
		for _, assignment := range node.Config.SetVariable.Assignments {
			value, evalErr := EvalExpression(assignment.Value, context)
			if evalErr != nil {
				return kernel.routeRecoverableFailure(transition, activation.ID, "state.expression_failed", evalErr)
			}
			if value == nil && assignment.Value.MayBeAbsent() {
				delete(candidate, assignment.State)
				continue
			}
			candidate[assignment.State] = value
		}
		normalized, validateErr := NormalizeContract("run.state", kernel.graph.Artifact.Contracts.State, candidate)
		if validateErr != nil {
			return kernel.routeRecoverableFailure(transition, activation.ID, "state.contract_failed", validateErr)
		}
		state.Variables = normalized
		return kernel.finishPorts(transition, activation.ID, []portValue{{Port: "completed", Value: Values{}}})
	case plan.OpcodeSwitchFirstMatch, plan.OpcodeSwitchAllMatch:
		ports := make([]string, 0, len(node.Config.Switch.Cases))
		defaultPort := ""
		for _, switchCase := range node.Config.Switch.Cases {
			if switchCase.Default {
				defaultPort = switchCase.Port
				continue
			}
			value, evalErr := EvalExpression(*switchCase.Condition, context)
			if evalErr != nil {
				return kernel.failActivation(transition, activation.ID, "switch.condition_failed", evalErr)
			}
			matched, ok := value.(bool)
			if !ok {
				return kernel.failActivation(transition, activation.ID, "switch.condition_failed", fmt.Errorf("condition returned %T", value))
			}
			if matched {
				ports = append(ports, switchCase.Port)
				if node.Opcode == plan.OpcodeSwitchFirstMatch {
					break
				}
			}
		}
		if len(ports) == 0 {
			ports = []string{defaultPort}
		}
		outputs := make([]portValue, len(ports))
		for index, port := range ports {
			outputs[index] = portValue{Port: port, Value: cloneValues(activation.Input)}
		}
		if node.Opcode == plan.OpcodeSwitchAllMatch {
			return kernel.finishParallelPorts(transition, activation.ID, ports, outputs)
		}
		return kernel.finishPorts(transition, activation.ID, outputs)
	case plan.OpcodeFork:
		ports := append([]string(nil), node.Config.Fork.Branches...)
		outputs := make([]portValue, len(ports))
		for index, port := range ports {
			outputs[index] = portValue{Port: port, Value: cloneValues(activation.Input)}
		}
		return kernel.finishParallelPorts(transition, activation.ID, ports, outputs)
	case plan.OpcodeFanOut:
		return kernel.executeFanOut(transition, activation.ID, context)
	case plan.OpcodeJoinAll, plan.OpcodeJoinAny:
		return kernel.executeJoin(transition, activation.ID)
	case plan.OpcodeLoop:
		return kernel.executeLoop(transition, activation.ID)
	case plan.OpcodeEndSucceeded:
		output, validateErr := NormalizeContract("run.output", kernel.graph.Artifact.Contracts.Output, activation.Input)
		if validateErr != nil {
			return kernel.failActivation(transition, activation.ID, "run.output_contract_failed", validateErr)
		}
		if activeFailureScopeRef(state, activation.Lineage) != nil {
			return kernel.failActivation(transition, activation.ID, "run.invalid_branch_terminal", fmt.Errorf("success terminal is inside an open branch/item scope"))
		}
		activation.Status = ActivationSucceeded
		activation.Output = output
		state.Output = output
		state.Status = StatusSucceeded
		cancelOpenWork(state, activation.ID)
		transition.Events = append(transition.Events, Event{Kind: "run_succeeded", ActivationID: activation.ID, NodeID: node.ID})
		return nil
	case plan.OpcodeEndFailed:
		activation.Status = ActivationFailed
		activation.Failure = failureForCodeAt(node.Config.End.Code, "reducer", node.ID)
		if scopeRef := activeFailureScopeRef(state, activation.Lineage); scopeRef != nil {
			switch scopeRef.Kind {
			case plan.ArtifactScopeFanOut:
				return kernel.failFanOutItem(transition, *scopeRef, WorkItemFailed, activation.Failure)
			case plan.ArtifactScopeParallel:
				if err := kernel.failParallelBranch(transition, *scopeRef, node.Config.End.Code); err != nil {
					return err
				}
				transition.Events = append(transition.Events, FailureEvent("branch_failed", activation.Failure, scopeRef.ScopeID, activation.ID, node.ID))
				return nil
			}
		}
		state.Status = StatusFailed
		state.Failure = cloneFailure(activation.Failure)
		cancelOpenWork(state, activation.ID)
		transition.Events = append(transition.Events, FailureEvent("run_failed", activation.Failure, "", activation.ID, node.ID))
		return nil
	case plan.OpcodeFailPlan:
		failure := failureForCodeAt(node.Config.FailPlan.FallbackCode, "reducer", node.ID)
		if activation.ParentActivationID != "" {
			parent, parentErr := state.activation(activation.ParentActivationID)
			if parentErr != nil {
				return parentErr
			}
			if parent.Failure != nil {
				failure = cloneFailure(parent.Failure)
				if strings.TrimSpace(failure.Code) == "" {
					failure.Code = node.Config.FailPlan.FallbackCode
					failure.Problem = firstNonEmpty(failure.Problem, FailureProblem(node.Config.FailPlan.FallbackCode))
					failure.Hint = firstNonEmpty(failure.Hint, FailureHint(node.Config.FailPlan.FallbackCode))
				}
			}
		}
		activation.Status = ActivationFailed
		activation.Failure = failure
		if scopeRef := activeFanOutRef(state, activation.Lineage); scopeRef != nil {
			return kernel.failFanOutItem(transition, *scopeRef, WorkItemFailed, failure)
		}
		state.Status = StatusFailed
		state.Failure = cloneFailure(failure)
		cancelOpenWork(state, activation.ID)
		failedEvent := FailureEvent("run_failed", failure, "", activation.ID, node.ID)
		failedEvent.Port = node.Config.FailPlan.SourcePort
		transition.Events = append(transition.Events, failedEvent)
		return nil
	default:
		return fmt.Errorf("unsupported runtime opcode %q", node.Opcode)
	}
}

func evalMessageCollaborationScopes(
	config *plan.ArtifactSendMessageConfig,
	context EvalContext,
) (WorkScopeMode, *domain.MessageScopeValue, []domain.MessageScopeValue, error) {
	if config.RelatedScopesArray != nil && len(config.RelatedScopes) != 0 {
		return "", nil, nil, fmt.Errorf("compiled message cannot contain both relatedScopes forms")
	}
	if len(config.RelatedScopes) > plan.MaxSendMessageRelatedScopes {
		return "", nil, nil, fmt.Errorf(
			"relatedScopes must not contain more than %d scopes",
			plan.MaxSendMessageRelatedScopes,
		)
	}
	var mode WorkScopeMode
	var workScope *domain.MessageScopeValue
	switch config.WorkScopeMode {
	case "":
	case plan.ArtifactWorkScopeAgent:
		mode = WorkScopeModeAgent
	case plan.ArtifactWorkScopeFixed:
		mode = WorkScopeModeFixed
		if config.WorkScope == nil {
			return "", nil, nil, fmt.Errorf("fixed workScope expression is missing")
		}
		value, err := EvalExpression(*config.WorkScope, context)
		if err != nil {
			return "", nil, nil, fmt.Errorf("evaluate workScope: %w", err)
		}
		scope, err := domain.NormalizeMessageScopeValue(value)
		if err != nil {
			return "", nil, nil, fmt.Errorf("normalize workScope: %w", err)
		}
		workScope = &scope
	default:
		return "", nil, nil, fmt.Errorf("unsupported workScope mode %q", config.WorkScopeMode)
	}
	if mode == "" && (len(config.RelatedScopes) != 0 || config.RelatedScopesArray != nil) {
		return "", nil, nil, fmt.Errorf("relatedScopes require workScope")
	}
	values := make([]any, 0, len(config.RelatedScopes))
	if config.RelatedScopesArray != nil {
		value, err := EvalExpression(*config.RelatedScopesArray, context)
		if err != nil {
			return "", nil, nil, fmt.Errorf("evaluate relatedScopes: %w", err)
		}
		array, ok := value.([]any)
		if !ok {
			return "", nil, nil, fmt.Errorf("relatedScopes evaluated to %T, want scope array", value)
		}
		values = append(values, array...)
	} else {
		for index, expression := range config.RelatedScopes {
			value, err := EvalExpression(expression, context)
			if err != nil {
				return "", nil, nil, fmt.Errorf("evaluate relatedScopes[%d]: %w", index, err)
			}
			values = append(values, value)
		}
	}
	if len(values) > plan.MaxSendMessageRelatedScopes {
		return "", nil, nil, fmt.Errorf(
			"relatedScopes must not contain more than %d scopes",
			plan.MaxSendMessageRelatedScopes,
		)
	}
	relatedScopes := make([]domain.MessageScopeValue, 0, len(values))
	seen := make(map[domain.MessageScopeValue]struct{}, len(values))
	for index, value := range values {
		scope, err := domain.NormalizeMessageScopeValue(value)
		if err != nil {
			return "", nil, nil, fmt.Errorf("normalize relatedScopes[%d]: %w", index, err)
		}
		if _, duplicate := seen[scope]; duplicate {
			continue
		}
		seen[scope] = struct{}{}
		relatedScopes = append(relatedScopes, scope)
	}
	return mode, workScope, relatedScopes, nil
}

func normalizeResponderArray(value any) ([]string, error) {
	raw, ok := value.([]any)
	if !ok {
		return nil, fmt.Errorf("allowedResponders evaluated to %T, want agent array", value)
	}
	if len(raw) == 0 {
		return nil, fmt.Errorf("allowedResponders must not be empty")
	}
	if len(raw) > plan.MaxWorkItems {
		return nil, fmt.Errorf("allowedResponders must not contain more than %d agents", plan.MaxWorkItems)
	}
	result := make([]string, 0, len(raw))
	seen := make(map[string]struct{}, len(raw))
	for index, item := range raw {
		responder, ok := item.(string)
		if !ok {
			return nil, fmt.Errorf("allowedResponders[%d] is %T, want agent string", index, item)
		}
		responder = strings.TrimSpace(responder)
		if responder == "" {
			return nil, fmt.Errorf("allowedResponders[%d] is blank", index)
		}
		if _, duplicate := seen[responder]; duplicate {
			continue
		}
		seen[responder] = struct{}{}
		result = append(result, responder)
	}
	return result, nil
}

// normalizeSelectOneCandidates is the runtime normalize for the ordered
// responder.selectOne candidate pool, enforcing the same MaxAllowedResponders
// bound Plan/Artifact validation applies to a static literal (a ref/dynamic
// pool cannot bypass it). Duplicates drop preserving first-occurrence order; a
// dynamic empty pool fails explicitly (cannot be statically rejected).
func normalizeSelectOneCandidates(value any) ([]string, error) {
	raw, ok := value.([]any)
	if !ok {
		return nil, fmt.Errorf("responder.selectOne evaluated to %T, want agent array", value)
	}
	if len(raw) == 0 {
		return nil, fmt.Errorf("responder.selectOne must not be empty")
	}
	if len(raw) > plan.MaxAllowedResponders {
		return nil, fmt.Errorf("responder.selectOne must not contain more than %d candidates", plan.MaxAllowedResponders)
	}
	result := make([]string, 0, len(raw))
	seen := make(map[string]struct{}, len(raw))
	for index, item := range raw {
		candidate, ok := item.(string)
		if !ok {
			return nil, fmt.Errorf("responder.selectOne[%d] is %T, want agent string", index, item)
		}
		candidate = strings.TrimSpace(candidate)
		if candidate == "" {
			return nil, fmt.Errorf("responder.selectOne[%d] is blank", index)
		}
		if _, duplicate := seen[candidate]; duplicate {
			continue
		}
		seen[candidate] = struct{}{}
		result = append(result, candidate)
	}
	return result, nil
}

// selectOneCandidatesFromObject extracts the ordered pool + additive strategy
// from a WorkItem responder `{selectOne: [...]}` (runtime counterpart of
// normalizedWorkItemResponder).
func selectOneCandidatesFromObject(object map[string]any) ([]string, plan.SelectOneStrategy, error) {
	raw, present := object["selectOne"]
	if !present {
		return nil, "", fmt.Errorf("responder object must be a selectOne candidate pool")
	}
	strategy := plan.SelectOneStrategy("")
	for key, rawValue := range object {
		if key == "selectOne" {
			continue
		}
		if key == "strategy" {
			strategyValue, ok := rawValue.(string)
			if !ok {
				return nil, "", fmt.Errorf("responder strategy must be a string")
			}
			parsed, err := plan.ParseSelectOneStrategyPresent(strategyValue)
			if err != nil {
				return nil, "", fmt.Errorf("responder strategy: %w", err)
			}
			strategy = parsed
			continue
		}
		return nil, "", fmt.Errorf("responder has unsupported field %q", key)
	}
	candidates, err := normalizeSelectOneCandidates(raw)
	if err != nil {
		return nil, "", err
	}
	if strategy.Normalized() == plan.SelectOneStrategyRandom && len(candidates) < 2 {
		return nil, "", fmt.Errorf("responder selectOne strategy %q requires at least two deduplicated candidates", strategy)
	}
	return candidates, strategy.Normalized(), nil
}

func (kernel *Kernel) routeRecoverableFailure(transition *Transition, activationID, code string, cause error) error {
	activation, _ := transition.State.activation(activationID)
	activation.Failure = failureFromCause(code, "reducer", cause)
	return kernel.finishPorts(transition, activationID, []portValue{{Port: "failed", Value: Values{}}})
}

func (kernel *Kernel) failActivation(transition *Transition, activationID, code string, cause error) error {
	state := transition.State
	activation, err := state.activation(activationID)
	if err != nil {
		return err
	}
	activation.Status = ActivationFailed
	activation.Failure = failureFromCause(code, "reducer", cause)
	if scopeRef := activeFailureScopeRef(state, activation.Lineage); scopeRef != nil {
		switch scopeRef.Kind {
		case plan.ArtifactScopeFanOut:
			return kernel.failFanOutItem(transition, *scopeRef, WorkItemFailed, activation.Failure)
		case plan.ArtifactScopeParallel:
			if err := kernel.failParallelBranch(transition, *scopeRef, code); err != nil {
				return err
			}
			return nil
		}
	}
	state.Status = StatusFailed
	state.Failure = cloneFailure(activation.Failure)
	cancelOpenWork(state, activation.ID)
	transition.Events = append(transition.Events, FailureEvent("run_failed", activation.Failure, "", activation.ID, activation.NodeID))
	return nil
}

func (kernel *Kernel) finishPorts(transition *Transition, activationID string, outputs []portValue) error {
	return kernel.commitPreparedRoutes(transition, activationID, func(prepared *Transition) error {
		return kernel.finishPortsInPlace(prepared, activationID, outputs)
	})
}

// finishParallelPorts makes opening the parallel scope and routing every
// selected branch one atomic reducer mutation. A binding failure on any branch
// cannot leave a waiting join or a ready sibling behind.
func (kernel *Kernel) finishParallelPorts(transition *Transition, activationID string, selected []string, outputs []portValue) error {
	return kernel.commitPreparedRoutes(transition, activationID, func(prepared *Transition) error {
		if err := kernel.openParallel(prepared, activationID, selected); err != nil {
			return err
		}
		return kernel.finishPortsInPlace(prepared, activationID, outputs)
	})
}

// commitPreparedRoutes evaluates a node completion against an isolated State
// clone. Only a completely prepared set of sibling routes is committed. This
// keeps the reducer replayable and prevents terminal Runs with newly-created
// live activations when a later edge binding fails.
func (kernel *Kernel) commitPreparedRoutes(transition *Transition, activationID string, prepare func(*Transition) error) error {
	prepared := &Transition{State: transition.State.Clone(), Intents: []Intent{}, Events: []Event{}}
	err := prepare(prepared)
	if err == nil {
		// Preserve the State pointer owned by Step/Complete while atomically
		// replacing its contents; callers may still hold that pointer for status
		// refresh and persistence after finishPorts returns.
		*transition.State = *prepared.State
		transition.Intents = append(transition.Intents, prepared.Intents...)
		transition.Events = append(transition.Events, prepared.Events...)
		return nil
	}
	if isFanOutResourceLimit(err) {
		return kernel.failFanOutActivationResource(transition, activationID, err)
	}
	var bindingFailure *routeBindingFailure
	if errors.As(err, &bindingFailure) {
		return kernel.failActivation(transition, activationID, "edge.binding_failed", bindingFailure.cause)
	}
	return err
}

func (kernel *Kernel) finishPortsInPlace(transition *Transition, activationID string, outputs []portValue) error {
	state := transition.State
	activation, err := state.activation(activationID)
	if err != nil {
		return err
	}
	_, item, fanOutAware := fanOutContextForActivation(state, activation)
	resourceCausal := fanOutAware || activation.FanOutResourceCausal
	if resourceCausal {
		for _, output := range outputs {
			if err := validateFanOutPayloadBudget(
				"fan_out activation "+activation.ID+" output "+output.Port,
				output.Value,
			); err != nil {
				return err
			}
		}
	}
	activation.Status = ActivationSucceeded
	activation.SelectedPorts = make([]string, len(outputs))
	if len(outputs) > 0 {
		activation.Output = cloneValues(outputs[0].Value)
	}
	for index, output := range outputs {
		activation.SelectedPorts[index] = output.Port
	}
	// Keep compiler-defined order for first_match; fork/all_match are already
	// deterministic. ScopeTransitions are inspected as a set by routeEdge.
	for _, output := range outputs {
		edges := kernel.graph.Outgoing(activation.NodeID, output.Port)
		if len(edges) == 0 {
			return fmt.Errorf("node %q output %q has no edge", activation.NodeID, output.Port)
		}
		for _, edge := range edges {
			if err := kernel.routeEdge(transition, activationID, edge, output.Value); err != nil {
				return err
			}
		}
	}
	// Every selected route has now received a copy of this causal carrier:
	// ordinary targets inherit it, parallel arrivals merge it into their join,
	// and fan-out boundaries freeze it on the WorkItem. Reacquire after route
	// appends may have reallocated the activation slice, then retire the source
	// carrier exactly once.
	activation, _ = state.activation(activationID)
	activation.TaskWatchFrontier = nil
	if resourceCausal {
		forceItemID := ""
		if item != nil {
			forceItemID = item.ID
		}
		var admissionErr error
		if fanOutAware {
			admissionErr = kernel.validateFanOutStateAdmission(
				"fan_out activation "+activation.ID+" routes",
				state,
				forceItemID,
			)
		} else {
			admissionErr = validateFanOutHardStateBudget(
				"fan_out causal activation "+activation.ID+" routes",
				state,
			)
		}
		if admissionErr != nil {
			return admissionErr
		}
	}
	return nil
}

func (kernel *Kernel) routeEdge(transition *Transition, sourceID string, edge *plan.ArtifactEdge, output Values) error {
	state := transition.State
	source, err := state.activation(sourceID)
	if err != nil {
		return err
	}
	targetNode, ok := kernel.graph.Node(edge.Target.NodeID)
	if !ok {
		return fmt.Errorf("edge %q target is missing", edge.ID)
	}
	contract, ok := inputPort(targetNode, edge.Target.Port)
	if !ok {
		return fmt.Errorf("edge %q target port is missing", edge.ID)
	}
	arrival, parallelEnter, fanOutBoundary, loopActions, err := classifyScopeTransitions(edge)
	if err != nil {
		return err
	}
	context, err := kernel.evalContext(state, source)
	if err != nil {
		return err
	}
	context.ActivationOutput = output
	_, _, fanOutAware := fanOutContextForActivation(state, source)
	resourceCausal := fanOutAware || source.FanOutResourceCausal
	var input Values
	if resourceCausal || fanOutBoundary != nil {
		input, err = evalFanOutEntryBindings("edge "+edge.ID, edge.Bindings, contract, context)
	} else {
		input, err = evalBindings("edge "+edge.ID, edge.Bindings, contract, context)
	}
	if err != nil {
		if fanOutBoundary != nil && isFanOutResourceLimit(err) {
			reference, referenceErr := fanOutRefForTransition(state, source, fanOutBoundary.OwnerID)
			if referenceErr != nil {
				return referenceErr
			}
			scope, scopeErr := state.scope(reference.ScopeID)
			if scopeErr != nil || scope.FanOut == nil {
				return fmt.Errorf("fan-out boundary scope %q is invalid", reference.ScopeID)
			}
			item := fanOutItemByID(scope, reference.ItemID)
			if item == nil {
				return fmt.Errorf("fan-out boundary item %q is invalid", reference.ItemID)
			}
			if failErr := kernel.failFanOutItemResource(
				transition, scope, item, source.TaskWatchFrontier,
			); failErr != nil {
				return failErr
			}
			return kernel.scheduleFanOut(transition, scope.ID)
		}
		if resourceCausal && isFanOutResourceLimit(err) {
			return err
		}
		return &routeBindingFailure{edgeID: edge.ID, cause: err}
	}
	if arrival != nil {
		if fanOutBoundary != nil {
			return fmt.Errorf("edge %q combines parallel arrival with fan-out boundary", edge.ID)
		}
		if len(loopActions) != 0 {
			return fmt.Errorf("edge %q combines parallel arrival with actionable loop transition %s.%s", edge.ID, loopActions[0].OwnerID, loopActions[0].Action)
		}
		scope, findErr := parallelScopeForTransition(state, source, *arrival)
		if findErr != nil {
			return findErr
		}
		if scope.Parallel == nil {
			return fmt.Errorf("edge %q parallel arrival scope %q is invalid", edge.ID, scope.ID)
		}
		if err := mergeTaskWatchFrontierIntoActivation(
			state,
			scope.Parallel.JoinActivationID,
			source.TaskWatchFrontier,
		); err != nil {
			return err
		}
		if source.FanOutResourceCausal || fanOutAware {
			join, joinErr := state.activation(scope.Parallel.JoinActivationID)
			if joinErr != nil {
				return joinErr
			}
			join.FanOutResourceCausal = true
		}
		if err := kernel.recordParallelArrival(transition, scope.ID, arrival.Branch, input); err != nil {
			return err
		}
		return nil
	}
	if fanOutBoundary != nil {
		return kernel.routeFanOutBoundary(
			transition,
			source,
			edge.ID,
			input,
			*fanOutBoundary,
			loopActions,
		)
	}

	return kernel.routeActivationTarget(
		transition, source, edge, input, parallelEnter, loopActions, fanOutAware,
	)
}

func (kernel *Kernel) routeActivationTarget(
	transition *Transition,
	source *Activation,
	edge *plan.ArtifactEdge,
	input Values,
	parallelEnter *plan.ArtifactScopeTransition,
	loopActions []*plan.ArtifactScopeTransition,
	fanOutAware bool,
) error {
	state := transition.State
	lineage := cloneScopeRefs(source.Lineage)
	triggerScopeID := ""
	if parallelEnter != nil {
		scope, findErr := parallelScopeForTransition(state, source, *parallelEnter)
		if findErr != nil {
			return findErr
		}
		lineage = append(lineage, ScopeRef{ScopeID: scope.ID, Kind: plan.ArtifactScopeParallel, OwnerID: scope.OwnerID, Branch: parallelEnter.Branch})
	}
	for _, loopAction := range loopActions {
		switch loopAction.Action {
		case "body":
			scope, findErr := loopScopeForSource(state, source, loopAction.OwnerID)
			if findErr != nil {
				return findErr
			}
			lineage = append(lineage, ScopeRef{ScopeID: scope.ID, Kind: plan.ArtifactScopeLoop, OwnerID: scope.OwnerID})
		case "continue", "break":
			scope, findErr := loopScopeForSource(state, source, loopAction.OwnerID)
			if findErr != nil {
				return findErr
			}
			if triggerScopeID != "" && triggerScopeID != scope.ID {
				return fmt.Errorf("edge %q has multiple loop boundary targets", edge.ID)
			}
			lineage = removeScopeRef(lineage, scope.ID)
			triggerScopeID = scope.ID
		default:
			return fmt.Errorf("edge %q has unsupported loop transition %q", edge.ID, loopAction.Action)
		}
	}
	target, err := state.appendActivation(edge.Target.NodeID, edge.Target.Port, input, lineage, ActivationReady)
	if err != nil {
		return err
	}
	target.ParentActivationID = source.ID
	target.TriggerScopeID = triggerScopeID
	target.FanOutResourceCausal = source.FanOutResourceCausal || fanOutAware
	target.TaskWatchFrontier, err = normalizedTaskWatchFrontier(state, source.TaskWatchFrontier)
	if err != nil {
		return err
	}
	transition.Events = append(transition.Events, Event{Kind: "activation_created", ActivationID: target.ID, NodeID: target.NodeID})
	return nil
}

func classifyScopeTransitions(
	edge *plan.ArtifactEdge,
) (
	arrival,
	parallelEnter,
	fanOutBoundary *plan.ArtifactScopeTransition,
	loopActions []*plan.ArtifactScopeTransition,
	err error,
) {
	loopActions = make([]*plan.ArtifactScopeTransition, 0, 2)
	for index := range edge.ScopeTransitions {
		item := &edge.ScopeTransitions[index]
		switch item.ScopeKind {
		case plan.ArtifactScopeParallel:
			if item.Action == "arrive" {
				arrival = item
			}
			if item.Action == "enter" {
				parallelEnter = item
			}
		case plan.ArtifactScopeLoop:
			// done/failed only describe leaving the source loop owner. The
			// owner activation already carries its parent lineage, so these
			// transitions are intentionally no-ops.
			if item.Action != "done" && item.Action != "failed" {
				loopActions = append(loopActions, item)
			}
		case plan.ArtifactScopeFanOut:
			fanOutBoundary, err = selectFanOutBoundary(edge.ID, fanOutBoundary, item)
			if err != nil {
				return nil, nil, nil, nil, err
			}
		}
	}
	// A zero-node loop body may carry body+continue (or body+break). Apply
	// entry before boundary exit; OwnerID ordering is not semantic.
	sort.SliceStable(loopActions, func(left, right int) bool {
		leftExit := loopActions[left].Action == "continue" || loopActions[left].Action == "break"
		rightExit := loopActions[right].Action == "continue" || loopActions[right].Action == "break"
		return !leftExit && rightExit
	})
	return arrival, parallelEnter, fanOutBoundary, loopActions, nil
}

func selectFanOutBoundary(
	edgeID string,
	current,
	candidate *plan.ArtifactScopeTransition,
) (*plan.ArtifactScopeTransition, error) {
	switch candidate.Action {
	case "arrive", "fail", "timeout":
		if current != nil {
			return nil, fmt.Errorf("edge %q has multiple fan-out boundary transitions", edgeID)
		}
		return candidate, nil
	case "item":
		return nil, fmt.Errorf("edge %q fan-out item entry must be routed by its owner", edgeID)
	default:
		return nil, fmt.Errorf("edge %q has unsupported fan-out transition %q", edgeID, candidate.Action)
	}
}

func (kernel *Kernel) routeFanOutBoundary(
	transition *Transition,
	source *Activation,
	edgeID string,
	input Values,
	boundary plan.ArtifactScopeTransition,
	loopActions []*plan.ArtifactScopeTransition,
) error {
	if len(loopActions) != 0 {
		return fmt.Errorf(
			"edge %q combines fan-out boundary with actionable loop transition %s.%s",
			edgeID,
			loopActions[0].OwnerID,
			loopActions[0].Action,
		)
	}
	reference, err := fanOutRefForTransition(transition.State, source, boundary.OwnerID)
	if err != nil {
		return err
	}
	if boundary.Action == "arrive" {
		return kernel.recordFanOutArrival(
			transition,
			reference.ScopeID,
			reference.ItemID,
			input,
			source.TaskWatchFrontier,
		)
	}
	status := WorkItemFailed
	fallbackCode := "run.unhandled_failure"
	failure := cloneFailure(source.Failure)
	if failure == nil {
		failure = failureForCodeAt(fallbackCode, "reducer", workItemFailurePathFromRef(*reference))
	} else if strings.TrimSpace(failure.Code) == "" {
		failure.Code = fallbackCode
		failure.Problem = firstNonEmpty(failure.Problem, FailureProblem(fallbackCode))
		failure.Hint = firstNonEmpty(failure.Hint, FailureHint(fallbackCode))
		if strings.TrimSpace(failure.Path) == "" {
			failure.Path = workItemFailurePathFromRef(*reference)
		}
	}
	return kernel.failFanOutItem(transition, *reference, status, failure)
}

func (kernel *Kernel) openParallel(transition *Transition, ownerActivationID string, selected []string) error {
	state := transition.State
	owner, err := state.activation(ownerActivationID)
	if err != nil {
		return err
	}
	artifactScope, ok := kernel.graph.Scope(plan.ArtifactScopeParallel, owner.NodeID)
	if !ok || artifactScope.Parallel == nil {
		return fmt.Errorf("parallel owner %q has no scope", owner.NodeID)
	}
	activated := make(map[string]bool, len(selected))
	for _, port := range selected {
		activated[port] = true
	}
	scope := state.appendScope(plan.ArtifactScopeParallel, owner.NodeID, owner.Lineage)
	branches := make([]BranchState, len(artifactScope.Parallel.Branches))
	for index, branch := range artifactScope.Parallel.Branches {
		status := BranchInactive
		if activated[branch.Port] {
			status = BranchActive
		}
		branches[index] = BranchState{Port: branch.Port, Status: status}
	}
	join, err := state.appendActivation(artifactScope.Parallel.JoinID, "", Values{}, owner.Lineage, ActivationWaiting)
	if err != nil {
		return err
	}
	join.ParentActivationID = ownerActivationID
	join.TriggerScopeID = scope.ID
	join.FanOutResourceCausal = owner.FanOutResourceCausal
	scope, _ = state.scope(scope.ID) // appendActivation may reallocate unrelated slice only; explicit refresh documents identity.
	scope.Parallel = &ParallelState{
		ActivationSet:    artifactScope.Parallel.ActivationSet,
		Mode:             artifactScope.Parallel.Mode,
		JoinNodeID:       artifactScope.Parallel.JoinID,
		JoinActivationID: join.ID,
		Branches:         branches,
	}
	owner, _ = state.activation(ownerActivationID)
	owner.OwnedScopeIDs = append(owner.OwnedScopeIDs, scope.ID)
	transition.Events = append(transition.Events, Event{Kind: "parallel_opened", ActivationID: owner.ID, ScopeID: scope.ID})
	return nil
}

func (kernel *Kernel) recordParallelArrival(transition *Transition, scopeID, branchPort string, value Values) error {
	state := transition.State
	scope, err := state.scope(scopeID)
	if err != nil {
		return err
	}
	if scope.Status != ScopeActive {
		return nil // a deterministic late loser arrival
	}
	branch := parallelBranch(scope, branchPort)
	if branch == nil {
		return fmt.Errorf("scope %q has no branch %q", scopeID, branchPort)
	}
	if branch.Status == BranchSucceeded {
		if reflect.DeepEqual(branch.Value, value) {
			return nil
		}
		return fmt.Errorf("branch %q replay has conflicting value", branchPort)
	}
	if branch.Status != BranchActive {
		return nil
	}
	branch.Status = BranchSucceeded
	branch.Value = cloneValues(value)
	transition.Events = append(transition.Events, Event{Kind: "branch_arrived", ScopeID: scope.ID, Port: branchPort})
	return kernel.resolveParallel(transition, scope.ID)
}

func (kernel *Kernel) failParallelBranch(transition *Transition, reference ScopeRef, code string) error {
	state := transition.State
	scope, err := state.scope(reference.ScopeID)
	if err != nil || scope.Parallel == nil {
		return fmt.Errorf("parallel failure scope %q is invalid", reference.ScopeID)
	}
	if scope.Status != ScopeActive {
		return nil
	}
	branch := parallelBranch(scope, reference.Branch)
	if branch == nil {
		return fmt.Errorf("parallel scope has no branch %q", reference.Branch)
	}
	if branch.Status == BranchActive {
		branch.Status = BranchFailed
		branch.Failure = &Failure{Code: code}
	}
	return kernel.resolveParallel(transition, scope.ID)
}

func (kernel *Kernel) resolveParallel(transition *Transition, scopeID string) error {
	state := transition.State
	scope, err := state.scope(scopeID)
	if err != nil || scope.Parallel == nil || scope.Status != ScopeActive {
		return err
	}
	parallel := scope.Parallel
	active, succeeded, failed := 0, 0, 0
	winner := ""
	for _, branch := range parallel.Branches {
		if branch.Status == BranchInactive {
			continue
		}
		active++
		switch branch.Status {
		case BranchSucceeded:
			succeeded++
			if winner == "" {
				winner = branch.Port
			}
		case BranchFailed, BranchCanceled:
			failed++
		}
	}
	resolved := false
	if parallel.Mode == plan.JoinAny {
		if winner != "" {
			parallel.Outcome = "completed"
			parallel.WinnerBranch = winner
			scope.Status = ScopeSucceeded
			resolved = true
		} else if active > 0 && failed == active {
			parallel.Outcome = "failed"
			scope.Status = ScopeFailed
			resolved = true
		}
	} else {
		if failed > 0 {
			parallel.Outcome = "failed"
			scope.Status = ScopeFailed
			resolved = true
		} else if active > 0 && succeeded == active {
			parallel.Outcome = "completed"
			scope.Status = ScopeSucceeded
			resolved = true
		}
	}
	if !resolved {
		return nil
	}
	if err := collapseScopeTaskWatchFrontiers(
		state,
		scope.ID,
		parallel.JoinActivationID,
		"",
	); err != nil {
		return err
	}
	scope, _ = state.scope(scopeID)
	parallel = scope.Parallel
	cancelScopeWork(state, scope.ID, "parallel.resolved")
	join, err := state.activation(parallel.JoinActivationID)
	if err != nil {
		return err
	}
	if join.Status == ActivationWaiting {
		join.Status = ActivationReady
	}
	transition.Events = append(transition.Events, Event{Kind: "parallel_resolved", ScopeID: scope.ID, Port: parallel.Outcome})
	return nil
}

func (kernel *Kernel) executeJoin(transition *Transition, activationID string) error {
	state := transition.State
	activation, _ := state.activation(activationID)
	scope, err := state.scope(activation.TriggerScopeID)
	if err != nil {
		return fmt.Errorf("join activation %q has no trigger scope", activationID)
	}
	outcome := ""
	switch {
	case scope.Parallel != nil:
		outcome = scope.Parallel.Outcome
	case scope.FanOut != nil:
		outcome = scope.FanOut.Outcome
	default:
		return fmt.Errorf("join activation %q trigger is not a join scope", activationID)
	}
	if outcome == "" {
		return fmt.Errorf("join activation %q is not resolved", activationID)
	}
	if outcome != "completed" {
		return kernel.finishPorts(transition, activationID, []portValue{{Port: outcome, Value: Values{}}})
	}
	node, _ := kernel.graph.Node(activation.NodeID)
	context, err := kernel.evalContext(state, activation)
	if err != nil {
		return err
	}
	var result Values
	if scope.FanOut != nil {
		result, err = evalFanOutEntryBindings(
			"join.merge",
			node.Config.Join.Merge,
			node.Config.Join.Result,
			context,
		)
	} else {
		result, err = evalBindings("join.merge", node.Config.Join.Merge, node.Config.Join.Result, context)
	}
	if err != nil {
		if isFanOutResourceLimit(err) {
			activation.Failure = &Failure{
				Code:    workItemResourceLimitFailureCode,
				Message: workItemResourceLimitMessage,
				Problem: FailureProblem(workItemResourceLimitFailureCode),
				Hint:    FailureHint(workItemResourceLimitFailureCode),
			}
		} else {
			activation.Failure = failureFromCause("join.merge_failed", "reducer", err)
		}
		return kernel.finishPorts(transition, activationID, []portValue{{Port: "failed", Value: Values{}}})
	}
	return kernel.finishPorts(transition, activationID, []portValue{{Port: "completed", Value: result}})
}

func (kernel *Kernel) executeLoop(transition *Transition, activationID string) error {
	state := transition.State
	activation, _ := state.activation(activationID)
	node, _ := kernel.graph.Node(activation.NodeID)
	config := node.Config.Loop
	var scope *Scope
	var err error
	switch activation.InputPort {
	case "enter":
		scope = state.appendScope(plan.ArtifactScopeLoop, node.ID, activation.Lineage)
		scope.Loop = &LoopState{
			Iteration: 1, Mode: config.Mode, Iterations: config.Iterations,
			State: cloneValues(activation.Input),
		}
		activation, _ = state.activation(activationID)
		activation.TriggerScopeID = scope.ID
		activation.OwnedScopeIDs = append(activation.OwnedScopeIDs, scope.ID)
		transition.Events = append(transition.Events, Event{Kind: "loop_opened", ActivationID: activation.ID, ScopeID: scope.ID})
		return kernel.finishPorts(transition, activationID, []portValue{{Port: "body", Value: cloneValues(scope.Loop.State)}})
	case "continue":
		scope, err = state.scope(activation.TriggerScopeID)
		if err != nil || scope.Loop == nil || scope.Status != ScopeActive {
			return fmt.Errorf("loop continuation scope is invalid")
		}
		scope.Loop.State = cloneValues(activation.Input)
		if scope.Loop.Iteration >= config.Iterations {
			scope.Status = ScopeSucceeded
			transition.Events = append(transition.Events, Event{Kind: "loop_completed", ActivationID: activation.ID, ScopeID: scope.ID})
			return kernel.finishPorts(transition, activationID, []portValue{{Port: "done", Value: cloneValues(scope.Loop.State)}})
		}
		scope.Loop.Iteration++
		transition.Events = append(transition.Events, Event{Kind: "loop_continued", ActivationID: activation.ID, ScopeID: scope.ID})
		return kernel.finishPorts(transition, activationID, []portValue{{Port: "body", Value: cloneValues(scope.Loop.State)}})
	case "break":
		scope, err = state.scope(activation.TriggerScopeID)
		if err != nil || scope.Loop == nil || scope.Status != ScopeActive {
			return fmt.Errorf("loop break scope is invalid")
		}
		scope.Loop.State = cloneValues(activation.Input)
		scope.Status = ScopeSucceeded
		transition.Events = append(transition.Events, Event{Kind: "loop_broken", ActivationID: activation.ID, ScopeID: scope.ID})
		return kernel.finishPorts(transition, activationID, []portValue{{Port: "done", Value: cloneValues(scope.Loop.State)}})
	default:
		return fmt.Errorf("loop activation has unsupported input port %q", activation.InputPort)
	}
}

func (kernel *Kernel) evalContext(state *State, activation *Activation) (EvalContext, error) {
	context := EvalContext{
		PlanInput:        state.Input,
		ActivationInput:  activation.Input,
		ActivationOutput: activation.Output,
		RunState:         state.Variables,
		Loops:            map[string]LoopValues{},
		Joins:            map[string]JoinValues{},
		FanOutItems:      map[string]Values{},
		JoinItems:        map[string][]any{},
	}
	seen := map[string]bool{}
	for _, reference := range activation.Lineage {
		if seen[reference.ScopeID] {
			continue
		}
		seen[reference.ScopeID] = true
		scope, err := state.scope(reference.ScopeID)
		if err != nil {
			return EvalContext{}, err
		}
		if scope.Loop != nil {
			context.Loops[scope.OwnerID] = kernel.loopValues(scope)
		}
		if scope.FanOut != nil {
			item := fanOutItemByID(scope, reference.ItemID)
			if item == nil {
				return EvalContext{}, fmt.Errorf(
					"activation %q fan-out lineage references unknown item %q",
					activation.ID,
					reference.ItemID,
				)
			}
			context.FanOutItems[scope.OwnerID] = fanOutItemValues(scope.FanOut, *item)
		}
	}
	if activation.TriggerScopeID != "" {
		scope, err := state.scope(activation.TriggerScopeID)
		if err != nil {
			return EvalContext{}, err
		}
		if scope.Loop != nil {
			context.Loops[scope.OwnerID] = kernel.loopValues(scope)
		}
		if scope.Parallel != nil {
			branches := map[string]Values{}
			for _, branch := range scope.Parallel.Branches {
				if branch.Status == BranchSucceeded {
					branches[branch.Port] = branch.Value
				}
			}
			winnerValue := Values(nil)
			if winner := parallelBranch(scope, scope.Parallel.WinnerBranch); winner != nil {
				winnerValue = winner.Value
			}
			context.Joins[scope.OwnerID] = JoinValues{Branches: branches, WinnerBranch: scope.Parallel.WinnerBranch, WinnerValue: winnerValue}
		}
		if scope.FanOut != nil {
			context.JoinItems[scope.OwnerID] = fanOutResultEnvelopes(scope)
		}
	}
	return context, nil
}

func (kernel *Kernel) loopValues(scope *Scope) LoopValues {
	values := LoopValues{
		State:     scope.Loop.State,
		Iteration: scope.Loop.Iteration,
	}
	if node, ok := kernel.graph.Node(scope.OwnerID); ok && node.Config.Loop != nil {
		values.IsLast = scope.Loop.Iteration >= node.Config.Loop.Iterations
	}
	return values
}

func evalBindings(path string, bindings []plan.ArtifactBinding, contract plan.ArtifactContract, context EvalContext) (Values, error) {
	values := make(Values, len(bindings))
	for _, binding := range bindings {
		value, err := EvalExpression(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
	}
	return NormalizeContract(path, contract, values)
}

func waitInstant(value any, valueType plan.ValueType, now time.Time) (time.Time, error) {
	switch valueType {
	case plan.TypeDuration:
		number, err := normalizeNumber(value)
		if err != nil {
			return time.Time{}, err
		}
		nanos, err := strconv.ParseInt(number.String(), 10, 64)
		if err != nil || nanos <= 0 {
			return time.Time{}, fmt.Errorf("duration is not positive int64 nanoseconds")
		}
		return now.Add(time.Duration(nanos)), nil
	case plan.TypeDatetime:
		text, ok := value.(string)
		if !ok {
			return time.Time{}, fmt.Errorf("datetime is %T", value)
		}
		return time.Parse(time.RFC3339Nano, text)
	default:
		return time.Time{}, fmt.Errorf("wait type %q is unsupported", valueType)
	}
}

func parallelScopeForTransition(state *State, source *Activation, transition plan.ArtifactScopeTransition) (*Scope, error) {
	for index := len(source.Lineage) - 1; index >= 0; index-- {
		reference := source.Lineage[index]
		if reference.Kind == plan.ArtifactScopeParallel && reference.OwnerID == transition.OwnerID {
			return state.scope(reference.ScopeID)
		}
	}
	for index := len(source.OwnedScopeIDs) - 1; index >= 0; index-- {
		scope, err := state.scope(source.OwnedScopeIDs[index])
		if err == nil && scope.Kind == plan.ArtifactScopeParallel && scope.OwnerID == transition.OwnerID {
			return scope, nil
		}
	}
	return nil, fmt.Errorf("activation %q has no parallel scope %q", source.ID, transition.OwnerID)
}

func loopScopeForSource(state *State, source *Activation, ownerID string) (*Scope, error) {
	if source.TriggerScopeID != "" {
		if scope, err := state.scope(source.TriggerScopeID); err == nil && scope.Kind == plan.ArtifactScopeLoop && scope.OwnerID == ownerID {
			return scope, nil
		}
	}
	for index := len(source.Lineage) - 1; index >= 0; index-- {
		reference := source.Lineage[index]
		if reference.Kind == plan.ArtifactScopeLoop && reference.OwnerID == ownerID {
			return state.scope(reference.ScopeID)
		}
	}
	for index := len(source.OwnedScopeIDs) - 1; index >= 0; index-- {
		if scope, err := state.scope(source.OwnedScopeIDs[index]); err == nil && scope.Kind == plan.ArtifactScopeLoop && scope.OwnerID == ownerID {
			return scope, nil
		}
	}
	return nil, fmt.Errorf("activation %q has no loop scope %q", source.ID, ownerID)
}

func fanOutRefForTransition(state *State, source *Activation, ownerID string) (*ScopeRef, error) {
	for index := len(source.Lineage) - 1; index >= 0; index-- {
		reference := source.Lineage[index]
		if reference.Kind != plan.ArtifactScopeFanOut || reference.OwnerID != ownerID {
			continue
		}
		scope, err := state.scope(reference.ScopeID)
		if err == nil && scope.FanOut != nil {
			copyRef := reference
			return &copyRef, nil
		}
	}
	return nil, fmt.Errorf("activation %q has no fan-out item scope %q", source.ID, ownerID)
}

func removeScopeRef(lineage []ScopeRef, scopeID string) []ScopeRef {
	result := make([]ScopeRef, 0, len(lineage))
	for _, reference := range lineage {
		if reference.ScopeID != scopeID {
			result = append(result, reference)
		}
	}
	return result
}

func activeFailureScopeRef(state *State, lineage []ScopeRef) *ScopeRef {
	for index := len(lineage) - 1; index >= 0; index-- {
		reference := lineage[index]
		if reference.Kind != plan.ArtifactScopeFanOut && reference.Kind != plan.ArtifactScopeParallel {
			continue
		}
		scope, err := state.scope(reference.ScopeID)
		if err == nil && scope.Status == ScopeActive {
			copyRef := reference
			return &copyRef
		}
	}
	return nil
}

func activeFanOutRef(state *State, lineage []ScopeRef) *ScopeRef {
	for index := len(lineage) - 1; index >= 0; index-- {
		reference := lineage[index]
		if reference.Kind != plan.ArtifactScopeFanOut {
			continue
		}
		scope, err := state.scope(reference.ScopeID)
		if err == nil && scope.Status == ScopeActive {
			copyRef := reference
			return &copyRef
		}
	}
	return nil
}

func parallelBranch(scope *Scope, port string) *BranchState {
	if scope == nil || scope.Parallel == nil {
		return nil
	}
	for index := range scope.Parallel.Branches {
		if scope.Parallel.Branches[index].Port == port {
			return &scope.Parallel.Branches[index]
		}
	}
	return nil
}

func cancelScopeWork(state *State, scopeID, 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 {
				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 {
				nested.Status = ScopeCanceled
				cancelFanOutItems(nested.FanOut, code)
				break
			}
		}
	}
	if scope, err := state.scope(scopeID); err == nil && scope.Parallel != nil {
		for index := range scope.Parallel.Branches {
			if scope.Parallel.Branches[index].Status == BranchActive {
				scope.Parallel.Branches[index].Status = BranchCanceled
			}
		}
	}
	if scope, err := state.scope(scopeID); err == nil && scope.FanOut != nil {
		cancelFanOutItems(scope.FanOut, code)
	}
}

func cancelOpenWork(state *State, exceptActivationID string) {
	for index := range state.Activations {
		activation := &state.Activations[index]
		if activation.ID != exceptActivationID && (activation.Status == ActivationReady || activation.Status == ActivationWaiting) {
			activation.Status = ActivationCanceled
		}
	}
	for index := range state.Scopes {
		if state.Scopes[index].Status == ScopeActive {
			state.Scopes[index].Status = ScopeCanceled
			cancelFanOutItems(state.Scopes[index].FanOut, "run.terminal")
		}
	}
}

func cancelFanOutItems(fanOut *FanOutState, code string) {
	if fanOut == nil {
		return
	}
	for index := range fanOut.Items {
		item := &fanOut.Items[index]
		if item.Status != WorkItemPending && item.Status != WorkItemRunning {
			continue
		}
		item.Status = WorkItemCanceled
		item.Failure = &Failure{Code: code}
	}
}

func refreshStatus(state *State) {
	if state.Status.Terminal() {
		return
	}
	ready, waiting := false, false
	for _, activation := range state.Activations {
		ready = ready || activation.Status == ActivationReady
		waiting = waiting || activation.Status == ActivationWaiting
	}
	if ready {
		state.Status = StatusRunning
		return
	}
	if waiting {
		state.Status = StatusWaiting
		return
	}
	state.Status = StatusFailed
	state.Failure = &Failure{Code: "run.graph_deadlocked", Message: "no terminal, ready, or waiting activation remains", Stage: "reducer", Problem: FailureProblem("run.graph_deadlocked"), Hint: FailureHint("run.graph_deadlocked")}
}

// StableReadyIDs is useful to bounded workers and tests without exposing
// mutable activation pointers.
func StableReadyIDs(state *State) []string {
	ids := []string{}
	if state == nil {
		return ids
	}
	for _, activation := range state.Activations {
		if activation.Status == ActivationReady {
			ids = append(ids, activation.ID)
		}
	}
	sort.Strings(ids)
	return ids
}

// JSONRoundTrip proves State contains only persistence-safe JSON values.
func JSONRoundTrip(state *State) (*State, error) {
	payload, err := json.Marshal(state)
	if err != nil {
		return nil, err
	}
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.UseNumber()
	var restored State
	if err := decoder.Decode(&restored); err != nil {
		return nil, err
	}
	return &restored, nil
}

// selectOneStrategyForConfig returns the config selectOne strategy, or empty
// when the compiled send_message carries no selectOne pool.
func selectOneStrategyForConfig(config *plan.ArtifactSendMessageConfig) plan.SelectOneStrategy {
	if config == nil || config.SelectOne == nil {
		return ""
	}
	switch plan.SelectOneStrategy(config.SelectOneStrategy).Normalized() {
	case plan.SelectOneStrategyRandom:
		return plan.SelectOneStrategyRandom
	default:
		return ""
	}
}

// effectiveSelectOneStrategy prefers dynamic (WorkItem), then config strategy,
// else empty for a fixed responder.
func effectiveSelectOneStrategy(dynamic plan.SelectOneStrategy, config plan.SelectOneStrategy) plan.SelectOneStrategy {
	if dynamic.Normalized() == plan.SelectOneStrategyRandom {
		return plan.SelectOneStrategyRandom
	}
	if config != "" && config.Normalized() == plan.SelectOneStrategyRandom {
		return plan.SelectOneStrategyRandom
	}
	return ""
}