// Package orchestrator owns the durable orchestration boundary around the pure
// Workflow Run graph kernel.  The kernel never performs network I/O; it emits
// effect intents which this package dispatches through idempotent adapters.
package orchestrator

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	"github.com/tidbcloud/AgentTeam/internal/core/domain"
	"github.com/tidbcloud/AgentTeam/internal/workflow/plan"
	runkernel "github.com/tidbcloud/AgentTeam/internal/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runartifact"
)

// MessageRequest is the immutable, fully rendered input to one
// send_message.sent or send_message.callback effect.  Runtime identity is
// copied from the Invocation snapshot, never resolved from the mutable
// Binding while a Run is executing.
type MessageRequest struct {
	WorkspaceID   string `json:"workspaceId"`
	PlanID        string `json:"planId"`
	PlanVersionID string `json:"planVersionId"`
	RunID         string `json:"runId"`
	ActivationID  string `json:"activationId"`
	EffectID      string `json:"effectId"`
	NodeID        string `json:"nodeId"`
	SourceKind    string `json:"sourceKind"`
	WorkflowName  string `json:"workflowName,omitempty"`

	OwnerUserID    string `json:"ownerUserId"`
	AuthorMemberID string `json:"authorMemberId"`

	// TeamID is the server-authorized Team frozen on the Run at Invocation
	// acceptance (Team-linked binding or dynamic-explicit teamId). It is empty
	// only for a dynamic Run whose caller omitted teamId. It is trusted transport
	// context, never derived from visible message text or a callback request.
	TeamID string `json:"teamId,omitempty"`

	TargetRef       string                    `json:"target,omitempty"`
	Scope           *domain.MessageScopeValue `json:"scope,omitempty"`
	Content         string                    `json:"content,omitempty"`
	ContentRef      *runartifact.Ref          `json:"contentRef,omitempty"`
	ClientMessageID string                    `json:"clientMessageId"`
	Callback        bool                      `json:"callback"`
	// Presentation is an optional server-owned envelope used by platform
	// generated messages such as automatic recovery nudges. It is validated
	// against the closed shared schema and never changes canonical Content.
	Presentation *domain.MessagePresentation `json:"presentation,omitempty"`
	// ResponderAgentRef is the immutable AgentRef rendered by the reducer. It may be
	// an Agent ID, member ID, or workspace-unique Agent name. The ordinary
	// message transaction resolves it against the destination visibility
	// boundary, materializes the resolved visible mention, and freezes callback
	// authority separately from the ordinary wake-recipient set.
	ResponderAgentRef string `json:"responderAgentRef,omitempty"`
	// ResponderCandidates is the frozen ordered candidate pool of
	// responder.selectOne, carried losslessly from the reducer Intent to the
	// capacity scheduler. It is non-empty exactly when the callback declared a
	// selectOne pool instead of a fixed ResponderAgentRef.
	ResponderCandidates []string `json:"responderCandidates,omitempty"`
	// SelectOneStrategy is the additive responder.selectOne strategy carried from
	// the Plan artifact ("" or "ordered" = legacy strict priority, "random" =
	// injectable unbiased randomization over currently available deduplicated
	// candidates). An absent field alone is the legacy ordered default; a present
	// field must resolve to exactly "ordered" or "random".
	SelectOneStrategy plan.SelectOneStrategy `json:"selectOneStrategy,omitempty"`
	// selectOneStrategyPresent records whether the selectOneStrategy field was
	// present on the wire (persisted RequestJSON/recovery). It lets Validate
	// reject an explicit empty/null/whitespace-padded/case-variant value that a
	// tampered request smuggles, while an absent field stays the legacy ordered
	// default. It is never serialized.
	selectOneStrategyPresent bool
	StartThread              bool                        `json:"startThread,omitempty"`
	WorkScopeMode            runkernel.WorkScopeMode     `json:"workScopeMode,omitempty"`
	WorkScope                *domain.MessageScopeValue   `json:"workScope,omitempty"`
	RelatedScopes            []domain.MessageScopeValue  `json:"relatedScopes,omitempty"`
	TaskWatchHandoff         *runkernel.TaskWatchHandoff `json:"taskWatchHandoff,omitempty"`
	// ExpectedCallbackResponderAgentID is a runtime-only trusted fence set by the
	// dispatcher on a copy of the frozen request immediately before the ordinary
	// message write. It is never serialized into the durable effect request and
	// is not author-facing; the ordinary message transaction compares it against
	// the resolved responder and rolls back on mismatch so no
	// message/outbox/fanout/wake side effect can be attributed to a drifted Agent.
	ExpectedCallbackResponderAgentID string `json:"-"`
	// SkipWorkflowRunMessageBinding is a runtime-only flag set by the automatic
	// recovery runner on the recovery nudge. The recovery nudge is a separate
	// wake message for an already-awaiting callback; it must not bind to the
	// awaiting effect (whose frozen request carries the original clientMessageId),
	// so the fanout writer omits the workflow_run contribution source while
	// preserving the workflow context metadata and the trusted responder fence.
	SkipWorkflowRunMessageBinding bool `json:"-"`
}

