package agent

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"sort"
	"strings"
	"time"
	"unicode/utf8"

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

// DeliveryOwnershipState records which side owns the next safe action for an
// accepted AgentTeam delivery.
type DeliveryOwnershipState string

type DeliveryOwnershipInflightReason string

type DeliveryOwnershipOutcome string

type deliveryOwnershipTransition struct {
	state          DeliveryOwnershipState
	inflightReason DeliveryOwnershipInflightReason
	outcome        DeliveryOwnershipOutcome
}

// errRuntimeInputMutationNotAttempted marks a delivery failure that happened
// before the runtime input mutation could cross its dispatch boundary. The
// wrapped cause can still be a timeout or transport-generation failure, but it
// is read-only/preflight ambiguity rather than ambiguity about whether the
// user input itself was accepted. Process may therefore return the durable
// owner to queued and retry it without risking a duplicate mutation.
var errRuntimeInputMutationNotAttempted = errors.New("runtime input mutation was not attempted")

const (
	DeliveryOwnershipQueued   DeliveryOwnershipState = "queued"
	DeliveryOwnershipInflight DeliveryOwnershipState = "inflight"
	DeliveryOwnershipTerminal DeliveryOwnershipState = "terminal"

	DeliveryInflightDispatching    DeliveryOwnershipInflightReason = "dispatching"
	DeliveryInflightOutcomeUnknown DeliveryOwnershipInflightReason = "outcome_unknown"

	DeliveryOutcomeAcked DeliveryOwnershipOutcome = "surfaced"
	// alternateDeliveryOutcomeAcked is accepted only when reading journals
	// written by prerelease builds that briefly used the semantic spelling.
	alternateDeliveryOutcomeAcked DeliveryOwnershipOutcome = "acked"
	DeliveryOutcomeFailed         DeliveryOwnershipOutcome = "failed"
)

var (
	deliveryOwnershipQueued      = deliveryOwnershipTransition{state: DeliveryOwnershipQueued}
	deliveryOwnershipDispatching = deliveryOwnershipTransition{
		state: DeliveryOwnershipInflight, inflightReason: DeliveryInflightDispatching,
	}
	deliveryOwnershipOutcomeUnknown = deliveryOwnershipTransition{
		state: DeliveryOwnershipInflight, inflightReason: DeliveryInflightOutcomeUnknown,
	}
	deliveryOwnershipAcked = deliveryOwnershipTransition{
		state: DeliveryOwnershipTerminal, outcome: DeliveryOutcomeAcked,
	}
	deliveryOwnershipFailed = deliveryOwnershipTransition{
		state: DeliveryOwnershipTerminal, outcome: DeliveryOutcomeFailed,
	}
)

func IsLiveDeliveryOwnershipState(state DeliveryOwnershipState) bool {
	return state == DeliveryOwnershipQueued || state == DeliveryOwnershipInflight
}

const (
	processPendingRuntimeInputMaxMessages  = processInboxMaxMessages
	processPendingReconcileMaxAttempts     = 6
	processPendingReconcileInitialDelay    = 100 * time.Millisecond
	processPendingReconcileMaxDelay        = 5 * time.Second
	processPendingReconcileSuccessGrace    = time.Second
	processPendingReconcilePendingElapsed  = 15 * time.Minute
	processPendingReconcileRequestTimeout  = 10 * time.Second
	processDeliveryOwnershipFailureDetails = "runtime delivery ownership could not be reconciled without risking duplicate input"
	deliveryOwnershipLastErrorMaxBytes     = 16 << 10
)

// DeliveryOwnershipRecord is the persisted, runtime-independent form of one
// accepted inbound delivery. RuntimeInput retains only transport correlation;
// Message remains the authoritative replay body.
type DeliveryOwnershipRecord struct {
	Message        types.InboxMessage              `json:"message"`
	Input          drivercontract.RuntimeInput     `json:"input,omitempty"`
	SessionID      string                          `json:"sessionId,omitempty"`
	WakeToken      string                          `json:"wakeToken,omitempty"`
	State          DeliveryOwnershipState          `json:"state"`
	InflightReason DeliveryOwnershipInflightReason `json:"inflightReason,omitempty"`
	Outcome        DeliveryOwnershipOutcome        `json:"outcome,omitempty"`
	AcceptedAt     time.Time                       `json:"acceptedAt"`
	UpdatedAt      time.Time                       `json:"updatedAt"`
	Attempts       int                             `json:"attempts,omitempty"`
	LastError      string                          `json:"lastError,omitempty"`
	FailureCode    string                          `json:"failureCode,omitempty"`
}

func (r DeliveryOwnershipRecord) MessageID() string {
	return strings.TrimSpace(firstNonEmptyString(r.Message.MessageID, r.Input.ID))
}

func (r DeliveryOwnershipRecord) IsQueued() bool {
	return r.State == DeliveryOwnershipQueued
}

func (r DeliveryOwnershipRecord) IsInflight(reason DeliveryOwnershipInflightReason) bool {
	return r.State == DeliveryOwnershipInflight && r.InflightReason == reason
}

func (r DeliveryOwnershipRecord) IsAcked() bool {
	r = NormalizeDeliveryOwnershipRecordForRead(r)
	return r.State == DeliveryOwnershipTerminal && r.Outcome == DeliveryOutcomeAcked
}

func (r DeliveryOwnershipRecord) IsFailed() bool {
	return r.State == DeliveryOwnershipTerminal && r.Outcome == DeliveryOutcomeFailed
}

// NormalizeDeliveryOwnershipRecordForRead accepts the prerelease alternate
// spelling while preserving the frozen persisted vocabulary on the next write.
func NormalizeDeliveryOwnershipRecordForRead(record DeliveryOwnershipRecord) DeliveryOwnershipRecord {
	if record.State == DeliveryOwnershipTerminal && record.Outcome == alternateDeliveryOutcomeAcked {
		record.Outcome = DeliveryOutcomeAcked
	}
	return record
}

