package invocation

import (
	"context"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/internal/core/identifiers"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/store/storeerr"
	domain "github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
)

const acceptDigestDomain = "loop.workflow.invocation.accept/v1"

// AcceptManual accepts a direct Plan occurrence without fabricating a Binding.
// Schedule acceptance deliberately remains transaction-owned by its cursor
// materializer through AcceptInTx.
func (store *Store) AcceptManual(ctx context.Context, params AcceptManualParams) (AcceptResult, error) {
	if err := store.Validate(); err != nil {
		return AcceptResult{}, err
	}
	normalized, err := normalizeManualAcceptParams(params)
	if err != nil {
		return AcceptResult{}, err
	}
	var result AcceptResult
	err = coredb.WithTx(ctx, store.db, func(tx coredb.Tx) error {
		var acceptErr error
		result, acceptErr = store.acceptManualInTx(ctx, tx, normalized)
		return acceptErr
	})
	var insertConflict invocationInsertConflictError
	if errors.As(err, &insertConflict) {
		replayed, found, replayErr := store.ReplayManual(ctx, ReplayManualParams{
			WorkspaceID: normalized.WorkspaceID, PlanID: normalized.PlanID,
			RequestedPlanVersionID: normalized.RequestedPlanVersionID,
			RequestInputJSON:       normalized.RequestInputJSON, InvokedBy: normalized.InvokedBy,
			IdempotencyKey: normalized.IdempotencyKey, TeamID: normalized.TeamID,
		})
		if replayErr != nil {
			return AcceptResult{}, replayErr
		}
		if found {
			return replayed, nil
		}
		return AcceptResult{}, storeerr.Conflict("workflow invocation already exists")
	}
	return result, err
}

// ReplayManual resolves an accepted direct request before any mutable Plan,
// default-version, membership, or Agent-manager read.
func (store *Store) ReplayManual(ctx context.Context, params ReplayManualParams) (AcceptResult, bool, error) {
	if err := store.Validate(); err != nil {
		return AcceptResult{}, false, err
	}
	params.WorkspaceID = strings.TrimSpace(params.WorkspaceID)
	params.PlanID = strings.TrimSpace(params.PlanID)
	params.RequestedPlanVersionID = strings.TrimSpace(params.RequestedPlanVersionID)
	params.IdempotencyKey = strings.TrimSpace(params.IdempotencyKey)
	params.InvokedBy.Type = strings.TrimSpace(params.InvokedBy.Type)
	params.InvokedBy.ID = strings.TrimSpace(params.InvokedBy.ID)
	params.InvokedBy.MemberID = strings.TrimSpace(params.InvokedBy.MemberID)
	if params.WorkspaceID == "" || params.PlanID == "" || params.IdempotencyKey == "" || len(params.IdempotencyKey) > 128 ||
		(params.InvokedBy.Type != ActorTypeHuman && params.InvokedBy.Type != ActorTypeAgent) ||
		params.InvokedBy.ID == "" || params.InvokedBy.MemberID == "" {
		return AcceptResult{}, false, storeerr.BadRequest("complete manual replay identity and a bounded idempotencyKey are required")
	}
	input, err := domain.DeterministicJSONObject("input", params.RequestInputJSON, domain.MaxInputBytes)
	if err != nil {
		return AcceptResult{}, false, storeerr.BadRequest(err.Error())
	}
	digest := manualRequestDigest(
		params.WorkspaceID, params.PlanID, params.RequestedPlanVersionID,
		params.TeamID, input, params.InvokedBy,
	)
	existing, found, err := loadInvocationByActorIdempotency(
		ctx, store.db, params.WorkspaceID, params.InvokedBy, params.IdempotencyKey,
	)
	if err != nil || !found {
		return AcceptResult{}, false, err
	}
	result, err := replayAcceptance(ctx, store.db, existing, digest)
	return result, true, err
}

func normalizeManualAcceptParams(params AcceptManualParams) (AcceptManualParams, error) {
	params.WorkspaceID = strings.TrimSpace(params.WorkspaceID)
	params.PlanID = strings.TrimSpace(params.PlanID)
	params.RequestedPlanVersionID = strings.TrimSpace(params.RequestedPlanVersionID)
	params.ExpectedPlanVersionID = strings.TrimSpace(params.ExpectedPlanVersionID)
	params.ExpectedVersionDigest = strings.TrimSpace(params.ExpectedVersionDigest)
	params.IdempotencyKey = strings.TrimSpace(params.IdempotencyKey)
	params.InvokedBy.Type = strings.TrimSpace(params.InvokedBy.Type)
	params.InvokedBy.ID = strings.TrimSpace(params.InvokedBy.ID)
	params.InvokedBy.MemberID = strings.TrimSpace(params.InvokedBy.MemberID)
	params.Principal.UserID = strings.TrimSpace(params.Principal.UserID)
	params.Principal.MemberID = strings.TrimSpace(params.Principal.MemberID)
	params.Principal.Role = strings.TrimSpace(params.Principal.Role)
	if params.WorkspaceID == "" || params.PlanID == "" || params.ExpectedPlanVersionID == "" ||
		params.ExpectedVersionDigest == "" || params.IdempotencyKey == "" || len(params.IdempotencyKey) > 128 {
		return AcceptManualParams{}, storeerr.BadRequest("workspaceId, planId, an exact PlanVersion fence, and a bounded idempotencyKey are required")
	}
	if params.InvokedBy.Type != ActorTypeHuman && params.InvokedBy.Type != ActorTypeAgent ||
		params.InvokedBy.ID == "" || params.InvokedBy.MemberID == "" {
		return AcceptManualParams{}, storeerr.BadRequest("manual invokedBy must be a complete human or agent actor")
	}
	if params.Principal.UserID == "" || params.Principal.MemberID == "" || params.Principal.Role == "" {
		return AcceptManualParams{}, storeerr.BadRequest("manual principal userId, memberId, and role are required")
	}
	input, err := domain.DeterministicJSONObject("input", params.InputJSON, domain.MaxInputBytes)
	if err != nil {
		return AcceptManualParams{}, storeerr.BadRequest(err.Error())
	}
	params.InputJSON = input
	if len(params.RequestInputJSON) == 0 {
		params.RequestInputJSON = input
	}
	requestInput, err := domain.DeterministicJSONObject("request input", params.RequestInputJSON, domain.MaxInputBytes)
	if err != nil {
		return AcceptManualParams{}, storeerr.BadRequest(err.Error())
	}
	params.RequestInputJSON = requestInput
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = params.Now.UTC().Truncate(time.Microsecond)
	return params, nil
}

