package agent

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
	"github.com/tidbcloud/AgentTeam/desktop/internal/consts"
	"github.com/tidbcloud/AgentTeam/desktop/internal/types"
)

const (
	// Inbox retry is a daemon-owned liveness path for messages that have not
	// yet been accepted by a runtime transport. The delay stays deliberately
	// small for short runtime state transitions, then saturates so an unavailable
	// runtime cannot turn the retry agentteam into a busy poll.
	processInboxRetryInitialDelay = 100 * time.Millisecond
	processInboxRetryMaxDelay     = 5 * time.Second
	processInboxRetryMaxAttempts  = 8
	processInboxRetryMaxElapsed   = time.Minute
	// Runtime boundaries such as review/approval and compaction are expected to
	// last much longer than a transient transport failure. They must not be
	// failed merely because the short retry-attempt budget was consumed, but a
	// bounded elapsed deadline still prevents an abandoned boundary from
	// holding the FIFO forever.
	processInboxRetryBoundaryMaxElapsed = 15 * time.Minute
)

type deliveryRetryClass string

const (
	deliveryRetryUnknown   deliveryRetryClass = "unknown"
	deliveryRetryTransient deliveryRetryClass = "transient"
	deliveryRetryBoundary  deliveryRetryClass = "runtime_boundary"
	deliveryRetryPermanent deliveryRetryClass = "permanent"
)

func classifyDeliveryRetryError(err error) deliveryRetryClass {
	if err == nil {
		return deliveryRetryUnknown
	}
	// Codex app-server may put the non-steerable turn kind exclusively in
	// structured error data while its human message remains generic. Preserve
	// review/approval/compaction as a long-lived runtime boundary rather than
	// exhausting the transient attempt budget in a few seconds.
	if _, ok := ackedActiveTurnNotSteerableKind(err); ok {
		return deliveryRetryBoundary
	}
	if isAmbiguousRuntimeDeliveryError(err) {
		return deliveryRetryTransient
	}
	if errors.Is(err, context.Canceled) {
		return deliveryRetryPermanent
	}
	lower := strings.ToLower(strings.TrimSpace(err.Error()))
	switch {
	case strings.Contains(lower, "produced empty input"),
		strings.Contains(lower, "requires a non-empty message id"),
		strings.Contains(lower, "missing runtime input id"),
		strings.Contains(lower, "requires a runtime input id"),
		strings.Contains(lower, "cannot accept acked delivery in \"notloaded\" state"),
		strings.Contains(lower, "cannot accept acked delivery in \"systemerror\" state"),
		strings.Contains(lower, "state is unavailable after refresh"):
		return deliveryRetryPermanent
	case strings.Contains(lower, "during compaction"),
		strings.Contains(lower, "during compact turn"),
		strings.Contains(lower, "during review turn"),
		strings.Contains(lower, "active but its active turn id is unavailable after authoritative read"),
		strings.Contains(lower, "active but its steerable turn is unavailable"):
		return deliveryRetryBoundary
	default:
		return deliveryRetryTransient
	}
}

func (ap *Process) noteInboxRetryFailureLocked(msg types.InboxMessage, err error) {
	if err == nil {
		return
	}
	messageID := inboxRetryMessageID(msg)
	if ap.inboxRetryMessageID == "" {
		ap.inboxRetryMessageID = messageID
		ap.inboxRetryStartedAt = time.Now()
	}
	if ap.inboxRetryMessageID != "" && ap.inboxRetryMessageID != messageID {
		return
	}
	ap.inboxRetryLastError = boundDeliveryOwnershipErrorDetail(err.Error())
	ap.inboxRetryLastClass = classifyDeliveryRetryError(err)
}

func defaultProcessInboxRetryDelay(attempt int) time.Duration {
	if attempt <= 1 {
		return processInboxRetryInitialDelay
	}
	delay := processInboxRetryInitialDelay
	for i := 1; i < attempt; i++ {
		if delay >= processInboxRetryMaxDelay/2 {
			return processInboxRetryMaxDelay
		}
		delay *= 2
	}
	if delay > processInboxRetryMaxDelay {
		return processInboxRetryMaxDelay
	}
	return delay
}

