package runtimeapi

import (
	"context"
	"errors"
	"net/http"
	"sync"
	"time"

	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	semantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/metadata"
)

// RuntimeCallerBinding is the operator workload authority's canonical identity
// selected from api-host's opaque handle. Catalog-owned Org/AppInstallation
// grants are deliberately absent and come from Runtime security persistence
// only after caller-bound signature verification.
type RuntimeCallerBinding struct {
	CallerWorkloadID string
}

// RuntimeBindingSource is a directly nullable function so assembly cannot
// accept a typed-nil authority. found=false is an unauthorized workload, not a
// license to fall back to configuration or delegation claims.
type RuntimeBindingSource func(context.Context, runtimegrpc.VerifiedCaller) (binding RuntimeCallerBinding, found bool, err error)

type RuntimeSessionState struct {
	User            models.UserSession
	AuthEpoch       uint64
	Subject         permissionstatev1.Subject
	PermissionFloor permissionstatev1.Version
}

// RuntimeSessionAuthority reuses AuthService's one durable session read path.
// It receives only verified delegation identifiers, never a raw user bearer or
// caller-authored UserSession.
type RuntimeSessionAuthority func(ctx context.Context, sessionID, principalID string) (*RuntimeSessionState, error)

type RuntimeReplicaSecurityState struct {
	OrgID              string
	RuntimeAudience    string
	SecurityEpoch      uint64
	LeaseSecurityEpoch uint64
	LeaseExpiresAt     time.Time
}

type RuntimeAdmissionSecurityState struct {
	StateFound         bool
	InstallationFound  bool
	InstallationActive bool
	Capabilities       []string
	SecurityEpoch      uint64
	LeaseFound         bool
	LeaseSecurityEpoch uint64
	LeaseExpiresAt     time.Time
	DelegationRevoked  bool
}

type RuntimeReplicaSecuritySource func(context.Context) (RuntimeReplicaSecurityState, bool, error)
type RuntimeAdmissionSecuritySource func(context.Context, delegationv1.Claims) (RuntimeAdmissionSecurityState, error)

type RuntimeSecurityStateConfig struct {
	Replica   RuntimeReplicaSecuritySource
	Admission RuntimeAdmissionSecuritySource
	Now       func() time.Time
}

// RuntimeSecurityState owns the lease-backed immutable replica context and the
// per-call authoritative admission read. Authoritative stale state invalidates
// readiness. A transport/read failure rejects that attempt without turning one
// failed observation into new authority about every later attempt; process
// eligibility remains bounded by the already-issued lease, while every later
// attempt still performs its own authoritative admission read.
type RuntimeSecurityState struct {
	replicaSource RuntimeReplicaSecuritySource
	replica       RuntimeReplicaSecurityState
	admission     RuntimeAdmissionSecuritySource
	now           func() time.Time
	mu            sync.RWMutex
	readiness     runtimeSecurityReadiness
	workloadReady bool
}

type runtimeSecurityReadiness uint8

const (
	runtimeSecurityUnavailable runtimeSecurityReadiness = iota
	runtimeSecurityStale
	runtimeSecurityReady
)

func NewRuntimeSecurityState(ctx context.Context, config RuntimeSecurityStateConfig) (*RuntimeSecurityState, error) {
	if config.Replica == nil || config.Admission == nil {
		return nil, errors.New("runtime security state: replica and admission sources are required")
	}
	if config.Now == nil {
		config.Now = time.Now
	}
	state := &RuntimeSecurityState{replicaSource: config.Replica, admission: config.Admission, now: config.Now}
	if err := state.Refresh(ctx); err != nil {
		return nil, err
	}
	return state, nil
}

type SecurityAdmissionConfig struct {
	Verifier *delegationv1.Verifier
	Bindings RuntimeBindingSource
	Sessions RuntimeSessionAuthority
	State    *RuntimeSecurityState
}

// SecurityAdmission owns the Runtime-specific gate after api-host has
// authenticated the transport workload. It is immutable after construction.
type SecurityAdmission struct {
	verifier *delegationv1.Verifier
	bindings RuntimeBindingSource
	sessions RuntimeSessionAuthority
	state    *RuntimeSecurityState
	reject   bool
}

func NewSecurityAdmission(config SecurityAdmissionConfig) (*SecurityAdmission, error) {
	if config.Verifier == nil || config.Bindings == nil || config.Sessions == nil || config.State == nil {
		return nil, errors.New("runtime security admission: verifier, bindings, sessions, and state are required")
	}
	return &SecurityAdmission{verifier: config.Verifier, bindings: config.Bindings, sessions: config.Sessions, state: config.State}, nil
}