// UnmarshalJSON decodes the durable MessageRequest and records whether the
// selectOneStrategy field was present on the wire, so Validate can reject an
// explicit empty/null/whitespace-padded/case-variant value a tampered persisted
// RequestJSON smuggles while an absent field stays the legacy ordered default.
func (request *MessageRequest) UnmarshalJSON(raw []byte) error {
	type wireMessageRequest MessageRequest
	var decoded wireMessageRequest
	if err := json.Unmarshal(raw, &decoded); err != nil {
		return err
	}
	*request = MessageRequest(decoded)
	var fields map[string]json.RawMessage
	if err := json.Unmarshal(raw, &fields); err != nil {
		return err
	}
	_, request.selectOneStrategyPresent = fields["selectOneStrategy"]
	return nil
}

func (request MessageRequest) Validate() error {
	required := map[string]string{
		"workspaceId":     request.WorkspaceID,
		"planId":          request.PlanID,
		"planVersionId":   request.PlanVersionID,
		"runId":           request.RunID,
		"activationId":    request.ActivationID,
		"effectId":        request.EffectID,
		"nodeId":          request.NodeID,
		"sourceKind":      request.SourceKind,
		"ownerUserId":     request.OwnerUserID,
		"authorMemberId":  request.AuthorMemberID,
		"clientMessageId": request.ClientMessageID,
	}
	for name, value := range required {
		if strings.TrimSpace(value) == "" {
			return fmt.Errorf("workflow Run message %s is required", name)
		}
	}
	if err := request.validateContent(); err != nil {
		return err
	}
	if request.Presentation != nil {
		if !request.Callback {
			return fmt.Errorf("non-callback workflow Run message cannot carry presentation")
		}
		if err := request.Presentation.Validate(); err != nil {
			return fmt.Errorf("workflow Run message presentation is invalid: %w", err)
		}
	}
	targetRef := strings.TrimSpace(request.TargetRef)
	if (targetRef == "") == (request.Scope == nil) {
		return fmt.Errorf("workflow Run message requires exactly one target or scope")
	}
	if targetRef != "" {
		target, err := domain.ParseMessageTargetReference(targetRef)
		if err != nil {
			return fmt.Errorf("workflow Run message target is invalid: %w", err)
		}
		if target.Kind == domain.MessageTargetKindWorkflow {
			return fmt.Errorf("workflow Run message target cannot be a Workflow Builder edit scope")
		}
	} else if err := domain.ValidateMessageScopeValue(*request.Scope); err != nil {
		return fmt.Errorf("workflow Run message scope is invalid: %w", err)
	}
	responderAgentRef := strings.TrimSpace(request.ResponderAgentRef)
	if request.ResponderAgentRef != responderAgentRef {
		return fmt.Errorf("workflow Run message responderAgentRef must be canonical")
	}
	if responderAgentRef != "" && !request.Callback {
		return fmt.Errorf("non-callback workflow Run message cannot carry responderAgentRef")
	}
	if request.selectOneStrategyPresent {
		// The strategy field was present on the wire (persisted RequestJSON /
		// recovery): it must resolve to exactly "ordered" or "random". An explicit
		// empty, null, whitespace-padded, case-variant, or unknown value fails
		// closed — an absent field alone is the legacy ordered default.
		if _, err := plan.ParseSelectOneStrategyPresent(string(request.SelectOneStrategy)); err != nil {
			return err
		}
	} else if request.SelectOneStrategy != "" {
		// Programmatic construction path: validate a non-empty strategy strictly.
		if _, err := plan.ParseSelectOneStrategy(string(request.SelectOneStrategy)); err != nil {
			return err
		}
	}
	if len(request.ResponderCandidates) > 0 {
		if !request.Callback {
			return fmt.Errorf("non-callback workflow Run message cannot carry responder candidates")
		}
		if responderAgentRef != "" {
			return fmt.Errorf("workflow Run message cannot carry both a fixed responder and responder candidates")
		}
		if err := validateMessageResponderCandidates(request.ResponderCandidates); err != nil {
			return err
		}
		if strategy := request.SelectOneStrategy.Normalized(); strategy == plan.SelectOneStrategyRandom {
			if len(normalizeMessageResponderCandidatesForCount(request.ResponderCandidates)) < 2 {
				return fmt.Errorf("workflow Run message selectOne strategy %q requires at least two deduplicated candidates", strategy)
			}
		}
	} else if request.SelectOneStrategy != "" {
		return fmt.Errorf("workflow Run message selectOne strategy is only valid with responder candidates")
	}
	if request.StartThread {
		if request.Callback {
			return fmt.Errorf("workflow Run message startThread does not support callback completion")
		}
		if targetRef != "" {
			target, err := domain.ParseMessageTargetReference(targetRef)
			if err != nil || target.Kind != domain.MessageTargetKindChannel {
				return fmt.Errorf("workflow Run message startThread requires a Channel root destination")
			}
		} else if request.Scope.Kind != domain.MessageScopeKindChannel {
			return fmt.Errorf("workflow Run message startThread requires a Channel root destination")
		}
	}
	if err := validateMessageCollaborationScopes(
		request.Callback,
		request.WorkScopeMode,
		request.WorkScope,
		request.RelatedScopes,
	); err != nil {
		return err
	}
	if err := validateTaskWatchHandoff(
		request.Callback,
		request.ResponderAgentRef,
		request.ResponderCandidates,
		request.Scope,
		request.WorkScopeMode,
		request.WorkScope,
		request.TaskWatchHandoff,
	); err != nil {
		return err
	}
	return nil
}