func (ap *Process) inboxRetryDelayForAttemptLocked(attempt int) time.Duration {
	if ap.inboxRetryDelay != nil {
		if delay := ap.inboxRetryDelay(attempt); delay > 0 {
			return delay
		}
		return 0
	}
	return defaultProcessInboxRetryDelay(attempt)
}

// scheduleInboxRetry arms the single daemon-owned retry worker. immediate is
// reserved for authoritative state changes (for example SessionInit or a
// runtime boundary); ordinary delivery failures use exponential backoff.
func (ap *Process) scheduleInboxRetry(immediate bool) {
	ap.mu.Lock()
	ap.scheduleInboxRetryLocked(immediate)
	ap.mu.Unlock()
}

func (ap *Process) scheduleInboxRetryLocked(immediate bool) {
	if ap.stopping || ap.state != processStateRunning || ap.driver == nil || len(ap.inbox) == 0 {
		if len(ap.inbox) == 0 {
			ap.clearInboxRetryBackoffLocked()
		}
		return
	}
	// A busy runtime that cannot accept direct/steered input needs a runtime
	// boundary, not a timer retry. Repeatedly popping and re-persisting the same
	// already-durable queued owner can otherwise exhaust its retry budget while
	// an older pending input legitimately owns the active turn. TurnEnd,
	// RuntimeIdle, SessionInit, or another authoritative state transition will
	// kick this scheduler again once the head is deliverable.
	if !ap.canAttemptInboxRetryLocked() {
		return
	}

	headID := inboxRetryMessageID(ap.inbox[0])
	if ap.inboxRetryMessageID != headID {
		ap.inboxRetryMessageID = headID
		ap.inboxRetryAttempt = 0
		ap.inboxRetryStartedAt = time.Now()
		ap.inboxRetryLastError = ""
		ap.inboxRetryLastClass = deliveryRetryUnknown
	}

	if ap.inboxRetryRunning {
		// The active worker will inspect the queue before it exits. Remember an
		// authoritative kick so a same-head failure is retried immediately rather
		// than waiting through the previous backoff generation.
		ap.inboxRetryImmediate = ap.inboxRetryImmediate || immediate
		return
	}
	if ap.inboxRetryTimer != nil {
		if !immediate {
			return
		}
		ap.inboxRetryTimer.Stop()
		ap.inboxRetryTimer = nil
	}

	delay := time.Duration(0)
	if !immediate {
		ap.inboxRetryAttempt++
		delay = ap.inboxRetryDelayForAttemptLocked(ap.inboxRetryAttempt)
	}
	ap.inboxRetryGeneration++
	generation := ap.inboxRetryGeneration
	ap.inboxRetryTimer = time.AfterFunc(delay, func() {
		ap.runInboxRetry(generation)
	})
}

