package orchestrator

import (
	"bytes"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"strings"
	"time"

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

const (
	EffectKindMessage  = "send_message"
	EffectKindWait     = "wait"
	EffectKindApproval = "request_approval"
)

// EffectRequest is the immutable durable command emitted by one reducer
// activation. Intent retains the compiler-derived retry/deadline semantics;
// Message contains the frozen Invocation identity needed by the normal Loop
// message path. A replay must execute these exact bytes rather than re-render
// the mutable Plan source or Binding.
type EffectRequest struct {
	Intent  runkernel.Intent `json:"intent"`
	Message *MessageRequest  `json:"message,omitempty"`
}

func (request EffectRequest) Validate() error {
	if strings.TrimSpace(request.Intent.ActivationID) == "" || strings.TrimSpace(request.Intent.NodeID) == "" {
		return fmt.Errorf("workflow Run effect requires activationId and nodeId")
	}
	if request.Intent.TextRef != nil {
		if err := request.Intent.TextRef.Validate(); err != nil {
			return fmt.Errorf("workflow Run effect content reference is invalid: %w", err)
		}
		if strings.TrimSpace(request.Intent.Text) != "" {
			return fmt.Errorf("workflow Run effect cannot carry inline text with content reference")
		}
	}
	if (request.Intent.WorkItemsAllowedResponders != nil) != (request.Intent.WorkItemsDependencyValueContract != nil) {
		return fmt.Errorf("workflow Run WorkItems intent requires both responder and dependency value contracts")
	}
	if request.Intent.Kind != runkernel.IntentApproval && request.Intent.FeedbackScope != nil {
		return fmt.Errorf("workflow Run feedbackScope is only valid on approval effects")
	}
	switch request.Intent.Kind {
	case runkernel.IntentSendMessage:
		if request.Message == nil {
			return fmt.Errorf("workflow Run message effect requires message request")
		}
		if err := request.Message.Validate(); err != nil {
			return err
		}
		if request.Message.ActivationID != request.Intent.ActivationID ||
			request.Message.NodeID != request.Intent.NodeID ||
			request.Message.Callback != request.Intent.Callback ||
			strings.TrimSpace(request.Message.ResponderAgentRef) != strings.TrimSpace(request.Intent.Responder) ||
			!sameResponderCandidates(request.Message.ResponderCandidates, request.Intent.ResponderCandidates) ||
			request.Message.StartThread != request.Intent.StartThread ||
			request.Message.WorkScopeMode != request.Intent.WorkScopeMode ||
			!sameMessageScopeValue(request.Message.WorkScope, request.Intent.WorkScope) ||
			!sameMessageScopeValues(request.Message.RelatedScopes, request.Intent.RelatedScopes) ||
			!sameTaskWatchHandoff(request.Message.TaskWatchHandoff, request.Intent.TaskWatchHandoff) ||
			!sameMessageContent(request.Message, request.Intent) ||
			!sameMessageScopeDestination(request.Message, request.Intent) {
			return fmt.Errorf("workflow Run message request does not match reducer intent")
		}
	case runkernel.IntentWait:
		if request.Message != nil || request.Intent.WakeAt == nil {
			return fmt.Errorf("workflow Run timer effect requires wakeAt and no message request")
		}
	case runkernel.IntentApproval:
		if request.Message != nil || strings.TrimSpace(request.Intent.Assignee) == "" || strings.TrimSpace(request.Intent.Summary) == "" {
			return fmt.Errorf("workflow Run approval effect requires assignee and summary")
		}
		if request.Intent.FeedbackScope != nil {
			if err := domain.ValidateMessageScopeValue(*request.Intent.FeedbackScope); err != nil {
				return fmt.Errorf("workflow Run approval feedbackScope is invalid: %w", err)
			}
		}
	default:
		return fmt.Errorf("unsupported workflow Run effect kind %q", request.Intent.Kind)
	}
	// Only wait may carry timing. send_message, request_approval, and join carry
	// no authored or synthetic runtime deadline and no wakeAt; their callbacks
	// park until the Agent reply or an infrastructure-level delivery failure
	// resolves them. A non-nil pointer is rejected even when it points at the
	// zero time: a caller must not smuggle a deadline or wakeAt in any form.
	if request.Intent.Kind != runkernel.IntentWait {
		if request.Intent.Deadline != nil {
			return fmt.Errorf("workflow Run %s effect cannot carry a deadline", request.Intent.Kind)
		}
		if request.Intent.WakeAt != nil {
			return fmt.Errorf("workflow Run %s effect cannot carry wakeAt", request.Intent.Kind)
		}
	}
	if request.Intent.Kind == runkernel.IntentWait &&
		(request.Intent.Deadline == nil || request.Intent.Deadline.IsZero()) {
		return fmt.Errorf("workflow Run timer effect requires a deadline")
	}
	if request.Intent.Retry.MaxAttempts < 1 || request.Intent.Retry.BackoffNanos < 1 {
		return fmt.Errorf("workflow Run effect retry policy is invalid")
	}
	return nil
}

// EffectResult is the complete durable receipt consumed by the reducer. The
// port is always one of the compiled node's output ports. Delivery metadata is
// preserved when an awaiting callback/approval later commits its decision so
// authorization and idempotency remain auditable from one row.
type EffectResult struct {
	Port                     string                    `json:"port,omitempty"`
	Output                   runkernel.Values          `json:"output,omitempty"`
	MessageID                string                    `json:"messageId,omitempty"`
	WakeAgentIDs             []string                  `json:"wakeAgentIds,omitempty"`
	CallbackResponderAgentID string                    `json:"callbackResponderAgentId,omitempty"`
	DeliveryScope            *domain.MessageScopeValue `json:"deliveryScope,omitempty"`
	ThreadScope              *domain.MessageScopeValue `json:"threadScope,omitempty"`
	WorkScope                *domain.MessageScopeValue `json:"workScope,omitempty"`
	NotificationID           string                    `json:"notificationId,omitempty"`
	NotificationRevision     int64                     `json:"notificationRevision,omitempty"`
	AssigneeMemberID         string                    `json:"assigneeMemberId,omitempty"`
	DecidedByMemberID        string                    `json:"decidedByMemberId,omitempty"`
	FailureCode              string                    `json:"failureCode,omitempty"`
	FailureMessage           string                    `json:"failureMessage,omitempty"`
	FailureStage             string                    `json:"failureStage,omitempty"`
	FailurePath              string                    `json:"failurePath,omitempty"`
	FailureProblem           string                    `json:"failureProblem,omitempty"`
	FailureHint              string                    `json:"failureHint,omitempty"`
	FailureRetryable         bool                      `json:"failureRetryable,omitempty"`
	FailureDiagnostics       []PersistedDiagnostic     `json:"failureDiagnostics,omitempty"`
	FailureWorkItem          *PersistedWorkItem        `json:"failureWorkItem,omitempty"`
	// FailureTruncation records the deterministic diagnostic budget cuts already
	// applied to FailureDiagnostics by an earlier persistence boundary. It is
	// carried on the durable EffectResult so recovery/replay preserves the exact
	// omittedCount/truncatedFields/serializedBytes instead of re-deriving them
	// from an already-bounded list (which would recompute the earlier loss away).
	FailureTruncation               *PersistedTruncation `json:"failureTruncation,omitempty"`
	IdempotencyKey                  string               `json:"idempotencyKey,omitempty"`
	ApprovalComment                 string               `json:"approvalComment,omitempty"`
	ApprovalFeedbackMessageID       string               `json:"approvalFeedbackMessageId,omitempty"`
	ApprovalFeedbackClientMessageID string               `json:"approvalFeedbackClientMessageId,omitempty"`
	CallbackAction                  string               `json:"callbackAction,omitempty"`
	CallbackActionIdempotencyKey    string               `json:"callbackActionIdempotencyKey,omitempty"`
	CompletedAt                     *time.Time           `json:"completedAt,omitempty"`
	// TeamMemberContext is the server-frozen minimal member context for the
	// callback responder, persisted on the durable effect receipt so retries,
	// callbacks, and replays reuse the same snapshot.
	TeamMemberContext *CallbackTeamMemberContext `json:"teamMemberContext,omitempty"`
}

func (result EffectResult) Normalize() EffectResult {
	// Identity fields (port, failure code/stage/path) are validated byte-for-byte
	// and fail closed to empty when malformed; they are never TrimSpace'd or
	// UTF-8-sanitized into a different valid identity. Free-text fields
	// (failure message/problem/hint) may trim.
	if port, ok := runkernel.PublicPlanPort(result.Port); ok {
		result.Port = port
	} else {
		result.Port = ""
	}
	result.MessageID = strings.TrimSpace(result.MessageID)
	result.CallbackResponderAgentID = strings.TrimSpace(result.CallbackResponderAgentID)
	result.NotificationID = strings.TrimSpace(result.NotificationID)
	result.AssigneeMemberID = strings.TrimSpace(result.AssigneeMemberID)
	result.DecidedByMemberID = strings.TrimSpace(result.DecidedByMemberID)
	if code, ok := runkernel.PublicFailureCode(result.FailureCode); ok {
		result.FailureCode = code
	} else {
		result.FailureCode = ""
	}
	result.FailureMessage = strings.TrimSpace(result.FailureMessage)
	if stage, ok := runkernel.PublicFailureStage(result.FailureStage); ok {
		result.FailureStage = stage
	} else {
		result.FailureStage = ""
	}
	if path, ok := runkernel.PublicFailurePath(result.FailurePath); ok {
		result.FailurePath = path
	} else {
		result.FailurePath = ""
	}
	result.FailureProblem = strings.TrimSpace(result.FailureProblem)
	result.FailureHint = strings.TrimSpace(result.FailureHint)
	result.IdempotencyKey = strings.TrimSpace(result.IdempotencyKey)
	result.ApprovalComment = strings.TrimSpace(result.ApprovalComment)
	result.ApprovalFeedbackMessageID = strings.TrimSpace(result.ApprovalFeedbackMessageID)
	result.ApprovalFeedbackClientMessageID = strings.TrimSpace(result.ApprovalFeedbackClientMessageID)
	if result.CallbackAction != CallbackActionPause && result.CallbackAction != CallbackActionPark {
		result.CallbackAction = ""
		result.CallbackActionIdempotencyKey = ""
	} else {
		result.CallbackActionIdempotencyKey = strings.TrimSpace(result.CallbackActionIdempotencyKey)
	}
	result.WakeAgentIDs = MessageResult{WakeAgentIDs: result.WakeAgentIDs}.Normalize().WakeAgentIDs
	if result.DeliveryScope != nil {
		scope := *result.DeliveryScope
		result.DeliveryScope = &scope
	}
	if result.ThreadScope != nil {
		scope := *result.ThreadScope
		result.ThreadScope = &scope
	}
	if result.WorkScope != nil {
		scope := *result.WorkScope
		result.WorkScope = &scope
	}
	if result.FailureTruncation != nil {
		truncation := *result.FailureTruncation
		truncation.TruncatedFields = append([]string(nil), result.FailureTruncation.TruncatedFields...)
		result.FailureTruncation = &truncation
	}
	if result.CompletedAt != nil {
		normalized := result.CompletedAt.UTC().Truncate(time.Microsecond)
		result.CompletedAt = &normalized
	}
	return result
}

func samePersistedTruncation(left, right *PersistedTruncation) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	if left.DiagnosticsTruncated != right.DiagnosticsTruncated ||
		left.OmittedCount != right.OmittedCount ||
		left.DroppedCount != right.DroppedCount ||
		left.SerializedBytes != right.SerializedBytes ||
		left.MinimalFallback != right.MinimalFallback ||
		len(left.TruncatedFields) != len(right.TruncatedFields) {
		return false
	}
	for index := range left.TruncatedFields {
		if left.TruncatedFields[index] != right.TruncatedFields[index] {
			return false
		}
	}
	return true
}