func (store *Store) acceptManualInTx(ctx context.Context, tx coredb.Tx, params AcceptManualParams) (AcceptResult, error) {
	digest := manualRequestDigest(
		params.WorkspaceID, params.PlanID, params.RequestedPlanVersionID,
		params.TeamID, params.RequestInputJSON, params.InvokedBy,
	)
	if replayed, found, err := replayActorAcceptanceInTx(
		ctx, tx, params.WorkspaceID, params.InvokedBy, params.IdempotencyKey, digest,
	); err != nil || found {
		return replayed, err
	}
	if err := lockManualAcceptancePlacement(ctx, tx, params.WorkspaceID); err != nil {
		return AcceptResult{}, err
	}

	material, err := lockManualPlanVersion(
		ctx, tx, params.WorkspaceID, params.PlanID, params.RequestedPlanVersionID,
		params.ExpectedPlanVersionID, params.ExpectedVersionDigest,
	)
	if err != nil {
		return AcceptResult{}, err
	}
	// The Plan lock serializes direct calls to this Plan. Recheck actor-scoped
	// replay after acquiring it so concurrent retries observe the winner.
	if replayed, found, err := replayActorAcceptanceInTx(
		ctx, tx, params.WorkspaceID, params.InvokedBy, params.IdempotencyKey, digest,
	); err != nil || found {
		return replayed, err
	}

	identity, err := lockAndAuthorizeManualPrincipal(ctx, tx, params.WorkspaceID, params.Principal, params.InvokedBy)
	if err != nil {
		return AcceptResult{}, err
	}
	now := params.Now
	occurrenceRef := domain.ManualOccurrenceRef(params.PlanID, params.InvokedBy, params.IdempotencyKey)
	provenance, err := canonicalManualSourceProvenance(
		material, params.InvokedBy, params.RequestInputJSON, occurrenceRef, now,
	)
	if err != nil {
		return AcceptResult{}, err
	}
	invocationID, err := identifiers.NewEntityID("wiv")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Invocation id: %w", err)
	}
	runID, err := identifiers.NewEntityID("wri")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Run id: %w", err)
	}
	eventID, err := identifiers.NewEntityID("wre")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Run event id: %w", err)
	}
	inputDigest := domain.InputDigest(params.InputJSON)
	invocation := Invocation{
		ID: invocationID, RunID: runID, WorkspaceID: params.WorkspaceID,
		PlanID: material.PlanID, PlanVersionID: material.PlanVersionID,
		WorkflowNameSnapshot: strings.TrimSpace(material.WorkflowName),
		SourceKind:           SourceKindManual, SourceOccurrenceRef: occurrenceRef, SourceOccurredAt: now,
		InputJSON: append([]byte(nil), params.InputJSON...), InputDigest: inputDigest,
		SourceProvenanceJSON: provenance, Identity: identity, InvokedBy: params.InvokedBy,
		IdempotencyKey: params.IdempotencyKey, RequestDigest: digest, AcceptedAt: now,
	}
	runTeamID, err := resolveRunTeamID(ctx, tx, params.WorkspaceID, material.PlanID)
	if err != nil {
		return AcceptResult{}, err
	}
	// Precedence: a Plan that is Team-linked through an approved, activated
	// relation freezes the authoritative Team ID; a caller-provided explicit
	// Team ID is server-authorized only for unlinked/dynamic Plans. A conflicting
	// explicit value on a Team-linked Plan is rejected so the durable relation
	// stays authoritative.
	if requestedTeamID := strings.TrimSpace(params.TeamID); requestedTeamID != "" {
		if runTeamID != "" && runTeamID != requestedTeamID {
			return AcceptResult{}, storeerr.Conflict("explicit teamId conflicts with the Plan's Team-linked relation")
		}
		if runTeamID == "" {
			if err := authorizeManualInvocationTeam(ctx, tx, params.WorkspaceID, requestedTeamID, params.InvokedBy); err != nil {
				return AcceptResult{}, err
			}
			runTeamID = requestedTeamID
		}
	}
	run := Run{
		ID: runID, WorkspaceID: params.WorkspaceID, InvocationID: invocationID,
		PlanID: invocation.PlanID, PlanVersionID: invocation.PlanVersionID,
		TeamID: runTeamID, WorkflowNameSnapshot: invocation.WorkflowNameSnapshot,
		Status: RunStatusQueued, InputJSON: append([]byte(nil), params.InputJSON...),
		InputDigest: inputDigest, Identity: identity, QueuedAt: now, CreatedAt: now, UpdatedAt: now,
	}
	eventPayload, err := acceptedRunEventPayload(invocation, run)
	if err != nil {
		return AcceptResult{}, err
	}
	event := RunEvent{
		ID: eventID, WorkspaceID: params.WorkspaceID, RunID: runID,
		Seq: 1, Type: domain.RunEventAccepted, PayloadJSON: eventPayload, CreatedAt: now,
	}
	if err := insertAcceptance(ctx, tx, invocation, run, event); err != nil {
		return AcceptResult{}, err
	}
	return AcceptResult{Invocation: invocation, Run: run, Event: event}, nil
}

