package plan

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/workflow/runlimit"
	"gopkg.in/yaml.v3"
)

// Compiler resource budgets are part of the Plan authoring contract. They
// bound both persisted executable complexity and the amount of work performed
// by graph/scope analysis before a PlanVersion can be published.
const (
	MaxSourceBytes       = 1 << 20
	MaxSourceYAMLNesting = 64
	MaxSourceYAMLNodes   = 100_000
	MaxGraphNodes        = 512
	MaxGraphEdges        = 2048
	MaxSchemaFields      = 2048
	// MaxInputBytes bounds one normalized runtime input after Plan defaults and
	// every Binding/occurrence layer have been merged. Individual legal layers
	// must not combine into an Invocation payload larger than this boundary.
	MaxInputBytes        = 1 << 20
	MaxLiteralNesting    = 32
	MaxLiteralValues     = 8192
	MaxExpressionBytes   = 8 << 10
	MaxExpressionNesting = 64
	MaxTemplateBytes     = 64 << 10
	MaxLoopIterations    = 10_000
	// Dynamic fan-out has compiler-frozen concurrency and a runtime-enforced
	// callback output limit. These are language/runtime contract constants,
	// not mutable deployment defaults.
	DefaultFanOutMaxConcurrency = 4
	MaxFanOutConcurrency        = 64
	MaxWorkItems                = 256
	MaxWorkItemKeyBytes         = 256
	// MaxAllowedResponders is the single safe-upper-bound contract for the
	// ordered candidate pool declared by `responder.selectOne` (send_message)
	// and by a WorkItem's selectOne responder pool. The same constant is
	// enforced by Plan validation, Artifact decode, and runtime normalize so an
	// authored literal, a decoded artifact, and an evaluated dynamic candidate
	// expression all share one bound. It is deliberately no larger than
	// MaxFanOutConcurrency: a responder pool is an ordered selection among
	// candidates, not a broadcast, and every candidate beyond the first is only
	// a capacity-occupancy fallback.
	MaxAllowedResponders = MaxFanOutConcurrency
	// Dynamic WorkItems are duplicated into one effect receipt and one
	// activation output, both of which use the ordinary 8192-value persistence
	// boundary. These aggregate limits keep the compiler-owned WorkItems value
	// itself below that boundary while leaving headroom for the receipt envelope
	// and authored callback output.
	MaxWorkItemDependencyEdges        = MaxGraphEdges
	MaxWorkItemRelatedScopeReferences = MaxWorkItems * 2
	MaxWorkItemsJSONValues            = 6 * 1024
	MaxWorkItemsJSONBytes             = 1 << 20
	// MaxRunActivations is the shared upper bound for one materialized Run.
	// Runtime pagination probes one record beyond it, while the pure reducer
	// refuses to create the first activation that would exceed it.
	MaxRunActivations = runlimit.MaxRunActivations
	MaxRetryAttempts  = 100
	MaxArtifactBytes  = 8 << 20
	// MaxArtifactMaterializationUnits bounds the number of independently
	// materialized IR records before lowering. MaxArtifactProjectedBytes is a
	// deliberately roomier, conservative heap/wire projection; the exact
	// normalized artifact still has to pass MaxArtifactBytes after lowering.
	MaxArtifactMaterializationUnits = 131_072
	MaxArtifactProjectedBytes       = 64 << 20
	MaxArtifactJSONNesting          = 256
	MaxArtifactSchemaFields         = 16_384
	MaxArtifactExpressions          = 32_768
	MaxArtifactExprNesting          = 128
)

func sourceBudgetDiagnostics(raw []byte) []Diagnostic {
	if len(raw) > MaxSourceBytes {
		return []Diagnostic{{
			Severity: SeverityError,
			Code:     "resource.source_too_large",
			Path:     "$",
			Message:  fmt.Sprintf("PlanVersion source must not exceed %d bytes", MaxSourceBytes),
		}}
	}
	return sourceStructureBudgetDiagnostics(raw)
}

func artifactWireBudgetDiagnostics(artifact *Artifact) ([]Diagnostic, error) {
	payload, err := artifact.DeterministicJSON()
	if err != nil {
		return nil, err
	}
	if len(payload) <= MaxArtifactBytes {
		return nil, nil
	}
	return []Diagnostic{{
		Severity: SeverityError,
		Code:     "resource.artifact_too_large",
		Path:     "$",
		Message:  fmt.Sprintf("compiled Plan artifact must not exceed %d bytes", MaxArtifactBytes),
	}}, nil
}

