package controlplane

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"errors"
	"fmt"
	"reflect"
	"sort"
	"strings"
	"time"
)

// ErrComponentManagedNoTargets reports that bounded target discovery
// completed successfully without finding managed work. Supporting hosts treat
// it as a successful no-op; they must not invent a global target or run
// mutation, effect, or evidence callbacks.
var ErrComponentManagedNoTargets = errors.New("managed transaction operation has no affected targets")

const (
	ComponentManagedMaximumOpaquePlanBytes = 1 << 20
	ComponentManagedMaximumTargetDomain    = 255
	ComponentManagedMaximumPhysicalKey     = 1024
)

type ComponentManagedScope string

const (
	ComponentManagedScopeItem               ComponentManagedScope = "item"
	ComponentManagedScopeSet                ComponentManagedScope = "set"
	ComponentManagedScopePlannedTransaction ComponentManagedScope = "planned_tx"
)

// Validate rejects undeclared managed operation scopes. Transaction ownership
// is expressed separately by the host entry point and is not an open-ended
// source of additional scope values.
func (s ComponentManagedScope) Validate() error {
	switch s {
	case ComponentManagedScopeItem,
		ComponentManagedScopeSet,
		ComponentManagedScopePlannedTransaction:
		return nil
	default:
		return fmt.Errorf("managed operation scope %q is unsupported", s)
	}
}

type ComponentManagedCapability string

const (
	ComponentManagedCapabilityExactSet            ComponentManagedCapability = "exact_set"
	ComponentManagedCapabilityTargetSerialization ComponentManagedCapability = "target_serialization"
	ComponentManagedCapabilityStructuredMutation  ComponentManagedCapability = "structured_mutation"
	ComponentManagedCapabilityEffects             ComponentManagedCapability = "transaction_bound_effects"
	ComponentManagedCapabilityEvidence            ComponentManagedCapability = "operation_evidence"
	ComponentManagedCapabilityFreshProjection     ComponentManagedCapability = "fresh_projection"
)

type ComponentManagedTransactionMode string

const (
	ComponentManagedTransactionPessimisticReadCommitted ComponentManagedTransactionMode = "pessimistic_read_committed"
)

type ComponentReadSurface struct {
	Table string
	Fact  string
}

type ComponentManagedTarget struct {
	TargetDomain     string
	PhysicalKeyBytes []byte
}

type ComponentManagedUsage struct {
	CurrentItems       int
	DesiredItems       int
	DeltaRows          int
	RowsRead           int
	RowsLocked         int
	RowsInserted       int
	RowsUpdated        int
	RowsDeleted        int
	MutationBytes      int
	MutationStatements int
	EvidenceWrites     int
	EffectCalls        int
	BatchRows          int
	LockWait           time.Duration
	Elapsed            time.Duration
	Attempts           int
}

type ComponentManagedBudget struct {
	MaxTargets            int
	MaxCurrentItems       int
	MaxDesiredItems       int
	MaxDeltaRows          int
	MaxMutationBytes      int
	MaxMutationStatements int
	MaxBatchRows          int
	MaxEvidenceWrites     int
	MaxEffectCalls        int
	MaxLockWait           time.Duration
	MaxElapsed            time.Duration
	MaxAttempts           int
}

type ComponentManagedEffectClass string

type ComponentManagedEffectRequirement struct {
	Class    ComponentManagedEffectClass
	Triggers []ComponentManagedMutationDisposition
}

// ComponentManagedMutationReceipt is the authoritative host-owned mutation
// meter. EffectClasses contains one entry for every effect call counted in
// Usage.EffectCalls.
type ComponentManagedMutationReceipt struct {
	Usage         ComponentManagedUsage
	EffectClasses []ComponentManagedEffectClass
}

type ComponentManagedPlanSpec struct {
	Targets              []ComponentManagedTarget
	RequiredCapabilities []ComponentManagedCapability
	Estimate             ComponentManagedUsage
	Effects              []ComponentManagedEffectRequirement
	ProjectionIdentity   []byte
	OpaquePlan           []byte
}

type ComponentManagedPlan struct {
	targets              []ComponentManagedTarget
	requiredCapabilities []ComponentManagedCapability
	estimate             ComponentManagedUsage
	effects              []ComponentManagedEffectRequirement
	projectionIdentity   []byte
	opaquePlan           []byte
	fingerprint          [sha256.Size]byte
}

func NewComponentManagedPlan(spec ComponentManagedPlanSpec) (ComponentManagedPlan, error) {
	targets, err := canonicalComponentManagedTargets(spec.Targets)
	if err != nil {
		return ComponentManagedPlan{}, err
	}
	capabilities, err := canonicalComponentManagedCapabilities(spec.RequiredCapabilities)
	if err != nil {
		return ComponentManagedPlan{}, err
	}
	if err := spec.Estimate.Validate("managed plan estimate"); err != nil {
		return ComponentManagedPlan{}, err
	}
	effects, err := canonicalComponentManagedEffects(spec.Effects)
	if err != nil {
		return ComponentManagedPlan{}, err
	}
	if len(spec.ProjectionIdentity) == 0 {
		return ComponentManagedPlan{}, fmt.Errorf("managed plan projection identity is required")
	}
	if len(spec.OpaquePlan) == 0 || len(spec.OpaquePlan) > ComponentManagedMaximumOpaquePlanBytes {
		return ComponentManagedPlan{}, fmt.Errorf("managed plan opaque plan must contain 1..%d bytes", ComponentManagedMaximumOpaquePlanBytes)
	}
	plan := ComponentManagedPlan{
		targets: targets, requiredCapabilities: capabilities, estimate: spec.Estimate,
		effects: effects, projectionIdentity: cloneManagedBytes(spec.ProjectionIdentity), opaquePlan: cloneManagedBytes(spec.OpaquePlan),
	}
	plan.fingerprint = fingerprintComponentManagedPlan(plan)
	return plan, nil
}