func cloneMessageScopeValue(scope *domain.MessageScopeValue) *domain.MessageScopeValue {
	if scope == nil {
		return nil
	}
	cloned := *scope
	return &cloned
}

func sameMessageScopeValue(left, right *domain.MessageScopeValue) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	return *left == *right
}

func sameResponderCandidates(left, right []string) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if left[index] != right[index] {
			return false
		}
	}
	return true
}

func sameMessageScopeValues(left, right []domain.MessageScopeValue) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if left[index] != right[index] {
			return false
		}
	}
	return true
}

func samePersistedWorkItem(left, right *PersistedWorkItem) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	if left.Key != right.Key || left.DependencyPath != right.DependencyPath {
		return false
	}
	if left.Index == nil || right.Index == nil {
		return left.Index == nil && right.Index == nil
	}
	return *left.Index == *right.Index
}

func samePersistedDiagnostics(left, right []PersistedDiagnostic) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		l, r := left[index], right[index]
		if l.Code != r.Code || l.Path != r.Path || l.Field != r.Field || l.Message != r.Message {
			return false
		}
		if l.Index == nil || r.Index == nil {
			if l.Index != r.Index {
				return false
			}
		} else if *l.Index != *r.Index {
			return false
		}
	}
	return true
}

func sameTaskWatchHandoff(left, right *runkernel.TaskWatchHandoff) bool {
	if left == nil || right == nil {
		return left == nil && right == nil
	}
	if left.TaskScope != right.TaskScope || left.WorkItemID != right.WorkItemID ||
		len(left.CausalPredecessorAgentIDs) != len(right.CausalPredecessorAgentIDs) {
		return false
	}
	for index := range left.CausalPredecessorAgentIDs {
		if left.CausalPredecessorAgentIDs[index] != right.CausalPredecessorAgentIDs[index] {
			return false
		}
	}
	return true
}