// sourceStructureBudgetDiagnostics uses an iterative walk before indexSource's
// recursive position index. Malformed YAML is left to Parse so it retains the
// normal syntax diagnostic and source location.
func sourceStructureBudgetDiagnostics(raw []byte) []Diagnostic {
	decoder := yaml.NewDecoder(bytes.NewReader(raw))
	for {
		var root yaml.Node
		if err := decoder.Decode(&root); err != nil {
			if errors.Is(err, io.EOF) {
				return nil
			}
			return nil
		}
		type pendingNode struct {
			node  *yaml.Node
			depth int
		}
		pending := []pendingNode{{node: &root}}
		seen := 0
		for len(pending) != 0 {
			last := len(pending) - 1
			item := pending[last]
			pending = pending[:last]
			if item.node == nil {
				continue
			}
			seen++
			if seen > MaxSourceYAMLNodes {
				return []Diagnostic{{
					Severity: SeverityError, Code: "resource.yaml_nodes", Path: "$",
					Message: fmt.Sprintf("PlanVersion YAML must not contain more than %d structural nodes", MaxSourceYAMLNodes),
				}}
			}
			if item.depth > MaxSourceYAMLNesting {
				return []Diagnostic{{
					Severity: SeverityError, Code: "resource.yaml_nesting", Path: "$",
					Message: fmt.Sprintf("PlanVersion YAML nesting must not exceed %d", MaxSourceYAMLNesting),
				}}
			}
			for index := len(item.node.Content) - 1; index >= 0; index-- {
				pending = append(pending, pendingNode{node: item.node.Content[index], depth: item.depth + 1})
			}
		}
	}
}

// documentBudgetDiagnostics runs before graph indexing and scope analysis.
// Returning immediately on a breached budget is intentional: diagnostics do
// not justify performing expensive work on a document that cannot be stored.
func documentBudgetDiagnostics(document *Document) []Diagnostic {
	if document == nil {
		return nil
	}
	diagnostics := make([]Diagnostic, 0, 4)
	if len(document.Graph.Nodes) > MaxGraphNodes {
		diagnostics = append(diagnostics, Diagnostic{
			Severity: SeverityError,
			Code:     "resource.too_many_nodes",
			Path:     "graph.nodes",
			Message:  fmt.Sprintf("graph must not contain more than %d nodes", MaxGraphNodes),
		})
	}
	if len(document.Graph.Edges) > MaxGraphEdges {
		diagnostics = append(diagnostics, Diagnostic{
			Severity: SeverityError,
			Code:     "resource.too_many_edges",
			Path:     "graph.edges",
			Message:  fmt.Sprintf("graph must not contain more than %d edges", MaxGraphEdges),
		})
	}
	if countDocumentSchemaFields(document, MaxSchemaFields+1) > MaxSchemaFields {
		diagnostics = append(diagnostics, Diagnostic{
			Severity: SeverityError,
			Code:     "resource.too_many_schema_fields",
			Path:     "$",
			Message:  fmt.Sprintf("PlanVersion must not declare more than %d schema fields", MaxSchemaFields),
		})
	}
	diagnostics = append(diagnostics, documentRelatedScopeBudgetDiagnostics(document)...)
	if len(diagnostics) != 0 {
		sortDiagnostics(diagnostics)
		return diagnostics
	}
	if nesting, values := documentLiteralBudget(document); nesting > MaxLiteralNesting || values > MaxLiteralValues {
		message := fmt.Sprintf("PlanVersion literals must not exceed nesting %d or %d total values", MaxLiteralNesting, MaxLiteralValues)
		diagnostics = append(diagnostics, Diagnostic{
			Severity: SeverityError,
			Code:     "resource.literal_budget",
			Path:     "$",
			Message:  message,
		})
	}
	if len(diagnostics) == 0 {
		diagnostics = append(diagnostics, artifactMaterializationBudgetDiagnostics(document, nil)...)
	}
	sortDiagnostics(diagnostics)
	return diagnostics
}

// artifactMaterializationBudgetDiagnostics projects the normalized IR before
// buildArtifact allocates it. In particular, it multiplies a contract's
// already-counted shape by the number of ports that materialize that contract;
// it never loops once per (field, port) or (field, edge) pair. A nil analysis
// is the pre-graph-analysis pass. The non-nil pass adds exact validated scope
// membership and transition costs.
func artifactMaterializationBudgetDiagnostics(document *Document, analysis *compilerAnalysis) []Diagnostic {
	if document == nil {
		return nil
	}
	projection := projectArtifactMaterialization(document, analysis)
	if projection.schemaFields <= MaxArtifactSchemaFields &&
		projection.expressions <= MaxArtifactExpressions &&
		projection.units <= MaxArtifactMaterializationUnits &&
		projection.projectedBytes <= MaxArtifactProjectedBytes {
		return nil
	}
	return []Diagnostic{{
		Severity: SeverityError,
		Code:     "resource.artifact_budget",
		Path:     "$",
		Message: fmt.Sprintf(
			"projected compiled Plan artifact exceeds the materialization budget (%d schema fields, %d expression instructions, %d IR units, %d projected bytes)",
			MaxArtifactSchemaFields, MaxArtifactExpressions, MaxArtifactMaterializationUnits, MaxArtifactProjectedBytes,
		),
	}}
}