func lockAndAuthorizeManualPrincipal(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	principal Principal,
	actor Actor,
) (ExecutionIdentitySnapshot, error) {
	var member lockedAcceptanceMember
	if err := tx.QueryRow(ctx, `
SELECT id, type, COALESCE(user_id, ''), COALESCE(workspace_role, ''), status
FROM members
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
`, workspaceID, principal.MemberID).Scan(
		&member.ID, &member.Type, &member.UserID, &member.WorkspaceRole, &member.Status,
	); err != nil {
		if coredb.IsNotFound(err) {
			return ExecutionIdentitySnapshot{}, ErrPrincipalUnavailable
		}
		return ExecutionIdentitySnapshot{}, fmt.Errorf("lock direct workflow Invocation principal: %w", err)
	}
	if member.Type != ActorTypeHuman || member.Status != "active" || member.UserID != principal.UserID || member.WorkspaceRole != principal.Role {
		return ExecutionIdentitySnapshot{}, ErrPrincipalUnavailable
	}
	if actor.Type == ActorTypeHuman {
		if actor.ID != principal.UserID || actor.MemberID != principal.MemberID {
			return ExecutionIdentitySnapshot{}, ErrPrincipalUnavailable
		}
	} else if err := authorizeAgentPrincipal(ctx, tx, workspaceID, principal, actor); err != nil {
		return ExecutionIdentitySnapshot{}, err
	}
	return ExecutionIdentitySnapshot{
		OwnerUserID: principal.UserID, OwnerMemberID: principal.MemberID,
		PrincipalMemberID: principal.MemberID, AuthorMemberID: principal.MemberID,
	}, nil
}

func canonicalManualSourceProvenance(
	material lockedManualPlanVersion,
	invokedBy Actor,
	requestInput []byte,
	occurrenceRef string,
	occurredAt time.Time,
) ([]byte, error) {
	type actorSnapshot struct {
		Type     string `json:"type"`
		ID       string `json:"id"`
		MemberID string `json:"memberId"`
	}
	actorValue := func(actor Actor) actorSnapshot {
		return actorSnapshot{Type: actor.Type, ID: actor.ID, MemberID: actor.MemberID}
	}
	payload := struct {
		SourceKind string `json:"sourceKind"`
		Source     struct {
			OccurrenceRef string          `json:"occurrenceRef"`
			OccurredAt    string          `json:"occurredAt"`
			RequestInput  json.RawMessage `json:"requestInput"`
		} `json:"source"`
		Plan struct {
			ID              string        `json:"id"`
			DisplayName     string        `json:"displayName"`
			VersionID       string        `json:"versionId"`
			CreatedBy       actorSnapshot `json:"createdBy"`
			PublishedBy     actorSnapshot `json:"publishedBy"`
			CompilerVersion string        `json:"compilerVersion"`
			Digest          string        `json:"digest"`
		} `json:"plan"`
		InvokedBy actorSnapshot `json:"invokedBy"`
	}{SourceKind: SourceKindManual}
	payload.Source.OccurrenceRef = occurrenceRef
	payload.Source.OccurredAt = occurredAt.UTC().Truncate(time.Microsecond).Format(time.RFC3339Nano)
	payload.Source.RequestInput = append(json.RawMessage(nil), requestInput...)
	payload.Plan.ID = material.PlanID
	payload.Plan.DisplayName = strings.TrimSpace(material.WorkflowName)
	payload.Plan.VersionID = material.PlanVersionID
	payload.Plan.CreatedBy = actorValue(material.PlanCreatedBy)
	payload.Plan.PublishedBy = actorValue(material.PlanPublishedBy)
	payload.Plan.CompilerVersion = strings.TrimSpace(material.PlanCompilerVersion)
	payload.Plan.Digest = strings.TrimSpace(material.PlanVersionDigest)
	payload.InvokedBy = actorValue(invokedBy)
	encoded, err := json.Marshal(payload)
	if err != nil {
		return nil, fmt.Errorf("encode direct workflow source provenance: %w", err)
	}
	return domain.DeterministicSourceProvenanceJSONObject("source provenance", encoded, domain.MaxSourceProvenanceBytes)
}