func sameMessageContent(message *MessageRequest, intent runkernel.Intent) bool {
	if message == nil {
		return false
	}
	if message.ContentRef != nil || intent.TextRef != nil {
		return message.ContentRef != nil && intent.TextRef != nil && *message.ContentRef == *intent.TextRef && strings.TrimSpace(message.Content) == "" && strings.TrimSpace(intent.Text) == ""
	}
	// Legacy fixtures and persisted v1 requests did not duplicate Content in
	// Intent validation. Keep that omission byte-compatible; the reference
	// path above is the only new case requiring an exact binding comparison.
	return true
}

func sameMessageScopeDestination(message *MessageRequest, intent runkernel.Intent) bool {
	if message.Scope == nil && intent.Scope == nil {
		// Preserve the v1 target protocol: historic durable requests validate
		// their rendered target independently but did not require duplicating it
		// into every hand-authored Intent fixture.
		return true
	}
	if message.Scope == nil || intent.Scope == nil {
		return false
	}
	return strings.TrimSpace(message.TargetRef) == "" &&
		strings.TrimSpace(intent.Target) == "" &&
		*message.Scope == *intent.Scope
}

func (result EffectResult) Completion(activationID string) (runkernel.Completion, error) {
	result = result.Normalize()
	activationID = strings.TrimSpace(activationID)
	if activationID == "" || result.Port == "" {
		return runkernel.Completion{}, fmt.Errorf("workflow Run effect result requires activationId and port")
	}
	output := result.Output
	if output == nil {
		output = runkernel.Values{}
	}
	var failure *runkernel.Failure
	if result.FailureCode != "" || result.FailureMessage != "" {
		failure = &runkernel.Failure{
			Code:        result.FailureCode,
			Message:     result.FailureMessage,
			Stage:       result.FailureStage,
			Path:        result.FailurePath,
			Problem:     result.FailureProblem,
			Hint:        result.FailureHint,
			Retryable:   result.FailureRetryable,
			Diagnostics: kernelFailureDiagnostics(result.FailureDiagnostics),
			Truncation:  kernelFailureTruncation(result.FailureTruncation),
		}
		if result.FailureWorkItem != nil {
			failure.WorkItem = &runkernel.FailureWorkItem{
				Key:            result.FailureWorkItem.Key,
				Index:          result.FailureWorkItem.Index,
				DependencyPath: result.FailureWorkItem.DependencyPath,
			}
		}
	}
	return runkernel.Completion{
		ActivationID: activationID, Port: result.Port, Output: output, Failure: failure,
		CallbackResponderAgentID: result.CallbackResponderAgentID,
	}, nil
}