type artifactMaterializationProjection struct {
	units          uint64
	projectedBytes uint64
	schemaFields   uint64
	expressions    uint64
}

type artifactSchemaProjection struct {
	units  uint64
	bytes  uint64
	fields uint64
}

type artifactNodeProjection struct {
	node           *Node
	input          artifactSchemaProjection
	callbackOutput artifactSchemaProjection
	joinResult     artifactSchemaProjection
	loopState      artifactSchemaProjection
}

const (
	artifactIRRecordBytes       = 48
	artifactSchemaRecordBytes   = 80
	artifactExpressionNodeBytes = 96
)

func projectArtifactMaterialization(document *Document, analysis *compilerAnalysis) artifactMaterializationProjection {
	projection := artifactMaterializationProjection{}
	projection.addUnits(5) // envelope, contracts, entry, nodes, and edges containers
	projection.addBytes(512)

	planInput := projectArtifactSchemaMap(document.Input)
	planOutput := projectArtifactSchemaMap(document.Output)
	planState := projectArtifactSchemaMap(document.State)
	projection.addContract(planInput, 1)
	projection.addContract(planOutput, 1)
	projection.addContract(planState, 1)

	nodes := make(map[string]artifactNodeProjection, len(document.Graph.Nodes))
	for index := range document.Graph.Nodes {
		node := &document.Graph.Nodes[index]
		value := artifactNodeProjection{node: node, input: projectArtifactSchemaMap(node.Input)}
		if node.SendMessage != nil {
			value.callbackOutput = projectArtifactSchemaMap(node.SendMessage.CallbackOutput)
		}
		if node.Join != nil {
			value.joinResult = projectArtifactSchemaMap(node.Join.Result)
		}
		if node.Loop != nil {
			value.loopState = projectArtifactSchemaMap(node.Loop.State)
		}
		nodes[node.ID] = value
	}

	projection.addUnits(uint64(len(document.Graph.Nodes) + len(document.Graph.Edges)))
	projection.addBytes(saturatedMultiply(uint64(len(document.Graph.Nodes)+len(document.Graph.Edges)), artifactIRRecordBytes, MaxArtifactProjectedBytes+1))

	entryTarget := nodes[document.Graph.Entry.NodeID]
	projection.addBytes(projectedJSONStringsContentBytes(document.Graph.Entry.NodeID, document.Graph.Entry.Port))
	if document.Graph.Entry.Input != nil {
		projection.addBindings(document.Graph.Entry.Input)
	} else if entryTarget.node != nil {
		target := projectedTargetContract(document, entryTarget, document.Graph.Entry.Port)
		projection.addImplicitBindings(projectedOmittedBindingUpperBound(len(document.Input), target.topLevelFields))
	}

	for index := range document.Graph.Nodes {
		node := &document.Graph.Nodes[index]
		value := nodes[node.ID]
		// Description is authoring-only and is not lowered into the artifact.
		projection.addBytes(projectedJSONStringsContentBytes(node.ID, string(node.Type)))

		// Input-port contracts.
		switch node.Type {
		case NodeLoop:
			projection.addPorts(3)
			projection.addContract(value.loopState, 3) // enter, continue, and break
		case NodeJoin:
			branches := projectedJoinBranchCount(nodes, node)
			projection.addPorts(branches)
			projection.addContract(value.input, branches)
		case NodeEnd:
			projection.addPorts(1)
			if node.End != nil && node.End.Outcome == EndSucceeded {
				projection.addContract(planOutput, 1)
			}
		default:
			projection.addPorts(1)
			projection.addContract(value.input, 1)
		}

		// Output-port contracts and typed config records.
		switch node.Type {
		case NodeSendMessage:
			projection.addPorts(3)
			if node.SendMessage != nil {
				projection.addSendMessageConfig(node.SendMessage, value.callbackOutput)
			}
		case NodeSetVariable:
			projection.addPorts(2)
			if node.SetVariable != nil {
				for _, assignment := range node.SetVariable.Assignments {
					projection.addUnits(1)
					projection.addBytes(saturatedAdd(artifactIRRecordBytes, projectedJSONStringContentBytes(assignment.State), MaxArtifactProjectedBytes+1))
					projection.addValueExpression(assignment.Value)
				}
			}
		case NodeWait:
			projection.addPorts(3)
			if node.Wait != nil {
				projection.addValueExpression(node.Wait.Until)
			}
		case NodeRequestApproval:
			projection.addPorts(4)
			if node.RequestApproval != nil {
				projection.addTemplate(node.RequestApproval.Assignee)
				projection.addTemplate(node.RequestApproval.Summary)
				if node.RequestApproval.FeedbackScope != nil {
					projection.addValueExpression(*node.RequestApproval.FeedbackScope)
				}
			}
		case NodeSwitch:
			if node.Switch != nil {
				projection.addPorts(uint64(len(node.Switch.Cases)))
				projection.addContract(value.input, uint64(len(node.Switch.Cases)))
				for _, switchCase := range node.Switch.Cases {
					projection.addUnits(1)
					projection.addBytes(saturatedAdd(artifactIRRecordBytes, projectedJSONStringContentBytes(switchCase.Port), MaxArtifactProjectedBytes+1))
					if switchCase.When != nil {
						projection.addExpressionSource(string(*switchCase.When))
					}
				}
			}
		case NodeFork:
			if node.Fork != nil {
				branches := uint64(len(node.Fork.Branches))
				projection.addPorts(branches)
				projection.addContract(value.input, branches)
				projection.addUnits(branches)
				for _, branch := range node.Fork.Branches {
					projection.addBytes(saturatedAdd(16, projectedJSONStringContentBytes(branch), MaxArtifactProjectedBytes+1))
				}
			}
		case NodeFanOut:
			projection.addPorts(1)
			projection.addContract(projectArtifactSchemaMap(FanOutItemContract()), 1)
			if node.FanOut != nil {
				projection.addValueExpression(node.FanOut.Items)
			}
		case NodeJoin:
			projection.addPorts(3)
			projection.addContract(value.joinResult, 1) // completed output
			projection.addContract(value.joinResult, 1) // config result
			if node.Join != nil {
				projection.addBindings(node.Join.Merge)
			}
		case NodeLoop:
			projection.addPorts(3)
			projection.addContract(value.loopState, 3) // body, done, and config state
		case NodeEnd:
			// End has no output ports.
		}
	}

	for index := range document.Graph.Edges {
		edge := &document.Graph.Edges[index]
		projection.addBytes(projectedJSONStringsContentBytes(edge.ID, edge.Source.NodeID, edge.Source.Port, edge.Target.NodeID, edge.Target.Port))
		if edge.Bindings != nil {
			projection.addBindings(edge.Bindings)
			continue
		}
		source, target := nodes[edge.Source.NodeID], nodes[edge.Target.NodeID]
		sourceFields, exposes := projectedSourceTopLevelFields(source, edge.Source.Port)
		if !exposes || target.node == nil {
			continue
		}
		targetFields := projectedTargetContract(document, target, edge.Target.Port).topLevelFields
		projection.addImplicitBindings(projectedOmittedBindingUpperBound(sourceFields, targetFields))
	}

	if analysis != nil {
		projection.addScopes(*analysis)
	}
	return projection
}