// NormalizeDeliveryOwnershipRecordForPersistence removes ephemeral runtime
// projects and bounds untrusted diagnostics before they enter the shared
// durable session-state file. An arbitrarily large provider error must never
// make a terminal transition permanently exceed the journal cap and block
// every later FIFO owner.
func NormalizeDeliveryOwnershipRecordForPersistence(record DeliveryOwnershipRecord) DeliveryOwnershipRecord {
	record = NormalizeDeliveryOwnershipRecordForRead(record)
	record.LastError = boundDeliveryOwnershipErrorDetail(record.LastError)
	// Message is the durable replay body. Input retains only transport
	// correlation; rendered prompt text is ephemeral and must not enter the
	// durable ownership journal.
	record.Input.CompiledPrompt = nil
	record.Input.Attachments = nil
	if record.State == DeliveryOwnershipTerminal {
		// Terminal evidence needs only stable deduplication, routing, freshness,
		// and diagnostic identity. The replay body must not keep consuming the
		// bounded live journal after ownership has ended.
		record.Message = types.InboxMessage{
			MessageID: record.Message.MessageID,
			Target:    record.Message.Target,
			TargetKey: record.Message.TargetKey,
			TargetSeq: record.Message.TargetSeq,
		}
		record.Input = drivercontract.RuntimeInput{ID: record.Input.ID}
		record.InflightReason = ""
	} else {
		record.Outcome = ""
	}
	return record
}

func ValidateDeliveryOwnershipRecord(record DeliveryOwnershipRecord) error {
	record = NormalizeDeliveryOwnershipRecordForRead(record)
	if record.MessageID() == "" {
		return fmt.Errorf("runtime delivery ownership requires message id")
	}
	switch record.State {
	case DeliveryOwnershipQueued:
		if record.InflightReason != "" || record.Outcome != "" {
			return fmt.Errorf("queued runtime delivery ownership carries inflight or terminal fields")
		}
	case DeliveryOwnershipInflight:
		if record.InflightReason != DeliveryInflightDispatching && record.InflightReason != DeliveryInflightOutcomeUnknown {
			return fmt.Errorf("inflight runtime delivery ownership has invalid reason %q", record.InflightReason)
		}
		if record.Outcome != "" {
			return fmt.Errorf("inflight runtime delivery ownership carries terminal outcome")
		}
	case DeliveryOwnershipTerminal:
		if record.Outcome != DeliveryOutcomeAcked && record.Outcome != DeliveryOutcomeFailed {
			return fmt.Errorf("terminal runtime delivery ownership has invalid outcome %q", record.Outcome)
		}
	default:
		return fmt.Errorf("runtime delivery ownership has invalid state %q", record.State)
	}
	return nil
}

func boundDeliveryOwnershipErrorDetail(detail string) string {
	detail = strings.TrimSpace(strings.ToValidUTF8(detail, "\uFFFD"))
	if len(detail) <= deliveryOwnershipLastErrorMaxBytes {
		return detail
	}
	end := deliveryOwnershipLastErrorMaxBytes
	for end > 0 && !utf8.RuneStart(detail[end]) {
		end--
	}
	return strings.TrimSpace(detail[:end])
}

// DeliveryOwnershipStore persists ownership using the dispatcher's existing
// atomic session-state store. Implementations must fsync an upsert before
// returning success: Put is the durability boundary behind agent:deliver:ack.
type DeliveryOwnershipStore interface {
	LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error)
	// PutDeliveryOwnership atomically writes one state transition. A new live
	// owner may reclaim the oldest live owners to remain within the bounded
	// journal; those durable failed tombstones are returned to the Process so it
	// can project the same transaction into its local containers.
	PutDeliveryOwnership(DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error)
	DeleteDeliveryOwnership(messageID string) error
}

// RuntimeInputReconciliation is the authoritative answer for a correlated
// clientUserMessageId after transport completion was ambiguous or a Process
// restarted with runtime-owned pending input.
type RuntimeInputReconciliation string

const (
	RuntimeInputReconciliationUnknown RuntimeInputReconciliation = "unknown"
	RuntimeInputReconciliationPending RuntimeInputReconciliation = "pending"
	RuntimeInputReconciliationAcked   RuntimeInputReconciliation = "surfaced" // legacy persisted value
	RuntimeInputReconciliationAbsent  RuntimeInputReconciliation = "absent"
)

// RuntimeInputReconciler is implemented by runtimes with a durable correlated
// input identity (Codex uses userMessage.clientId). An absent answer must be
// authoritative; callers will replay the exact original body only then.
type RuntimeInputReconciler interface {
	ReconcileRuntimeInput(ctx context.Context, inputID, sessionID string) (RuntimeInputReconciliation, error)
}

// SuccessfulAckedBusyRuntimeInputReconciliationPolicy lets a runtime opt out
// of the grace-period history check after an active-turn input request returned
// success. Direct turn starts retain their ordinary reconciliation lifecycle.
// Immediate reconciliation for ambiguous outcomes and restored ownership is
// unaffected.
type SuccessfulAckedBusyRuntimeInputReconciliationPolicy interface {
	ReconcileSuccessfulAckedBusyRuntimeInput() bool
}

// SetDeliveryOwnershipStore attaches durable accepted-delivery ownership and
// restores records from a prior Process/daemon. Call it immediately after a
// no-prompt launch is published and before any post-publication handoff can
// mutate the runtime. Use SetDeliveryOwnershipStoreForHandoff when one queued
// record has already been reserved for that first explicit handoff.
func (ap *Process) SetDeliveryOwnershipStore(store DeliveryOwnershipStore) error {
	return ap.setDeliveryOwnershipStore(store, "")
}

// SetDeliveryOwnershipStoreForHandoff installs the durable store while
// reserving one already-fsynced queued record for an explicit post-launch
// DeliverForHandoff call. This lets the runtime publish a session/thread id
// before the first mutating request.
func (ap *Process) SetDeliveryOwnershipStoreForHandoff(store DeliveryOwnershipStore, messageID string) error {
	return ap.setDeliveryOwnershipStore(store, strings.TrimSpace(messageID))
}