func (p ComponentManagedPlan) Targets() []ComponentManagedTarget {
	return cloneComponentManagedTargets(p.targets)
}

func (p ComponentManagedPlan) RequiredCapabilities() []ComponentManagedCapability {
	return append([]ComponentManagedCapability(nil), p.requiredCapabilities...)
}

func (p ComponentManagedPlan) Estimate() ComponentManagedUsage { return p.estimate }

func (p ComponentManagedPlan) Effects() []ComponentManagedEffectRequirement {
	return cloneComponentManagedEffects(p.effects)
}

func (p ComponentManagedPlan) ProjectionIdentity() []byte {
	return cloneManagedBytes(p.projectionIdentity)
}
func (p ComponentManagedPlan) OpaquePlan() []byte  { return cloneManagedBytes(p.opaquePlan) }
func (p ComponentManagedPlan) Fingerprint() []byte { return cloneManagedBytes(p.fingerprint[:]) }

func (p ComponentManagedPlan) SameEnvelope(other ComponentManagedPlan) bool {
	return p.fingerprint == other.fingerprint
}

func (p ComponentManagedPlan) Validate() error {
	validated, err := NewComponentManagedPlan(ComponentManagedPlanSpec{
		Targets:              p.targets,
		RequiredCapabilities: p.requiredCapabilities,
		Estimate:             p.estimate,
		Effects:              p.effects,
		ProjectionIdentity:   p.projectionIdentity,
		OpaquePlan:           p.opaquePlan,
	})
	if err != nil {
		return err
	}
	if !reflect.DeepEqual(p.targets, validated.targets) ||
		!reflect.DeepEqual(p.requiredCapabilities, validated.requiredCapabilities) ||
		p.estimate != validated.estimate ||
		!reflect.DeepEqual(p.effects, validated.effects) ||
		!reflect.DeepEqual(p.projectionIdentity, validated.projectionIdentity) ||
		!reflect.DeepEqual(p.opaquePlan, validated.opaquePlan) ||
		p.fingerprint != validated.fingerprint {
		return fmt.Errorf("managed plan is not a canonical initialized envelope")
	}
	return nil
}

type ComponentManagedLifecycleRequest struct {
	Item                *ComponentManagedRequest
	Set                 *ComponentManagedSetRequest
	DesiredBusinessKeys []string
}

func NewComponentManagedItemLifecycleRequest(request ComponentManagedRequest) ComponentManagedLifecycleRequest {
	clone := cloneComponentManagedRequest(request)
	return ComponentManagedLifecycleRequest{Item: &clone}
}

func NewComponentManagedSetLifecycleRequest(request ComponentManagedSetRequest, desiredBusinessKeys []string) (ComponentManagedLifecycleRequest, error) {
	keys, err := canonicalComponentManagedBusinessKeys(desiredBusinessKeys)
	if err != nil {
		return ComponentManagedLifecycleRequest{}, err
	}
	clone := request
	return ComponentManagedLifecycleRequest{Set: &clone, DesiredBusinessKeys: keys}, nil
}

func (r ComponentManagedLifecycleRequest) Clone() ComponentManagedLifecycleRequest {
	clone := ComponentManagedLifecycleRequest{DesiredBusinessKeys: append([]string(nil), r.DesiredBusinessKeys...)}
	if r.Item != nil {
		item := cloneComponentManagedRequest(*r.Item)
		clone.Item = &item
	}
	if r.Set != nil {
		set := *r.Set
		clone.Set = &set
	}
	return clone
}

func (r ComponentManagedLifecycleRequest) Validate(scope ComponentManagedScope) error {
	switch scope {
	case ComponentManagedScopeItem:
		if r.Item == nil || r.Set != nil || len(r.DesiredBusinessKeys) != 0 {
			return fmt.Errorf("item lifecycle request must contain exactly one item request")
		}
	case ComponentManagedScopeSet:
		if r.Set == nil || r.Item != nil {
			return fmt.Errorf("set lifecycle request must contain exactly one set request")
		}
		keys, err := canonicalComponentManagedBusinessKeys(r.DesiredBusinessKeys)
		if err != nil {
			return err
		}
		for index := range keys {
			if keys[index] != r.DesiredBusinessKeys[index] {
				return fmt.Errorf("managed desired business keys must be sorted and unique")
			}
		}
	default:
		return fmt.Errorf("managed lifecycle scope %q is unsupported", scope)
	}
	return nil
}

type ComponentManagedLifecycleProvider struct {
	Plan    func(context.Context, ComponentManagedScope, ComponentManagedLifecycleRequest, ComponentManagedReadStore) (ComponentManagedPlan, error)
	Mutate  func(context.Context, ComponentManagedScope, ComponentManagedLifecycleRequest, ComponentManagedPlan, ComponentManagedMutationStore) (ComponentManagedMutationResult, error)
	Project func(context.Context, ComponentManagedScope, ComponentManagedLifecycleRequest, ComponentManagedReadStore) (ComponentManagedProjection, error)
}

func (p ComponentManagedLifecycleProvider) Validate() error {
	missing := make([]string, 0, 3)
	if p.Plan == nil {
		missing = append(missing, "Plan")
	}
	if p.Mutate == nil {
		missing = append(missing, "Mutate")
	}
	if p.Project == nil {
		missing = append(missing, "Project")
	}
	if len(missing) != 0 {
		return fmt.Errorf("managed lifecycle provider is missing callbacks: %s", strings.Join(missing, ", "))
	}
	return nil
}

type ComponentManagedLifecycleRegistrationSpec struct {
	Kind            ComponentKind
	ReadSet         []ComponentReadSurface
	Capabilities    []ComponentManagedCapability
	EffectClasses   []ComponentManagedEffectClass
	TransactionMode ComponentManagedTransactionMode
	Budget          ComponentManagedBudget
	Lifecycle       ComponentManagedLifecycleProvider
}