type projectedContract struct {
	topLevelFields int
}

func projectedTargetContract(document *Document, node artifactNodeProjection, port string) projectedContract {
	if node.node == nil {
		return projectedContract{}
	}
	switch node.node.Type {
	case NodeLoop:
		if node.node.Loop == nil {
			return projectedContract{}
		}
		return projectedContract{topLevelFields: len(node.node.Loop.State)}
	case NodeEnd:
		if node.node.End != nil && node.node.End.Outcome == EndSucceeded {
			return projectedContract{topLevelFields: len(document.Output)}
		}
		return projectedContract{}
	default:
		return projectedContract{topLevelFields: len(node.node.Input)}
	}
}

func projectedSourceTopLevelFields(node artifactNodeProjection, port string) (int, bool) {
	if node.node == nil {
		return 0, false
	}
	switch node.node.Type {
	case NodeFork, NodeSwitch:
		return len(node.node.Input), true
	case NodeSendMessage:
		if port == "completed" && node.node.SendMessage != nil {
			fields := len(node.node.SendMessage.CallbackOutput)
			if node.node.SendMessage.StartThread {
				fields += 2
			}
			if node.node.SendMessage.hasWorkScope() {
				fields++
			}
			if node.node.SendMessage.WorkItems != nil {
				fields++
			}
			return fields, node.node.SendMessage.Completion == CompletionCallback || node.node.SendMessage.StartThread
		}
	case NodeFanOut:
		if port == "item" {
			return len(FanOutItemContract()), true
		}
	case NodeJoin:
		if port == "completed" && node.node.Join != nil {
			return len(node.node.Join.Result), true
		}
	case NodeLoop:
		if node.node.Loop != nil {
			switch port {
			case "body", "done":
				return len(node.node.Loop.State), true
			}
		}
	}
	return 0, false
}

func projectedJoinBranchCount(nodes map[string]artifactNodeProjection, join *Node) uint64 {
	if join == nil || join.Join == nil {
		return 0
	}
	if join.Join.FanOutID != "" {
		return 1
	}
	owner := nodes[join.Join.ForkID].node
	if owner == nil {
		return 0
	}
	if owner.Type == NodeFork && owner.Fork != nil {
		return uint64(len(owner.Fork.Branches))
	}
	if owner.Type == NodeSwitch && owner.Switch != nil && owner.Switch.Mode == SwitchAllMatch {
		return uint64(len(owner.Switch.Cases))
	}
	return 0
}