func (ap *Process) setDeliveryOwnershipStore(store DeliveryOwnershipStore, handoffMessageID string) error {
	if ap == nil || store == nil {
		return nil
	}
	records, err := store.LoadDeliveryOwnership()
	if err != nil {
		return fmt.Errorf("load runtime delivery ownership: %w", err)
	}
	sort.SliceStable(records, func(i, j int) bool {
		if records[i].AcceptedAt.Equal(records[j].AcceptedAt) {
			left, right := records[i].Message, records[j].Message
			if left.TargetKey == right.TargetKey && left.TargetSeq > 0 && right.TargetSeq > 0 && left.TargetSeq != right.TargetSeq {
				return left.TargetSeq < right.TargetSeq
			}
			return records[i].MessageID() < records[j].MessageID()
		}
		if records[i].AcceptedAt.IsZero() {
			return false
		}
		if records[j].AcceptedAt.IsZero() {
			return true
		}
		return records[i].AcceptedAt.Before(records[j].AcceptedAt)
	})
	for _, record := range records {
		if err := ValidateDeliveryOwnershipRecord(record); err != nil {
			return err
		}
	}

	restoredInputs := make(map[string]drivercontract.RuntimeInput)
	for _, record := range records {
		if record.State != DeliveryOwnershipInflight || record.Input.CompiledPrompt != nil {
			continue
		}
		compiled, compileErr := ap.compileDeliveryPromptWithContext(context.Background(), record.Message, "restored_delivery", record.WakeToken)
		if compileErr != nil {
			return fmt.Errorf("compile restored delivery %s: %w", record.MessageID(), compileErr)
		}
		restoredInputs[record.MessageID()] = ap.runtimeInputForDelivery(context.Background(), record.Message, compiled)
	}

	ap.mu.Lock()
	ap.deliveryOwnershipStore = store
	ap.deliveryOwnershipStoreEpoch++
	ap.deliveryOwnershipHandoffID = handoffMessageID
	resumeSafetyUnsafe := handoffMessageID != ""
	now := time.Now().UTC()
	var launchAcked []DeliveryOwnershipRecord
	for _, record := range records {
		messageID := record.MessageID()
		if messageID == "" {
			continue
		}
		ap.rememberRuntimeDeliveryReceiptLocked(record.Input.ID)
		// Failed is a durable terminal owner just like acked, but it must not
		// advance freshness: the runtime never observed this input. Restore it
		// before consulting launch-acked hints so an old failed delivery can
		// never be resurrected as a new initial prompt.
		if record.State == DeliveryOwnershipTerminal && record.Outcome == DeliveryOutcomeFailed {
			ap.rememberFailedMessageLocked(messageID)
			continue
		}
		if _, acked := ap.ackedSet[messageID]; acked {
			record.State = DeliveryOwnershipTerminal
			record.Outcome = DeliveryOutcomeAcked
			record.InflightReason = ""
			record.UpdatedAt = now
			launchAcked = append(launchAcked, record)
			continue
		}
		restoredInput := record.Input
		if input, ok := restoredInputs[messageID]; ok {
			restoredInput = input
		}
		switch record.State {
		case DeliveryOwnershipTerminal:
			ap.rememberRuntimeAckedMessageLocked(messageID)
			ap.freshness.MarkObserved(record.Message.TargetKey, record.Message.TargetSeq, messageID)
		case DeliveryOwnershipInflight:
			resumeSafetyUnsafe = true
			if ap.pendingRuntimeInputs == nil {
				ap.pendingRuntimeInputs = make(map[string]pendingRuntimeInput)
			}
			input := restoredInput
			if strings.TrimSpace(input.ID) == "" {
				input.ID = messageID
			}
			createdAt := record.AcceptedAt
			if createdAt.IsZero() {
				createdAt = now
			}
			pending := pendingRuntimeInput{
				Message:           record.Message,
				Input:             input,
				SessionID:         strings.TrimSpace(record.SessionID),
				WakeToken:         strings.TrimSpace(record.WakeToken),
				CreatedAt:         createdAt,
				InflightReason:    DeliveryInflightOutcomeUnknown,
				ReconcileAttempts: record.Attempts,
				Restored:          true,
			}
			ap.pendingRuntimeInputs[input.ID] = pending
			ap.rememberUnobservedFrontierHintLocked(record.Message)
		case DeliveryOwnershipQueued:
			if messageID == handoffMessageID {
				continue
			}
			if _, acked := ap.ackedSet[messageID]; acked {
				continue
			}
			if stale, persistErr := ap.durablySuppressStaleDeliveryLocked(record.Message, "already_observed_before_delivery_ownership_restore"); stale {
				if persistErr != nil {
					ap.mu.Unlock()
					return fmt.Errorf("persist stale restored delivery %s as acked: %w", messageID, persistErr)
				}
				continue
			}
			if !inboxContainsMessageID(ap.inbox, messageID) {
				resumeSafetyUnsafe = true
				ap.inbox = append(ap.inbox, record.Message)
				ap.rememberUnobservedFrontierHintLocked(record.Message)
			} else {
				resumeSafetyUnsafe = true
			}
		}
	}
	ap.mu.Unlock()
	if resumeSafetyUnsafe {
		// SessionInit can arrive before Dispatcher installs this store and may
		// provisionally observe an empty idle process. Restoring any live owner
		// must supersede that snapshot so callback replay remains fail-closed.
		ap.markResumeSafety(false)
	}

	for _, record := range launchAcked {
		if err := ap.persistDeliveryOwnership(record); err != nil {
			return fmt.Errorf("persist launch-acked runtime delivery ownership: %w", err)
		}
	}
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
	if handoffMessageID == "" {
		ap.scheduleInboxRetry(true)
	}
	return nil
}

func inboxContainsMessageID(messages []types.InboxMessage, messageID string) bool {
	messageID = strings.TrimSpace(messageID)
	for _, msg := range messages {
		if strings.TrimSpace(msg.MessageID) == messageID {
			return true
		}
	}
	return false
}

func (ap *Process) ownershipRecord(pending pendingRuntimeInput, transition deliveryOwnershipTransition, attempts int, lastErr error) DeliveryOwnershipRecord {
	acceptedAt := pending.CreatedAt
	if acceptedAt.IsZero() {
		acceptedAt = time.Now().UTC()
	}
	record := DeliveryOwnershipRecord{
		Message:        pending.Message,
		Input:          pending.Input,
		SessionID:      strings.TrimSpace(pending.SessionID),
		WakeToken:      strings.TrimSpace(pending.WakeToken),
		State:          transition.state,
		InflightReason: transition.inflightReason,
		Outcome:        transition.outcome,
		AcceptedAt:     acceptedAt.UTC(),
		UpdatedAt:      time.Now().UTC(),
		Attempts:       attempts,
	}
	if lastErr != nil {
		record.LastError = boundDeliveryOwnershipErrorDetail(lastErr.Error())
	}
	return NormalizeDeliveryOwnershipRecordForPersistence(record)
}