// AcceptInTx atomically creates one immutable Invocation and its one queued
// Run. Schedule callers retain ownership of the outer transaction so cursor
// advancement/one-shot closure can be committed with this acceptance.
func (store *Store) AcceptInTx(ctx context.Context, tx coredb.Tx, params AcceptParams) (AcceptResult, error) {
	if store == nil || tx == nil {
		return AcceptResult{}, fmt.Errorf("workflow invocation transaction is not configured")
	}
	normalized, err := normalizeAcceptParams(params)
	if err != nil {
		return AcceptResult{}, err
	}
	digest := acceptRequestDigest(normalized)
	occurrenceRef := domain.ScheduleOccurrenceRef(normalized.Expected.BindingSourceID, *normalized.OccurrenceAt)
	replayed, found, err := replayOccurrenceAcceptanceInTx(
		ctx, tx, normalized.WorkspaceID, normalized.Expected.BindingSourceID, occurrenceRef, digest,
	)
	if err != nil {
		return AcceptResult{}, err
	}
	if found {
		return replayed, nil
	}

	deployment, err := loadDeployment(
		ctx, tx, normalized.WorkspaceID, normalized.BindingID, normalized.Expected.BindingSourceID, true,
	)
	if err != nil {
		return AcceptResult{}, err
	}
	if err := requireDeploymentFence(deployment, normalized); err != nil {
		return AcceptResult{}, err
	}

	// The source row lock serializes acceptance for one activation generation.
	// Recheck occurrence identity after acquiring it so concurrent retries see
	// the winner before attempting an insert.
	replayed, found, err = replayOccurrenceAcceptanceInTx(
		ctx, tx, normalized.WorkspaceID, deployment.BindingSourceID, occurrenceRef, digest,
	)
	if err != nil {
		return AcceptResult{}, err
	}
	if found {
		return replayed, nil
	}

	// Every mutable membership row used to authorize a fresh acceptance is
	// locked once, in member-id order. In particular, do not lock the owner and
	// principal in semantic order: two Bindings may use each other's owner as
	// principal and would otherwise acquire the same pair in opposite orders.
	members, err := lockAcceptanceMembers(ctx, tx, deployment, normalized)
	if err != nil {
		return AcceptResult{}, err
	}
	identity, err := executionIdentityFromLockedMembers(deployment, members)
	if err != nil {
		return AcceptResult{}, err
	}

	now := normalized.Now
	occurredAt := now
	if normalized.OccurrenceAt != nil {
		occurredAt = *normalized.OccurrenceAt
	}
	provenance, err := canonicalSourceProvenance(deployment, occurrenceRef, occurredAt)
	if err != nil {
		return AcceptResult{}, err
	}
	invocationID, err := identifiers.NewEntityID("wiv")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Invocation id: %w", err)
	}
	runID, err := identifiers.NewEntityID("wri")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Run id: %w", err)
	}
	eventID, err := identifiers.NewEntityID("wre")
	if err != nil {
		return AcceptResult{}, fmt.Errorf("generate workflow Run event id: %w", err)
	}
	inputDigest := domain.InputDigest(normalized.InputJSON)
	invocation := Invocation{
		ID: invocationID, RunID: runID, WorkspaceID: normalized.WorkspaceID,
		BindingID: deployment.BindingID, BindingRevisionID: deployment.BindingRevisionID,
		BindingRevision: deployment.BindingRevision, BindingSourceID: deployment.BindingSourceID,
		BindingActivation: deployment.BindingActivation,
		PlanID:            deployment.PlanID, PlanVersionID: deployment.PlanVersionID,
		WorkflowNameSnapshot: strings.TrimSpace(deployment.WorkflowName),
		SourceKind:           normalized.SourceKind, SourceOccurrenceRef: occurrenceRef, SourceOccurredAt: occurredAt,
		InputJSON: append([]byte(nil), normalized.InputJSON...), InputDigest: inputDigest, SourceProvenanceJSON: provenance,
		Identity:       identity,
		IdempotencyKey: normalized.IdempotencyKey, RequestDigest: digest, AcceptedAt: now,
	}
	runTeamID, err := resolveRunTeamID(ctx, tx, normalized.WorkspaceID, deployment.PlanID)
	if err != nil {
		return AcceptResult{}, err
	}
	run := Run{
		ID: runID, WorkspaceID: normalized.WorkspaceID, InvocationID: invocationID,
		PlanID: invocation.PlanID, PlanVersionID: invocation.PlanVersionID,
		TeamID: runTeamID, WorkflowNameSnapshot: invocation.WorkflowNameSnapshot,
		Status: RunStatusQueued, InputJSON: append([]byte(nil), normalized.InputJSON...), InputDigest: inputDigest, Identity: identity,
		QueuedAt: now, CreatedAt: now, UpdatedAt: now,
	}
	eventPayload, err := acceptedRunEventPayload(invocation, run)
	if err != nil {
		return AcceptResult{}, err
	}
	event := RunEvent{
		ID: eventID, WorkspaceID: normalized.WorkspaceID, RunID: runID,
		Seq: 1, Type: domain.RunEventAccepted, PayloadJSON: eventPayload, CreatedAt: now,
	}
	if err := insertAcceptance(ctx, tx, invocation, run, event); err != nil {
		return AcceptResult{}, err
	}
	return AcceptResult{Invocation: invocation, Run: run, Event: event}, nil
}