func EncodeEffectRequest(request EffectRequest) (json.RawMessage, error) {
	if err := request.Validate(); err != nil {
		return nil, err
	}
	payload, err := json.Marshal(request)
	if err != nil {
		return nil, fmt.Errorf("encode workflow Run effect request: %w", err)
	}
	return payload, nil
}

func DecodeEffectRequest(payload []byte) (EffectRequest, error) {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	decoder.UseNumber()
	var request EffectRequest
	if err := decoder.Decode(&request); err != nil {
		return EffectRequest{}, fmt.Errorf("decode workflow Run effect request: %w", err)
	}
	if err := requireJSONEOF(decoder); err != nil {
		return EffectRequest{}, fmt.Errorf("decode workflow Run effect request: %w", err)
	}
	if err := request.Validate(); err != nil {
		return EffectRequest{}, err
	}
	return request, nil
}

func EncodeEffectResult(result EffectResult) (json.RawMessage, error) {
	result = result.Normalize()
	// Bound the failure diagnostics exactly once at this durable boundary so the
	// persisted ResultJSON carries the deterministic cuts and the truncation
	// metadata that replay needs to carry forward.
	result.FailureDiagnostics, result.FailureTruncation = boundEffectResultFailureDiagnostics(
		result.FailureDiagnostics,
		result.FailureTruncation,
	)
	if result.Output == nil {
		result.Output = runkernel.Values{}
	}
	payload, err := json.Marshal(result)
	if err != nil {
		return nil, fmt.Errorf("encode workflow Run effect result: %w", err)
	}
	return payload, nil
}