func (ap *Process) runInboxRetry(generation uint64) {
	ap.mu.Lock()
	if generation != ap.inboxRetryGeneration {
		ap.mu.Unlock()
		return
	}
	ap.inboxRetryTimer = nil
	if ap.stopping || ap.state != processStateRunning || ap.driver == nil || len(ap.inbox) == 0 || !ap.canAttemptInboxRetryLocked() {
		if len(ap.inbox) == 0 {
			ap.clearInboxRetryBackoffLocked()
		}
		ap.mu.Unlock()
		return
	}
	ap.inboxRetryRunning = true
	ap.inboxRetryImmediate = false
	ctx, cancel := context.WithCancel(context.Background())
	ap.inboxRetryCancel = cancel
	ap.mu.Unlock()

	claimed := types.InboxMessage{}
	hasClaimed := false
	var popErr error
	if err := ap.acquireDeliveryGate(ctx); err == nil {
		ap.mu.Lock()
		if generation == ap.inboxRetryGeneration &&
			!ap.stopping &&
			ap.state == processStateRunning &&
			ap.driver != nil {
			claimed, hasClaimed, popErr = ap.popNextDeliverableInboxMessageLocked("already_observed_before_inbox_retry_dequeue")
		}
		ap.mu.Unlock()
		if hasClaimed {
			_ = ap.deliverMessageWithContextSerialized(ctx, claimed, true)
		}
		ap.releaseDeliveryGate()
	}
	_ = popErr // the locked pop records this failure on the head for retry policy
	cancel()

	ap.mu.Lock()
	ap.inboxRetryCancel = nil
	if generation != ap.inboxRetryGeneration || ap.stopping || ap.state != processStateRunning {
		ap.inboxRetryRunning = false
		ap.mu.Unlock()
		return
	}
	if len(ap.inbox) == 0 {
		ap.inboxRetryRunning = false
		ap.clearInboxRetryBackoffLocked()
		ap.mu.Unlock()
		return
	}

	headID := inboxRetryMessageID(ap.inbox[0])
	claimedID := inboxRetryMessageID(claimed)
	if ap.shouldTerminateInboxRetryHeadLocked(headID) {
		failed := ap.inbox[0]
		detail := boundDeliveryOwnershipErrorDetail(ap.inboxRetryFailureDetailLocked())
		attempts := ap.inboxRetryAttempt
		ap.mu.Unlock()

		// Keep inboxRetryRunning and the head owner intact while the terminal
		// transition is being fsynced. A duplicate kick cannot start another
		// runtime attempt, and q2 cannot pass q1 until failed ownership is durable.
		record := DeliveryOwnershipRecord{
			Message:     failed,
			State:       DeliveryOwnershipTerminal,
			Outcome:     DeliveryOutcomeFailed,
			AcceptedAt:  time.Now().UTC(),
			UpdatedAt:   time.Now().UTC(),
			Attempts:    attempts,
			LastError:   detail,
			FailureCode: "delivery_retry_exhausted",
		}
		if err := ap.persistTerminalDeliveryOwnership(record); err != nil {
			ap.mu.Lock()
			ap.inboxRetryRunning = false
			ap.mu.Unlock()
			return
		}

		ap.mu.Lock()
		ap.inboxRetryRunning = false
		ap.rememberFailedMessageLocked(strings.TrimSpace(failed.MessageID))
		if len(ap.inbox) > 0 && inboxRetryMessageID(ap.inbox[0]) == headID {
			ap.inbox = ap.inbox[1:]
		} else {
			ap.inbox = removeInboxMessageByID(ap.inbox, headID)
		}
		canAdvance := generation == ap.inboxRetryGeneration && !ap.stopping && ap.state == processStateRunning
		ap.clearInboxRetryBackoffLocked()
		ap.mu.Unlock()
		// The failed tombstone is durable and the local head is gone. Promote
		// resume safety only through the common quiescence predicate so a newer
		// queued/pending owner keeps the session fail-closed.
		ap.markResumeSafetyIfQuiescent()
		ap.emitFailedInboxRetryDelivery(failed, detail)
		if !canAdvance {
			return
		}
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
		ap.scheduleInboxRetry(true)
		return
	}
	ap.inboxRetryRunning = false
	immediate := ap.inboxRetryImmediate || !hasClaimed || headID != claimedID
	ap.inboxRetryImmediate = false
	if headID != ap.inboxRetryMessageID {
		ap.inboxRetryMessageID = headID
		ap.inboxRetryAttempt = 0
		ap.inboxRetryStartedAt = time.Now()
		ap.inboxRetryLastError = ""
		ap.inboxRetryLastClass = deliveryRetryUnknown
		immediate = true
	}
	ap.scheduleInboxRetryLocked(immediate)
	ap.mu.Unlock()
}

func (ap *Process) canAttemptInboxRetryLocked() bool {
	if ap.transientRuntimeRetryPending || ap.hasUncertainRuntimeInputLocked() || ap.driver == nil ||
		ap.pendingWakeLifecycleOwnerCountLocked() >= processInboxMaxMessages {
		return false
	}
	_, hasProtocolDriver := ap.directMessageDriverLocked()
	_, hasAckedDriver := ap.ackedBusyRuntimeInputDriverLocked()
	supportsBusyStdinDelivery := false
	if hasProtocolDriver && !ap.isIdle {
		if busyDriver, ok := ap.driver.(drivercontract.BusyStdinDeliveryDriver); ok {
			supportsBusyStdinDelivery = busyDriver.SupportsBusyStdinDelivery()
		}
	}
	decision := decideProcessDelivery(deliveryDecisionInput{
		HasProtocolMessageDriver:       hasProtocolDriver,
		SupportsBusyStdinDelivery:      supportsBusyStdinDelivery,
		HasAckedBusyRuntimeInputDriver: hasAckedDriver,
		SupportsStdinNotification:      ap.driver.SupportsStdinNotification(),
		HasSession:                     strings.TrimSpace(ap.SessionID) != "",
		IsIdle:                         ap.isIdle,
	})
	// A dispatching q1 may still be followed by q2 when the runtime has an
	// explicit busy queue (the runtime preserves that FIFO). If the current
	// state would force q2 through daemon inbox/retry instead, q1 must first
	// receive its correlated acknowledgement or reconciliation result; otherwise q2
	// could use a post-settlement idle window and overtake q1.
	if ap.hasBlockingPendingRuntimeInputLocked() &&
		(ap.isIdle || decision.Kind == deliveryDecisionQueue || ap.hasRestoredBlockingRuntimeInputLocked()) {
		return false
	}
	return decision.Kind != deliveryDecisionQueue
}