func normalizeAcceptParams(params AcceptParams) (AcceptParams, error) {
	params.WorkspaceID = strings.TrimSpace(params.WorkspaceID)
	params.BindingID = strings.TrimSpace(params.BindingID)
	params.SourceKind = strings.TrimSpace(params.SourceKind)
	params.IdempotencyKey = strings.TrimSpace(params.IdempotencyKey)
	params.Expected.BindingSourceID = strings.TrimSpace(params.Expected.BindingSourceID)
	params.Expected.BindingRevisionID = strings.TrimSpace(params.Expected.BindingRevisionID)
	params.Expected.PlanVersionID = strings.TrimSpace(params.Expected.PlanVersionID)
	params.Expected.RevisionDigest = strings.TrimSpace(params.Expected.RevisionDigest)
	if params.WorkspaceID == "" || params.BindingID == "" || params.IdempotencyKey == "" || len(params.IdempotencyKey) > 128 {
		return AcceptParams{}, storeerr.BadRequest("workspaceId, bindingId, and a bounded idempotencyKey are required")
	}
	if params.Expected.BindingSourceID == "" || params.Expected.BindingRevisionID == "" ||
		params.Expected.PlanVersionID == "" || params.Expected.RevisionDigest == "" {
		return AcceptParams{}, storeerr.BadRequest("an exact deployment fence is required")
	}
	deterministicInput, err := domain.DeterministicJSONObject("input", params.InputJSON, domain.MaxInputBytes)
	if err != nil {
		return AcceptParams{}, storeerr.BadRequest(err.Error())
	}
	params.InputJSON = deterministicInput
	if len(params.RequestInputJSON) == 0 {
		params.RequestInputJSON = params.InputJSON
	}
	canonicalRequestInput, err := domain.DeterministicJSONObject("request input", params.RequestInputJSON, domain.MaxInputBytes)
	if err != nil {
		return AcceptParams{}, storeerr.BadRequest(err.Error())
	}
	params.RequestInputJSON = canonicalRequestInput
	if params.SourceKind != SourceKindSchedule || params.OccurrenceAt == nil {
		return AcceptParams{}, storeerr.BadRequest("schedule acceptance requires source kind schedule and occurrenceAt")
	}
	occurrence := params.OccurrenceAt.UTC().Truncate(time.Microsecond)
	params.OccurrenceAt = &occurrence
	if params.Now.IsZero() {
		params.Now = time.Now()
	}
	params.Now = params.Now.UTC().Truncate(time.Microsecond)
	return params, nil
}

func requireDeploymentFence(deployment lockedDeployment, params AcceptParams) error {
	if name := strings.TrimSpace(deployment.WorkflowName); name == "" || len(name) > 255 {
		return ErrDeploymentChanged
	}
	if deployment.BindingState != "enabled" || deployment.SourceStatus != "active" ||
		deployment.ActiveSourceID != deployment.BindingSourceID || deployment.ActiveRevisionID != deployment.BindingRevisionID {
		return ErrInactiveDeployment
	}
	if deployment.SourceKind != params.SourceKind {
		return ErrDeploymentChanged
	}
	if deployment.BindingSourceID != params.Expected.BindingSourceID ||
		deployment.BindingRevisionID != params.Expected.BindingRevisionID ||
		deployment.PlanVersionID != params.Expected.PlanVersionID ||
		deployment.RevisionDigest != params.Expected.RevisionDigest {
		return ErrDeploymentChanged
	}
	if deployment.NextOccurrenceAt == nil || !deployment.NextOccurrenceAt.Equal(*params.OccurrenceAt) {
		return ErrDeploymentChanged
	}
	return nil
}

type lockedAcceptanceMember struct {
	ID            string
	Type          string
	UserID        string
	WorkspaceRole string
	Status        string
}

func lockAcceptanceMembers(
	ctx context.Context,
	tx coredb.Tx,
	deployment lockedDeployment,
	params AcceptParams,
) (map[string]lockedAcceptanceMember, error) {
	ids := []string{deployment.OwnerMemberID}
	ids = sortedUniqueNonEmpty(ids)
	if len(ids) == 0 {
		return nil, fmt.Errorf("workflow invocation acceptance has no execution owner")
	}

	query := `
SELECT id, type, COALESCE(user_id, ''), COALESCE(workspace_role, ''), status
FROM members
WHERE workspace_id = $1 AND id IN (`
	args := make([]any, 0, len(ids)+1)
	args = append(args, deployment.WorkspaceID)
	for index, id := range ids {
		if index > 0 {
			query += ","
		}
		query += fmt.Sprintf("$%d", index+2)
		args = append(args, id)
	}
	query += `)
ORDER BY id ASC
FOR UPDATE`

	rows, err := tx.Query(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("lock workflow invocation acceptance members: %w", err)
	}
	defer rows.Close()
	values := make(map[string]lockedAcceptanceMember, len(ids))
	for rows.Next() {
		var value lockedAcceptanceMember
		if err := rows.Scan(&value.ID, &value.Type, &value.UserID, &value.WorkspaceRole, &value.Status); err != nil {
			return nil, fmt.Errorf("scan locked workflow invocation acceptance member: %w", err)
		}
		values[value.ID] = value
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate locked workflow invocation acceptance members: %w", err)
	}
	return values, nil
}

func sortedUniqueNonEmpty(values []string) []string {
	seen := make(map[string]struct{}, len(values))
	result := make([]string, 0, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		result = append(result, value)
	}
	sort.Strings(result)
	return result
}

func executionIdentityFromLockedMembers(
	deployment lockedDeployment,
	members map[string]lockedAcceptanceMember,
) (ExecutionIdentitySnapshot, error) {
	owner, found := members[deployment.OwnerMemberID]
	if !found || owner.Type != "human" || owner.Status != "active" || owner.UserID == "" || owner.UserID != deployment.OwnerUserID {
		return ExecutionIdentitySnapshot{}, ErrOwnerUnavailable
	}
	return ExecutionIdentitySnapshot{
		OwnerUserID: deployment.OwnerUserID, OwnerMemberID: deployment.OwnerMemberID,
		PrincipalMemberID: deployment.OwnerMemberID, AuthorMemberID: deployment.OwnerMemberID,
	}, nil
}