type ComponentManagedLifecycleRegistration struct {
	kind            ComponentKind
	readSet         []ComponentReadSurface
	capabilities    []ComponentManagedCapability
	effectClasses   []ComponentManagedEffectClass
	transactionMode ComponentManagedTransactionMode
	budget          ComponentManagedBudget
	lifecycle       ComponentManagedLifecycleProvider
}

// ComponentKindManagedLifecycleRegistration binds an existing component kind
// registration to its additive Plan/Mutate/Project lifecycle without changing
// the field layout of ComponentKindRegistration.
type ComponentKindManagedLifecycleRegistration struct {
	registration ComponentKindRegistration
	lifecycle    ComponentManagedLifecycleRegistration
}

// WithManagedLifecycle returns an immutable lifecycle binding for this
// component registration. ManagedProvider may remain populated during an
// additive migration; lifecycle-aware hosts use the returned binding.
func (r ComponentKindRegistration) WithManagedLifecycle(lifecycle ComponentManagedLifecycleRegistration) (ComponentKindManagedLifecycleRegistration, error) {
	if strings.TrimSpace(string(r.Contract.Kind)) == "" {
		return ComponentKindManagedLifecycleRegistration{}, fmt.Errorf("component registration kind is required")
	}
	if strings.TrimSpace(string(lifecycle.Kind())) == "" {
		return ComponentKindManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle registration is required")
	}
	if r.Contract.Kind != lifecycle.Kind() {
		return ComponentKindManagedLifecycleRegistration{}, fmt.Errorf("component registration kind %q does not match managed lifecycle kind %q", r.Contract.Kind, lifecycle.Kind())
	}
	return ComponentKindManagedLifecycleRegistration{registration: r, lifecycle: lifecycle}, nil
}

// Registration returns the existing component kind registration.
func (r ComponentKindManagedLifecycleRegistration) Registration() ComponentKindRegistration {
	return r.registration
}

// ManagedLifecycle returns the lifecycle bound to the component kind.
func (r ComponentKindManagedLifecycleRegistration) ManagedLifecycle() ComponentManagedLifecycleRegistration {
	return r.lifecycle
}

func NewComponentManagedLifecycleRegistration(spec ComponentManagedLifecycleRegistrationSpec) (ComponentManagedLifecycleRegistration, error) {
	if strings.TrimSpace(string(spec.Kind)) == "" || strings.TrimSpace(string(spec.Kind)) != string(spec.Kind) {
		return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle component kind is required and must be canonical")
	}
	readSet, err := canonicalComponentReadSurfaces(spec.ReadSet)
	if err != nil {
		return ComponentManagedLifecycleRegistration{}, err
	}
	capabilities, err := canonicalComponentManagedCapabilities(spec.Capabilities)
	if err != nil {
		return ComponentManagedLifecycleRegistration{}, err
	}
	for _, required := range []ComponentManagedCapability{
		ComponentManagedCapabilityTargetSerialization,
		ComponentManagedCapabilityStructuredMutation,
		ComponentManagedCapabilityEvidence,
		ComponentManagedCapabilityFreshProjection,
	} {
		if !containsComponentManagedCapability(capabilities, required) {
			return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle registration is missing required capability %q", required)
		}
	}
	effectClasses, err := canonicalComponentManagedEffectClasses(spec.EffectClasses)
	if err != nil {
		return ComponentManagedLifecycleRegistration{}, err
	}
	hasEffectsCapability := containsComponentManagedCapability(capabilities, ComponentManagedCapabilityEffects)
	if len(effectClasses) != 0 && !hasEffectsCapability {
		return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle effect classes require capability %q", ComponentManagedCapabilityEffects)
	}
	if hasEffectsCapability && len(effectClasses) == 0 {
		return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle capability %q requires at least one effect class", ComponentManagedCapabilityEffects)
	}
	if spec.TransactionMode != ComponentManagedTransactionPessimisticReadCommitted {
		return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle transaction mode %q is unsupported", spec.TransactionMode)
	}
	if err := spec.Budget.Validate(); err != nil {
		return ComponentManagedLifecycleRegistration{}, fmt.Errorf("managed lifecycle budget: %w", err)
	}
	if err := spec.Lifecycle.Validate(); err != nil {
		return ComponentManagedLifecycleRegistration{}, err
	}
	return ComponentManagedLifecycleRegistration{
		kind: spec.Kind, readSet: readSet, capabilities: capabilities, effectClasses: effectClasses,
		transactionMode: spec.TransactionMode, budget: spec.Budget, lifecycle: spec.Lifecycle,
	}, nil
}

func (r ComponentManagedLifecycleRegistration) Kind() ComponentKind { return r.kind }
func (r ComponentManagedLifecycleRegistration) ReadSet() []ComponentReadSurface {
	return append([]ComponentReadSurface(nil), r.readSet...)
}
func (r ComponentManagedLifecycleRegistration) Capabilities() []ComponentManagedCapability {
	return append([]ComponentManagedCapability(nil), r.capabilities...)
}
func (r ComponentManagedLifecycleRegistration) EffectClasses() []ComponentManagedEffectClass {
	return append([]ComponentManagedEffectClass(nil), r.effectClasses...)
}
func (r ComponentManagedLifecycleRegistration) TransactionMode() ComponentManagedTransactionMode {
	return r.transactionMode
}
func (r ComponentManagedLifecycleRegistration) Budget() ComponentManagedBudget { return r.budget }
func (r ComponentManagedLifecycleRegistration) Lifecycle() ComponentManagedLifecycleProvider {
	return r.lifecycle
}