func projectArtifactSchemaMap(values map[string]Schema) artifactSchemaProjection {
	projection := artifactSchemaProjection{}
	for name, schema := range values {
		first := true
		for current := &schema; current != nil; current = current.Items {
			projection.fields = saturatedAdd(projection.fields, 1, MaxArtifactSchemaFields+1)
			projection.units = saturatedAdd(projection.units, 1, MaxArtifactMaterializationUnits+1)
			bytes := saturatedAdd(artifactSchemaRecordBytes, projectedJSONStringContentBytes(string(current.Type)), MaxArtifactProjectedBytes+1)
			if first {
				bytes = saturatedAdd(bytes, projectedJSONStringContentBytes(name), MaxArtifactProjectedBytes+1)
				first = false
			}
			for _, value := range current.Values {
				projection.units = saturatedAdd(projection.units, 1, MaxArtifactMaterializationUnits+1)
				bytes = saturatedAdd(bytes, saturatedAdd(8, projectedJSONStringContentBytes(value), MaxArtifactProjectedBytes+1), MaxArtifactProjectedBytes+1)
			}
			if current.Default != nil {
				literalUnits, literalBytes := projectArtifactLiteral(current.Default)
				projection.units = saturatedAdd(projection.units, literalUnits, MaxArtifactMaterializationUnits+1)
				bytes = saturatedAdd(bytes, literalBytes, MaxArtifactProjectedBytes+1)
			}
			projection.bytes = saturatedAdd(projection.bytes, bytes, MaxArtifactProjectedBytes+1)
		}
	}
	return projection
}

func projectArtifactLiteral(value any) (units, bytes uint64) {
	pending := []any{value}
	for len(pending) != 0 {
		last := len(pending) - 1
		current := pending[last]
		pending = pending[:last]
		units = saturatedAdd(units, 1, MaxArtifactMaterializationUnits+1)
		switch typed := current.(type) {
		case nil:
			bytes = saturatedAdd(bytes, 4, MaxArtifactProjectedBytes+1)
		case string:
			bytes = saturatedAdd(bytes, projectedJSONStringBytes(typed), MaxArtifactProjectedBytes+1)
		case bool:
			bytes = saturatedAdd(bytes, 5, MaxArtifactProjectedBytes+1)
		case []any:
			bytes = saturatedAdd(bytes, 2+uint64(len(typed)), MaxArtifactProjectedBytes+1)
			pending = append(pending, typed...)
		case map[string]any:
			bytes = saturatedAdd(bytes, 2+uint64(len(typed)), MaxArtifactProjectedBytes+1)
			for key, child := range typed {
				bytes = saturatedAdd(bytes, saturatedAdd(1, projectedJSONStringBytes(key), MaxArtifactProjectedBytes+1), MaxArtifactProjectedBytes+1)
				pending = append(pending, child)
			}
		case map[any]any:
			bytes = saturatedAdd(bytes, 2+uint64(len(typed))*8, MaxArtifactProjectedBytes+1)
			for _, child := range typed {
				pending = append(pending, child)
			}
		default:
			// Numbers and the remaining supported scalar forms have a bounded
			// normalized representation. Unsupported values are rejected later.
			bytes = saturatedAdd(bytes, 32, MaxArtifactProjectedBytes+1)
		}
		if units > MaxArtifactMaterializationUnits || bytes > MaxArtifactProjectedBytes {
			return units, bytes
		}
	}
	return units, bytes
}

func (projection *artifactMaterializationProjection) addContract(source artifactSchemaProjection, copies uint64) {
	if projection == nil || copies == 0 {
		return
	}
	projection.schemaFields = saturatedAdd(
		projection.schemaFields,
		saturatedMultiply(source.fields, copies, MaxArtifactSchemaFields+1),
		MaxArtifactSchemaFields+1,
	)
	projection.addUnits(saturatedMultiply(saturatedAdd(source.units, 1, MaxArtifactMaterializationUnits+1), copies, MaxArtifactMaterializationUnits+1))
	projection.addBytes(saturatedMultiply(saturatedAdd(source.bytes, 16, MaxArtifactProjectedBytes+1), copies, MaxArtifactProjectedBytes+1))
}

func (projection *artifactMaterializationProjection) addPorts(count uint64) {
	projection.addUnits(count)
	projection.addBytes(saturatedMultiply(count, artifactIRRecordBytes, MaxArtifactProjectedBytes+1))
}

func (projection *artifactMaterializationProjection) addBindings(bindings map[string]ValueExpr) {
	for target, value := range bindings {
		projection.addUnits(1)
		projection.addBytes(saturatedAdd(artifactIRRecordBytes, projectedJSONStringContentBytes(target), MaxArtifactProjectedBytes+1))
		projection.addValueExpression(value)
	}
}