func (request MessageRequest) validateContent() error {
	if request.ContentRef == nil && strings.TrimSpace(request.Content) == "" {
		return fmt.Errorf("workflow Run message content is required")
	}
	if request.ContentRef != nil {
		if err := request.ContentRef.Validate(); err != nil {
			return fmt.Errorf("workflow Run message content reference is invalid: %w", err)
		}
		if strings.TrimSpace(request.Content) != "" {
			return fmt.Errorf("workflow Run message cannot carry inline content with contentRef")
		}
	}
	return nil
}

func validateTaskWatchHandoff(
	callback bool,
	responderAgentRef string,
	responderCandidates []string,
	deliveryScope *domain.MessageScopeValue,
	workScopeMode runkernel.WorkScopeMode,
	workScope *domain.MessageScopeValue,
	handoff *runkernel.TaskWatchHandoff,
) error {
	if handoff == nil {
		return nil
	}
	if !callback || (strings.TrimSpace(responderAgentRef) == "" && len(responderCandidates) == 0) {
		return fmt.Errorf("workflow Run Task Watch handoff requires a callback responder")
	}
	if workScopeMode != runkernel.WorkScopeModeFixed || workScope == nil || deliveryScope == nil ||
		handoff.TaskScope.Kind != domain.MessageScopeKindTask ||
		*workScope != handoff.TaskScope || *deliveryScope != handoff.TaskScope {
		return fmt.Errorf("workflow Run Task Watch handoff requires one fixed Task delivery/work scope")
	}
	if strings.TrimSpace(handoff.WorkItemID) == "" || handoff.WorkItemID != strings.TrimSpace(handoff.WorkItemID) {
		return fmt.Errorf("workflow Run Task Watch handoff workItemId is required and canonical")
	}
	if handoff.CausalPredecessorAgentIDs == nil {
		return fmt.Errorf("workflow Run Task Watch handoff predecessor IDs must be an array")
	}
	seen := make(map[string]struct{}, len(handoff.CausalPredecessorAgentIDs))
	for _, agentID := range handoff.CausalPredecessorAgentIDs {
		if agentID == "" || agentID != strings.TrimSpace(agentID) {
			return fmt.Errorf("workflow Run Task Watch handoff predecessor Agent IDs must be canonical")
		}
		if _, duplicate := seen[agentID]; duplicate {
			return fmt.Errorf("workflow Run Task Watch handoff predecessor Agent IDs must be unique")
		}
		seen[agentID] = struct{}{}
	}
	if len(seen) > domain.MaxWorkflowTaskWatchHandoffPredecessors {
		return fmt.Errorf(
			"workflow Run Task Watch handoff must not contain more than %d unique predecessor Agent IDs",
			domain.MaxWorkflowTaskWatchHandoffPredecessors,
		)
	}
	return nil
}