func (r ComponentManagedLifecycleRegistration) ValidatePlanCapabilities(plan ComponentManagedPlan) error {
	if err := plan.Validate(); err != nil {
		return fmt.Errorf("managed plan: %w", err)
	}
	if err := r.validatePlanBudget(plan); err != nil {
		return err
	}
	declared := make(map[ComponentManagedCapability]struct{}, len(r.capabilities))
	for _, capability := range r.capabilities {
		declared[capability] = struct{}{}
	}
	for _, capability := range plan.requiredCapabilities {
		if _, ok := declared[capability]; !ok {
			return fmt.Errorf("managed plan requires undeclared capability %q", capability)
		}
	}
	declaredEffects := make(map[ComponentManagedEffectClass]struct{}, len(r.effectClasses))
	for _, class := range r.effectClasses {
		declaredEffects[class] = struct{}{}
	}
	for _, effect := range plan.effects {
		if _, ok := declaredEffects[effect.Class]; !ok {
			return fmt.Errorf("managed plan requires undeclared effect class %q", effect.Class)
		}
	}
	return nil
}

func (r ComponentManagedLifecycleRegistration) validatePlanBudget(plan ComponentManagedPlan) error {
	if len(plan.targets) > r.budget.MaxTargets {
		return fmt.Errorf("managed plan targets %d exceeds registration budget %d", len(plan.targets), r.budget.MaxTargets)
	}
	return r.budget.validateUsageLimits("managed plan estimate", plan.estimate)
}

// ValidateMutationReceipt validates the host-metered usage snapshot after the
// provider mutation, declared effects, and evidence write have completed and
// before the host commits its transaction. Actual usage must come from the
// host-owned mutation store, never from a provider callback return value.
func (r ComponentManagedLifecycleRegistration) ValidateMutationReceipt(plan ComponentManagedPlan, result ComponentManagedMutationResult, receipt ComponentManagedMutationReceipt) error {
	if err := r.ValidatePlanCapabilities(plan); err != nil {
		return err
	}
	if err := result.Validate(); err != nil {
		return err
	}
	actual := receipt.Usage
	if err := actual.Validate("managed mutation actual usage"); err != nil {
		return err
	}
	actualEffectClasses, err := canonicalComponentManagedEffectClasses(receipt.EffectClasses)
	if err != nil {
		return fmt.Errorf("managed mutation actual effect classes: %w", err)
	}
	if actual.EffectCalls != len(receipt.EffectClasses) {
		return fmt.Errorf("managed mutation actual effect calls %d do not match recorded effect classes %d", actual.EffectCalls, len(receipt.EffectClasses))
	}
	if actual.RowsRead > actual.CurrentItems || actual.RowsLocked != actual.CurrentItems-actual.RowsRead {
		return fmt.Errorf("managed mutation actual current items must equal rows read plus rows locked")
	}
	if actual.RowsInserted > actual.DeltaRows || actual.RowsUpdated > actual.DeltaRows-actual.RowsInserted || actual.RowsDeleted != actual.DeltaRows-actual.RowsInserted-actual.RowsUpdated {
		return fmt.Errorf("managed mutation actual delta rows must equal rows inserted, updated, and deleted")
	}
	if err := r.budget.validateUsageLimits("managed mutation actual usage", actual); err != nil {
		return err
	}
	if actual.EvidenceWrites != 1 {
		return fmt.Errorf("managed mutation requires exactly one host evidence write; got %d", actual.EvidenceWrites)
	}
	var expectedEffectClasses []ComponentManagedEffectClass
	for _, effect := range plan.effects {
		for _, trigger := range effect.Triggers {
			if trigger == result.Disposition {
				expectedEffectClasses = append(expectedEffectClasses, effect.Class)
			}
		}
	}
	if !reflect.DeepEqual(actualEffectClasses, expectedEffectClasses) {
		return fmt.Errorf("managed mutation actual effect classes %q do not match required classes %q", actualEffectClasses, expectedEffectClasses)
	}
	switch result.Disposition {
	case ComponentManagedMutationNoChange:
		if actual.DeltaRows != 0 || actual.MutationStatements != 0 || actual.EffectCalls != 0 {
			return fmt.Errorf("managed mutation disposition %q contradicts actual usage", result.Disposition)
		}
	case ComponentManagedMutationDesiredProjectionStaged, ComponentManagedMutationSafeProjectionStaged:
		if actual.DeltaRows == 0 && actual.EffectCalls == 0 {
			return fmt.Errorf("managed mutation disposition %q requires an actual row or effect change", result.Disposition)
		}
	}
	return nil
}

func (b ComponentManagedBudget) validateUsageLimits(label string, usage ComponentManagedUsage) error {
	limits := []struct {
		field         string
		actual, limit int
	}{
		{field: "current items", actual: usage.CurrentItems, limit: b.MaxCurrentItems},
		{field: "desired items", actual: usage.DesiredItems, limit: b.MaxDesiredItems},
		{field: "delta rows", actual: usage.DeltaRows, limit: b.MaxDeltaRows},
		{field: "mutation bytes", actual: usage.MutationBytes, limit: b.MaxMutationBytes},
		{field: "mutation statements", actual: usage.MutationStatements, limit: b.MaxMutationStatements},
		{field: "evidence writes", actual: usage.EvidenceWrites, limit: b.MaxEvidenceWrites},
		{field: "effect calls", actual: usage.EffectCalls, limit: b.MaxEffectCalls},
		{field: "batch rows", actual: usage.BatchRows, limit: b.MaxBatchRows},
		{field: "attempts", actual: usage.Attempts, limit: b.MaxAttempts},
	}
	for _, entry := range limits {
		if entry.actual > entry.limit {
			return fmt.Errorf("%s %s %d exceeds registration budget %d", label, entry.field, entry.actual, entry.limit)
		}
	}
	if usage.LockWait > b.MaxLockWait {
		return fmt.Errorf("%s lock wait %s exceeds registration budget %s", label, usage.LockWait, b.MaxLockWait)
	}
	if usage.Elapsed > b.MaxElapsed {
		return fmt.Errorf("%s elapsed %s exceeds registration budget %s", label, usage.Elapsed, b.MaxElapsed)
	}
	return nil
}

type ComponentManagedOrderDirection string