func (projection *artifactMaterializationProjection) addImplicitBindings(count int) {
	if count <= 0 {
		return
	}
	value := uint64(count)
	projection.addUnits(value)
	projection.addBytes(saturatedMultiply(value, artifactIRRecordBytes+32, MaxArtifactProjectedBytes+1))
	projection.addExpressionInstructions(value, saturatedMultiply(value, 32, MaxArtifactProjectedBytes+1))
}

func (projection *artifactMaterializationProjection) addValueExpression(value ValueExpr) {
	switch {
	case strings.TrimSpace(value.Ref) != "":
		projection.addExpressionInstructions(1, projectedJSONStringContentBytes(value.Ref))
	case value.Value != nil:
		units, bytes := projectArtifactLiteral(*value.Value)
		projection.addExpressionInstructions(1, bytes)
		projection.addUnits(units)
	case strings.TrimSpace(value.Expr) != "":
		projection.addExpressionSource(value.Expr)
	default:
		// Invalid one-of values are diagnosed by semantic validation. Counting
		// one placeholder keeps this pass total and deterministic.
		projection.addExpressionInstructions(1, 0)
	}
}

func (projection *artifactMaterializationProjection) addExpressionSource(source string) {
	trimmed := strings.TrimSpace(source)
	projection.addExpressionInstructions(projectedExpressionInstructions(trimmed), projectedJSONStringContentBytes(trimmed))
}

func (projection *artifactMaterializationProjection) addExpressionInstructions(count, payloadBytes uint64) {
	if count == 0 {
		count = 1
	}
	projection.expressions = saturatedAdd(projection.expressions, count, MaxArtifactExpressions+1)
	// One unit is the AST instruction and a second conservatively covers the
	// derived state-effect/load entry that the instruction may materialize.
	projection.addUnits(saturatedMultiply(count, 2, MaxArtifactMaterializationUnits+1))
	projection.addBytes(saturatedAdd(
		saturatedMultiply(count, artifactExpressionNodeBytes, MaxArtifactProjectedBytes+1),
		payloadBytes,
		MaxArtifactProjectedBytes+1,
	))
}

func (projection *artifactMaterializationProjection) addTemplate(source string) {
	position := 0
	for position < len(source) {
		start := strings.Index(source[position:], "{{")
		if start < 0 {
			if position < len(source) {
				projection.addUnits(1)
				projection.addBytes(saturatedAdd(32, projectedJSONStringContentLengthBytes(len(source)-position), MaxArtifactProjectedBytes+1))
			}
			return
		}
		start += position
		if start > position {
			projection.addUnits(1)
			projection.addBytes(saturatedAdd(32, projectedJSONStringContentLengthBytes(start-position), MaxArtifactProjectedBytes+1))
		}
		end := strings.Index(source[start+2:], "}}")
		if end < 0 {
			projection.addUnits(1)
			projection.addBytes(saturatedAdd(32, projectedJSONStringContentLengthBytes(len(source)-start), MaxArtifactProjectedBytes+1))
			return
		}
		end += start + 2
		projection.addUnits(1) // expression segment
		projection.addBytes(32)
		projection.addExpressionSource(source[start+2 : end])
		position = end + 2
	}
}

// projectedExpressionInstructions is an allocation-free lexical upper bound
// on expression AST nodes. Parentheses and commas do not materialize; array
// openings, operands, functions, unary operators, and binary operators do.
func projectedExpressionInstructions(source string) uint64 {
	var count uint64
	for index := 0; index < len(source); {
		character := source[index]
		switch {
		case character == ' ' || character == '\t' || character == '\r' || character == '\n':
			index++
		case character == '(' || character == ')' || character == ']' || character == ',':
			index++
		case character == '[':
			count++
			index++
		case character == '\'' || character == '"':
			count++
			quote := character
			index++
			for index < len(source) {
				if source[index] == '\\' {
					index += 2
					continue
				}
				if source[index] == quote {
					index++
					break
				}
				index++
			}
		case isProjectedExpressionWord(character):
			count++
			for index < len(source) && isProjectedExpressionWord(source[index]) {
				index++
			}
		default:
			count++
			if index+1 < len(source) && (source[index+1] == '=' || source[index+1] == character) {
				index += 2
			} else {
				index++
			}
		}
		if count > MaxArtifactExpressions {
			return MaxArtifactExpressions + 1
		}
	}
	if count == 0 {
		return 1
	}
	return count
}

func isProjectedExpressionWord(character byte) bool {
	return character >= 'a' && character <= 'z' ||
		character >= 'A' && character <= 'Z' ||
		character >= '0' && character <= '9' ||
		character == '_' || character == '.' || character == '$'
}