func (ap *Process) persistDeliveryOwnership(record DeliveryOwnershipRecord) error {
	record, messageID, err := prepareDeliveryOwnershipRecord(record)
	if err != nil {
		return err
	}
	ap.mu.Lock()
	store := ap.deliveryOwnershipStore
	ap.mu.Unlock()
	ap.deliveryOwnershipMutationMu.Lock()
	reclaimed, err := putDeliveryOwnershipRecord(store, record, messageID)
	ap.deliveryOwnershipMutationMu.Unlock()
	if err != nil {
		return err
	}
	ap.applyReclaimedDeliveryOwnership(reclaimed)
	return nil
}

// persistDeliveryOwnershipLocked commits one ownership record while ap.mu is
// held. Reconciliation transitions deliberately keep this lock across the
// store Put so a live RuntimeInputAcked event and a stale reconciliation
// result cannot cross and let the stale write become the final durable phase.
func (ap *Process) persistDeliveryOwnershipLocked(record DeliveryOwnershipRecord) error {
	record, messageID, err := prepareDeliveryOwnershipRecord(record)
	if err != nil {
		return err
	}
	ap.deliveryOwnershipMutationMu.Lock()
	defer ap.deliveryOwnershipMutationMu.Unlock()
	return putDeliveryOwnershipTransition(ap.deliveryOwnershipStore, record, messageID)
}

func prepareDeliveryOwnershipRecord(record DeliveryOwnershipRecord) (DeliveryOwnershipRecord, string, error) {
	record = NormalizeDeliveryOwnershipRecordForPersistence(record)
	if err := ValidateDeliveryOwnershipRecord(record); err != nil {
		return DeliveryOwnershipRecord{}, "", err
	}
	messageID := record.MessageID()
	if record.AcceptedAt.IsZero() {
		record.AcceptedAt = time.Now().UTC()
	}
	if record.UpdatedAt.IsZero() {
		record.UpdatedAt = time.Now().UTC()
	}
	return record, messageID, nil
}

func putDeliveryOwnershipRecord(store DeliveryOwnershipStore, record DeliveryOwnershipRecord, messageID string) ([]DeliveryOwnershipRecord, error) {
	if store == nil {
		return nil, nil
	}
	var (
		reclaimed []DeliveryOwnershipRecord
		err       error
	)
	reclaimed, err = store.PutDeliveryOwnership(record)
	if err != nil {
		return nil, fmt.Errorf("persist runtime delivery ownership %s as %s: %w", messageID, record.State, err)
	}
	return reclaimed, nil
}

func putDeliveryOwnershipTransition(store DeliveryOwnershipStore, record DeliveryOwnershipRecord, messageID string) error {
	if store == nil {
		return nil
	}
	if _, err := store.PutDeliveryOwnership(record); err != nil {
		return fmt.Errorf("persist runtime delivery ownership %s as %s: %w", messageID, record.State, err)
	}
	return nil
}

func (ap *Process) deleteDeliveryOwnership(messageID string) error {
	ap.mu.Lock()
	store := ap.deliveryOwnershipStore
	ap.mu.Unlock()
	return ap.deleteDeliveryOwnershipFromStore(store, messageID)
}

// deleteDeliveryOwnershipLocked is the ap.mu-held variant used by rollback
// paths. It avoids recursively taking ap.mu while retaining the same journal
// serialization as ordinary deletes.
func (ap *Process) deleteDeliveryOwnershipLocked(messageID string) error {
	return ap.deleteDeliveryOwnershipFromStore(ap.deliveryOwnershipStore, messageID)
}

func (ap *Process) deleteDeliveryOwnershipFromStore(store DeliveryOwnershipStore, messageID string) error {
	if store == nil || strings.TrimSpace(messageID) == "" {
		return nil
	}
	ap.deliveryOwnershipMutationMu.Lock()
	defer ap.deliveryOwnershipMutationMu.Unlock()
	return store.DeleteDeliveryOwnership(strings.TrimSpace(messageID))
}

// persistTerminalDeliveryOwnership retries until the terminal transition is
// durably committed or the Process is explicitly stopping. Callers must keep
// the existing queued/uncertain owner intact until this returns nil. This is
// the critical ordering that lets q2 advance only after q1's explicit failure
// can survive a daemon crash.
func (ap *Process) persistTerminalDeliveryOwnership(record DeliveryOwnershipRecord) error {
	var lastErr error
	for attempt := 1; ; attempt++ {
		if err := ap.persistDeliveryOwnership(record); err == nil {
			return nil
		} else {
			lastErr = err
			slog.Error("persist terminal runtime delivery ownership failed; retaining live owner",
				"agent_id", ap.AgentID,
				"message_id", record.MessageID(),
				"state", record.State,
				"error", err)
		}

		ap.mu.Lock()
		stopping := ap.stopping || ap.state == processStateStopped
		delay := ap.inboxRetryDelayForAttemptLocked(attempt)
		done := ap.exited
		ap.mu.Unlock()
		if stopping {
			return lastErr
		}
		if delay <= 0 {
			delay = processInboxRetryInitialDelay
		}
		timer := time.NewTimer(delay)
		if done == nil {
			<-timer.C
			continue
		}
		select {
		case <-timer.C:
		case <-done:
			timer.Stop()
			return lastErr
		}
	}
}

func (ap *Process) persistQueuedDelivery(msg types.InboxMessage, acceptedAt time.Time, attempts int, lastErr error) error {
	inputID := strings.TrimSpace(msg.MessageID)
	return ap.persistDeliveryOwnership(ap.ownershipRecord(pendingRuntimeInput{
		Message:   msg,
		Input:     drivercontract.RuntimeInput{ID: inputID},
		CreatedAt: acceptedAt,
	}, deliveryOwnershipQueued, attempts, lastErr))
}