func inboxRetryMessageID(msg types.InboxMessage) string {
	return strings.TrimSpace(msg.MessageID)
}

func (ap *Process) clearInboxRetryBackoffLocked() {
	ap.inboxRetryAttempt = 0
	ap.inboxRetryMessageID = ""
	ap.inboxRetryImmediate = false
	ap.inboxRetryStartedAt = time.Time{}
	ap.inboxRetryLastError = ""
	ap.inboxRetryLastClass = deliveryRetryUnknown
}

func (ap *Process) shouldTerminateInboxRetryHeadLocked(headID string) bool {
	if headID == "" || headID != ap.inboxRetryMessageID || ap.inboxRetryLastError == "" {
		return false
	}
	if ap.inboxRetryLastClass == deliveryRetryPermanent {
		return true
	}
	if ap.inboxRetryLastClass == deliveryRetryBoundary {
		maxElapsed := ap.inboxRetryBoundaryMaxElapsed
		if maxElapsed <= 0 {
			maxElapsed = processInboxRetryBoundaryMaxElapsed
		}
		return !ap.inboxRetryStartedAt.IsZero() && time.Since(ap.inboxRetryStartedAt) >= maxElapsed
	}
	maxAttempts := ap.inboxRetryMaxAttempts
	if maxAttempts <= 0 {
		maxAttempts = processInboxRetryMaxAttempts
	}
	if ap.inboxRetryAttempt >= maxAttempts {
		return true
	}
	maxElapsed := ap.inboxRetryMaxElapsed
	if maxElapsed <= 0 {
		maxElapsed = processInboxRetryMaxElapsed
	}
	return !ap.inboxRetryStartedAt.IsZero() && time.Since(ap.inboxRetryStartedAt) >= maxElapsed
}

func (ap *Process) inboxRetryFailureDetailLocked() string {
	detail := fmt.Sprintf("runtime delivery failed after %d retry attempt(s)", ap.inboxRetryAttempt)
	if ap.inboxRetryLastClass != "" {
		detail += "; class=" + string(ap.inboxRetryLastClass)
	}
	if ap.inboxRetryLastError != "" {
		detail += "; latest error: " + ap.inboxRetryLastError
	}
	return detail
}

func (ap *Process) emitFailedInboxRetryDelivery(msg types.InboxMessage, detail string) {
	ap.emitRuntimeEvent(RuntimeEvent{
		MessageID:  strings.TrimSpace(msg.MessageID),
		EventClass: consts.RuntimeEventClassWakeFailed,
		Target:     strings.TrimSpace(msg.Target),
		Detail:     detail,
	})
}

func removeInboxMessageByID(messages []types.InboxMessage, messageID string) []types.InboxMessage {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return messages
	}
	next := messages[:0]
	for _, msg := range messages {
		if strings.TrimSpace(msg.MessageID) == messageID {
			continue
		}
		next = append(next, msg)
	}
	return next
}

// stopInboxRetryLocked invalidates both a pending timer and an in-flight retry
// context. A driver that honors cancellation returns promptly; the generation
// check still prevents late completion from scheduling more work.
func (ap *Process) stopInboxRetryLocked() {
	ap.inboxRetryGeneration++
	if ap.inboxRetryTimer != nil {
		ap.inboxRetryTimer.Stop()
		ap.inboxRetryTimer = nil
	}
	if ap.inboxRetryCancel != nil {
		ap.inboxRetryCancel()
		ap.inboxRetryCancel = nil
	}
	ap.clearInboxRetryBackoffLocked()
}