// NewRejectingSecurityAdmission is the only production-safe P6c bootstrap.
// It carries no verifier key, static grant, Org, AppInstallation, or session.
func NewRejectingSecurityAdmission() *SecurityAdmission { return &SecurityAdmission{reject: true} }

// Ready reports only whether this admission has a configured verifier and
// authority ports. P6d extends the same owner with epoch/lease readiness.
func (a *SecurityAdmission) Ready() bool {
	return a != nil && !a.reject && a.verifier != nil && a.bindings != nil && a.sessions != nil && a.state != nil && a.state.Ready()
}

func (s *RuntimeSecurityState) Ready() bool {
	if s == nil || s.now == nil {
		return false
	}
	s.mu.RLock()
	replica, readiness := s.replica, s.readiness
	s.mu.RUnlock()
	return readiness == runtimeSecurityReady && validReplicaSecurityState(replica, s.now())
}

// PreparedIdentity returns the lease-backed Runtime identity used when
// preparing ServingState. Requests still revalidate the independent
// SecurityEpoch through TrustedCall; this accessor is never a per-request
// substitute for SecurityAdmission.
func (s *RuntimeSecurityState) PreparedIdentity() (orgID, runtimeAudience string, securityEpoch uint64, ok bool) {
	if !s.Ready() {
		return "", "", 0, false
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.replica.OrgID, s.replica.RuntimeAudience, s.replica.SecurityEpoch, true
}

// Refresh rebinds readiness from the authoritative replica lease owner. Lease
// renewal/serving-state controllers call this after publishing a new lease; no
// implicit refresh interval or staleness value is invented before P6-D.
func (s *RuntimeSecurityState) Refresh(ctx context.Context) error {
	if s == nil || s.replicaSource == nil || s.now == nil {
		return errors.New("runtime security state: replica source is unavailable")
	}
	replica, found, err := s.replicaSource(ctx)
	if err != nil {
		// The error is not authoritative evidence that the already-issued
		// replica lease was revoked. Preserve the existing readiness state only
		// while that lease remains valid; Ready still closes at the exact lease
		// expiry and no request can bypass its independent admission read.
		s.mu.Lock()
		s.workloadReady = false
		if !validReplicaSecurityState(s.replica, s.now()) {
			s.readiness = runtimeSecurityUnavailable
		}
		s.mu.Unlock()
		return err
	}
	s.mu.Lock()
	switch {
	case !found:
		s.readiness = runtimeSecurityUnavailable
		s.workloadReady = false
	case validReplicaSecurityState(replica, s.now()):
		s.replica = replica
		s.readiness = runtimeSecurityReady
		s.workloadReady = true
	case validReplicaIdentity(replica):
		switch {
		case !validReplicaIdentity(s.replica):
			s.replica = replica
		case sameReplicaIdentity(s.replica, replica):
			// Epoch/lease facts are authoritative for the same identity even
			// when they prove this replica stale.
			s.replica.SecurityEpoch = replica.SecurityEpoch
			s.replica.LeaseSecurityEpoch = replica.LeaseSecurityEpoch
			s.replica.LeaseExpiresAt = replica.LeaseExpiresAt
		}
		s.readiness = runtimeSecurityStale
		s.workloadReady = false
	default:
		// Preserve the last authoritative Org/audience identity. Invalid
		// refresh data may stop readiness but cannot erase the context used
		// to authenticate malformed credentials before reporting staleness.
		s.readiness = runtimeSecurityStale
		s.workloadReady = false
	}
	s.mu.Unlock()
	return nil
}

func (s *RuntimeSecurityState) Replica() (RuntimeReplicaSecurityState, bool) {
	if !s.Ready() {
		return RuntimeReplicaSecurityState{}, false
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.replica, true
}

// workloadReplica returns authority only when the replica owner has just
// confirmed the process lease. Workload-consumer calls have no delegated-user
// admission read of their own, so a failed replica refresh must fence them
// until the authoritative replica source succeeds again.
func (s *RuntimeSecurityState) workloadReplica() (RuntimeReplicaSecurityState, bool) {
	if s == nil || s.now == nil {
		return RuntimeReplicaSecurityState{}, false
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.replica, s.readiness == runtimeSecurityReady && s.workloadReady && validReplicaSecurityState(s.replica, s.now())
}

// verificationReplica returns only a previously authoritative immutable
// identity context. It remains usable to authenticate a delegation after its
// lease expires or the state owner becomes unavailable; it never makes the
// replica ready and authorize still rejects the call as stale/unavailable.
// This preserves the public precedence that an absent/malformed delegation is
// unauthenticated while STALE_STATE applies only to an otherwise valid token.
func (s *RuntimeSecurityState) verificationReplica() (RuntimeReplicaSecurityState, bool) {
	if s == nil {
		return RuntimeReplicaSecurityState{}, false
	}
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.replica, s.replica.OrgID != "" && s.replica.RuntimeAudience != "" && s.replica.SecurityEpoch > 0
}

func validReplicaIdentity(state RuntimeReplicaSecurityState) bool {
	return state.OrgID != "" && state.RuntimeAudience != "" && state.SecurityEpoch > 0
}

func sameReplicaIdentity(left, right RuntimeReplicaSecurityState) bool {
	return left.OrgID == right.OrgID && left.RuntimeAudience == right.RuntimeAudience
}

func validReplicaSecurityState(state RuntimeReplicaSecurityState, now time.Time) bool {
	return state.OrgID != "" && state.RuntimeAudience != "" && state.SecurityEpoch > 0 &&
		state.LeaseSecurityEpoch == state.SecurityEpoch && !state.LeaseExpiresAt.IsZero() && now.Before(state.LeaseExpiresAt)
}

func (s *RuntimeSecurityState) authorize(ctx context.Context, claims delegationv1.Claims, requiredCapability string) error {
	replica, ready := s.Replica()
	if !ready {
		return s.readinessFailure()
	}
	if claims.OrgID != replica.OrgID {
		return &semantic.Failure{Category: semantic.FailureForbidden}
	}
	state, err := s.admission(ctx, claims)
	if err != nil {
		// This attempt cannot establish current authority and fails closed. The
		// absence of a result is not a global authority fact: later attempts
		// must retry their own read instead of being rejected from cached error
		// state. Known epoch/lease conflicts below still poison readiness.
		return admissionOwnerError(ctx, err)
	}
	if state.DelegationRevoked {
		return &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	if !state.StateFound || !state.InstallationFound || !state.InstallationActive || !containsCapability(state.Capabilities, requiredCapability) {
		return &semantic.Failure{Category: semantic.FailureForbidden}
	}
	observedAt := s.now()
	if !state.LeaseFound || state.SecurityEpoch != replica.SecurityEpoch || state.LeaseSecurityEpoch != state.SecurityEpoch ||
		!observedAt.Before(state.LeaseExpiresAt) {
		s.invalidateUnlessLeaseObservationWasSuperseded(replica, state, observedAt)
		served := claims.SecurityEpoch
		if state.LeaseFound {
			served = state.LeaseSecurityEpoch
		}
		return staleSecurityFailure(state.SecurityEpoch, served)
	}
	// A caller minted before a security rotation is stale, but the Runtime
	// replica is not. Reject only that delegation. Marking process readiness
	// stale here lets one old client poison every concurrent current caller
	// until the controller refreshes the same already-valid lease.
	if state.SecurityEpoch != claims.SecurityEpoch {
		return staleSecurityFailure(state.SecurityEpoch, claims.SecurityEpoch)
	}
	return nil
}

// invalidateUnlessLeaseObservationWasSuperseded fences the process for a
// current authoritative conflict, but does not let a delayed admission read
// poison a newer lease already published by the replica renewal owner. The
// admission attempt still fails closed; only an independently renewed,
// same-epoch lease keeps later attempts eligible to perform their own read.
func (s *RuntimeSecurityState) invalidateUnlessLeaseObservationWasSuperseded(
	startedWith RuntimeReplicaSecurityState,
	observed RuntimeAdmissionSecurityState,
	observedAt time.Time,
) {
	if s == nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	current := s.replica
	if observed.LeaseFound &&
		!observed.LeaseExpiresAt.IsZero() &&
		sameReplicaIdentity(startedWith, current) &&
		observed.SecurityEpoch == current.SecurityEpoch &&
		observed.LeaseSecurityEpoch == observed.SecurityEpoch &&
		observed.LeaseExpiresAt.Before(current.LeaseExpiresAt) &&
		validReplicaSecurityState(current, observedAt) {
		return
	}
	s.readiness = runtimeSecurityStale
	s.workloadReady = false
}

func (s *RuntimeSecurityState) readinessFailure() error {
	if s == nil {
		return &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	s.mu.RLock()
	readiness := s.readiness
	s.mu.RUnlock()
	if readiness == runtimeSecurityUnavailable {
		return &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	return s.staleFailure(0)
}

func (s *RuntimeSecurityState) staleFailure(served uint64) error {
	required := uint64(0)
	if s != nil {
		s.mu.RLock()
		required = s.replica.SecurityEpoch
		if served == 0 {
			served = s.replica.LeaseSecurityEpoch
		}
		s.mu.RUnlock()
	}
	return staleSecurityFailure(required, served)
}

func staleSecurityFailure(required, served uint64) error {
	return &semantic.Failure{
		Category: semantic.FailureStaleState, StaleAxis: semantic.StaleAxisSecurity,
		RequiredRevision: required, ServedRevision: served,
	}
}

func containsCapability(capabilities []string, required string) bool {
	for _, capability := range capabilities {
		if capability == required {
			return true
		}
	}
	return false
}

func (a *SecurityAdmission) UnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context, request any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		if err := ctx.Err(); err != nil {
			return nil, runtimegrpc.StatusError(ctx, err)
		}
		// Identity exchange has already passed api-host workload authentication
		// and owns bearer/session/security validation in Runtime Identity. It has
		// no delegation yet, so data-RPC SecurityAdmission must not require one.
		if info != nil && info.FullMethod == runtimev1.IdentityContextService_Exchange_FullMethodName {
			return handler(ctx, request)
		}
		methodClass, ok := runtimegrpc.RequestMethodClass(ctx)
		if !ok {
			return nil, runtimegrpc.StatusError(ctx, &semantic.Failure{Category: semantic.FailureInternal})
		}
		if methodClass == runtimegrpc.MethodClassWorkloadConsumer {
			admitted, err := a.admitWorkload(ctx)
			if err != nil {
				return nil, runtimegrpc.StatusError(ctx, err)
			}
			return handler(admitted, request)
		}
		if methodClass != runtimegrpc.MethodClassDelegatedUserData {
			return nil, runtimegrpc.StatusError(ctx, &semantic.Failure{Category: semantic.FailureInternal})
		}
		admitted, err := a.admit(ctx)
		if err != nil {
			return nil, runtimegrpc.StatusError(ctx, err)
		}
		return handler(admitted, request)
	}
}

// StreamServerInterceptor applies exactly the same security state machine to
// a client-streaming command. Keeping one admission implementation prevents a
// stream from receiving a weaker identity, delegation, session, or cancellation
// policy than a unary Runtime capability.
func (a *SecurityAdmission) StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(server any, stream grpc.ServerStream, _ *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
		if stream == nil {
			return runtimegrpc.StatusError(context.Background(), &semantic.Failure{Category: semantic.FailureInternal})
		}
		ctx := stream.Context()
		if err := ctx.Err(); err != nil {
			return runtimegrpc.StatusError(ctx, err)
		}
		methodClass, ok := runtimegrpc.RequestMethodClass(ctx)
		if !ok {
			return runtimegrpc.StatusError(ctx, &semantic.Failure{Category: semantic.FailureInternal})
		}
		if methodClass == runtimegrpc.MethodClassWorkloadConsumer {
			admitted, err := a.admitWorkload(ctx)
			if err != nil {
				return runtimegrpc.StatusError(ctx, err)
			}
			return handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: admitted})
		}
		if methodClass != runtimegrpc.MethodClassDelegatedUserData {
			return runtimegrpc.StatusError(ctx, &semantic.Failure{Category: semantic.FailureInternal})
		}
		admitted, err := a.admit(ctx)
		if err != nil {
			return runtimegrpc.StatusError(ctx, err)
		}
		return handler(server, &runtimeContextServerStream{ServerStream: stream, ctx: admitted})
	}
}

type trustedWorkloadCallContextKey struct{}

// TrustedWorkloadCall is the request-scoped authority for workload-consumer
// methods. It intentionally has no principal, session, or end-user delegation:
// the host-authenticated caller and Runtime's durable workload binding are the
// only identity inputs for this method class.
type TrustedWorkloadCall struct {
	orgID            string
	callerWorkloadID string
	capability       string
	securityEpoch    uint64
}

// RuntimeWorkloadCall is the read-only projection of host-authenticated
// workload authority made available to Runtime-owned workload engines. The
// private context key remains owned here, so callers cannot manufacture this
// value by placing fields in a request message.
type RuntimeWorkloadCall struct {
	OrgID            string
	CallerWorkloadID string
	Capability       string
	SecurityEpoch    uint64
}

// RuntimeWorkloadCallFromContext returns authority only after the api-host and
// SecurityAdmission interceptors have authenticated the transport workload.
func RuntimeWorkloadCallFromContext(ctx context.Context) (RuntimeWorkloadCall, bool) {
	trusted, ok := trustedWorkloadCallFromContext(ctx)
	if !ok {
		return RuntimeWorkloadCall{}, false
	}
	return RuntimeWorkloadCall{
		OrgID: trusted.orgID, CallerWorkloadID: trusted.callerWorkloadID,
		Capability: trusted.capability, SecurityEpoch: trusted.securityEpoch,
	}, true
}

func (a *SecurityAdmission) admitWorkload(ctx context.Context) (context.Context, error) {
	capability, ok := runtimegrpc.Capability(ctx)
	if !ok {
		return nil, &semantic.Failure{Category: semantic.FailureInternal}
	}
	if a == nil || a.reject || a.bindings == nil || a.state == nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	caller, ok := runtimegrpc.Caller(ctx)
	if !ok {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	bindingCtx, finishBinding := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseWorkloadBinding)
	binding, found, err := a.bindings(bindingCtx, caller)
	finishBinding()
	if err := admissionOwnerError(ctx, err); err != nil {
		return nil, err
	}
	if !found || binding.CallerWorkloadID == "" {
		return nil, &semantic.Failure{Category: semantic.FailureForbidden}
	}
	replica, ready := a.state.workloadReplica()
	if !ready {
		if a.state.Ready() {
			return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
		}
		return nil, a.state.readinessFailure()
	}
	trusted := TrustedWorkloadCall{
		orgID: replica.OrgID, callerWorkloadID: binding.CallerWorkloadID,
		capability: capability, securityEpoch: replica.SecurityEpoch,
	}
	return context.WithValue(ctx, trustedWorkloadCallContextKey{}, trusted), nil
}

func trustedWorkloadCallFromContext(ctx context.Context) (TrustedWorkloadCall, bool) {
	if ctx == nil {
		return TrustedWorkloadCall{}, false
	}
	call, ok := ctx.Value(trustedWorkloadCallContextKey{}).(TrustedWorkloadCall)
	return call, ok && call.orgID != "" && call.callerWorkloadID != "" &&
		call.capability != "" && call.securityEpoch != 0
}

func (a *SecurityAdmission) admit(ctx context.Context) (context.Context, error) {
	requiredCapability, ok := runtimegrpc.Capability(ctx)
	if !ok {
		return nil, &semantic.Failure{Category: semantic.FailureInternal}
	}
	if a == nil || a.reject || a.verifier == nil || a.bindings == nil || a.sessions == nil || a.state == nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	workload, ok := runtimegrpc.Caller(ctx)
	if !ok {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	token, ok := delegationMetadata(ctx)
	if !ok || len(token) > delegationv1.MaxCompactTokenBytes {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}

	bindingCtx, finishBinding := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseWorkloadBinding)
	binding, found, err := a.bindings(bindingCtx, workload)
	finishBinding()
	if err := admissionOwnerError(ctx, err); err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if !found {
		binding = rejectionBinding()
	}
	replica, known := a.state.verificationReplica()
	if !known {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	_, finishVerification := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseDelegationVerification)
	claims, err := a.verifier.VerifyCallerBound(token, a.state.now(), delegationv1.ExpectedCallerBindings{
		CallerWorkloadID: binding.CallerWorkloadID, RuntimeAudience: replica.RuntimeAudience,
		RequiredCapability: requiredCapability,
	})
	finishVerification()
	if err != nil {
		return nil, classifyDelegationFailure(err)
	}
	if !found {
		return nil, &semantic.Failure{Category: semantic.FailureForbidden}
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	stateCtx, finishSecurityState := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseSecurityState)
	authorizeErr := a.state.authorize(stateCtx, claims, requiredCapability)
	finishSecurityState()
	if authorizeErr != nil {
		return nil, authorizeErr
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}

	sessionCtx, finishSession := beginRuntimeRPCProfilePhase(ctx, runtimeRPCPhaseSession)
	session, err := a.sessions(sessionCtx, claims.SessionID, claims.PrincipalID)
	finishSession()
	if err := admissionOwnerError(ctx, err); err != nil {
		return nil, err
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if session == nil || session.AuthEpoch == 0 || session.AuthEpoch != claims.AuthEpoch ||
		session.User.ID == "" || session.User.ID != claims.PrincipalID {
		return nil, &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	if err := session.Subject.Validate(); err != nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	if err := session.PermissionFloor.Validate(); err != nil {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	if session.Subject.UserID() != session.User.ID ||
		session.Subject.ProfileID() != session.User.ProfileID {
		return nil, &semantic.Failure{Category: semantic.FailureUnavailable}
	}
	trusted := newTrustedCall(claims, session.User)
	trusted.subject = session.Subject
	trusted.permissionFloor = session.PermissionFloor
	return context.WithValue(ctx, trustedCallContextKey{}, trusted), nil
}

func delegationMetadata(ctx context.Context) (string, bool) {
	values, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		return "", false
	}
	tokens := values.Get(envelope.RuntimeDelegationMetadataKey)
	if len(tokens) != 1 || tokens[0] == "" {
		return "", false
	}
	return tokens[0], true
}

func rejectionBinding() RuntimeCallerBinding {
	return RuntimeCallerBinding{CallerWorkloadID: "unbound-workload"}
}

func classifyDelegationFailure(err error) error {
	var failure *delegationv1.Failure
	if !errors.As(err, &failure) {
		return &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	if failure.Category == delegationv1.FailureForbidden {
		return &semantic.Failure{Category: semantic.FailureForbidden}
	}
	return &semantic.Failure{Category: semantic.FailureUnauthenticated}
}

func admissionOwnerError(ctx context.Context, err error) error {
	if ctxErr := ctx.Err(); ctxErr != nil {
		return ctxErr
	}
	if err == nil {
		return nil
	}
	var statusError interface{ HTTPStatus() int }
	if errors.As(err, &statusError) && statusError.HTTPStatus() == http.StatusUnauthorized {
		return &semantic.Failure{Category: semantic.FailureUnauthenticated}
	}
	return &semantic.Failure{Category: semantic.FailureUnavailable}
}

type trustedCallContextKey struct{}

// TrustedCall is the one immutable, request-scoped Runtime authority view. Its
// fields are private even inside the internal package; accessors defensively
// copy pointer/slice-backed state.
type TrustedCall struct {
	delegationID      string
	orgID             string
	principalID       string
	sessionID         string
	callerWorkloadID  string
	appInstallationID string
	runtimeAudience   string
	capabilities      []string
	authEpoch         uint64
	securityEpoch     uint64
	user              models.UserSession
	subject           permissionstatev1.Subject
	permissionFloor   permissionstatev1.Version
}

func newTrustedCall(claims delegationv1.Claims, user models.UserSession) TrustedCall {
	return TrustedCall{
		delegationID: claims.DelegationID, orgID: claims.OrgID, principalID: claims.PrincipalID,
		sessionID: claims.SessionID, callerWorkloadID: claims.CallerWorkloadID,
		appInstallationID: claims.AppInstallationID, runtimeAudience: claims.RuntimeAudience,
		capabilities: append([]string(nil), claims.Capabilities...), authEpoch: claims.AuthEpoch,
		securityEpoch: claims.SecurityEpoch, user: cloneUserSession(user),
	}
}

func trustedCallFromContext(ctx context.Context) (TrustedCall, bool) {
	if ctx == nil {
		return TrustedCall{}, false
	}
	call, ok := ctx.Value(trustedCallContextKey{}).(TrustedCall)
	if !ok || call.principalID == "" || call.user.ID == "" {
		return TrustedCall{}, false
	}
	call.capabilities = append([]string(nil), call.capabilities...)
	call.user = cloneUserSession(call.user)
	return call, true
}

func (c TrustedCall) UserSession() models.UserSession { return cloneUserSession(c.user) }
func (c TrustedCall) Capabilities() []string          { return append([]string(nil), c.capabilities...) }
func (c TrustedCall) Subject() permissionstatev1.Subject {
	return c.subject
}
func (c TrustedCall) PermissionFloor() permissionstatev1.Version {
	return c.permissionFloor
}

func cloneUserSession(user models.UserSession) models.UserSession {
	clone := user
	clone.Email = cloneStringPointer(user.Email)
	clone.RoleID = cloneStringPointer(user.RoleID)
	clone.RoleName = cloneStringPointer(user.RoleName)
	return clone
}

func cloneStringPointer(value *string) *string {
	if value == nil {
		return nil
	}
	clone := *value
	return &clone
}