const (
	ComponentManagedOrderAscending  ComponentManagedOrderDirection = "ascending"
	ComponentManagedOrderDescending ComponentManagedOrderDirection = "descending"
)

type ComponentManagedOrder struct {
	Column    string
	Direction ComponentManagedOrderDirection
}

type ComponentManagedRowQuery struct {
	Table             string
	Columns           []string
	Equal             ComponentRow
	OrderBy           []ComponentManagedOrder
	LogicalKeyColumns []string
	MaximumRows       int
}

func (q ComponentManagedRowQuery) Validate(maxBatchRows int) error {
	if err := (ComponentRowQuery{Table: q.Table, Columns: q.Columns, Equal: q.Equal}).Validate(); err != nil {
		return err
	}
	if maxBatchRows <= 0 || q.MaximumRows <= 0 || q.MaximumRows > maxBatchRows {
		return fmt.Errorf("maximum rows must be within 1..%d", maxBatchRows)
	}
	seen := make(map[string]struct{}, len(q.OrderBy))
	for index, order := range q.OrderBy {
		if strings.TrimSpace(order.Column) == "" || strings.TrimSpace(order.Column) != order.Column {
			return fmt.Errorf("order column %d is not canonical", index)
		}
		if order.Direction != ComponentManagedOrderAscending && order.Direction != ComponentManagedOrderDescending {
			return fmt.Errorf("order direction %q is unsupported", order.Direction)
		}
		if _, exists := seen[order.Column]; exists {
			return fmt.Errorf("order column %q is duplicated", order.Column)
		}
		seen[order.Column] = struct{}{}
	}
	if err := validateComponentColumnsOptional("logical key columns", q.LogicalKeyColumns); err != nil {
		return err
	}
	projected := make(map[string]struct{}, len(q.Columns))
	for _, column := range q.Columns {
		projected[strings.TrimSpace(column)] = struct{}{}
	}
	for _, column := range q.LogicalKeyColumns {
		if _, ok := projected[strings.TrimSpace(column)]; !ok {
			return fmt.Errorf("logical key column %q must be included in projected columns", strings.TrimSpace(column))
		}
	}
	return nil
}

type ComponentManagedKeysetRowQuery struct {
	Query         ComponentManagedRowQuery
	ContinueAfter ComponentRow
}

// ComponentManagedKeysetRowPage returns one bounded page and whether the
// predicate was exhausted in the same transaction. Exhausted is
// explicit because a full final page is otherwise indistinguishable from a
// page with more rows, especially when the operation has exactly consumed its
// current-item budget.
type ComponentManagedKeysetRowPage struct {
	Rows      []ComponentRow
	Exhausted bool
}

func (q ComponentManagedKeysetRowQuery) Validate(maxBatchRows int) error {
	if err := q.Query.Validate(maxBatchRows); err != nil {
		return err
	}
	if err := validateComponentRow("continuation", q.ContinueAfter, false); err != nil {
		return err
	}
	if len(q.Query.OrderBy) == 0 || len(q.Query.OrderBy) != len(q.Query.LogicalKeyColumns) {
		return fmt.Errorf("keyset pagination requires order columns to exactly match logical key columns")
	}
	if len(q.ContinueAfter) != 0 && len(q.ContinueAfter) != len(q.Query.OrderBy) {
		return fmt.Errorf("continuation must contain exactly one value for every order column")
	}
	for index, order := range q.Query.OrderBy {
		if order.Column != q.Query.LogicalKeyColumns[index] {
			return fmt.Errorf("keyset order column %q must match logical key column %q at index %d", order.Column, q.Query.LogicalKeyColumns[index], index)
		}
		if len(q.ContinueAfter) == 0 {
			continue
		}
		value, ok := q.ContinueAfter[order.Column]
		if !ok {
			return fmt.Errorf("continuation is missing order column %q", order.Column)
		}
		if value == nil {
			return fmt.Errorf("continuation order column %q cannot be nil", order.Column)
		}
	}
	return nil
}

type ComponentManagedInsertRow struct {
	LogicalKey ComponentRow
	Values     ComponentRow
}

type ComponentManagedInsertBatch struct {
	Table                   string
	Rows                    []ComponentManagedInsertRow
	CurrentTimestampColumns []string
}

func (b ComponentManagedInsertBatch) Validate(maxBatchRows int) error {
	if err := validateComponentTable(b.Table); err != nil {
		return err
	}
	if maxBatchRows <= 0 || len(b.Rows) == 0 || len(b.Rows) > maxBatchRows {
		return fmt.Errorf("insert batch rows must contain 1..%d rows", maxBatchRows)
	}
	if err := validateComponentColumnsOptional("current timestamp columns", b.CurrentTimestampColumns); err != nil {
		return err
	}
	for index, row := range b.Rows {
		if err := validateComponentRow("logical key", row.LogicalKey, true); err != nil {
			return fmt.Errorf("insert row %d: %w", index, err)
		}
		if err := validateComponentRow("values", row.Values, true); err != nil {
			return fmt.Errorf("insert row %d: %w", index, err)
		}
		if err := rejectComponentColumnOverlap(row.Values, b.CurrentTimestampColumns); err != nil {
			return fmt.Errorf("insert row %d: %w", index, err)
		}
		for column, value := range row.LogicalKey {
			if actual, ok := row.Values[column]; !ok || !reflect.DeepEqual(actual, value) {
				return fmt.Errorf("insert row %d logical key column %q must exactly match values", index, column)
			}
		}
	}
	return nil
}

type ComponentManagedUpdate struct {
	Table                   string
	Match                   ComponentRow
	Values                  ComponentRow
	CurrentTimestampColumns []string
	MaximumAffectedRows     int
}