func validateMessageCollaborationScopes(
	callback bool,
	mode runkernel.WorkScopeMode,
	workScope *domain.MessageScopeValue,
	relatedScopes []domain.MessageScopeValue,
) error {
	if len(relatedScopes) > plan.MaxSendMessageRelatedScopes {
		return fmt.Errorf(
			"workflow Run relatedScopes must not contain more than %d scopes",
			plan.MaxSendMessageRelatedScopes,
		)
	}
	if !callback {
		if mode != "" || workScope != nil || len(relatedScopes) != 0 {
			return fmt.Errorf("non-callback workflow Run message cannot carry workScope or relatedScopes")
		}
		return nil
	}
	switch mode {
	case "":
		if workScope != nil || len(relatedScopes) != 0 {
			return fmt.Errorf("legacy workflow Run callback cannot carry workScope or relatedScopes")
		}
		return nil
	case runkernel.WorkScopeModeAgent:
		if workScope != nil {
			return fmt.Errorf("agent-resolved workflow Run workScope must start as null")
		}
	case runkernel.WorkScopeModeFixed:
		if workScope == nil {
			return fmt.Errorf("fixed workflow Run workScope is required")
		}
		if err := domain.ValidateMessageScopeValue(*workScope); err != nil {
			return fmt.Errorf("fixed workflow Run workScope is invalid: %w", err)
		}
	default:
		return fmt.Errorf("workflow Run workScope mode %q is invalid", mode)
	}
	seen := make(map[domain.MessageScopeValue]struct{}, len(relatedScopes))
	for index, scope := range relatedScopes {
		if err := domain.ValidateMessageScopeValue(scope); err != nil {
			return fmt.Errorf("workflow Run relatedScopes[%d] is invalid: %w", index, err)
		}
		if _, duplicate := seen[scope]; duplicate {
			return fmt.Errorf("workflow Run relatedScopes[%d] is duplicated", index)
		}
		seen[scope] = struct{}{}
	}
	return nil
}

// validateMessageResponderCandidates validates the frozen ordered candidate
// pool of responder.selectOne on the wire. It mirrors the reducer's normalize
// contract: non-empty, deduplicated, within MaxAllowedResponders, and every
// candidate a normalized non-blank agent string.
func validateMessageResponderCandidates(candidates []string) error {
	if len(candidates) == 0 {
		return fmt.Errorf("workflow Run responder candidates must not be empty")
	}
	if len(candidates) > plan.MaxAllowedResponders {
		return fmt.Errorf(
			"workflow Run responder candidates must not contain more than %d agents",
			plan.MaxAllowedResponders,
		)
	}
	seen := make(map[string]struct{}, len(candidates))
	for index, candidate := range candidates {
		trimmed := strings.TrimSpace(candidate)
		if trimmed == "" {
			return fmt.Errorf("workflow Run responder candidates[%d] is blank", index)
		}
		if trimmed != candidate {
			return fmt.Errorf("workflow Run responder candidates[%d] must be canonical", index)
		}
		if _, duplicate := seen[trimmed]; duplicate {
			return fmt.Errorf("workflow Run responder candidates[%d] is duplicated", index)
		}
		seen[trimmed] = struct{}{}
	}
	return nil
}

// MessageResult is the durable receipt returned after the normal Loop message
// path commits. WakeAgentIDs contains the ordinary routing result and may
// include the callback responder plus other mentions, defaults, or Task
// watchers. CallbackResponderAgentID separately freezes the one Agent with
// callback authority; DeliveryScope is the normalized committed scope.
type MessageResult struct {
	MessageID                string                    `json:"messageId"`
	WakeAgentIDs             []string                  `json:"wakeAgentIds"`
	CallbackResponderAgentID string                    `json:"callbackResponderAgentId,omitempty"`
	DeliveryScope            *domain.MessageScopeValue `json:"deliveryScope,omitempty"`
	ThreadScope              *domain.MessageScopeValue `json:"threadScope,omitempty"`
	IdempotentReplay         bool                      `json:"idempotentReplay"`
	// TeamMemberContext is the server-frozen minimal member context resolved at
	// step delivery for any Run with a frozen Team ID (Team-linked or
	// dynamic-explicit). It is copied into the durable effect receipt so
	// callback/retry/replay reuse the same snapshot.
	TeamMemberContext *CallbackTeamMemberContext `json:"teamMemberContext,omitempty"`
}

func (result MessageResult) Normalize() MessageResult {
	result.MessageID = strings.TrimSpace(result.MessageID)
	result.CallbackResponderAgentID = strings.TrimSpace(result.CallbackResponderAgentID)
	seen := make(map[string]struct{}, len(result.WakeAgentIDs))
	agents := make([]string, 0, len(result.WakeAgentIDs))
	for _, value := range result.WakeAgentIDs {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, ok := seen[value]; ok {
			continue
		}
		seen[value] = struct{}{}
		agents = append(agents, value)
	}
	sort.Strings(agents)
	result.WakeAgentIDs = agents
	if result.DeliveryScope != nil {
		scope := *result.DeliveryScope
		result.DeliveryScope = &scope
	}
	if result.ThreadScope != nil {
		scope := *result.ThreadScope
		result.ThreadScope = &scope
	}
	return result
}