// authorizeAgentPrincipal follows the runtime-resource lock order already
// used by workspace-member offboarding: human members (locked by the caller),
// then machine, Agent, and finally the Agent member. Agent deletion locks
// Agent before Agent member and does not lock the machine, so this order is
// compatible with both destructive paths and cannot form member<->Agent or
// member<->machine lock cycles.
func authorizeAgentPrincipal(
	ctx context.Context,
	tx coredb.Tx,
	workspaceID string,
	principal Principal,
	actor Actor,
) error {
	// The machine id is needed before either runtime row can be locked. This
	// snapshot read grants no authority: every relationship is re-read and
	// checked below under locks. A concurrent delete/rebind therefore fails
	// closed rather than using this preliminary value.
	var candidateMachineID string
	if err := tx.QueryRow(ctx, `
SELECT machine_id
FROM agents
WHERE workspace_id = $1 AND id = $2
`, workspaceID, actor.ID).Scan(&candidateMachineID); err != nil {
		if coredb.IsNotFound(err) {
			return ErrPrincipalUnavailable
		}
		return fmt.Errorf("resolve workflow invocation Agent machine: %w", err)
	}
	candidateMachineID = strings.TrimSpace(candidateMachineID)
	if candidateMachineID == "" {
		return ErrPrincipalUnavailable
	}

	var lockedMachineID, machineStatus, createdByUserID string
	if err := tx.QueryRow(ctx, `
SELECT id, status, created_by_user_id
FROM machines
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
	`, workspaceID, candidateMachineID).Scan(&lockedMachineID, &machineStatus, &createdByUserID); err != nil {
		if coredb.IsNotFound(err) {
			return ErrPrincipalUnavailable
		}
		return fmt.Errorf("lock workflow invocation Agent machine: %w", err)
	}
	if lockedMachineID != candidateMachineID || machineStatus != "active" || createdByUserID != principal.UserID {
		return ErrPrincipalUnavailable
	}

	var lockedAgentMachineID, lockedAgentMemberID, agentStatus string
	if err := tx.QueryRow(ctx, `
SELECT machine_id, member_id, status
FROM agents
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
	`, workspaceID, actor.ID).Scan(&lockedAgentMachineID, &lockedAgentMemberID, &agentStatus); err != nil {
		if coredb.IsNotFound(err) {
			return ErrPrincipalUnavailable
		}
		return fmt.Errorf("lock workflow invocation Agent: %w", err)
	}
	if lockedAgentMachineID != candidateMachineID || lockedAgentMemberID != actor.MemberID || agentStatus == "archived" {
		return ErrPrincipalUnavailable
	}

	var agentMember lockedAcceptanceMember
	if err := tx.QueryRow(ctx, `
SELECT id, type, COALESCE(user_id, ''), COALESCE(workspace_role, ''), status
FROM members
WHERE workspace_id = $1 AND id = $2
FOR UPDATE
	`, workspaceID, actor.MemberID).Scan(
		&agentMember.ID, &agentMember.Type, &agentMember.UserID, &agentMember.WorkspaceRole, &agentMember.Status,
	); err != nil {
		if coredb.IsNotFound(err) {
			return ErrPrincipalUnavailable
		}
		return fmt.Errorf("lock workflow invocation Agent member: %w", err)
	}
	if agentMember.ID != actor.MemberID || agentMember.Type != ActorTypeAgent || agentMember.Status != "active" {
		return ErrPrincipalUnavailable
	}
	return nil
}

func canonicalSourceProvenance(deployment lockedDeployment, ref string, occurredAt time.Time) ([]byte, error) {
	type actorSnapshot struct {
		Type     string `json:"type"`
		ID       string `json:"id"`
		MemberID string `json:"memberId"`
	}
	actorValue := func(actor Actor) actorSnapshot {
		return actorSnapshot{Type: actor.Type, ID: actor.ID, MemberID: actor.MemberID}
	}
	var config struct {
		InputPolicy json.RawMessage `json:"inputPolicy"`
		Source      json.RawMessage `json:"source"`
	}
	if err := json.Unmarshal(deployment.ConfigJSON, &config); err != nil || len(config.InputPolicy) == 0 || len(config.Source) == 0 {
		return nil, ErrDeploymentChanged
	}
	payload := struct {
		SourceKind string `json:"sourceKind"`
		Source     struct {
			ID            string          `json:"id"`
			Activation    int64           `json:"activation"`
			OccurrenceRef string          `json:"occurrenceRef"`
			OccurredAt    string          `json:"occurredAt"`
			ActivatedBy   actorSnapshot   `json:"activatedBy"`
			Config        json.RawMessage `json:"config"`
		} `json:"source"`
		Binding struct {
			ID          string          `json:"id"`
			RevisionID  string          `json:"revisionId"`
			Revision    int64           `json:"revision"`
			CreatedBy   actorSnapshot   `json:"createdBy"`
			InputPolicy json.RawMessage `json:"inputPolicy"`
			Config      json.RawMessage `json:"bindingConfig"`
		} `json:"binding"`
		Plan struct {
			ID              string        `json:"id"`
			DisplayName     string        `json:"displayName"`
			VersionID       string        `json:"versionId"`
			CreatedBy       actorSnapshot `json:"createdBy"`
			PublishedBy     actorSnapshot `json:"publishedBy"`
			CompilerVersion string        `json:"compilerVersion"`
			Digest          string        `json:"digest"`
		} `json:"plan"`
	}{SourceKind: deployment.SourceKind}
	payload.Source.ID = deployment.BindingSourceID
	payload.Source.Activation = deployment.BindingActivation
	payload.Source.OccurrenceRef = ref
	payload.Source.OccurredAt = occurredAt.UTC().Truncate(time.Microsecond).Format(time.RFC3339Nano)
	payload.Source.ActivatedBy = actorValue(deployment.SourceActivatedBy)
	payload.Source.Config = append(json.RawMessage(nil), config.Source...)
	payload.Binding.ID = deployment.BindingID
	payload.Binding.RevisionID = deployment.BindingRevisionID
	payload.Binding.Revision = deployment.BindingRevision
	payload.Binding.CreatedBy = actorValue(deployment.BindingCreatedBy)
	payload.Binding.InputPolicy = append(json.RawMessage(nil), config.InputPolicy...)
	payload.Binding.Config = append(json.RawMessage(nil), deployment.ConfigJSON...)
	payload.Plan.ID = deployment.PlanID
	payload.Plan.DisplayName = strings.TrimSpace(deployment.WorkflowName)
	payload.Plan.VersionID = deployment.PlanVersionID
	payload.Plan.CreatedBy = actorValue(deployment.PlanCreatedBy)
	payload.Plan.PublishedBy = actorValue(deployment.PlanPublishedBy)
	payload.Plan.CompilerVersion = deployment.PlanCompilerVersion
	payload.Plan.Digest = deployment.PlanVersionDigest
	encoded, err := json.Marshal(payload)
	if err != nil {
		return nil, fmt.Errorf("encode workflow source provenance: %w", err)
	}
	return domain.DeterministicSourceProvenanceJSONObject("source provenance", encoded, domain.MaxSourceProvenanceBytes)
}