func (ap *Process) persistAckedDelivery(msg types.InboxMessage, input drivercontract.RuntimeInput, sessionID, wakeToken string, acceptedAt time.Time) error {
	return ap.persistDeliveryOwnership(ap.ownershipRecord(pendingRuntimeInput{
		Message:   msg,
		Input:     input,
		SessionID: sessionID,
		WakeToken: wakeToken,
		CreatedAt: acceptedAt,
	}, deliveryOwnershipAcked, 0, nil))
}

func (ap *Process) hasUncertainRuntimeInputLocked() bool {
	for _, pending := range ap.pendingRuntimeInputs {
		if !pending.reclaimed && pending.InflightReason == DeliveryInflightOutcomeUnknown {
			return true
		}
	}
	return false
}

// hasLivePendingRuntimeInputLocked reports whether any pending runtime input
// still holds live delivery ownership. A reclaimed inflight entry is retained
// only as a bounded correlation lease so an exact late ACK can upgrade
// its durable failed tombstone; it is no longer a live owner and must not
// reserve wake capacity, hold idle/recovery, cover freshness, or gate paging.
func (ap *Process) hasLivePendingRuntimeInputLocked() bool {
	for _, pending := range ap.pendingRuntimeInputs {
		if !pending.reclaimed {
			return true
		}
	}
	return false
}

// hasBlockingPendingRuntimeInputLocked reports correlated owners that have
// crossed the runtime mutation boundary. Zero-value pending entries are used
// by a few legacy/in-memory callers as a liveness hint, but are not durable
// ownership and must not block an inbox retry.
func (ap *Process) hasBlockingPendingRuntimeInputLocked() bool {
	for _, pending := range ap.pendingRuntimeInputs {
		if pending.reclaimed {
			continue
		}
		if pending.InflightReason == DeliveryInflightDispatching ||
			pending.InflightReason == DeliveryInflightOutcomeUnknown {
			return true
		}
	}
	return false
}

// hasRestoredBlockingRuntimeInputLocked reports whether a recovered owner is
// still holding the FIFO head. A live same-generation owner may be safely
// followed by a runtime-native busy queue (the provider preserves that
// ordering), but a replacement has no correlation with the old mutation. It
// must reconcile or fail the restored owner before admitting q2, even when
// the replacement Pi process currently looks idle/busy-capable.
func (ap *Process) hasRestoredBlockingRuntimeInputLocked() bool {
	for _, pending := range ap.pendingRuntimeInputs {
		if pending.reclaimed || !pending.Restored {
			continue
		}
		if pending.InflightReason == DeliveryInflightDispatching ||
			pending.InflightReason == DeliveryInflightOutcomeUnknown {
			return true
		}
	}
	return false
}

type pendingRuntimeInputMutationToken struct {
	inputID    string
	createdAt  time.Time
	generation uint64
}

func (ap *Process) currentPendingRuntimeInputForTokenLocked(token pendingRuntimeInputMutationToken) (pendingRuntimeInput, bool) {
	pending, ok := ap.pendingRuntimeInputs[token.inputID]
	if !ok || !pending.CreatedAt.Equal(token.createdAt) || !pending.reconcileRunning || pending.reconcileGeneration != token.generation {
		return pendingRuntimeInput{}, false
	}
	return pending, true
}

// claimPendingRuntimeInputTransitionLocked invalidates the currently-running
// reconciler generation and reserves the record for an authoritative
// transition (for example a live runtime_input_acked event). Keeping
// reconcileRunning=true blocks a replacement reconciler until this transition
// either commits or explicitly releases the reservation after a failed
// non-terminal write.
func (ap *Process) claimPendingRuntimeInputTransitionLocked(
	inputID string,
	createdAt time.Time,
	expectedGeneration *uint64,
) (pendingRuntimeInput, pendingRuntimeInputMutationToken, bool) {
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if !ok || !pending.CreatedAt.Equal(createdAt) {
		return pendingRuntimeInput{}, pendingRuntimeInputMutationToken{}, false
	}
	// A final-hop lifecycle capability was admitted before Stop made the
	// Process non-accepting and keeps retirement/replacement fenced. Its complete
	// response write is therefore still authorized to durably commit acked;
	// ordinary late runtime events remain rejected.
	if (ap.stopping || ap.state == processStateStopped) && pending.finalHopLifecycleLease == nil {
		return pendingRuntimeInput{}, pendingRuntimeInputMutationToken{}, false
	}
	if expectedGeneration != nil && (!pending.reconcileRunning || pending.reconcileGeneration != *expectedGeneration) {
		return pendingRuntimeInput{}, pendingRuntimeInputMutationToken{}, false
	}
	pending.reconcileGeneration++
	pending.reconcileRunning = true
	ap.pendingRuntimeInputs[inputID] = pending
	return pending, pendingRuntimeInputMutationToken{
		inputID:    inputID,
		createdAt:  pending.CreatedAt,
		generation: pending.reconcileGeneration,
	}, true
}

func (ap *Process) persistReconcileProgressIfCurrent(
	token pendingRuntimeInputMutationToken,
	transition deliveryOwnershipTransition,
	attempts int,
	lastErr error,
) (bool, error) {
	ap.mu.Lock()
	defer ap.mu.Unlock()
	pending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
	if !ok || ap.stopping || ap.state == processStateStopped {
		return false, nil
	}
	return true, ap.persistDeliveryOwnershipLocked(ap.ownershipRecord(pending, transition, attempts, lastErr))
}