func (u ComponentManagedUpdate) Validate(maxBatchRows int) error {
	if err := (ComponentUpdate{Table: u.Table, Match: u.Match, Values: u.Values, CurrentTimestampColumns: u.CurrentTimestampColumns}).Validate(); err != nil {
		return err
	}
	if maxBatchRows <= 0 || u.MaximumAffectedRows <= 0 || u.MaximumAffectedRows > maxBatchRows {
		return fmt.Errorf("maximum affected rows must be within 1..%d", maxBatchRows)
	}
	return nil
}

type ComponentManagedDelete struct {
	Table               string
	Match               ComponentRow
	MaximumAffectedRows int
}

func (d ComponentManagedDelete) Validate(maxBatchRows int) error {
	if err := (ComponentDelete{Table: d.Table, Match: d.Match}).Validate(); err != nil {
		return err
	}
	if maxBatchRows <= 0 || d.MaximumAffectedRows <= 0 || d.MaximumAffectedRows > maxBatchRows {
		return fmt.Errorf("maximum affected rows must be within 1..%d", maxBatchRows)
	}
	return nil
}

type ComponentManagedReadStore interface {
	ReadRows(context.Context, ComponentManagedRowQuery) ([]ComponentRow, error)
}

type ComponentManagedMutationStore interface {
	ComponentManagedReadStore
	LockRows(context.Context, ComponentManagedRowQuery) ([]ComponentRow, error)
	NewID() (string, error)
	InsertRows(context.Context, ComponentManagedInsertBatch) (int64, error)
	UpdateRows(context.Context, ComponentManagedUpdate) (int64, error)
	DeleteRows(context.Context, ComponentManagedDelete) (int64, error)
	// ActualReceipt returns the authoritative host-owned meter snapshot. Store
	// implementations must reject an operation before execution when its
	// declared row bound or the resulting usage would exceed a budget. BatchRows
	// is the largest row count returned or affected by one store call. The host
	// adds elapsed time, attempts, effect calls, effect classes, and evidence work
	// to the same receipt, then validates it with ValidateMutationReceipt before
	// commit.
	ActualReceipt() ComponentManagedMutationReceipt
}

type ComponentManagedTransactionReadStore interface {
	ComponentManagedReadStore
	ReadRowsPage(context.Context, ComponentManagedKeysetRowQuery) ([]ComponentRow, error)
}

type ComponentManagedTransactionMutationStore interface {
	ComponentManagedMutationStore
	ReadRowsPage(context.Context, ComponentManagedKeysetRowQuery) ([]ComponentRow, error)
	LockRowsPage(context.Context, ComponentManagedKeysetRowQuery) ([]ComponentRow, error)
}

// ComponentManagedKeysetTransactionReadStore is the additive exact-exhaustion
// extension implemented by hosts that support bounded keyset iteration. A
// provider must require this interface before relying on Exhausted.
type ComponentManagedKeysetTransactionReadStore interface {
	ComponentManagedTransactionReadStore
	ReadRowsKeysetPage(context.Context, ComponentManagedKeysetRowQuery) (ComponentManagedKeysetRowPage, error)
}

// ComponentManagedKeysetTransactionMutationStore extends the v0.2.14 mutation
// store without changing that published interface.
type ComponentManagedKeysetTransactionMutationStore interface {
	ComponentManagedTransactionMutationStore
	ComponentManagedKeysetTransactionReadStore
	LockRowsKeysetPage(context.Context, ComponentManagedKeysetRowQuery) (ComponentManagedKeysetRowPage, error)
}

// ComponentManagedTransactionOperationProvider supplies the component-specific
// mutation and staged projection for one immutable plan executed inside a
// caller-owned transaction. The host owns target serialization, the structured
// store, effects, evidence, receipt validation, and transaction outcome.
type ComponentManagedTransactionOperationProvider struct {
	Mutate  func(context.Context, ComponentManagedPlan, ComponentManagedTransactionMutationStore) (ComponentManagedMutationResult, error)
	Project func(context.Context, ComponentManagedPlan, ComponentManagedTransactionReadStore) (ComponentManagedProjection, error)
}

// ComponentManagedPlannedTransactionOperationProvider discovers its immutable
// target plan inside the caller-owned transaction. This is used when an
// already-locked source mutation determines the affected targets. Hosts must
// call Plan before and after target serialization and reject target drift.
type ComponentManagedPlannedTransactionOperationProvider struct {
	Plan    func(context.Context, ComponentManagedTransactionReadStore) (ComponentManagedPlan, error)
	Mutate  func(context.Context, ComponentManagedPlan, ComponentManagedTransactionMutationStore) (ComponentManagedMutationResult, error)
	Project func(context.Context, ComponentManagedPlan, ComponentManagedTransactionReadStore) (ComponentManagedProjection, error)
}

func (p ComponentManagedPlannedTransactionOperationProvider) Validate() error {
	missing := make([]string, 0, 3)
	if p.Plan == nil {
		missing = append(missing, "Plan")
	}
	if p.Mutate == nil {
		missing = append(missing, "Mutate")
	}
	if p.Project == nil {
		missing = append(missing, "Project")
	}
	if len(missing) != 0 {
		return fmt.Errorf("managed planned transaction operation provider is missing callbacks: %s", strings.Join(missing, ", "))
	}
	return nil
}

func (p ComponentManagedTransactionOperationProvider) Validate() error {
	missing := make([]string, 0, 2)
	if p.Mutate == nil {
		missing = append(missing, "Mutate")
	}
	if p.Project == nil {
		missing = append(missing, "Project")
	}
	if len(missing) != 0 {
		return fmt.Errorf("managed transaction operation provider is missing callbacks: %s", strings.Join(missing, ", "))
	}
	return nil
}

type ComponentManagedMutationDisposition string

const (
	ComponentManagedMutationNoChange                ComponentManagedMutationDisposition = "no_change"
	ComponentManagedMutationDesiredProjectionStaged ComponentManagedMutationDisposition = "desired_projection_staged"
	ComponentManagedMutationSafeProjectionStaged    ComponentManagedMutationDisposition = "safe_projection_staged"
)