func DecodeEffectResult(payload []byte) (EffectResult, error) {
	decoder := json.NewDecoder(bytes.NewReader(payload))
	decoder.DisallowUnknownFields()
	decoder.UseNumber()
	var result EffectResult
	if err := decoder.Decode(&result); err != nil {
		return EffectResult{}, fmt.Errorf("decode workflow Run effect result: %w", err)
	}
	if err := requireJSONEOF(decoder); err != nil {
		return EffectResult{}, fmt.Errorf("decode workflow Run effect result: %w", err)
	}
	return result.Normalize(), nil
}

func requireJSONEOF(decoder *json.Decoder) error {
	var trailing json.RawMessage
	err := decoder.Decode(&trailing)
	if err == io.EOF {
		return nil
	}
	if err != nil {
		return fmt.Errorf("invalid trailing JSON: %w", err)
	}
	return fmt.Errorf("trailing JSON value")
}

// StableChildID produces a globally unique deterministic row id from the Run
// id and reducer-local logical key. Reducer state can therefore be replayed
// without persisting a second random-id map, while independent Runs never
// collide on logical ids such as a000.../s000....
func StableChildID(prefix, runID, logicalKey string) (string, error) {
	prefix, runID, logicalKey = strings.TrimSpace(prefix), strings.TrimSpace(runID), strings.TrimSpace(logicalKey)
	if prefix == "" || runID == "" || logicalKey == "" {
		return "", fmt.Errorf("workflow Run child id requires prefix, runId, and logical key")
	}
	if len(prefix) > 8 {
		return "", fmt.Errorf("workflow Run child id prefix is too long")
	}
	hash := sha256.Sum256([]byte("loop.workflow.run.child/v1\x00" + prefix + "\x00" + runID + "\x00" + logicalKey))
	return prefix + "_" + hex.EncodeToString(hash[:16]), nil
}

func StableEffectIdempotencyKey(runID, activationID string, kind runkernel.IntentKind) (string, error) {
	runID, activationID = strings.TrimSpace(runID), strings.TrimSpace(activationID)
	if runID == "" || activationID == "" || strings.TrimSpace(string(kind)) == "" {
		return "", fmt.Errorf("workflow Run effect idempotency requires runId, activationId, and kind")
	}
	hash := sha256.Sum256([]byte("loop.workflow.run.effect/v1\x00" + runID + "\x00" + activationID + "\x00" + string(kind)))
	return "workflow-run:" + hex.EncodeToString(hash[:]), nil
}