// commitPendingRuntimeInputTransition conditionally persists and applies a
// terminal/ownership-transfer mutation while holding ap.mu across both steps.
// Retry mode is used for terminal records, whose previous live owner must
// remain intact until either durability succeeds, another authoritative
// generation wins, or the Process stops.
func (ap *Process) commitPendingRuntimeInputTransition(
	token pendingRuntimeInputMutationToken,
	record DeliveryOwnershipRecord,
	retry bool,
	mutateLocked func(pendingRuntimeInput),
) (bool, error) {
	var lastErr error
	for attempt := 1; ; attempt++ {
		ap.mu.Lock()
		pending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
		admittedFinalHop := ok && pending.finalHopLifecycleLease != nil
		if !ok || ((ap.stopping || ap.state == processStateStopped) && !admittedFinalHop) {
			ap.mu.Unlock()
			return false, nil
		}
		err := ap.persistDeliveryOwnershipLocked(record)
		if err == nil {
			if mutateLocked != nil {
				mutateLocked(pending)
			}
			ap.mu.Unlock()
			return true, nil
		}
		lastErr = err
		stopping := (ap.stopping || ap.state == processStateStopped) && !admittedFinalHop
		delay := ap.inboxRetryDelayForAttemptLocked(attempt)
		done := ap.exited
		ap.mu.Unlock()

		if !retry || stopping {
			return false, lastErr
		}
		slog.Error("persist conditional terminal runtime delivery ownership failed; retaining live owner",
			"agent_id", ap.AgentID,
			"message_id", record.MessageID(),
			"state", record.State,
			"error", err)
		if delay <= 0 {
			delay = processInboxRetryInitialDelay
		}
		timer := time.NewTimer(delay)
		if done == nil {
			<-timer.C
			continue
		}
		select {
		case <-timer.C:
		case <-done:
			timer.Stop()
			return false, lastErr
		}
	}
}

func (ap *Process) schedulePendingRuntimeInputReconcile(inputID string, createdAt time.Time, immediate bool) {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return
	}
	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if !ok ||
		!pending.CreatedAt.Equal(createdAt) ||
		pending.reconcileRunning ||
		ap.stopping ||
		ap.state == processStateStopped {
		ap.mu.Unlock()
		return
	}
	pending.reconcileGeneration++
	generation := pending.reconcileGeneration
	pending.reconcileRunning = true
	ap.pendingRuntimeInputs[inputID] = pending
	ap.mu.Unlock()

	go ap.runPendingRuntimeInputReconcile(inputID, createdAt, generation, immediate)
}

// scheduleOldestPendingRuntimeInputReconcile preserves the durable ownership
// FIFO across restart. Restored pending/uncertain records are authoritative but
// read-only reconciliation of q2 must not race q1: q1 may become authoritatively
// absent and require replay before q2 can move. A queued head or reserved
// two-phase handoff therefore blocks every pending reconciler behind it.
func (ap *Process) scheduleOldestPendingRuntimeInputReconcile(immediate bool) {
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped || strings.TrimSpace(ap.deliveryOwnershipHandoffID) != "" {
		ap.mu.Unlock()
		return
	}
	for _, pending := range ap.pendingRuntimeInputs {
		if pending.reconcileRunning && !pending.reclaimed {
			ap.mu.Unlock()
			return
		}
	}
	var oldest pendingRuntimeInput
	found := false
	for _, pending := range ap.pendingRuntimeInputs {
		if pending.reclaimed || strings.TrimSpace(pending.Input.ID) == "" {
			continue
		}
		if !found || pending.CreatedAt.Before(oldest.CreatedAt) ||
			(pending.CreatedAt.Equal(oldest.CreatedAt) && pending.Input.ID < oldest.Input.ID) {
			oldest = pending
			found = true
		}
	}
	if found && oldest.FinalHopDeferred {
		ap.mu.Unlock()
		return
	}
	if found && !oldest.Restored {
		if _, supportsReconcile := ap.driver.(RuntimeInputReconciler); !supportsReconcile {
			// ACK-only runtimes keep a successfully dispatched current-
			// generation input live until that input's own correlated
			// acknowledgement arrives. A scheduler kick from an earlier sibling
			// must not reinterpret this owner as an unsupported-reconciliation
			// failure. Explicit failure paths call schedulePending... directly,
			// while restored owners still fail closed below.
			ap.mu.Unlock()
			return
		}
	}
	// A pending owner is itself the oldest FIFO head and must reconcile before
	// any queued successor. The previous "any inbox blocks reconciliation"
	// shortcut stranded a restored/accepted q1 forever behind q2: q1 could not
	// become absent/acked/failed, while q2 was correctly prevented from
	// overtaking it. Only a queue with no pending owner is blocked here. A tool
	// legacy delivery receipt remains allowed through the same path for its final-hop
	// lifecycle.
	if len(ap.inbox) > 0 && !found {
		ap.mu.Unlock()
		return
	}
	ap.mu.Unlock()
	if found {
		ap.schedulePendingRuntimeInputReconcile(oldest.Input.ID, oldest.CreatedAt, immediate)
	}
}

func pendingReconcileDelay(attempt int) time.Duration {
	if attempt <= 1 {
		return processPendingReconcileInitialDelay
	}
	delay := processPendingReconcileInitialDelay
	for i := 1; i < attempt; i++ {
		delay *= 2
		if delay >= processPendingReconcileMaxDelay {
			return processPendingReconcileMaxDelay
		}
	}
	return delay
}

func (ap *Process) runPendingRuntimeInputReconcile(inputID string, createdAt time.Time, generation uint64, immediate bool) {
	token := pendingRuntimeInputMutationToken{inputID: inputID, createdAt: createdAt, generation: generation}
	if !immediate {
		ap.mu.Lock()
		grace := ap.pendingReconcileSuccessGrace
		done := ap.exited
		ap.mu.Unlock()
		if grace <= 0 {
			grace = processPendingReconcileSuccessGrace
		}
		timer := time.NewTimer(grace)
		if done == nil {
			<-timer.C
		} else {
			select {
			case <-timer.C:
			case <-done:
				timer.Stop()
				return
			}
		}
	}
	for {
		ap.mu.Lock()
		pending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
		if !ok || ap.stopping || ap.state == processStateStopped {
			ap.mu.Unlock()
			return
		}
		attempt := pending.ReconcileAttempts + 1
		pending.ReconcileAttempts = attempt
		ap.pendingRuntimeInputs[inputID] = pending
		driver, supportsReconcile := ap.driver.(RuntimeInputReconciler)
		ap.mu.Unlock()

		if attempt > 1 {
			timer := time.NewTimer(pendingReconcileDelay(attempt))
			select {
			case <-timer.C:
			case <-ap.processDone():
				timer.Stop()
				return
			}
		}
		if !ap.runPendingRuntimeInputReconcileAttempt(token, pending, attempt, driver, supportsReconcile) {
			return
		}
	}
}