// Validate verifies the trusted delivery receipt returned by MessageWriter.
// In particular, DeliveryScope is not inferred from the authored request: the
// writer must report the normalized scope that the ordinary message path
// actually committed so callback capabilities cannot be bound to an
// unverified or stale destination.
func (result MessageResult) Validate() error {
	if strings.TrimSpace(result.MessageID) == "" {
		return fmt.Errorf("workflow Run message result messageId is required")
	}
	if result.DeliveryScope == nil {
		return fmt.Errorf("workflow Run message result deliveryScope is required")
	}
	if err := domain.ValidateMessageScopeValue(*result.DeliveryScope); err != nil {
		return fmt.Errorf("workflow Run message result deliveryScope is invalid: %w", err)
	}
	if result.ThreadScope != nil {
		if err := domain.ValidateMessageScopeValue(*result.ThreadScope); err != nil || result.ThreadScope.Kind != domain.MessageScopeKindThread {
			return fmt.Errorf("workflow Run message result threadScope is invalid")
		}
	}
	return nil
}

// ResolveCallbackResponderAuthority returns the one authorized Agent
// to complete a callback. New receipts freeze that authority separately from
// ordinary routing, which may wake additional mentioned/default/watching
// Agents. A missing field remains readable only for legacy receipts whose wake
// set itself proves one identity.
func ResolveCallbackResponderAuthority(
	callback bool,
	callbackResponderAgentID string,
	wakeAgentIDs []string,
) (string, error) {
	callbackResponderAgentID = strings.TrimSpace(callbackResponderAgentID)
	agents := (MessageResult{WakeAgentIDs: wakeAgentIDs}).Normalize().WakeAgentIDs
	if !callback {
		if callbackResponderAgentID != "" {
			return "", fmt.Errorf("non-callback workflow Run message cannot carry callback responder authority")
		}
		return "", nil
	}
	if callbackResponderAgentID == "" {
		if len(agents) != 1 {
			return "", fmt.Errorf("legacy workflow Run callback receipt must wake exactly one agent")
		}
		return agents[0], nil
	}
	for _, agentID := range agents {
		if agentID == callbackResponderAgentID {
			return callbackResponderAgentID, nil
		}
	}
	return "", fmt.Errorf("workflow Run callback responder authority must be present in the wake set")
}

// MessageWriter dispatches one durable message effect. Implementations must
// pass ClientMessageID to the downstream message store so replay after a
// process crash returns the same committed message instead of creating a
// duplicate. A successful result with a MessageID must come from the native
// message transaction that atomically commits the transcript message and its
// Workflow Run source binding. The dispatcher intentionally cannot add that
// binding after WriteRunMessage returns because doing so would expose an
// unbound visible message across the transaction boundary.
type MessageWriter interface {
	WriteRunMessage(context.Context, MessageRequest) (MessageResult, error)
}

// MessageReceiptRecoverer reconstructs the immutable delivery receipt for a
// message whose native transaction already committed and bound MessageID to
// the durable Workflow Run Effect. Recovery must read only durable receipt
// facts identified by the frozen request and bound message ID; in particular,
// it must not resolve the current owner, destination, membership, or routing
// topology and it must never attempt a second message write.
//
// MessageWriter implementations may omit this interface. A dispatcher that
// reclaims a bound Effect without a recoverer keeps the Effect pending rather
// than replaying WriteRunMessage or manufacturing a message-less terminal
// result.
type MessageReceiptRecoverer interface {
	RecoverRunMessageReceipt(context.Context, MessageRequest, string) (MessageResult, error)
}

// normalizeMessageResponderCandidatesForCount returns the deduplicated candidate
// count for the selectOne random lower-bound check (duplicate candidates are
// collapsed before deciding whether a random pool has at least two distinct
// options).
func normalizeMessageResponderCandidatesForCount(candidates []string) []string {
	seen := make(map[string]struct{}, len(candidates))
	deduped := make([]string, 0, len(candidates))
	for _, candidate := range candidates {
		trimmed := strings.TrimSpace(candidate)
		if trimmed == "" {
			continue
		}
		if _, exists := seen[trimmed]; exists {
			continue
		}
		seen[trimmed] = struct{}{}
		deduped = append(deduped, trimmed)
	}
	return deduped
}