func projectedOmittedBindingUpperBound(sourceFields, targetFields int) int {
	if sourceFields < targetFields {
		return max(sourceFields, 0)
	}
	return max(targetFields, 0)
}

func (projection *artifactMaterializationProjection) addScopes(analysis compilerAnalysis) {
	membershipCounts := map[string]uint64{}
	addMembership := func(nodeID string) {
		membershipCounts[nodeID] = saturatedAdd(membershipCounts[nodeID], 1, MaxArtifactMaterializationUnits+1)
		// membership map entry, nodePaths frame, and final node ScopePath frame
		projection.addUnits(3)
		projection.addBytes(saturatedAdd(3*artifactIRRecordBytes, projectedJSONStringContentBytes(nodeID), MaxArtifactProjectedBytes+1))
	}
	addStringList := func(values []string) {
		projection.addUnits(uint64(len(values)))
		for _, value := range values {
			projection.addBytes(saturatedAdd(16, projectedJSONStringContentBytes(value), MaxArtifactProjectedBytes+1))
		}
	}

	scopeOwners := make([]string, 0, len(analysis.parallelScopes)+len(analysis.loopScopes)+len(analysis.fanOutScopes))
	for _, scope := range analysis.parallelScopes {
		scopeOwners = append(scopeOwners, scope.ownerID)
		projection.addUnits(2) // scope and parallel payload
		projection.addBytes(saturatedAdd(2*artifactIRRecordBytes, projectedJSONStringsContentBytes(scope.ownerID, scope.joinID, scope.activation), MaxArtifactProjectedBytes+1))
		for _, branch := range scope.branches {
			projection.addUnits(2) // branch payload and lowering membership record
			projection.addBytes(saturatedAdd(2*artifactIRRecordBytes, projectedJSONStringContentBytes(branch.port), MaxArtifactProjectedBytes+1))
			addStringList(branch.members)
			addStringList(branch.entryEdges)
			addStringList(branch.successArrivals)
			addStringList(branch.failedTerminals)
			for _, member := range branch.members {
				addMembership(member)
			}
			// Failed Ends can be shared. Counting every occurrence is a safe
			// upper bound and avoids a branch×terminal uniqueness pass.
			for _, terminal := range branch.failedTerminals {
				addMembership(terminal)
			}
			transitions := uint64(len(branch.entryEdges) + len(branch.successArrivals))
			projection.addUnits(transitions)
			projection.addBytes(saturatedMultiply(transitions, artifactIRRecordBytes, MaxArtifactProjectedBytes+1))
		}
	}
	for _, scope := range analysis.loopScopes {
		scopeOwners = append(scopeOwners, scope.ownerID)
		projection.addUnits(2) // scope and loop payload
		projection.addBytes(saturatedAdd(2*artifactIRRecordBytes, projectedJSONStringContentBytes(scope.ownerID), MaxArtifactProjectedBytes+1))
		for _, member := range scope.members {
			addMembership(member)
		}
		addStringList(scope.bodyEntryEdges)
		addStringList(scope.members)
		addStringList(scope.continueEdges)
		addStringList(scope.breakEdges)
		addStringList(scope.doneEdges)
		addStringList(scope.failedEdges)
		transitions := uint64(len(scope.bodyEntryEdges) + len(scope.continueEdges) + len(scope.breakEdges) + len(scope.doneEdges) + len(scope.failedEdges))
		projection.addUnits(transitions)
		projection.addBytes(saturatedMultiply(transitions, artifactIRRecordBytes, MaxArtifactProjectedBytes+1))
	}
	for _, scope := range analysis.fanOutScopes {
		scopeOwners = append(scopeOwners, scope.ownerID)
		projection.addUnits(2) // scope and fan_out payload
		projection.addBytes(saturatedAdd(2*artifactIRRecordBytes, projectedJSONStringsContentBytes(scope.ownerID, scope.joinID), MaxArtifactProjectedBytes+1))
		for _, member := range scope.members {
			addMembership(member)
		}
		for _, terminal := range scope.failedTerminals {
			addMembership(terminal)
		}
		addStringList(scope.members)
		addStringList(scope.bodyEntryEdges)
		addStringList(scope.successArrivals)
		addStringList(scope.failedTerminals)
		// Every non-terminal body member may materialize both compiler-owned
		// failed and timed_out edges. Under fan-out each such edge also carries
		// one scope-settlement transition, so include the worst case even
		// though authored recovery edges can make the actual count smaller.
		transitions := uint64(len(scope.bodyEntryEdges) + len(scope.successArrivals) + 2*len(scope.members))
		projection.addUnits(transitions)
		projection.addBytes(saturatedMultiply(transitions, artifactIRRecordBytes, MaxArtifactProjectedBytes+1))
	}
	for _, ownerID := range scopeOwners {
		parentFrames := membershipCounts[ownerID]
		projection.addUnits(parentFrames)
		projection.addBytes(saturatedMultiply(parentFrames, artifactIRRecordBytes, MaxArtifactProjectedBytes+1))
	}
}