func acceptRequestDigest(params AcceptParams) string {
	return requestDigest(
		params.WorkspaceID, params.BindingID, params.SourceKind,
		params.RequestInputJSON, Actor{}, params.OccurrenceAt,
	)
}

func manualRequestDigest(workspaceID, planID, requestedPlanVersionID, teamID string, requestInput []byte, actor Actor) string {
	selector := "default"
	if requestedPlanVersionID = strings.TrimSpace(requestedPlanVersionID); requestedPlanVersionID != "" {
		selector = "version:" + requestedPlanVersionID
	}
	if teamID = strings.TrimSpace(teamID); teamID != "" {
		selector += "|team:" + teamID
	}
	return requestDigest(workspaceID, planID, selector, requestInput, actor, nil)
}

func requestDigest(workspaceID string, bindingID string, sourceKind string, requestInput []byte, actor Actor, occurrenceAt *time.Time) string {
	parts := []string{acceptDigestDomain, workspaceID, bindingID, sourceKind, string(requestInput), actor.Type, actor.ID, actor.MemberID}
	if occurrenceAt != nil {
		parts = append(parts, occurrenceAt.UTC().Truncate(time.Microsecond).Format(time.RFC3339Nano))
	}
	hash := sha256.New()
	var length [8]byte
	for _, part := range parts {
		binary.BigEndian.PutUint64(length[:], uint64(len(part)))
		_, _ = hash.Write(length[:])
		_, _ = hash.Write([]byte(part))
	}
	return hex.EncodeToString(hash.Sum(nil))
}

func replayAcceptance(ctx context.Context, queryer coredb.QueryRower, existing Invocation, digest string) (AcceptResult, error) {
	if existing.RequestDigest != digest {
		return AcceptResult{}, IdempotencyConflictError{}
	}
	run, err := loadRunByInvocation(ctx, queryer, existing.WorkspaceID, existing.ID)
	if err != nil {
		return AcceptResult{}, err
	}
	existing.RunID = run.ID
	event, err := loadAcceptedRunEvent(ctx, queryer, existing.WorkspaceID, run.ID)
	if err != nil {
		return AcceptResult{}, err
	}
	return AcceptResult{Invocation: existing, Run: run, Event: event, Replayed: true}, nil
}

// replayActorAcceptanceInTx performs the complete current-read chain for one
// actor/idempotency retry. Each row is acquired separately in the same order
// used by acceptance inserts: Invocation -> Run -> accepted Event.
func replayActorAcceptanceInTx(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	actor Actor,
	idempotencyKey string,
	digest string,
) (AcceptResult, bool, error) {
	existing, found, err := lockInvocationByActorIdempotency(ctx, queryer, workspaceID, actor, idempotencyKey)
	if err != nil || !found {
		return AcceptResult{}, found, err
	}
	result, err := replayLockedAcceptance(ctx, queryer, existing, digest)
	return result, true, err
}

// replayOccurrenceAcceptanceInTx is the schedule-occurrence equivalent of
// replayActorAcceptanceInTx. It preserves the occurrence-specific conflict
// type while using the same three current point reads.
func replayOccurrenceAcceptanceInTx(
	ctx context.Context,
	queryer coredb.QueryRower,
	workspaceID string,
	sourceID string,
	occurrenceRef string,
	digest string,
) (AcceptResult, bool, error) {
	existing, found, err := lockInvocationByOccurrence(ctx, queryer, workspaceID, sourceID, occurrenceRef)
	if err != nil || !found {
		return AcceptResult{}, found, err
	}
	if existing.RequestDigest != digest {
		return AcceptResult{}, true, OccurrenceConflictError{}
	}
	result, err := replayLockedAcceptance(ctx, queryer, existing, digest)
	return result, true, err
}

func replayLockedAcceptance(ctx context.Context, queryer coredb.QueryRower, existing Invocation, digest string) (AcceptResult, error) {
	if existing.RequestDigest != digest {
		return AcceptResult{}, IdempotencyConflictError{}
	}
	run, err := lockRunByInvocation(ctx, queryer, existing.WorkspaceID, existing.ID)
	if err != nil {
		return AcceptResult{}, err
	}
	existing.RunID = run.ID
	event, err := lockAcceptedRunEvent(ctx, queryer, existing.WorkspaceID, run.ID)
	if err != nil {
		return AcceptResult{}, err
	}
	return AcceptResult{Invocation: existing, Run: run, Event: event, Replayed: true}, nil
}