// runPendingRuntimeInputReconcileAttempt holds the reentrant lifecycle lease
// only while one active reconciler request and its conditional ownership
// mutation are in flight. Stop first cancels the Process admission lifetime,
// then drains this lease; a canceled request therefore retains its prior live
// owner for the replacement Process instead of committing a late phase change.
func (ap *Process) runPendingRuntimeInputReconcileAttempt(
	token pendingRuntimeInputMutationToken,
	pending pendingRuntimeInput,
	attempt int,
	driver RuntimeInputReconciler,
	supportsReconcile bool,
) bool {
	if !ap.beginLifecycleMutation() {
		return false
	}
	defer ap.endLifecycleMutation()

	ap.mu.Lock()
	currentPending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
	stopping := ap.stopping || ap.state == processStateStopped
	ap.mu.Unlock()
	if !ok || stopping {
		return false
	}
	pending = currentPending
	if !supportsReconcile {
		ap.failUncertainRuntimeInput(token, attempt, errors.New("runtime does not support correlated input reconciliation"))
		return false
	}

	admissionCtx, cancelAdmission := ap.contextWithDeliveryAdmissionLifetime(context.Background())
	ctx, cancelTimeout := context.WithTimeout(admissionCtx, processPendingReconcileRequestTimeout)
	result, err := driver.ReconcileRuntimeInput(ctx, token.inputID, pending.SessionID)
	cancelTimeout()
	cancelAdmission()

	ap.mu.Lock()
	stopping = ap.stopping || ap.state == processStateStopped
	ap.mu.Unlock()
	if stopping || ap.deliveryAdmissionLifetimeContext().Err() != nil {
		return false
	}
	if err != nil {
		if ap.pendingReconcileShortBudgetExhausted(attempt) {
			ap.failUncertainRuntimeInput(token, attempt, err)
			return false
		}
		current, _ := ap.persistReconcileProgressIfCurrent(token, reconciliationPersistencePhase(pending), attempt, err)
		return current
	}

	switch result {
	case RuntimeInputReconciliationAcked:
		if pending.Restored || ap.reconciledInputBelongsToPriorSession(pending) {
			ap.finalizeHistoricalRuntimeInputAcked(token)
			return false
		}
		ap.handleRuntimeInputAckedForReconcile(drivercontract.ParsedEvent{
			Kind:           consts.ParsedEventKindRuntimeInputAcked,
			RuntimeInputID: token.inputID,
			SessionID:      pending.SessionID,
		}, token.generation)
		return false
	case RuntimeInputReconciliationAbsent:
		ap.requeueAbsentRuntimeInput(token, attempt)
		return false
	case RuntimeInputReconciliationPending:
		if ap.pendingReconcileElapsedBudgetExhausted(pending) {
			ap.failUncertainRuntimeInput(token, attempt, fmt.Errorf("runtime reconciliation remained %s beyond elapsed budget", result))
			return false
		}
		current, _ := ap.persistReconcileProgressIfCurrent(token, reconciliationPersistencePhase(pending), attempt, nil)
		return current
	case RuntimeInputReconciliationUnknown:
		if ap.pendingReconcileShortBudgetExhausted(attempt) {
			ap.failUncertainRuntimeInput(token, attempt, fmt.Errorf("runtime reconciliation remained %s", result))
			return false
		}
		current, _ := ap.persistReconcileProgressIfCurrent(token, reconciliationPersistencePhase(pending), attempt, nil)
		return current
	default:
		if ap.pendingReconcileShortBudgetExhausted(attempt) {
			ap.failUncertainRuntimeInput(token, attempt, fmt.Errorf("runtime reconciliation returned invalid state %q", result))
			return false
		}
		return true
	}
}

func reconciliationPersistencePhase(pending pendingRuntimeInput) deliveryOwnershipTransition {
	if pending.InflightReason == DeliveryInflightDispatching && !pending.Restored {
		return deliveryOwnershipDispatching
	}
	return deliveryOwnershipOutcomeUnknown
}

func (ap *Process) pendingReconcileShortBudgetExhausted(attempt int) bool {
	ap.mu.Lock()
	maxAttempts := ap.pendingReconcileMaxAttempts
	ap.mu.Unlock()
	if maxAttempts <= 0 {
		maxAttempts = processPendingReconcileMaxAttempts
	}
	return attempt >= maxAttempts
}

func (ap *Process) pendingReconcileElapsedBudgetExhausted(pending pendingRuntimeInput) bool {
	ap.mu.Lock()
	maxElapsed := ap.pendingReconcileMaxElapsed
	ap.mu.Unlock()
	if maxElapsed <= 0 {
		maxElapsed = processPendingReconcilePendingElapsed
	}
	return !pending.CreatedAt.IsZero() && time.Since(pending.CreatedAt) >= maxElapsed
}

func (ap *Process) reconciledInputBelongsToPriorSession(pending pendingRuntimeInput) bool {
	ap.mu.Lock()
	currentSessionID := strings.TrimSpace(ap.SessionID)
	ap.mu.Unlock()
	pendingSessionID := strings.TrimSpace(pending.SessionID)
	return currentSessionID != "" && pendingSessionID != "" && currentSessionID != pendingSessionID
}

func (ap *Process) finalizeHistoricalRuntimeInputAcked(reconcileToken pendingRuntimeInputMutationToken) {
	ap.mu.Lock()
	pending, transitionToken, ok := ap.claimPendingRuntimeInputTransitionLocked(
		reconcileToken.inputID,
		reconcileToken.createdAt,
		&reconcileToken.generation,
	)
	ap.mu.Unlock()
	if !ok {
		return
	}
	committed, err := ap.commitPendingRuntimeInputTransition(
		transitionToken,
		ap.ownershipRecord(pending, deliveryOwnershipAcked, pending.ReconcileAttempts, nil),
		true,
		func(current pendingRuntimeInput) {
			delete(ap.pendingRuntimeInputs, transitionToken.inputID)
			if len(ap.pendingRuntimeInputs) == 0 {
				ap.pendingRuntimeInputs = nil
			}
			ap.rememberRuntimeAckedMessageLocked(strings.TrimSpace(current.Message.MessageID))
			ap.freshness.MarkObserved(current.Message.TargetKey, current.Message.TargetSeq, current.Message.MessageID)
		},
	)
	if err != nil || !committed {
		return
	}
	ap.commitDeliveryPromptCompileMetrics(transitionToken.inputID)
	ap.markResumeSafetyIfQuiescent()

	detail := "runtime input was already acknowledged in the previous runtime session; the current process did not replay it"
	ap.emitRuntimeEvent(RuntimeEvent{
		MessageID:  strings.TrimSpace(pending.Message.MessageID),
		WakeToken:  strings.TrimSpace(pending.WakeToken),
		EventClass: consts.RuntimeEventClassWakeFailed,
		Target:     strings.TrimSpace(pending.Message.Target),
		Detail:     detail,
	})
	ap.scheduleInboxRetry(true)
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
}