type ComponentManagedMutationResult struct {
	Disposition ComponentManagedMutationDisposition
}

func (r ComponentManagedMutationResult) Validate() error {
	switch r.Disposition {
	case ComponentManagedMutationNoChange, ComponentManagedMutationDesiredProjectionStaged, ComponentManagedMutationSafeProjectionStaged:
		return nil
	default:
		return fmt.Errorf("managed mutation disposition %q is unsupported", r.Disposition)
	}
}

type ComponentManagedProjection struct {
	Identity []byte
	Expected []byte
	Live     []byte
}

func (p ComponentManagedProjection) Validate() error {
	if len(p.Identity) == 0 || len(p.Expected) == 0 || len(p.Live) == 0 {
		return fmt.Errorf("managed projection identity, expected, and live bytes are required")
	}
	return nil
}

func (u ComponentManagedUsage) Validate(label string) error {
	values := []int{u.CurrentItems, u.DesiredItems, u.DeltaRows, u.RowsRead, u.RowsLocked, u.RowsInserted, u.RowsUpdated, u.RowsDeleted, u.MutationBytes, u.MutationStatements, u.EvidenceWrites, u.EffectCalls, u.BatchRows, u.Attempts}
	for _, value := range values {
		if value < 0 {
			return fmt.Errorf("%s cannot contain negative values", label)
		}
	}
	if u.LockWait < 0 || u.Elapsed < 0 {
		return fmt.Errorf("%s cannot contain negative durations", label)
	}
	return nil
}

func (b ComponentManagedBudget) Validate() error {
	if b.MaxTargets <= 0 || b.MaxCurrentItems <= 0 || b.MaxDesiredItems <= 0 || b.MaxDeltaRows <= 0 || b.MaxMutationBytes <= 0 || b.MaxMutationStatements <= 0 || b.MaxBatchRows <= 0 || b.MaxBatchRows > 500 || b.MaxEvidenceWrites <= 0 || b.MaxEffectCalls <= 0 || b.MaxLockWait <= 0 || b.MaxElapsed <= 0 || b.MaxAttempts <= 0 {
		return fmt.Errorf("all limits must be positive and max batch rows must not exceed 500")
	}
	return nil
}

func canonicalComponentManagedTargets(input []ComponentManagedTarget) ([]ComponentManagedTarget, error) {
	if len(input) == 0 {
		return nil, fmt.Errorf("managed plan targets are required")
	}
	targets := cloneComponentManagedTargets(input)
	for index, target := range targets {
		if strings.TrimSpace(target.TargetDomain) == "" || strings.TrimSpace(target.TargetDomain) != target.TargetDomain || len(target.TargetDomain) > ComponentManagedMaximumTargetDomain {
			return nil, fmt.Errorf("managed target %d target domain is not canonical or exceeds %d bytes", index, ComponentManagedMaximumTargetDomain)
		}
		if len(target.PhysicalKeyBytes) == 0 || len(target.PhysicalKeyBytes) > ComponentManagedMaximumPhysicalKey {
			return nil, fmt.Errorf("managed target %d physical key must contain 1..%d bytes", index, ComponentManagedMaximumPhysicalKey)
		}
	}
	sort.Slice(targets, func(i, j int) bool {
		if targets[i].TargetDomain != targets[j].TargetDomain {
			return targets[i].TargetDomain < targets[j].TargetDomain
		}
		return string(targets[i].PhysicalKeyBytes) < string(targets[j].PhysicalKeyBytes)
	})
	out := targets[:0]
	for _, target := range targets {
		if len(out) != 0 && out[len(out)-1].TargetDomain == target.TargetDomain && string(out[len(out)-1].PhysicalKeyBytes) == string(target.PhysicalKeyBytes) {
			continue
		}
		out = append(out, target)
	}
	return out, nil
}

func cloneComponentManagedTargets(input []ComponentManagedTarget) []ComponentManagedTarget {
	out := make([]ComponentManagedTarget, len(input))
	for index, target := range input {
		out[index] = ComponentManagedTarget{TargetDomain: target.TargetDomain, PhysicalKeyBytes: cloneManagedBytes(target.PhysicalKeyBytes)}
	}
	return out
}

func canonicalComponentManagedCapabilities(input []ComponentManagedCapability) ([]ComponentManagedCapability, error) {
	values := append([]ComponentManagedCapability(nil), input...)
	for _, value := range values {
		switch value {
		case ComponentManagedCapabilityExactSet, ComponentManagedCapabilityTargetSerialization, ComponentManagedCapabilityStructuredMutation, ComponentManagedCapabilityEffects, ComponentManagedCapabilityEvidence, ComponentManagedCapabilityFreshProjection:
		default:
			return nil, fmt.Errorf("managed capability %q is unsupported", value)
		}
	}
	sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
	for index := 1; index < len(values); index++ {
		if values[index] == values[index-1] {
			return nil, fmt.Errorf("managed capability %q is duplicated", values[index])
		}
	}
	return values, nil
}

func containsComponentManagedCapability(values []ComponentManagedCapability, target ComponentManagedCapability) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func canonicalComponentManagedEffects(input []ComponentManagedEffectRequirement) ([]ComponentManagedEffectRequirement, error) {
	out := cloneComponentManagedEffects(input)
	for index := range out {
		if strings.TrimSpace(string(out[index].Class)) == "" || strings.TrimSpace(string(out[index].Class)) != string(out[index].Class) {
			return nil, fmt.Errorf("managed effect %d class is not canonical", index)
		}
		if len(out[index].Triggers) == 0 {
			return nil, fmt.Errorf("managed effect %q triggers are required", out[index].Class)
		}
		sort.Slice(out[index].Triggers, func(i, j int) bool { return out[index].Triggers[i] < out[index].Triggers[j] })
		for triggerIndex, trigger := range out[index].Triggers {
			if err := (ComponentManagedMutationResult{Disposition: trigger}).Validate(); err != nil || trigger == ComponentManagedMutationNoChange {
				return nil, fmt.Errorf("managed effect %q trigger %q is unsupported", out[index].Class, trigger)
			}
			if triggerIndex > 0 && trigger == out[index].Triggers[triggerIndex-1] {
				return nil, fmt.Errorf("managed effect %q trigger %q is duplicated", out[index].Class, trigger)
			}
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].Class < out[j].Class })
	for index := 1; index < len(out); index++ {
		if out[index].Class == out[index-1].Class {
			return nil, fmt.Errorf("managed effect class %q is duplicated", out[index].Class)
		}
	}
	return out, nil
}