func insertInvocation(ctx context.Context, execer coredb.QueryExecer, value Invocation) error {
	var memberID, actorType, actorID any
	if value.InvokedBy.Type != "" {
		memberID, actorType, actorID = value.InvokedBy.MemberID, value.InvokedBy.Type, value.InvokedBy.ID
	}
	var bindingID, bindingRevisionID, bindingRevision, bindingSourceID, bindingActivation any
	if value.BindingID != "" {
		bindingID = value.BindingID
	}
	if value.BindingRevisionID != "" {
		bindingRevisionID = value.BindingRevisionID
	}
	if value.BindingRevision > 0 {
		bindingRevision = value.BindingRevision
	}
	if value.BindingSourceID != "" {
		bindingSourceID = value.BindingSourceID
	}
	if value.BindingActivation > 0 {
		bindingActivation = value.BindingActivation
	}
	_, err := execer.Exec(ctx, `
INSERT INTO workflow_invocations (
    id, workspace_id, binding_id, binding_revision_id, binding_revision_number,
    binding_source_id, binding_activation_number,
    plan_id, plan_version_id, workflow_name_snapshot,
    source_kind, source_occurrence_ref, source_occurred_at,
    input_json, input_digest, source_provenance_json,
    owner_user_id_snapshot, owner_member_id_snapshot,
    principal_member_id_snapshot, author_member_id_snapshot,
    invoked_by_member_id, invoked_by_actor_type, invoked_by_actor_id,
    idempotency_key, request_digest, accepted_at
) VALUES (
    $1, $2, $3, $4, $5,
    $6, $7,
    $8, $9, $10,
    $11, $12, $13,
    $14, $15, $16,
    $17, $18,
    $19, $20,
    $21, $22, $23,
    $24, $25, $26
)
`, value.ID, value.WorkspaceID, bindingID, bindingRevisionID, bindingRevision,
		bindingSourceID, bindingActivation,
		value.PlanID, value.PlanVersionID, value.WorkflowNameSnapshot,
		value.SourceKind, value.SourceOccurrenceRef, value.SourceOccurredAt,
		value.InputJSON, value.InputDigest, value.SourceProvenanceJSON,
		value.Identity.OwnerUserID, value.Identity.OwnerMemberID,
		value.Identity.PrincipalMemberID, value.Identity.AuthorMemberID,
		memberID, actorType, actorID,
		value.IdempotencyKey, value.RequestDigest, value.AcceptedAt)
	if err != nil {
		if coredb.IsUniqueViolation(err) {
			constraint, _ := coredb.UniqueViolationConstraint(err)
			return invocationInsertConflictError{Constraint: constraint}
		}
		if coredb.IsForeignKeyViolation(err) {
			return ErrDeploymentChanged
		}
		return fmt.Errorf("insert workflow invocation: %w", err)
	}
	return nil
}

// invocationInsertConflictError leaves manual race resolution to
// AcceptManual after its transaction has rolled back. Querying the winner
// from inside a transaction whose insert failed is not portable across SQL
// engines and would also keep unrelated aggregate locks held while waiting.
type invocationInsertConflictError struct{ Constraint string }

func (err invocationInsertConflictError) Error() string {
	if strings.TrimSpace(err.Constraint) == "" {
		return "workflow invocation already exists"
	}
	return "workflow invocation already exists for " + err.Constraint
}

func (invocationInsertConflictError) Unwrap() error { return ErrConflict }

// resolveRunTeamID and insertRun live in run_insert.go so this file stays under
// the code-shape soft line ceiling.

func acceptedRunEventPayload(invocation Invocation, run Run) ([]byte, error) {
	payload, err := json.Marshal(struct {
		InvocationID      string `json:"invocationId"`
		BindingRevisionID string `json:"bindingRevisionId,omitempty"`
		PlanVersionID     string `json:"planVersionId"`
		Status            string `json:"status"`
		InputDigest       string `json:"inputDigest"`
	}{
		InvocationID: invocation.ID, BindingRevisionID: invocation.BindingRevisionID,
		PlanVersionID: invocation.PlanVersionID, Status: run.Status, InputDigest: invocation.InputDigest,
	})
	if err != nil {
		return nil, fmt.Errorf("encode accepted workflow Run event: %w", err)
	}
	canonical, err := domain.DeterministicJSONObject("accepted workflow Run event", payload, domain.MaxSourceProvenanceBytes)
	if err != nil {
		return nil, fmt.Errorf("canonicalize accepted workflow Run event: %w", err)
	}
	return canonical, nil
}

func insertAcceptance(ctx context.Context, execer coredb.QueryExecer, invocation Invocation, run Run, event RunEvent) error {
	if err := insertInvocation(ctx, execer, invocation); err != nil {
		return err
	}
	if err := insertRun(ctx, execer, run); err != nil {
		return err
	}
	return insertRunEvent(ctx, execer, event)
}

func insertRunEvent(ctx context.Context, execer coredb.QueryExecer, value RunEvent) error {
	_, err := execer.Exec(ctx, `
INSERT INTO workflow_run_instance_events (
    id, workspace_id, run_id, seq, event_type, payload_json, created_at
) VALUES ($1, $2, $3, $4, $5, $6, $7)
`, value.ID, value.WorkspaceID, value.RunID, value.Seq, value.Type, value.PayloadJSON, value.CreatedAt)
	if err != nil {
		if coredb.IsUniqueViolation(err) {
			return storeerr.Conflict("workflow Run event already exists")
		}
		if coredb.IsForeignKeyViolation(err) {
			return ErrDeploymentChanged
		}
		return fmt.Errorf("insert accepted workflow Run event: %w", err)
	}
	return nil
}