func (projection *artifactMaterializationProjection) addUnits(value uint64) {
	if projection != nil {
		projection.units = saturatedAdd(projection.units, value, MaxArtifactMaterializationUnits+1)
	}
}

func (projection *artifactMaterializationProjection) addBytes(value uint64) {
	if projection != nil {
		projection.projectedBytes = saturatedAdd(projection.projectedBytes, value, MaxArtifactProjectedBytes+1)
	}
}

// encoding/json escapes each byte of the HTML-sensitive ASCII characters
// '&', '<', and '>' as a six-byte \u00XX sequence. Control bytes have the
// same maximum expansion, so six bytes per source byte is a conservative
// bound for every arbitrary string that can survive into compiled JSON.
func projectedJSONStringContentBytes(value string) uint64 {
	return projectedJSONStringContentLengthBytes(len(value))
}

func projectedJSONStringContentLengthBytes(length int) uint64 {
	if length <= 0 {
		return 0
	}
	return saturatedMultiply(uint64(length), 6, MaxArtifactProjectedBytes+1)
}

func projectedJSONStringBytes(value string) uint64 {
	return saturatedAdd(2, projectedJSONStringContentBytes(value), MaxArtifactProjectedBytes+1)
}

func projectedJSONStringsContentBytes(values ...string) uint64 {
	var total uint64
	for _, value := range values {
		total = saturatedAdd(total, projectedJSONStringContentBytes(value), MaxArtifactProjectedBytes+1)
	}
	return total
}

func saturatedAdd(left, right, cap uint64) uint64 {
	if left >= cap || right >= cap-left {
		return cap
	}
	return left + right
}

func saturatedMultiply(left, right, cap uint64) uint64 {
	if left == 0 || right == 0 {
		return 0
	}
	if left >= cap || right > cap/left {
		return cap
	}
	return left * right
}

func documentLiteralBudget(document *Document) (maximumDepth int, totalValues int) {
	if document == nil {
		return 0, 0
	}
	pending := make([]pendingLiteralValue, 0)
	appendSchemaDefaults := func(values map[string]Schema) {
		for _, schema := range values {
			for current := &schema; current != nil; current = current.Items {
				if current.Default != nil {
					pending = append(pending, pendingLiteralValue{value: current.Default})
				}
			}
		}
	}
	appendValueExpressions := func(values map[string]ValueExpr) {
		for _, expression := range values {
			if expression.Value != nil {
				pending = append(pending, pendingLiteralValue{value: *expression.Value})
			}
		}
	}
	appendSchemaDefaults(document.Input)
	appendSchemaDefaults(document.Output)
	appendSchemaDefaults(document.State)
	appendValueExpressions(document.Graph.Entry.Input)
	for index := range document.Graph.Nodes {
		node := &document.Graph.Nodes[index]
		appendSchemaDefaults(node.Input)
		if node.SendMessage != nil {
			appendSchemaDefaults(node.SendMessage.CallbackOutput)
			pending = appendSendMessageLiteralValues(pending, node.SendMessage)
		}
		if node.SetVariable != nil {
			for _, assignment := range node.SetVariable.Assignments {
				if assignment.Value.Value != nil {
					pending = append(pending, pendingLiteralValue{value: *assignment.Value.Value})
				}
			}
		}
		if node.Wait != nil && node.Wait.Until.Value != nil {
			pending = append(pending, pendingLiteralValue{value: *node.Wait.Until.Value})
		}
		if node.Join != nil {
			appendSchemaDefaults(node.Join.Result)
			appendValueExpressions(node.Join.Merge)
		}
		if node.FanOut != nil && node.FanOut.Items.Value != nil {
			pending = append(pending, pendingLiteralValue{value: *node.FanOut.Items.Value})
		}
		if node.Loop != nil {
			appendSchemaDefaults(node.Loop.State)
		}
	}
	for _, edge := range document.Graph.Edges {
		appendValueExpressions(edge.Bindings)
	}

	for len(pending) != 0 {
		last := len(pending) - 1
		item := pending[last]
		pending = pending[:last]
		totalValues++
		maximumDepth = max(maximumDepth, item.depth)
		if totalValues > MaxLiteralValues || maximumDepth > MaxLiteralNesting {
			return maximumDepth, totalValues
		}
		switch value := item.value.(type) {
		case []any:
			for _, child := range value {
				pending = append(pending, pendingLiteralValue{value: child, depth: item.depth + 1})
			}
		case map[string]any:
			for _, child := range value {
				pending = append(pending, pendingLiteralValue{value: child, depth: item.depth + 1})
			}
		case map[any]any:
			for _, child := range value {
				pending = append(pending, pendingLiteralValue{value: child, depth: item.depth + 1})
			}
		}
	}
	return maximumDepth, totalValues
}