func cloneComponentManagedEffects(input []ComponentManagedEffectRequirement) []ComponentManagedEffectRequirement {
	out := make([]ComponentManagedEffectRequirement, len(input))
	for index, effect := range input {
		out[index] = ComponentManagedEffectRequirement{Class: effect.Class, Triggers: append([]ComponentManagedMutationDisposition(nil), effect.Triggers...)}
	}
	return out
}

func canonicalComponentManagedEffectClasses(input []ComponentManagedEffectClass) ([]ComponentManagedEffectClass, error) {
	values := append([]ComponentManagedEffectClass(nil), input...)
	for index, value := range values {
		if strings.TrimSpace(string(value)) == "" || strings.TrimSpace(string(value)) != string(value) {
			return nil, fmt.Errorf("managed effect class %d is not canonical", index)
		}
	}
	sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
	for index := 1; index < len(values); index++ {
		if values[index] == values[index-1] {
			return nil, fmt.Errorf("managed effect class %q is duplicated", values[index])
		}
	}
	return values, nil
}

func canonicalComponentReadSurfaces(input []ComponentReadSurface) ([]ComponentReadSurface, error) {
	if len(input) == 0 {
		return nil, fmt.Errorf("managed lifecycle read set is required")
	}
	out := append([]ComponentReadSurface(nil), input...)
	seen := make(map[string]struct{}, len(out))
	for index, surface := range out {
		if strings.TrimSpace(surface.Table) == "" || strings.TrimSpace(surface.Table) != surface.Table || strings.TrimSpace(surface.Fact) == "" {
			return nil, fmt.Errorf("managed lifecycle read set entry %d is incomplete or non-canonical", index)
		}
		key := strings.ToLower(surface.Table)
		if _, exists := seen[key]; exists {
			return nil, fmt.Errorf("managed lifecycle read set table %q appears more than once", surface.Table)
		}
		seen[key] = struct{}{}
	}
	sort.Slice(out, func(i, j int) bool { return strings.ToLower(out[i].Table) < strings.ToLower(out[j].Table) })
	return out, nil
}

func canonicalComponentManagedBusinessKeys(input []string) ([]string, error) {
	out := append([]string(nil), input...)
	for index, key := range out {
		if key == "" || strings.TrimSpace(key) != key {
			return nil, fmt.Errorf("managed desired business key %d is not canonical", index)
		}
	}
	sort.Strings(out)
	for index := 1; index < len(out); index++ {
		if out[index] == out[index-1] {
			return nil, fmt.Errorf("managed desired business key %q is duplicated", out[index])
		}
	}
	return out, nil
}

func cloneComponentManagedRequest(input ComponentManagedRequest) ComponentManagedRequest {
	input.Payload = cloneManagedBytes(input.Payload)
	input.ExpectedProjection = cloneManagedBytes(input.ExpectedProjection)
	input.Component.Payload = cloneManagedBytes(input.Component.Payload)
	input.Component.DependsOn = append([]ComponentReference(nil), input.Component.DependsOn...)
	return input
}

func fingerprintComponentManagedPlan(plan ComponentManagedPlan) [sha256.Size]byte {
	hash := sha256.New()
	writeUint64 := func(value uint64) {
		var encoded [8]byte
		binary.BigEndian.PutUint64(encoded[:], value)
		_, _ = hash.Write(encoded[:])
	}
	writeBytes := func(value []byte) {
		writeUint64(uint64(len(value)))
		_, _ = hash.Write(value)
	}
	writeBytes([]byte("nexuscrm/component-managed-plan/v1"))
	writeUint64(uint64(len(plan.targets)))
	for _, target := range plan.targets {
		writeBytes([]byte(target.TargetDomain))
		writeBytes(target.PhysicalKeyBytes)
	}
	writeUint64(uint64(len(plan.requiredCapabilities)))
	for _, capability := range plan.requiredCapabilities {
		writeBytes([]byte(capability))
	}
	writeUint64(uint64(len(plan.effects)))
	for _, effect := range plan.effects {
		writeBytes([]byte(effect.Class))
		writeUint64(uint64(len(effect.Triggers)))
		for _, trigger := range effect.Triggers {
			writeBytes([]byte(trigger))
		}
	}
	for _, value := range []int{
		plan.estimate.CurrentItems,
		plan.estimate.DesiredItems,
		plan.estimate.DeltaRows,
		plan.estimate.RowsRead,
		plan.estimate.RowsLocked,
		plan.estimate.RowsInserted,
		plan.estimate.RowsUpdated,
		plan.estimate.RowsDeleted,
		plan.estimate.MutationBytes,
		plan.estimate.MutationStatements,
		plan.estimate.EvidenceWrites,
		plan.estimate.EffectCalls,
		plan.estimate.BatchRows,
		plan.estimate.Attempts,
	} {
		writeUint64(uint64(value))
	}
	writeUint64(uint64(plan.estimate.LockWait))
	writeUint64(uint64(plan.estimate.Elapsed))
	writeBytes(plan.projectionIdentity)
	writeBytes(plan.opaquePlan)
	var out [sha256.Size]byte
	copy(out[:], hash.Sum(nil))
	return out
}

func cloneManagedBytes(input []byte) []byte { return append([]byte(nil), input...) }