func (ap *Process) requeueAbsentRuntimeInput(token pendingRuntimeInputMutationToken, attempts int) bool {
	ap.mu.Lock()
	pending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
	if !ok || ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return true
	}
	record := ap.ownershipRecord(pendingRuntimeInput{
		Message:   pending.Message,
		Input:     drivercontract.RuntimeInput{ID: strings.TrimSpace(pending.Message.MessageID)},
		CreatedAt: pending.CreatedAt,
	}, deliveryOwnershipQueued, attempts, nil)
	err := ap.persistDeliveryOwnershipLocked(record)
	if err == nil {
		ap.discardUnackedPendingDeliveryLocked(pending.WakeToken, pending.Message.MessageID)
		delete(ap.pendingRuntimeInputs, token.inputID)
		if len(ap.pendingRuntimeInputs) == 0 {
			ap.pendingRuntimeInputs = nil
		}
		ap.inbox = requeueInboxMessage(ap.inbox, pending.Message, true)
		ap.rememberUnobservedFrontierHintLocked(pending.Message)
		ap.mu.Unlock()
		ap.discardDeliveryPromptCompileMetrics(token.inputID)
		ap.scheduleInboxRetry(true)
		return true
	}
	ap.mu.Unlock()
	ap.failUncertainRuntimeInput(token, attempts, err)
	return true
}

func (ap *Process) failUncertainRuntimeInput(token pendingRuntimeInputMutationToken, attempts int, cause error) {
	detail := processDeliveryOwnershipFailureDetails
	if cause != nil {
		detail = boundDeliveryOwnershipErrorDetail(detail + ": " + strings.TrimSpace(cause.Error()))
	}
	ap.mu.Lock()
	pending, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
	stopping := ap.stopping || ap.state == processStateStopped
	ap.mu.Unlock()
	if !ok || stopping {
		return
	}
	record := ap.ownershipRecord(pending, deliveryOwnershipFailed, attempts, cause)
	record.FailureCode = "runtime_input_reconciliation_exhausted"
	committed, err := ap.commitPendingRuntimeInputTransition(token, record, true, func(current pendingRuntimeInput) {
		delete(ap.pendingRuntimeInputs, token.inputID)
		if len(ap.pendingRuntimeInputs) == 0 {
			ap.pendingRuntimeInputs = nil
		}
		ap.rememberFailedMessageLocked(strings.TrimSpace(current.Message.MessageID))
	})
	if err != nil || !committed {
		// A newer authoritative generation won, or the Process is stopping.
		// In both cases retain whichever durable owner is now authoritative.
		return
	}
	ap.commitDeliveryPromptCompileMetrics(token.inputID)
	ap.markResumeSafetyIfQuiescent()
	ap.emitRuntimeEvent(RuntimeEvent{
		MessageID:  strings.TrimSpace(pending.Message.MessageID),
		WakeToken:  strings.TrimSpace(pending.WakeToken),
		EventClass: consts.RuntimeEventClassWakeFailed,
		Target:     strings.TrimSpace(pending.Message.Target),
		Detail:     detail,
	})
	ap.scheduleInboxRetry(true)
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
}

func (ap *Process) rememberFailedMessageLocked(messageID string) {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return
	}
	if ap.failedSet == nil {
		ap.failedSet = make(map[string]struct{}, 1)
	}
	if _, exists := ap.failedSet[messageID]; exists {
		return
	}
	ap.failedMessages = append(ap.failedMessages, messageID)
	ap.failedSet[messageID] = struct{}{}
	if len(ap.failedMessages) <= maxTerminalMessages {
		return
	}
	evicted := ap.failedMessages[0]
	ap.failedMessages = ap.failedMessages[1:]
	delete(ap.failedSet, evicted)
}

func isAmbiguousRuntimeDeliveryError(err error) bool {
	if err == nil {
		return false
	}
	// A read-only preflight may fail with the same transport sentinel used for
	// mutating RPCs. Its typed outer boundary is authoritative: no turn/start or
	// turn/steer frame was dispatched, so the durable owner must remain retryable
	// instead of entering correlated uncertainty.
	if errors.Is(err, errRuntimeInputMutationNotAttempted) {
		return false
	}
	if errors.Is(err, context.DeadlineExceeded) {
		return true
	}
	if errors.Is(err, errCodexAppServerTransportAmbiguous) {
		return true
	}
	if errors.Is(err, errKimiWireTransportAmbiguous) {
		return true
	}
	if errors.Is(err, errKimiACPTransportAmbiguous) {
		return true
	}
	if errors.Is(err, errOpenCodePromptTransportAmbiguous) {
		return true
	}
	if errors.Is(err, errPiRPCTransportAmbiguous) {
		return true
	}
	if errors.Is(err, errPiRPCTransportClosed) {
		// sendCommand uses errRuntimeInputMutationNotAttempted for the
		// preflight/no-write case above. A bare transport-closed result means
		// the frame was attempted and may have been consumed before the pipe
		// disappeared, so retain ownership and reconcile instead of replaying.
		return true
	}
	lower := strings.ToLower(strings.TrimSpace(err.Error()))
	return strings.Contains(lower, "deadline exceeded") ||
		strings.Contains(lower, "timeout") ||
		strings.Contains(lower, "timed out") ||
		strings.Contains(lower, "unexpected eof") ||
		strings.Contains(lower, "connection reset") ||
		strings.Contains(lower, "broken pipe")
}