package agent

import (
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"strings"
	"time"

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

type deliveryDecisionKind string

var (
	ErrProcessNotAcceptingDelivery = errors.New("agent process is stopping or stopped")
	// ErrInterruptDeliveryNeedsRestart means the durable message owner was
	// retained, but native runtime mutation did not reach an ACK-safe success
	// boundary. The caller must reject strict handoff or replace the Process.
	ErrInterruptDeliveryNeedsRestart = errors.New("native interrupt-and-deliver did not accept the runtime input")
	errStdinDeliveryOutcomeUncertain = errors.New("stdin delivery write outcome is uncertain")
)

const (
	deliveryDecisionQueue     deliveryDecisionKind = "queue"
	deliveryDecisionProtocol  deliveryDecisionKind = "protocol_direct"
	deliveryDecisionAckedBusy deliveryDecisionKind = "acked_busy_runtime_input"
	deliveryDecisionIdle      deliveryDecisionKind = "idle_stdin"

	// Protocol delivery can block on runtime HTTP/RPC calls. Bound it so a
	// stuck runtime cannot hold the process delivery path indefinitely.
	processProtocolDeliveryTimeout = 120 * time.Second
	// Prompt compilation/rendering is local and should finish quickly; the
	// timeout prevents malformed inputs or template regressions from wedging
	// wake delivery forever.
	processDeliveryPromptTimeout = 30 * time.Second

	processInboxMaxMessages     = 512
	processInboxMaxBytes        = 8 << 20
	processInboxMaxMessageBytes = 512 << 10

	// maxReclaimedCorrelationLeases bounds the in-memory reclaimed inflight
	// correlation entries retained for a late ACK to upgrade a durable
	// failed tombstone. The durable tombstone itself lives in the resume state
	// store; this cap only bounds the in-memory upgrade hint.
	maxReclaimedCorrelationLeases = 64
	// reclaimedCorrelationLeaseTTL bounds the lifetime of a reclaimed correlation
	// lease. A late ACK arriving after the TTL still upgrades the durable
	// tombstone via the store's failed->acked transition.
	reclaimedCorrelationLeaseTTL = 10 * time.Minute

	pendingRuntimeInputDetail = "New message delivered to agent."
	runtimeInputAckedDetail   = "Agent acknowledged message input."
	runtimeCompletedTurnIDKey = "completed_turn_id"
)

type deliveryDecisionInput struct {
	HasProtocolMessageDriver       bool
	SupportsBusyStdinDelivery      bool
	HasAckedBusyRuntimeInputDriver bool
	SupportsStdinNotification      bool
	HasSession                     bool
	IsIdle                         bool
}

type deliveryDecision struct {
	Kind deliveryDecisionKind
}

func decideProcessDelivery(in deliveryDecisionInput) deliveryDecision {
	if in.HasProtocolMessageDriver && in.SupportsStdinNotification && in.HasSession && in.IsIdle {
		return deliveryDecision{Kind: deliveryDecisionProtocol}
	}
	if in.HasAckedBusyRuntimeInputDriver && in.HasSession && !in.IsIdle {
		return deliveryDecision{Kind: deliveryDecisionAckedBusy}
	}
	if in.HasProtocolMessageDriver && in.SupportsBusyStdinDelivery && in.HasSession && !in.IsIdle {
		return deliveryDecision{Kind: deliveryDecisionProtocol}
	}
	if in.SupportsStdinNotification && in.HasSession {
		if in.IsIdle {
			return deliveryDecision{Kind: deliveryDecisionIdle}
		}
	}
	return deliveryDecision{Kind: deliveryDecisionQueue}
}

// Deliver sends one inbound message to the process inbox.
// If the process is idle and the runtime supports stdin delivery, the message
// is written directly to stdin instead of queued.
func (ap *Process) Deliver(msg types.InboxMessage) {
	if err := ap.deliverMessage(msg, false); err != nil && strings.TrimSpace(msg.MessageID) == "" {
		ap.emitRuntimeEvent(RuntimeEvent{
			EventClass: consts.RuntimeEventClassWakeFailed,
			Target:     strings.TrimSpace(msg.Target),
			Detail:     err.Error(),
		})
	}
}

// DeliverForHandoff sends one inbound message and returns an error when the
// message could not be accepted by runtime transport or the daemon-local inbox.
func (ap *Process) DeliverForHandoff(ctx context.Context, msg types.InboxMessage) error {
	if err := ap.acquireDeliveryGate(ctx); err != nil {
		return err
	}
	defer ap.releaseDeliveryGate()
	ap.mu.Lock()
	fromDurableQueue := strings.TrimSpace(ap.deliveryOwnershipHandoffID) != "" &&
		strings.TrimSpace(ap.deliveryOwnershipHandoffID) == strings.TrimSpace(msg.MessageID)
	if fromDurableQueue {
		ap.deliveryOwnershipHandoffID = ""
	}
	ap.mu.Unlock()
	err := ap.deliverMessageWithContextSerialized(ctx, msg, fromDurableQueue)
	if err != nil && fromDurableQueue {
		if errors.Is(err, ErrProcessNotAcceptingDelivery) {
			// The durable queued record remains authoritative for the replacement
			// Process. Never convert an explicit stop boundary into a successful
			// ACK owned only by a Process that is already exiting.
			return err
		}
		// The dispatcher fsynced this owner before launch. If prompt rendering or
		// the pending-phase fsync cannot proceed yet, keep the same durable body
		// in the Process FIFO instead of failing session publication and opening a
		// start/exit race. Returning nil is valid acceptance: the daemon-owned
		// queue, not the runtime transport, owns the exact body.
		ap.mu.Lock()
		if !ap.deliveryMessageAlreadyOwnedLocked(msg) {
			ap.inbox = requeueInboxMessage(ap.inbox, msg, true)
			ap.rememberUnobservedFrontierHintLocked(msg)
		}
		ap.mu.Unlock()
		ap.scheduleInboxRetry(false)
		return nil
	}
	if err == nil && fromDurableQueue {
		ap.scheduleInboxRetry(true)
	}
	return err
}

// InterruptAndDeliver tries to use a runtime-native explicit interrupt path for
// one authoritative inbound message. It returns handled=false when the runtime
// has no such capability so callers can fall back to process restart.
func (ap *Process) InterruptAndDeliver(ctx context.Context, msg types.InboxMessage) (bool, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ap.acquireDeliveryGate(ctx); err != nil {
		return true, err
	}
	defer ap.releaseDeliveryGate()
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return true, ErrProcessNotAcceptingDelivery
	}
	driver, ok := ap.driver.(drivercontract.InterruptingRuntimeInputDriver)
	sessionID := strings.TrimSpace(ap.SessionID)
	if !ok || sessionID == "" {
		ap.mu.Unlock()
		return false, nil
	}
	if strings.TrimSpace(msg.MessageID) == "" {
		ap.mu.Unlock()
		return true, fmt.Errorf("runtime delivery requires a non-empty message id")
	}
	if ap.deliveryMessageAlreadyOwnedLocked(msg) {
		ap.mu.Unlock()
		return true, nil
	}
	if !ap.canAcceptWakeLifecycleOwnerLocked(msg.MessageID) {
		ap.mu.Unlock()
		// Native interrupt delivery is a separate admission entrypoint from the
		// ordinary serialized decision tree. Apply the same wake-lifecycle bound
		// before the runtime can observe the replacement input.
		if err := ap.requeueDeliveryMessage(msg, false); err != nil {
			return true, err
		}
		return true, fmt.Errorf("%w: current wake lifecycle owner cannot be interrupted safely", ErrInterruptDeliveryNeedsRestart)
	}
	if inboxMessageBytes(msg) > processInboxMaxMessageBytes {
		ap.mu.Unlock()
		return true, ap.rejectInboxBackpressure(msg)
	}
	previousWake := ap.wakeSnapshotLocked()
	wakeEvent, hasWakeEvent := ap.beginWakeDeliveryForMessageLocked(msg, "received")
	ap.isIdle = false
	// Enter the handover window before the runtime RPC: a tool call racing
	// inside the interrupt delivery must not read the previous turn's scope.
	scopeHandover := ap.beginSendScopeHandoverOwnerLocked(msg.MessageID)
	turnHandover := ap.beginTurnDeliveryHandoverLocked(msg.MessageID)
	ap.mu.Unlock()
	scopeHandoverCommitted := false
	defer func() {
		if !scopeHandoverCommitted {
			ap.mu.Lock()
			ap.abortSendScopeHandoverLocked(scopeHandover)
			ap.restoreTurnDeliveryLocked(turnHandover)
			ap.mu.Unlock()
		}
	}()
	ap.markResumeSafety(false)
	compiled, renderErr := ap.compileDeliveryPromptWithContext(ctx, msg, "interrupt_direct", wakeEvent.WakeToken)
	if renderErr != nil {
		ap.restoreWakeSnapshot(previousWake)
		return true, renderErr
	}
	deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(ctx)
	input := ap.runtimeInputForDelivery(deliveryCtx, msg, compiled)
	if input.Empty() {
		cancel()
		ap.restoreWakeSnapshot(previousWake)
		return true, fmt.Errorf("interrupt delivery produced empty input")
	}
	if strings.TrimSpace(input.ID) == "" {
		cancel()
		ap.restoreWakeSnapshot(previousWake)
		return true, fmt.Errorf("interrupt delivery requires a runtime input id")
	}
	pending := pendingRuntimeInput{
		Message:        msg,
		Input:          input,
		SessionID:      sessionID,
		WakeToken:      wakeEvent.WakeToken,
		CreatedAt:      time.Now().UTC(),
		InflightReason: DeliveryInflightDispatching,
	}
	if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipDispatching, 0, nil)); err != nil {
		cancel()
		ap.restoreWakeSnapshot(previousWake)
		return true, err
	}
	// The compiler/ownership phase is allowed to finish after retirement has
	// fenced new admissions. Do not issue a runtime mutation to a generation
	// that waitProcess has already marked stopped; the durable dispatching owner
	// is enough for its replacement to reconcile.
	if !ap.AcceptingDeliveries() {
		cancel()
		return true, nil
	}
	ap.rememberPendingRuntimeInput(pending)
	_, correlatedDelivery := ap.driver.(RuntimeInputReconciler)
	var pendingStatusDone chan struct{}
	pendingStatusOK := false
	if correlatedDelivery {
		pendingStatusDone, pendingStatusOK = ap.beginRuntimeInputPendingEvent(input.ID, pending.CreatedAt)
	}
	if !ap.beginLifecycleMutation() {
		ap.forgetPendingRuntimeInput(pending.Input.ID, pending.CreatedAt)
		cancel()
		return true, nil
	}
	deliverErr := driver.DeliverInterruptingRuntimeInput(deliveryCtx, input, sessionID)
	ap.endLifecycleMutation()
	cancel()
	if deliverErr != nil {
		if pendingStatusOK {
			ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
			ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
			if !ap.pendingRuntimeInputExists(input.ID, pending.CreatedAt) {
				return true, nil
			}
		}
		if isAmbiguousRuntimeDeliveryError(deliverErr) {
			ap.mu.Lock()
			ap.abortSendScopeHandoverAmbiguousLocked()
			ap.turnDelivery.poisoned = true
			ap.mu.Unlock()
			scopeHandoverCommitted = true
			pending.InflightReason = DeliveryInflightOutcomeUnknown
			if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipOutcomeUnknown, 0, deliverErr)); err != nil {
				slog.Error("persist ambiguous interrupt ownership failed; retaining durable pending owner",
					"agent_id", ap.AgentID, "message_id", input.ID, "error", err)
			}
			ap.mu.Lock()
			if current, exists := ap.pendingRuntimeInputs[input.ID]; exists && current.CreatedAt.Equal(pending.CreatedAt) {
				current.InflightReason = DeliveryInflightOutcomeUnknown
				ap.pendingRuntimeInputs[input.ID] = current
			}
			ap.mu.Unlock()
			ap.emitRuntimeInputPendingEvent(pending)
			ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
			return true, fmt.Errorf("%w: %w", ErrInterruptDeliveryNeedsRestart, deliverErr)
		}
		if err := ap.persistQueuedDelivery(msg, pending.CreatedAt, 0, deliverErr); err != nil {
			// The pre-RPC pending phase is already durable. Retain it as uncertain
			// rather than returning the message to the server and risking a second
			// replacement turn after an ACK race.
			ap.mu.Lock()
			if current, exists := ap.pendingRuntimeInputs[input.ID]; exists && current.CreatedAt.Equal(pending.CreatedAt) {
				current.InflightReason = DeliveryInflightOutcomeUnknown
				ap.pendingRuntimeInputs[input.ID] = current
			}
			ap.mu.Unlock()
			ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
			return true, fmt.Errorf("%w: interrupt failed: %v; retain pending ownership: %w", ErrInterruptDeliveryNeedsRestart, deliverErr, err)
		}
		ap.discardDeliveryPromptCompileMetrics(input.ID)
		ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
		ap.restoreWakeSnapshot(previousWake)
		if err := ap.requeueDeliveryMessageAfterFailure(msg, false, deliverErr); err != nil {
			return true, err
		}
		return true, fmt.Errorf("%w: %w", ErrInterruptDeliveryNeedsRestart, deliverErr)
	}
	if correlatedDelivery {
		// The interrupt starts the next authoritative request without
		// advancing the surface epoch: retire the previous turn's scope and
		// restamp this pending to the new generation so its exact delivered ack
		// records against the live turn.
		ap.mu.Lock()
		// Correlated runtimes keep the handover window unavailable until their
		// exact acknowledgement commits the replacement scope.
		if current, exists := ap.pendingRuntimeInputs[input.ID]; exists && current.CreatedAt.Equal(pending.CreatedAt) {
			current.sendScopeGen = ap.sendScope.generation
			ap.pendingRuntimeInputs[input.ID] = current
		}
		ap.mu.Unlock()
		scopeHandoverCommitted = true
		ap.markRuntimeHandoffSucceeded()
		if pendingStatusOK {
			ap.emitRuntimeInputPendingEvent(pending)
			ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
			ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
		}
		if _, ok := ap.driver.(RuntimeInputReconciler); ok && ap.hasDeliveryOwnershipStore() {
			ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, false)
		}
		return true, nil
	}
	// The interrupt starts the next authoritative request without advancing
	// the ack epoch: retire the previous turn's scope and, in the same
	// critical section, record the new body's targets so no handover-empty
	// window can be read as a proactive allowance.
	ap.mu.Lock()
	if ap.sendScope.handover != nil && ap.sendScope.handover.mode == "replace" {
		ap.commitSendScopeHandoverLocked(msg)
	}
	ap.commitTurnDeliveryAckLocked(msg.MessageID, msg.Target)
	ap.mu.Unlock()
	scopeHandoverCommitted = true
	// The uncorrelated transport may already be issuing its first Exec call.
	// Publish the pending authority before any terminal ownership fsync can fail
	// or block. Correlated runtimes publish only from their acknowledgement.
	ap.PrepareRuntimeMessageDelivery(msg, RuntimeDeliveryPathProtocol)
	ap.markRuntimeHandoffSucceeded()
	ap.commitDeliveryPromptCompileMetrics(input.ID)
	if err := ap.persistTerminalDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipAcked, 0, nil)); err != nil {
		return true, err
	}
	ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
	if hasWakeEvent {
		ap.emitWakeStatus(wakeEvent)
	}
	ap.RememberRuntimeAckedMessage(strings.TrimSpace(msg.MessageID))
	ap.emitParsedEvent(newPromptInputParsedEvent(
		input.PromptText(),
		"delivery_text",
		ap.driver.Name(),
		deliveryPromptEventType(msg),
		sessionID,
	))
	ap.RememberObservedFrontier(msg)
	ap.markResumeSafetyIfQuiescent()
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
	return true, nil
}

// InterruptActiveRun interrupts the current runtime turn/run without delivering
// a replacement input. It returns handled=false when the runtime has no native
// keep-session interrupt capability.
func (ap *Process) InterruptActiveRun(ctx context.Context) (bool, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	// Interrupt is the precondition for a busy live/detached restore. Serialize
	// it with ordinary delivery so the runtime cannot be interrupted concurrently
	// with a new handoff before RestoreRecoveryPoint performs its ownership proof.
	if err := ap.acquireDeliveryGate(ctx); err != nil {
		return true, err
	}
	defer ap.releaseDeliveryGate()
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return true, ErrProcessNotAcceptingDelivery
	}
	driver, ok := ap.driver.(drivercontract.ActiveRunInterruptDriver)
	sessionID := strings.TrimSpace(ap.SessionID)
	ap.mu.Unlock()
	if !ok {
		return false, nil
	}
	if sessionID == "" {
		sessionless, supportsSessionless := driver.(interface {
			SupportsSessionlessActiveRunInterrupt() bool
		})
		if !supportsSessionless || !sessionless.SupportsSessionlessActiveRunInterrupt() {
			return false, nil
		}
	}

	if err := ap.markResumeSafetyUnsafe(); err != nil {
		return true, err
	}

	interruptCtx, cancel := ap.newProtocolDeliveryContextFor(ctx)
	if !ap.beginLifecycleMutation() {
		cancel()
		return true, ErrProcessNotAcceptingDelivery
	}
	err := driver.InterruptActiveRun(interruptCtx, sessionID)
	ap.endLifecycleMutation()
	cancel()
	if err != nil {
		return true, err
	}

	ap.mu.Lock()
	ap.isIdle = true
	// A successful interrupt ends the interrupted turn's authoritative
	// request: retire its send scope immediately (and fence late acks via the
	// generation bump) instead of waiting for a TurnEnd/Idle event the
	// interrupted runtime may never emit.
	ap.resetSendScopeLocked()
	ap.resetTurnDeliveryLocked()
	ap.mu.Unlock()
	ap.markResumeSafetyIfQuiescent()
	return true, nil
}

func (ap *Process) deliverMessage(msg types.InboxMessage, fromQueue bool) error {
	ctx := context.Background()
	return ap.deliverMessageWithContext(ctx, msg, fromQueue)
}

func (ap *Process) deliverMessageWithContext(ctx context.Context, msg types.InboxMessage, fromQueue bool) error {
	if err := ap.acquireDeliveryGate(ctx); err != nil {
		return err
	}
	defer ap.releaseDeliveryGate()
	return ap.deliverMessageWithContextSerialized(ctx, msg, fromQueue)
}

// deliverMessageWithContextSerialized performs one ordered admission while
// deliveryGate is held. Internal ordered-dequeue recursion must call this helper
// directly rather than reacquiring the non-reentrant mutex.
func (ap *Process) deliverMessageWithContextSerialized(ctx context.Context, msg types.InboxMessage, fromQueue bool) error {
	if ctx == nil {
		ctx = context.Background()
	}
	if strings.TrimSpace(msg.MessageID) == "" {
		return fmt.Errorf("runtime delivery requires a non-empty message id")
	}
	ap.mu.Lock()
	if ap.stopping || ap.state == processStateStopped {
		ap.mu.Unlock()
		return ErrProcessNotAcceptingDelivery
	}
	if ap.deliveryMessageAlreadyOwnedLocked(msg) {
		ap.mu.Unlock()
		return nil
	}
	if stale, err := ap.durablySuppressStaleDeliveryLocked(msg, "already_observed_before_dispatch"); stale {
		ap.mu.Unlock()
		return err
	}
	// The retry worker may have passed its admission check just before a
	// replacement restored q1. Re-check at the concrete dispatch boundary so a
	// queued q2 can never overtake a restored/ambiguous Pi owner during that
	// scheduling race. Fresh foreground deliveries retain their existing lane;
	// only daemon-owned FIFO retries are fenced here.
	if fromQueue && ap.hasRestoredBlockingRuntimeInputLocked() {
		ap.mu.Unlock()
		return ap.requeueDeliveryMessage(msg, fromQueue)
	}
	if !ap.canAcceptWakeLifecycleOwnerLocked(msg.MessageID) {
		ap.mu.Unlock()
		// Wake lifecycle capacity is a pre-runtime admission boundary. Keep the
		// exact body in the durable daemon FIFO; never let the runtime observe an
		// input that cannot receive its own correlated reply lifecycle.
		return ap.requeueDeliveryMessage(msg, fromQueue)
	}
	// A transient retry still owns the runtime handoff slot for the failed
	// message. Do not let a newly routed body observe the driver's temporary
	// idle snapshot and start/steer ahead of the rollback + redelivery. The
	// retry helper deliberately bypasses this ordinary-delivery entrypoint and
	// holds deliveryGate around the complete recovery operation.
	if ap.transientRuntimeRetryPending {
		ap.mu.Unlock()
		return ap.requeueDeliveryMessage(msg, fromQueue)
	}
	// Reconciliation is a background ownership concern, not a foreground
	// admission gate. A historical outcome_unknown input may never become
	// knowable, but it must not stop an independently identified new message
	// from reaching a runtime that can accept it. The old owner remains
	// non-replayable and continues its own bounded reconciliation.
	wasIdleAtDispatch := ap.isIdle
	protocolDriver, hasProtocolDriver := ap.directMessageDriverLocked()
	ackedDriver, hasAckedDriver := ap.ackedBusyRuntimeInputDriverLocked()
	supportsStdinNotification := ap.driver != nil && ap.driver.SupportsStdinNotification()
	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:      supportsStdinNotification,
		HasSession:                     strings.TrimSpace(ap.SessionID) != "",
		IsIdle:                         ap.isIdle,
	})
	// Fresh routed work is a foreground lane. The inbox is recovery/backlog
	// work and therefore cannot force a newly admitted message behind an old
	// poison head. If the runtime can accept the fresh message now, deliver it
	// now; the bounded retry worker resumes backlog afterward. A runtime that is
	// intrinsically unable to accept input in its current state still queues the
	// fresh body through the ordinary deliveryDecisionQueue path below.

	switch decision.Kind {
	case deliveryDecisionAckedBusy:
		// Active-turn acked delivery uses the runtime's correlated pending-input
		// model: transport success only means the runtime accepted the message
		// into its own pending queue. AgentTeam keeps freshness held and waits for the
		// runtime to echo RuntimeInput.ID back as runtime_input_acked before
		// releasing ownership or advancing the observed frontier.
		sessionID := ap.SessionID
		wakeToken := generateWakeToken()
		ap.isIdle = false
		ap.rememberUnobservedFrontierHintLocked(msg)
		ap.mu.Unlock()
		ap.markResumeSafety(false)
		compiled, renderErr := ap.compileDeliveryPromptWithContext(ctx, msg, "turn_direct", wakeToken)
		if renderErr != nil {
			slog.Debug("acked active delivery compile failed, requeuing",
				"agent_id", ap.AgentID, "error", renderErr)
			return ap.requeueAckedDeliveryMessage(msg, fromQueue)
		}
		deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(ctx)
		input := ap.runtimeInputForDelivery(deliveryCtx, msg, compiled)
		if input.Empty() {
			cancel()
			slog.Debug("acked active delivery produced empty input, requeuing",
				"agent_id", ap.AgentID)
			return ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, errors.New("acked active delivery produced empty input"))
		}
		if strings.TrimSpace(input.ID) == "" {
			cancel()
			slog.Debug("acked active delivery missing runtime input id, requeuing",
				"agent_id", ap.AgentID)
			return ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, errors.New("acked active delivery missing runtime input id"))
		}
		pending := pendingRuntimeInput{
			Message:        msg,
			Input:          input,
			SessionID:      sessionID,
			WakeToken:      wakeToken,
			CreatedAt:      time.Now().UTC(),
			InflightReason: DeliveryInflightDispatching,
		}
		if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipDispatching, 0, nil)); err != nil {
			cancel()
			return err
		}
		// Keep the accepted owner durable, but never send it to a runtime whose
		// Process generation has already been fenced by Stop/Sleep or waitProcess.
		if !ap.AcceptingDeliveries() {
			cancel()
			return nil
		}
		ap.rememberTransientRuntimeRetryInput(input.ID, input, sessionID)
		ap.rememberPendingRuntimeInput(pending)
		pendingStatusDone, pendingStatusOK := ap.beginRuntimeInputPendingEvent(input.ID, pending.CreatedAt)
		// AckedBusyRuntimeInputDriver's contract makes an accepted RPC
		// pending-only for EVERY driver (Claude echoes UUIDs without
		// implementing the reconciler interface), so scope waits for the
		// exact acknowledgement in all cases.
		ap.mu.Lock()
		ap.beginSendScopeAppendLocked()
		ap.beginTurnDeliveryAppendLocked("")
		ap.mu.Unlock()
		if !ap.beginLifecycleMutation() {
			cancel()
			return nil
		}
		deliverErr := ackedDriver.DeliverAckedBusyRuntimeInput(deliveryCtx, input, sessionID)
		ap.endLifecycleMutation()
		cancel()
		ap.mu.Lock()
		switch {
		case deliverErr != nil:
			ap.abortSendScopeAppendLocked()
			ap.turnDelivery.handover = nil
		default:
			// Same-turn busy delivery keeps the existing targets; the
			// acknowledged target joins only at the exact ack.
			ap.abortSendScopeAppendLocked()
			ap.turnDelivery.handover = nil
		}
		ap.mu.Unlock()
		if deliverErr != nil {
			if pendingStatusOK {
				ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
				ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
				if !ap.pendingRuntimeInputExists(input.ID, pending.CreatedAt) {
					return nil
				}
			}
			if isAmbiguousRuntimeDeliveryError(deliverErr) {
				pending.InflightReason = DeliveryInflightOutcomeUnknown
				if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipOutcomeUnknown, 0, deliverErr)); err != nil {
					slog.Error("persist ambiguous acked ownership failed; retaining durable pending owner",
						"agent_id", ap.AgentID, "message_id", input.ID, "error", err)
				}
				ap.mu.Lock()
				if current, ok := ap.pendingRuntimeInputs[input.ID]; ok && current.CreatedAt.Equal(pending.CreatedAt) {
					current.InflightReason = DeliveryInflightOutcomeUnknown
					ap.pendingRuntimeInputs[input.ID] = current
				}
				ap.mu.Unlock()
				ap.emitRuntimeInputPendingEvent(pending)
				if _, ok := ap.driver.(RuntimeInputReconciler); ok {
					ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
				}
				// A driver without an authoritative read reconciler still must
				// not replay a mutation whose request may have reached the
				// runtime. Keep the exact owner fail-closed for a late correlated
				// ACK; capacity reclaim or process replacement bounds its
				// lifetime without manufacturing a duplicate input.
				return nil
			}
			if err := ap.persistQueuedDelivery(msg, pending.CreatedAt, 0, deliverErr); err != nil {
				pending.InflightReason = DeliveryInflightOutcomeUnknown
				ap.mu.Lock()
				if current, ok := ap.pendingRuntimeInputs[input.ID]; ok && current.CreatedAt.Equal(pending.CreatedAt) {
					current.InflightReason = DeliveryInflightOutcomeUnknown
					ap.pendingRuntimeInputs[input.ID] = current
				}
				ap.mu.Unlock()
				ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
				return nil
			}
			ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
			ap.discardDeliveryPromptCompileMetrics(input.ID)
			slog.Debug("acked active delivery failed, requeuing",
				"agent_id", ap.AgentID, "error", deliverErr)
			return ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, deliverErr)
		}
		ap.markRuntimeHandoffSucceeded()
		if pendingStatusOK {
			ap.emitRuntimeInputPendingEvent(pending)
			ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
			ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
		}
		// A lost/delayed userMessage echo must not leave accepted ownership in
		// pending forever. Reconcile after a grace window unless the runtime's
		// successful active-turn contract owns the input until its correlated
		// echo. An echo removes the record first and makes the scheduled worker a
		// no-op. Ambiguous and restored ownership still use immediate reconciliation.
		if policy, ok := ap.driver.(SuccessfulAckedBusyRuntimeInputReconciliationPolicy); ok &&
			!policy.ReconcileSuccessfulAckedBusyRuntimeInput() {
			return nil
		}
		if _, ok := ap.driver.(RuntimeInputReconciler); ok && ap.hasDeliveryOwnershipStore() {
			ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, false)
		}
		return nil

	case deliveryDecisionProtocol:
		// For runtimes with a protocol-level direct delivery implementation,
		// prefer the accept/reject capability path while idle / starting a new
		// turn, or while active only when the driver explicitly reports that
		// real routed-message input can be accepted by the busy turn.
		sessionID := ap.SessionID
		attemptSnapshot := ap.deliveryAttemptSnapshotLocked()
		wakeEvent, hasWakeEvent := ap.beginWakeDeliveryForMessageLocked(msg, "received")

		ap.isIdle = false
		ap.mu.Unlock()
		ap.markResumeSafety(false)
		compiled, renderErr := ap.compileDeliveryPromptWithContext(ctx, msg, "turn_direct", wakeEvent.WakeToken)
		if renderErr != nil {
			ap.restoreDeliveryAttempt(attemptSnapshot)
			slog.Debug("direct delivery compile failed, requeuing",
				"agent_id", ap.AgentID, "error", renderErr)
		}
		if renderErr == nil {
			deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(ctx)
			input := ap.runtimeInputForDelivery(deliveryCtx, msg, compiled)
			if input.Empty() {
				cancel()
				ap.restoreDeliveryAttempt(attemptSnapshot)
				slog.Debug("direct delivery produced empty input, requeuing",
					"agent_id", ap.AgentID)
				requeueErr := ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, errors.New("direct delivery produced empty input"))
				ap.markResumeSafetyIfQuiescent()
				return requeueErr
			}
			pending := pendingRuntimeInput{
				Message:        msg,
				Input:          input,
				SessionID:      sessionID,
				WakeToken:      wakeEvent.WakeToken,
				CreatedAt:      time.Now().UTC(),
				InflightReason: DeliveryInflightDispatching,
			}
			if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipDispatching, 0, nil)); err != nil {
				cancel()
				ap.restoreDeliveryAttempt(attemptSnapshot)
				return err
			}
			// An admitted delivery may persist its first owner after the child
			// exits, but it must not mutate that retired runtime generation.
			if !ap.AcceptingDeliveries() {
				cancel()
				return nil
			}
			ap.rememberPendingRuntimeInput(pending)
			_, correlatedDelivery := ap.driver.(RuntimeInputReconciler)
			if capability, ok := ap.driver.(drivercontract.CorrelatedRuntimeInputDeliveryDriver); ok {
				correlatedDelivery = correlatedDelivery || capability.SupportsCorrelatedRuntimeInputDelivery()
			}
			var pendingStatusDone chan struct{}
			pendingStatusOK := false
			if correlatedDelivery {
				pendingStatusDone, pendingStatusOK = ap.beginRuntimeInputPendingEvent(input.ID, pending.CreatedAt)
			}
			rollbackFrontier := func() {}
			if !correlatedDelivery {
				rollbackFrontier = ap.markObservedFrontierForRuntimeHandoff(msg)
			}
			ap.mu.Lock()
			var idleSendHandoverSaved savedSendScope
			var idleTurnHandoverSaved savedTurnDelivery
			if wasIdleAtDispatch {
				// Idle protocol delivery starts a NEW authoritative request:
				// replace handover owned by this input until its ack.
				idleSendHandoverSaved = ap.beginSendScopeHandoverOwnerLocked(msg.MessageID)
				idleTurnHandoverSaved = ap.beginTurnDeliveryHandoverLocked(msg.MessageID)
			} else {
				// Busy protocol delivery is SAME-TURN: keep the existing
				// targets and append at ack time.
				ap.beginSendScopeAppendOwnerLocked(msg.MessageID)
				ap.beginTurnDeliveryAppendLocked(msg.MessageID)
			}
			ap.mu.Unlock()
			if !ap.beginLifecycleMutation() {
				cancel()
				return nil
			}
			deliverErr := ap.deliverProtocolRuntimeInput(deliveryCtx, protocolDriver, input, sessionID)
			ap.endLifecycleMutation()
			cancel()
			ap.mu.Lock()
			switch {
			case deliverErr != nil && wasIdleAtDispatch && isAmbiguousRuntimeDeliveryError(deliverErr):
				// The runtime may have accepted the replacement without an ack.
				// Keep both authority views unavailable until later evidence
				// resolves the handover; restoring the old turn would be unsafe.
				ap.abortSendScopeHandoverAmbiguousLocked()
			case deliverErr != nil && wasIdleAtDispatch:
				// A failed idle replace rolls the previous turn's turn-delivery
				// set back; restore is a no-op once an in-RPC ack committed it.
				ap.abortSendScopeHandoverLocked(idleSendHandoverSaved)
				ap.restoreTurnDeliveryLocked(idleTurnHandoverSaved)
			case deliverErr != nil:
				ap.abortSendScopeAppendLocked()
				ap.turnDelivery.handover = nil
			case !correlatedDelivery:
				ap.commitSendScopeHandoverLocked(msg)
				ap.commitTurnDeliveryAckLocked(msg.MessageID, msg.Target)
			}
			ap.mu.Unlock()
			if deliverErr == nil {
				if correlatedDelivery {
					ap.markRuntimeHandoffSucceeded()
					if pendingStatusOK {
						ap.emitRuntimeInputPendingEvent(pending)
						ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
						ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
					}
					if _, ok := ap.driver.(RuntimeInputReconciler); ok && ap.hasDeliveryOwnershipStore() {
						ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, false)
					}
					return nil
				}
				ap.commitDeliveryPromptCompileMetrics(input.ID)
				// The uncorrelated transport has definitively accepted the body. Publish
				// the exact pending callback context before ownership persistence, since
				// the child can call Exec before this RPC returns.
				ap.PrepareRuntimeMessageDelivery(msg, RuntimeDeliveryPathProtocol)
				if err := ap.persistTerminalDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipAcked, 0, nil)); err != nil {
					return err
				}
				ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
				ap.markRuntimeHandoffSucceeded()
				ap.rememberTransientRuntimeRetryInput(msg.MessageID, input, sessionID)
				if hasWakeEvent {
					ap.emitWakeStatus(wakeEvent)
				}
				ap.RememberRuntimeAckedMessage(strings.TrimSpace(msg.MessageID))
				ap.emitParsedEvent(newPromptInputParsedEvent(
					input.PromptText(),
					"delivery_text",
					ap.driver.Name(),
					deliveryPromptEventType(msg),
					sessionID,
				))
				ap.RememberObservedFrontier(msg)
				ap.markResumeSafetyIfQuiescent()
				ap.scheduleOldestPendingRuntimeInputReconcile(true)
				return nil
			}
			if pendingStatusOK {
				ap.finishRuntimeInputPendingEvent(input.ID, pending.CreatedAt, pendingStatusDone)
				ap.flushRuntimeInputAckedEvent(input.ID, pending.CreatedAt)
				if !ap.pendingRuntimeInputExists(input.ID, pending.CreatedAt) {
					return nil
				}
			}
			rollbackFrontier()
			if isAmbiguousRuntimeDeliveryError(deliverErr) {
				pending.InflightReason = DeliveryInflightOutcomeUnknown
				if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipOutcomeUnknown, 0, deliverErr)); err != nil {
					slog.Error("persist ambiguous direct ownership failed; retaining durable pending owner",
						"agent_id", ap.AgentID, "message_id", input.ID, "error", err)
				}
				ap.mu.Lock()
				if current, ok := ap.pendingRuntimeInputs[input.ID]; ok && current.CreatedAt.Equal(pending.CreatedAt) {
					current.InflightReason = DeliveryInflightOutcomeUnknown
					ap.pendingRuntimeInputs[input.ID] = current
				}
				ap.mu.Unlock()
				ap.emitRuntimeInputPendingEvent(pending)
				if _, ok := ap.driver.(RuntimeInputReconciler); ok {
					ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
				}
				// A response-less mutating request may already have reached an
				// ACK-only runtime. Keep the exact owner fail-closed for its late
				// streamed acknowledgement; never put the body back in the replay
				// queue merely because read-only reconciliation is unavailable.
				return nil
			}
			ap.discardDeliveryPromptCompileMetrics(input.ID)
			if err := ap.persistQueuedDelivery(msg, pending.CreatedAt, 0, deliverErr); err != nil {
				ap.mu.Lock()
				if current, ok := ap.pendingRuntimeInputs[input.ID]; ok && current.CreatedAt.Equal(pending.CreatedAt) {
					current.InflightReason = DeliveryInflightOutcomeUnknown
					ap.pendingRuntimeInputs[input.ID] = current
				}
				ap.mu.Unlock()
				ap.schedulePendingRuntimeInputReconcile(input.ID, pending.CreatedAt, true)
				return nil
			}
			ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
			ap.restoreDeliveryAttempt(attemptSnapshot)
			slog.Debug("direct delivery failed, requeuing",
				"agent_id", ap.AgentID, "error", deliverErr)
			requeueErr := ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, deliverErr)
			ap.markResumeSafetyIfQuiescent()
			return requeueErr
		}

		// Delivery failed or text was empty — requeue.
		requeueErr := ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, renderErr)
		ap.markResumeSafetyIfQuiescent()
		return requeueErr

	// If the process is idle and supports stdin delivery, deliver directly.
	case deliveryDecisionIdle:
		attemptSnapshot := ap.deliveryAttemptSnapshotLocked()
		wakeEvent, hasWakeEvent := ap.beginWakeDeliveryForMessageLocked(msg, "received")
		ap.isIdle = false
		ap.mu.Unlock()

		ap.markResumeSafety(false)
		if err := ap.persistQueuedDelivery(msg, time.Time{}, 0, nil); err != nil {
			ap.restoreDeliveryAttempt(attemptSnapshot)
			return err
		}
		ap.prepareRuntimeMessageDeliveryRecovery(msg, RuntimeDeliveryPathStdin)
		rollbackFrontier := ap.markObservedFrontierForRuntimeHandoff(msg)
		ap.mu.Lock()
		stdinScopeHandover := ap.beginSendScopeHandoverLocked()
		stdinTurnHandover := ap.beginTurnDeliveryHandoverLocked("")
		ap.mu.Unlock()
		stdinPending, err := ap.deliverMessageViaStdinWithContext(ctx, msg, "stdin_delivery", wakeEvent.WakeToken)
		ap.mu.Lock()
		switch {
		case err != nil:
			ap.abortSendScopeHandoverLocked(stdinScopeHandover)
			ap.restoreTurnDeliveryLocked(stdinTurnHandover)
		case ap.sendScope.handover != nil:
			ap.commitSendScopeHandoverLocked(msg)
			ap.commitTurnDeliveryAckLocked(msg.MessageID, msg.Target)
		}
		ap.mu.Unlock()
		if err != nil {
			rollbackFrontier()
			if errors.Is(err, errStdinDeliveryOutcomeUncertain) {
				ap.notifyRuntimeMessageDeliveryForTurn(msg, nil)
				// Some bytes may already belong to the runtime, so restoring idle or
				// putting the exact body back in the daemon FIFO could duplicate user
				// input. Raw stdin has no correlated reconciliation capability, so
				// fail-close the owner and retire the now-corrupted byte stream as an
				// unexpected exit. A replacement Process can then continue newer
				// owners without ever replaying this body.
				ap.retireAmbiguousStdinTransport(stdinPending, err)
				return nil
			}
			ap.restoreDeliveryAttempt(attemptSnapshot)
			slog.Debug("stdin delivery compile failed, re-queued message", "agent_id", ap.AgentID, "error", err)
			requeueErr := ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, err)
			ap.markResumeSafetyIfQuiescent()
			return requeueErr
		}
		ap.notifyRuntimeMessageDeliveryForTurn(msg, nil)
		if err := ap.persistTerminalDeliveryOwnership(ap.ownershipRecord(stdinPending, deliveryOwnershipAcked, 0, nil)); err != nil {
			return err
		}
		ap.forgetPendingRuntimeInput(stdinPending.Input.ID, stdinPending.CreatedAt)
		ap.commitDeliveryPromptCompileMetrics(stdinPending.Input.ID)
		ap.markRuntimeHandoffSucceeded()
		if hasWakeEvent {
			ap.emitWakeStatus(wakeEvent)
		}
		ap.RememberRuntimeAckedMessage(strings.TrimSpace(msg.MessageID))
		ap.RememberObservedFrontier(msg)
		ap.markResumeSafetyIfQuiescent()
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
		return nil

	}

	if inboxMessageBytes(msg) > processInboxMaxMessageBytes {
		ap.mu.Unlock()
		return ap.rejectInboxBackpressure(msg)
	}
	ap.mu.Unlock()
	return ap.requeueDeliveryMessage(msg, fromQueue)
}

func (ap *Process) acquireDeliveryGate(ctx context.Context) error {
	callerCtx := ctx
	admissionCtx, cancel := ap.contextWithDeliveryAdmissionLifetime(ctx)
	defer cancel()
	err := ap.acquireDeliveryGateForDrain(admissionCtx)
	if err != nil && ap.deliveryAdmissionLifetimeContext().Err() != nil && (callerCtx == nil || callerCtx.Err() == nil) {
		return ErrProcessNotAcceptingDelivery
	}
	return err
}

func (ap *Process) acquireDeliveryGateForDrain(ctx context.Context) error {
	if ctx == nil {
		ctx = context.Background()
	}
	ap.deliveryGateOnce.Do(func() {
		ap.deliveryGate = make(chan struct{}, 1)
		ap.deliveryGate <- struct{}{}
	})
	select {
	case <-ap.deliveryGate:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}

func (ap *Process) deliveryAdmissionLifetimeContext() context.Context {
	ap.deliveryAdmissionLifetimeOnce.Do(func() {
		ap.deliveryAdmissionLifetime, ap.deliveryAdmissionLifetimeCancel = context.WithCancel(context.Background())
	})
	return ap.deliveryAdmissionLifetime
}

func (ap *Process) cancelDeliveryAdmissionLifetime() {
	ap.deliveryAdmissionLifetimeContext()
	if ap.deliveryAdmissionLifetimeCancel != nil {
		ap.deliveryAdmissionLifetimeCancel()
	}
}

// contextWithDeliveryAdmissionLifetime merges a caller's cancellation with the
// Process retirement lifetime without a goroutine-per-RPC waiter. context.AfterFunc
// registers directly on cancelCtx and the returned cleanup unregisters it when
// the bounded operation finishes normally.
func (ap *Process) contextWithDeliveryAdmissionLifetime(parent context.Context) (context.Context, context.CancelFunc) {
	if parent == nil {
		parent = context.Background()
	}
	ctx, cancel := context.WithCancel(parent)
	stop := context.AfterFunc(ap.deliveryAdmissionLifetimeContext(), cancel)
	return ctx, func() {
		stop()
		cancel()
	}
}

func (ap *Process) releaseDeliveryGate() {
	ap.deliveryGate <- struct{}{}
}

// markRuntimeHandoffSucceeded advances the generation observed by queued
// runtime-boundary events. A TurnEnd/RuntimeIdle that was emitted for the
// preceding turn must not mark this newly accepted start/steer idle.
func (ap *Process) markRuntimeHandoffSucceeded() {
	ap.mu.Lock()
	ap.runtimeHandoffEpoch++
	ap.isIdle = false
	ap.mu.Unlock()
	// A successful transport handoff releases the FIFO head. If newer daemon-
	// owned messages are waiting, try the next one without waiting for a runtime
	// boundary. Acked inputs remain runtime-owned in pendingRuntimeInputs; this
	// kick never moves them back into the inbox.
	ap.scheduleInboxRetry(true)
}

func (ap *Process) currentRuntimeHandoffEpoch() uint64 {
	ap.mu.Lock()
	defer ap.mu.Unlock()
	return ap.runtimeHandoffEpoch
}

func (ap *Process) runtimeBoundarySuperseded(entryEpoch uint64, ev drivercontract.ParsedEvent) bool {
	activeTurnID := ""
	if provider, ok := ap.driver.(drivercontract.ActiveTurnIDProvider); ok {
		activeTurnID = strings.TrimSpace(provider.CurrentActiveTurnID())
	}
	ap.mu.Lock()
	epochChanged := ap.runtimeHandoffEpoch != entryEpoch
	ap.mu.Unlock()
	if epochChanged {
		return true
	}

	completedTurnID := ""
	if len(ev.Metadata) > 0 {
		completedTurnID = strings.TrimSpace(fmt.Sprint(ev.Metadata[runtimeCompletedTurnIDKey]))
	}
	if ev.Kind == consts.ParsedEventKindTurnEnd && completedTurnID != "" && activeTurnID != "" && activeTurnID != completedTurnID {
		return true
	}
	if ev.Kind == consts.ParsedEventKindRuntimeIdle && activeTurnID != "" {
		return true
	}
	return false
}

func newProtocolDeliveryContextFor(parent context.Context) (context.Context, context.CancelFunc) {
	if parent == nil {
		parent = context.Background()
	}
	return context.WithTimeout(parent, processProtocolDeliveryTimeout)
}

func (ap *Process) newProtocolDeliveryContextFor(parent context.Context) (context.Context, context.CancelFunc) {
	admissionCtx, cancelAdmission := ap.contextWithDeliveryAdmissionLifetime(parent)
	ctx, cancelTimeout := newProtocolDeliveryContextFor(admissionCtx)
	return ctx, func() {
		cancelTimeout()
		cancelAdmission()
	}
}

func newDeliveryPromptContext() (context.Context, context.CancelFunc) {
	return newDeliveryPromptContextFor(context.Background())
}

func newDeliveryPromptContextFor(parent context.Context) (context.Context, context.CancelFunc) {
	if parent == nil {
		parent = context.Background()
	}
	return context.WithTimeout(parent, processDeliveryPromptTimeout)
}

func (ap *Process) requeueDeliveryMessage(msg types.InboxMessage, fromQueue bool) error {
	return ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, nil)
}

func (ap *Process) requeueDeliveryMessageAfterFailure(msg types.InboxMessage, fromQueue bool, deliveryErr error) error {
	ap.discardDeliveryPromptCompileMetrics(strings.TrimSpace(msg.MessageID))
	ap.mu.Lock()
	if ap.deliveryMessageAlreadyOwnedLocked(msg) {
		ap.mu.Unlock()
		return nil
	}
	if stale, err := ap.durablySuppressStaleDeliveryLocked(msg, "already_observed_before_requeue"); stale {
		ap.mu.Unlock()
		return err
	}
	if !ap.canQueueInboxMessageLocked(msg) {
		ap.mu.Unlock()
		return ap.rejectInboxBackpressure(msg)
	}
	ap.mu.Unlock()

	if err := ap.persistQueuedDelivery(msg, time.Time{}, 0, deliveryErr); err != nil {
		ap.emitRuntimeEvent(RuntimeEvent{
			MessageID:  strings.TrimSpace(msg.MessageID),
			EventClass: consts.RuntimeEventClassWakeFailed,
			Target:     strings.TrimSpace(msg.Target),
			Detail:     err.Error(),
		})
		return err
	}
	ap.mu.Lock()
	if ap.deliveryMessageAlreadyOwnedLocked(msg) {
		ap.mu.Unlock()
		return nil
	}
	if stale, err := ap.durablySuppressStaleDeliveryLocked(msg, "already_observed_before_requeue"); stale {
		ap.mu.Unlock()
		return err
	}
	if !ap.canQueueInboxMessageLocked(msg) {
		ap.mu.Unlock()
		_ = ap.deleteDeliveryOwnershipLocked(strings.TrimSpace(msg.MessageID))
		return ap.rejectInboxBackpressure(msg)
	}
	ap.inbox = requeueInboxMessage(ap.inbox, msg, fromQueue)
	if deliveryErr == nil {
		ap.rememberUnobservedFrontierHintLocked(msg)
	}
	ap.noteInboxRetryFailureLocked(msg, deliveryErr)
	ap.mu.Unlock()
	ap.scheduleInboxRetry(false)
	return nil
}

// requeueAckedDeliveryMessage preserves the original body after an acked
// transport rejection. The common daemon retry worker, rather than a generic
// queue notification, is responsible for making progress without a later
// runtime boundary.
func (ap *Process) requeueAckedDeliveryMessage(msg types.InboxMessage, fromQueue bool) error {
	return ap.requeueDeliveryMessageAfterFailure(msg, fromQueue, nil)
}

// deliveryMessageAlreadyOwnedLocked reports whether a concrete AgentTeam delivery
// already has exactly one owner in the process: queued, accepted-but-unacked,
// or acknowledged. It prevents concurrent retry/runtime-boundary delivery and
// duplicate upstream delivery from producing two runtime inputs for one
// MessageID. Must be called with ap.mu held.
func (ap *Process) deliveryMessageAlreadyOwnedLocked(msg types.InboxMessage) bool {
	messageID := strings.TrimSpace(msg.MessageID)
	if messageID == "" {
		return false
	}
	if _, _, ok := ap.pendingRuntimeInputByMessageIDLocked(messageID); ok {
		return true
	}
	if _, ok := ap.ackedSet[messageID]; ok {
		return true
	}
	if _, ok := ap.failedSet[messageID]; ok {
		return true
	}
	if ap.pendingDeliveryIndexLocked("", messageID, "") >= 0 {
		return true
	}
	if _, ok := ap.deliveryReconcileReservations[messageID]; ok {
		return true
	}
	for _, queued := range ap.inbox {
		if strings.TrimSpace(queued.MessageID) == messageID {
			return true
		}
	}
	return false
}

// pendingRuntimeInputByMessageIDLocked resolves the durable AgentTeam delivery
// identity independently of the runtime correlation key. Most direct inputs
// use MessageID as Input.ID, while final-hop MCP slots intentionally use an
// opaque receipt as Input.ID; both must still represent one owner for q1.
// Must be called with ap.mu held.
func (ap *Process) pendingRuntimeInputByMessageIDLocked(messageID string) (string, pendingRuntimeInput, bool) {
	messageID = strings.TrimSpace(messageID)
	if messageID == "" {
		return "", pendingRuntimeInput{}, false
	}
	if pending, ok := ap.pendingRuntimeInputs[messageID]; ok && strings.TrimSpace(pending.Message.MessageID) == messageID {
		return messageID, pending, true
	}
	for inputID, pending := range ap.pendingRuntimeInputs {
		if strings.TrimSpace(pending.Message.MessageID) == messageID {
			return inputID, pending, true
		}
	}
	return "", pendingRuntimeInput{}, false
}

// authorizesRuntimeAckWake reports whether an exact correlated runtime
// acknowledgement for this pending input is allowed to begin a wake and activate
// delivery lifecycle state in the current live acknowledgement epoch. A normal live
// admission (fresh busy steer for Claude/Codex/Kimi) authorizes it; a Restored
// (recovered) pending input or a reclaimed correlation lease does not — its
// acknowledgement only upgrades durable acked evidence and freshness/frontier
// bookkeeping, never wake state.
func (pending pendingRuntimeInput) authorizesRuntimeAckWake() bool {
	if pending.Restored || pending.reclaimed {
		return false
	}
	return true
}

func (ap *Process) rememberPendingRuntimeInput(pending pendingRuntimeInput) {
	inputID := strings.TrimSpace(pending.Input.ID)
	if inputID == "" {
		return
	}
	if pending.CreatedAt.IsZero() {
		pending.CreatedAt = time.Now()
	}

	ap.mu.Lock()
	pending.ackEpoch = ap.ackEpoch
	pending.sendScopeGen = ap.sendScope.generation
	if ap.pendingRuntimeInputs == nil {
		ap.pendingRuntimeInputs = make(map[string]pendingRuntimeInput, 1)
	}
	ap.pendingRuntimeInputs[inputID] = pending
	ap.mu.Unlock()
}

func (ap *Process) forgetPendingRuntimeInput(inputID string, createdAt time.Time) {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return
	}
	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if ok && pending.CreatedAt.Equal(createdAt) {
		delete(ap.pendingRuntimeInputs, inputID)
		if len(ap.pendingRuntimeInputs) == 0 {
			ap.pendingRuntimeInputs = nil
		}
	}
	ap.mu.Unlock()
}

func (ap *Process) beginRuntimeInputPendingEvent(inputID string, createdAt time.Time) (chan struct{}, bool) {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return nil, false
	}

	ap.mu.Lock()
	defer ap.mu.Unlock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if !ok || !pending.CreatedAt.Equal(createdAt) {
		return nil, false
	}
	if pending.pendingStatusDone != nil {
		return nil, false
	}
	done := make(chan struct{})
	pending.pendingStatusDone = done
	ap.pendingRuntimeInputs[inputID] = pending
	return done, true
}

func (ap *Process) finishRuntimeInputPendingEvent(inputID string, createdAt time.Time, done chan struct{}) {
	if done == nil {
		return
	}
	inputID = strings.TrimSpace(inputID)
	if inputID != "" {
		ap.mu.Lock()
		pending, ok := ap.pendingRuntimeInputs[inputID]
		if ok && pending.CreatedAt.Equal(createdAt) && pending.pendingStatusDone == done {
			pending.pendingStatusDone = nil
			ap.pendingRuntimeInputs[inputID] = pending
		}
		ap.mu.Unlock()
	}
	close(done)
}

func (ap *Process) pendingRuntimeInputExists(inputID string, createdAt time.Time) bool {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return false
	}
	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	ap.mu.Unlock()
	return ok && pending.CreatedAt.Equal(createdAt)
}

func (ap *Process) hasDeliveryOwnershipStore() bool {
	ap.mu.Lock()
	hasStore := ap.deliveryOwnershipStore != nil
	ap.mu.Unlock()
	return hasStore
}

func (ap *Process) flushRuntimeInputAckedEvent(inputID string, createdAt time.Time) {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return
	}
	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if !ok || !pending.CreatedAt.Equal(createdAt) || pending.pendingAckedEvent == nil {
		ap.mu.Unlock()
		return
	}
	ev := *pending.pendingAckedEvent
	pending.pendingAckedEvent = nil
	ap.pendingRuntimeInputs[inputID] = pending
	ap.mu.Unlock()
	ap.handleRuntimeInputAcked(ev)
}

func (ap *Process) handleRuntimeInputAcked(ev drivercontract.ParsedEvent) bool {
	return ap.handleRuntimeInputAckedWithExpectedGeneration(ev, nil)
}

func (ap *Process) handleRuntimeInputAckedForReconcile(ev drivercontract.ParsedEvent, generation uint64) bool {
	return ap.handleRuntimeInputAckedWithExpectedGeneration(ev, &generation)
}

func (ap *Process) handleRuntimeInputAckedWithExpectedGeneration(ev drivercontract.ParsedEvent, expectedGeneration *uint64) bool {
	inputID := strings.TrimSpace(ev.RuntimeInputID)
	if inputID == "" {
		return true
	}

	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs[inputID]
	if !ok {
		ap.mu.Unlock()
		// Reconciliation may have exhausted its bounded safety budget and
		// durably recorded a failed tombstone before Pi finally flushed the
		// correlated user marker. Preserve terminal monotonicity by upgrading
		// that exact failed owner to acked; never re-open a wake cycle or
		// replay its body after late positive evidence.
		ap.upgradeLateFailedDeliveryOwnership(inputID, ev)
		return true
	}
	if expectedGeneration != nil && (!pending.reconcileRunning || pending.reconcileGeneration != *expectedGeneration) {
		ap.mu.Unlock()
		return true
	}
	if pending.pendingStatusDone != nil {
		evCopy := ev
		pending.pendingAckedEvent = &evCopy
		// The acknowledgement itself is the runtime-delivery evidence: install
		// the send scope NOW, atomically, so the first tool call racing after
		// this ack (and before the delivery RPC returns) sees the live turn's
		// targets — but only after the same trusted correlation checks the
		// normal path applies: a session-mismatched ack never authorizes a
		// target. Durability and wake side effects stay deferred to
		// flushRuntimeInputAckedEvent.
		if !runtimeInputAckedSessionMismatch(pending, ev) {
			ap.commitAckedSendScopeLocked(pending)
			ap.commitTurnDeliveryAckLocked(pending.Input.ID, pending.Message.Target)
		}
		// The live/synthetic acknowledgement owns the next transition. Invalidate
		// any in-flight read immediately even though status ordering defers the
		// durable terminal write until flushRuntimeInputAckedEvent.
		pending.reconcileGeneration++
		pending.reconcileRunning = true
		ap.pendingRuntimeInputs[inputID] = pending
		ap.mu.Unlock()
		return true
	}
	pending, transitionToken, ok := ap.claimPendingRuntimeInputTransitionLocked(inputID, pending.CreatedAt, expectedGeneration)
	if !ok {
		ap.mu.Unlock()
		return true
	}
	if runtimeInputAckedSessionMismatch(pending, ev) {
		ap.mu.Unlock()
		mismatchErr := fmt.Errorf("runtime input acknowledged in mismatched session %q", strings.TrimSpace(ev.SessionID))
		_ = mismatchErr // scope never authorized below; mismatch path unchanged
		queuedRecord := ap.ownershipRecord(pendingRuntimeInput{
			Message:   pending.Message,
			Input:     drivercontract.RuntimeInput{ID: strings.TrimSpace(pending.Message.MessageID)},
			CreatedAt: pending.CreatedAt,
		}, deliveryOwnershipQueued, pending.ReconcileAttempts, mismatchErr)
		committed, persistErr := ap.commitPendingRuntimeInputTransition(
			transitionToken,
			queuedRecord,
			false,
			func(current pendingRuntimeInput) {
				cancelRuntimeDeliveryLeaseTimerLocked(&current)
				delete(ap.pendingRuntimeInputs, inputID)
				if len(ap.pendingRuntimeInputs) == 0 {
					ap.pendingRuntimeInputs = nil
				}
				ap.inbox = requeueInboxMessage(ap.inbox, current.Message, true)
				ap.rememberUnobservedFrontierHintLocked(current.Message)
			},
		)
		if persistErr != nil {
			ap.mu.Lock()
			current, stillClaimed := ap.currentPendingRuntimeInputForTokenLocked(transitionToken)
			if stillClaimed {
				current.InflightReason = DeliveryInflightOutcomeUnknown
				current.reconcileRunning = false
				ap.pendingRuntimeInputs[inputID] = current
			}
			ap.mu.Unlock()
			if stillClaimed {
				ap.schedulePendingRuntimeInputReconcile(inputID, pending.CreatedAt, true)
			}
			return true
		}
		if !committed {
			return true
		}
		ap.discardDeliveryPromptCompileMetrics(inputID)
		if pending.finalHopLifecycleLease != nil {
			// Keep the exact final-hop owner through every admitted post-commit
			// side effect below. Stop may already be waiting for this lease, but it
			// cannot publish retirement/replacement before this callback exits.
			defer pending.finalHopLifecycleLease.resolve(inputID)
		}
		slog.Debug("suppressing runtime input acknowledgement with mismatched session",
			"agent_id", ap.AgentID,
			"runtime_input_id", inputID,
			"pending_session_id", strings.TrimSpace(pending.SessionID),
			"ack_session_id", strings.TrimSpace(ev.SessionID),
		)
		ap.scheduleInboxRetry(false)
		return true
	}
	ap.mu.Unlock()
	// Exact correlated acknowledgement is already authoritative before the
	// durable ownership transition below: the send scope updates atomically
	// now (with the generation and liveness fences inside), so a tool call
	// racing the ownership fsync sees the acknowledged target immediately.
	// Only Workflow maintenance suppression is deferred so a tool call racing
	// the fsync cannot report completion before its Step becomes active; wake
	// state still waits for the normal committed live-ack path.
	ap.mu.Lock()
	ap.commitAckedSendScopeLocked(pending)
	ap.commitTurnDeliveryAckLocked(pending.Input.ID, pending.Message.Target)
	ap.mu.Unlock()
	ap.workflowCompaction.observeWorkflowRuntimeMessage(pending.Message)
	ackedRecord := ap.ownershipRecord(pending, deliveryOwnershipAcked, 0, nil)
	ackedRecord.SessionID = firstNonEmptyString(ev.SessionID, pending.SessionID)
	var (
		suppressed   bool
		staleTurn    bool
		ackNoWake    bool
		wakeEvent    WakeStatusEvent
		hasWakeEvent bool
		liveAckEpoch uint64
	)
	committed, err := ap.commitPendingRuntimeInputTransition(
		transitionToken,
		ackedRecord,
		true,
		func(current pendingRuntimeInput) {
			cancelRuntimeDeliveryLeaseTimerLocked(&current)
			delete(ap.pendingRuntimeInputs, inputID)
			if len(ap.pendingRuntimeInputs) == 0 {
				ap.pendingRuntimeInputs = nil
			}
			messageID := strings.TrimSpace(current.Message.MessageID)
			// A trustworthy exact correlated runtime acknowledgement (Claude
			// type=user+isReplay+uuid, Codex clientUserMessageId echo, Kimi
			// TurnBegin/SteerInput correlation) is proof the message became
			// model-visible in the CURRENT live provider turn, even when the
			// pending input was admitted in an earlier turn. Capture the live
			// acknowledgement epoch so delivery lifecycle state activates in that
			// live epoch, not the stale admission epoch. Only an
			// ordinary final-hop delivery owner admitted before its turn retired
			// and acknowledged after retirement completes durable ownership
			// without reopening wake state.
			liveAckEpoch = ap.ackEpoch
			if current.finalHopLifecycleLease != nil && current.ackEpoch != liveAckEpoch {
				// The durable final-hop owner may finish after its provider turn
				// retired, but it cannot reopen wake state.
				staleTurn = true
				ap.rememberRuntimeAckedMessageLocked(messageID)
				return
			}
			if !current.authorizesRuntimeAckWake() {
				// A Restored (recovered) pending input or a reclaimed correlation
				// lease is not a live admission: its exact acknowledgement upgrades
				// durable acked evidence and the contractually allowed freshness/frontier
				// bookkeeping, but it must never begin a wake or make the Process busy
				// in the same or a later epoch.
				ackNoWake = true
				ap.rememberRuntimeAckedMessageLocked(messageID)
				ap.freshness.MarkObserved(current.Message.TargetKey, current.Message.TargetSeq, messageID)
				return
			}
			if ap.suppressStaleDeliveryLocked(current.Message, "already_observed_before_runtime_input_acked_event") {
				suppressed = true
				return
			}
			// Atomically migrate the final owner from pending to acked before
			// any wake/status callback can re-enter delivery with the same ID.
			ap.rememberRuntimeAckedMessageLocked(messageID)
			wakeEvent, hasWakeEvent = ap.beginWakeDeliveryWithTokenForMessageLocked(current.Message, current.WakeToken, "received")
			if !ap.usesExplicitTurnLifecycle() {
				ap.isIdle = false
			}
		},
	)
	if err != nil || !committed {
		return true
	}
	ap.commitDeliveryPromptCompileMetrics(inputID)
	if pending.finalHopLifecycleLease != nil {
		// The durable terminal record now precedes lifecycle release. Stop may
		// have marked the Process non-accepting while the final response write was
		// in flight, but replacement cannot proceed until this entire admitted
		// callback (wake/event/frontier/retry side effects included) returns.
		defer pending.finalHopLifecycleLease.resolve(inputID)
	}
	if suppressed {
		ap.markResumeSafetyIfQuiescent()
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
		return true
	}
	if staleTurn {
		ap.RememberObservedFrontier(pending.Message)
		ap.markResumeSafetyIfQuiescent()
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
		ap.scheduleInboxRetry(true)
		return true
	}
	ap.mu.Lock()
	stopped := ap.stopping || ap.state == processStateStopped
	ap.mu.Unlock()
	if stopped {
		// The already-admitted final-hop owner is allowed to finish its durable
		// commit after Stop marks the Process non-accepting, but it must not start
		// a new wake/status callback cycle on the retired Process generation.
		ap.RememberObservedFrontier(pending.Message)
		ap.markResumeSafetyIfQuiescent()
		return true
	}
	if ackNoWake {
		// A Restored/reclaimed acknowledgement upgraded the durable record and
		// freshness/frontier above but must never begin a wake, emit received
		// status or make the Process busy.
		ap.RememberObservedFrontier(pending.Message)
		ap.markResumeSafetyIfQuiescent()
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
		return true
	}

	if hasWakeEvent {
		ap.emitWakeStatus(wakeEvent)
	}
	if ap.prepareRuntimeMessageDeliveryRecovery(pending.Message, RuntimeDeliveryPathProtocol) {
		// Observe the runtime message only in the live acknowledgement epoch where
		// the trustworthy exact correlated ACK occurred, not the stale admission
		// epoch from when the message was queued. A stale final-hop owner from a
		// retired turn settles durability only and must not re-authorize its
		// targets into the live turn's scope, so skip the notification for it.
		ap.mu.Lock()
		staleFinalHop := pending.FinalHopDeferred &&
			pending.sendScopeGen != ap.sendScope.generation
		ap.mu.Unlock()
		if !staleFinalHop {
			ap.notifyRuntimeMessageDeliveryForTurn(pending.Message, &liveAckEpoch)
		}
	}
	ap.emitParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: inputID,
		Runtime:        ap.driver.Name(),
		SessionID:      strings.TrimSpace(firstNonEmptyString(ev.SessionID, pending.SessionID)),
		Message:        runtimeInputAckedDetail,
	})
	ap.RememberObservedFrontier(pending.Message)
	ap.markResumeSafetyIfQuiescent()
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
	return true
}

// upgradeLateFailedDeliveryOwnership applies late positive evidence after the
// in-memory pending lease has been retired. The durable store deliberately
// permits failed -> acked but forbids acked -> failed; keeping this
// transition here closes the child-exit/reconcile race without resurrecting a
// wake lifecycle on a retired Process.
func (ap *Process) upgradeLateFailedDeliveryOwnership(inputID string, ev drivercontract.ParsedEvent) {
	inputID = strings.TrimSpace(inputID)
	if inputID == "" {
		return
	}
	ap.mu.Lock()
	store := ap.deliveryOwnershipStore
	ap.mu.Unlock()
	if store == nil {
		return
	}
	ap.deliveryOwnershipMutationMu.Lock()
	records, err := store.LoadDeliveryOwnership()
	if err != nil {
		ap.deliveryOwnershipMutationMu.Unlock()
		slog.Warn("load failed runtime delivery ownership for late acknowledgement",
			"agent_id", ap.AgentID, "runtime_input_id", inputID, "error", err)
		return
	}
	var upgraded DeliveryOwnershipRecord
	found := false
	for _, record := range records {
		if !record.IsFailed() || record.MessageID() != inputID {
			continue
		}
		if strings.TrimSpace(record.SessionID) != "" && strings.TrimSpace(ev.SessionID) != "" &&
			strings.TrimSpace(record.SessionID) != strings.TrimSpace(ev.SessionID) {
			continue
		}
		upgraded = record
		upgraded.State = DeliveryOwnershipTerminal
		upgraded.InflightReason = ""
		upgraded.Outcome = DeliveryOutcomeAcked
		if strings.TrimSpace(ev.SessionID) != "" {
			upgraded.SessionID = strings.TrimSpace(ev.SessionID)
		}
		upgraded.UpdatedAt = time.Now().UTC()
		found = true
		break
	}
	if found {
		upgraded = NormalizeDeliveryOwnershipRecordForPersistence(upgraded)
		if _, err = store.PutDeliveryOwnership(upgraded); err != nil {
			found = false
		}
	}
	ap.deliveryOwnershipMutationMu.Unlock()
	if !found {
		if err != nil {
			slog.Warn("persist late acked runtime delivery ownership failed",
				"agent_id", ap.AgentID, "runtime_input_id", inputID, "error", err)
		}
		return
	}
	ap.mu.Lock()
	ap.rememberRuntimeAckedMessageLocked(strings.TrimSpace(upgraded.Message.MessageID))
	ap.freshness.MarkObserved(upgraded.Message.TargetKey, upgraded.Message.TargetSeq, upgraded.Message.MessageID)
	ap.mu.Unlock()
	ap.markResumeSafetyIfQuiescent()
}

func runtimeInputAckedSessionMismatch(pending pendingRuntimeInput, ev drivercontract.ParsedEvent) bool {
	pendingSessionID := strings.TrimSpace(pending.SessionID)
	ackSessionID := strings.TrimSpace(ev.SessionID)
	return pendingSessionID != "" && ackSessionID != "" && pendingSessionID != ackSessionID
}

func (ap *Process) emitRuntimeInputPendingEvent(pending pendingRuntimeInput) {
	ap.emitParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputPending,
		RuntimeInputID: strings.TrimSpace(pending.Input.ID),
		Runtime:        ap.driver.Name(),
		SessionID:      strings.TrimSpace(pending.SessionID),
		Text:           strings.TrimSpace(pending.Input.PromptText()),
		PromptKind:     consts.PromptKindDeliveryText,
		EventType:      string(deliveryPromptEventType(pending.Message)),
		Message:        pendingRuntimeInputDetail,
	})
}

func (ap *Process) canQueueInboxMessageLocked(msg types.InboxMessage) bool {
	// Aggregate capacity belongs to the durable admission transaction, which can
	// terminalize older owners before accepting this MessageID. The Process
	// mirror must not reject new work merely because it has not projected that
	// transaction locally yet. Only an intrinsically oversized body is rejected.
	return inboxMessageBytes(msg) <= processInboxMaxMessageBytes
}

// applyReclaimedDeliveryOwnership projects one completed durable admission
// transaction into this Process. The store has already converted every record
// below to a failed tombstone, so local queued/reconcile owners must no longer
// retain capacity or ordering authority. An inflight pending input is kept only
// as a bounded correlation lease so late authoritative acknowledgement can
// upgrade the tombstone; it is no longer a durable live owner.
func (ap *Process) applyReclaimedDeliveryOwnership(records []DeliveryOwnershipRecord) {
	if len(records) == 0 {
		return
	}
	var wakeEvents []WakeStatusEvent
	var runtimeEvents []RuntimeEvent

	ap.mu.Lock()
	now := time.Now().UTC()
	for _, record := range records {
		messageID := record.MessageID()
		if messageID == "" {
			continue
		}
		// Stale-projection gate, evaluated before any side effect. The durable
		// reclaim says this delivery was failed to make room for a newer one, but
		// an authoritative concurrent acked transition may have already
		// committed this messageID as acked, or the pending slot may now hold
		// a different admission (identity mismatch). Either makes the projection
		// stale: it must be a full no-op for failure side effects and must not
		// re-terminate the wake, re-mark the owner failed, or emit wake_failed. Only a still-current
		// owner receives the full failure projection.
		_, acked := ap.ackedSet[messageID]
		pendingIdentityStale := false
		if inputID, pending, ok := ap.pendingRuntimeInputByMessageIDLocked(messageID); ok {
			if pending.CreatedAt.Equal(record.AcceptedAt) {
				// Still the current owner. Keep a bounded correlation lease so an
				// exact late ACK can upgrade the durable failed tombstone.
				// The lease is bounded by sweepReclaimedCorrelationLeasesLocked.
				pending.reconcileGeneration++
				pending.reconcileRunning = false
				pending.reclaimed = true
				pending.reclaimedAt = now
				ap.pendingRuntimeInputs[inputID] = pending
			} else {
				// The pending slot now holds a different admission; this reclaim
				// refers to a stale one and must not corrupt the current owner.
				pendingIdentityStale = true
			}
		}
		// Idempotent cleanups apply to both stale and current projections: a
		// reclaimed messageID must not remain in the inbox, retry, or reconcile
		// reservation regardless of whether its failure is re-projected.
		ap.inbox = removeInboxMessageByID(ap.inbox, messageID)
		if len(ap.pendingRuntimeInputs) == 0 {
			ap.pendingRuntimeInputs = nil
		}
		delete(ap.deliveryReconcileReservations, messageID)
		if ap.inboxRetryMessageID == messageID {
			ap.stopInboxRetryLocked()
			ap.clearInboxRetryBackoffLocked()
		}
		ap.clearTransientRuntimeRetryStateForWakeLocked(messageID)
		if acked || pendingIdentityStale {
			continue
		}
		if index := ap.pendingDeliveryIndexLocked("", messageID, ""); index == 0 {
			wakeEvents = append(wakeEvents, ap.terminalWakeStatusLocked(
				consts.WakeStatusFailed, "", "older runtime delivery was reclaimed for a newer message",
			))
		} else if index > 0 {
			cycle := ap.pendingDeliveryQueue[index-1]
			ap.pendingDeliveryQueue = append(ap.pendingDeliveryQueue[:index-1], ap.pendingDeliveryQueue[index:]...)
			wakeEvents = append(wakeEvents, wakeStatusForDelivery(
				cycle, consts.WakeStatusFailed, "", "older runtime delivery was reclaimed for a newer message",
			))
		}
		ap.rememberFailedMessageLocked(messageID)
		runtimeEvents = append(runtimeEvents, RuntimeEvent{
			MessageID: messageID, EventClass: consts.RuntimeEventClassWakeFailed,
			Target: record.Message.Target,
			Detail: "older runtime delivery was failed to keep newer delivery available",
		})
	}
	ap.sweepReclaimedCorrelationLeasesLocked(now)
	ap.scheduleReclaimedCorrelationLeaseSweepLocked(now)
	ap.mu.Unlock()

	for _, event := range wakeEvents {
		ap.emitWakeStatus(event)
	}
	for _, event := range runtimeEvents {
		ap.emitRuntimeEvent(event)
	}
}

// sweepReclaimedCorrelationLeasesLocked bounds the reclaimed inflight
// correlation leases retained in pendingRuntimeInputs. A reclaimed entry is
// kept only so an exact late ACK can upgrade its durable failed
// tombstone; the durable tombstone itself lives in the resume state store and
// the store still allows a later failed->acked upgrade, so dropping an
// expired or over-cap lease here does not lose ownership. Entries past
// reclaimedCorrelationLeaseTTL are evicted first; if the remainder still exceeds
// maxReclaimedCorrelationLeases the oldest are dropped. Must be called with
// ap.mu held.
func (ap *Process) sweepReclaimedCorrelationLeasesLocked(now time.Time) {
	if len(ap.pendingRuntimeInputs) == 0 {
		return
	}
	reclaimedIDs := make([]string, 0, len(ap.pendingRuntimeInputs))
	for id, pending := range ap.pendingRuntimeInputs {
		if !pending.reclaimed {
			continue
		}
		if !pending.reclaimedAt.IsZero() && now.Sub(pending.reclaimedAt) > ap.reclaimedCorrelationLeaseTTLValue() {
			delete(ap.pendingRuntimeInputs, id)
			continue
		}
		reclaimedIDs = append(reclaimedIDs, id)
	}
	for len(reclaimedIDs) > maxReclaimedCorrelationLeases {
		oldestIdx := 0
		oldest := ap.pendingRuntimeInputs[reclaimedIDs[oldestIdx]]
		for i := 1; i < len(reclaimedIDs); i++ {
			candidate := ap.pendingRuntimeInputs[reclaimedIDs[i]]
			if candidate.reclaimedAt.Before(oldest.reclaimedAt) ||
				(candidate.reclaimedAt.Equal(oldest.reclaimedAt) && reclaimedIDs[i] < reclaimedIDs[oldestIdx]) {
				oldestIdx = i
				oldest = candidate
			}
		}
		delete(ap.pendingRuntimeInputs, reclaimedIDs[oldestIdx])
		reclaimedIDs = append(reclaimedIDs[:oldestIdx], reclaimedIDs[oldestIdx+1:]...)
	}
	if len(ap.pendingRuntimeInputs) == 0 {
		ap.pendingRuntimeInputs = nil
	}
}

// reclaimedCorrelationLeaseTTLValue returns the configured reclaimed lease TTL,
// falling back to the package default when the Process-level override is unset.
func (ap *Process) reclaimedCorrelationLeaseTTLValue() time.Duration {
	if ap.reclaimedCorrelationLeaseTTL > 0 {
		return ap.reclaimedCorrelationLeaseTTL
	}
	return reclaimedCorrelationLeaseTTL
}

// scheduleReclaimedCorrelationLeaseSweepLocked arms a production deadline timer
// that evicts expired reclaimed correlation leases even when no further reclaim
// admission arrives. Without it, a reclaimed lease could outlive its TTL until
// the next applyReclaimedDeliveryOwnership call (or indefinitely if no later
// reclamation happens). Must be called with ap.mu held.
func (ap *Process) scheduleReclaimedCorrelationLeaseSweepLocked(now time.Time) {
	ttl := ap.reclaimedCorrelationLeaseTTLValue()
	var earliest time.Time
	for _, pending := range ap.pendingRuntimeInputs {
		if !pending.reclaimed || pending.reclaimedAt.IsZero() {
			continue
		}
		expiry := pending.reclaimedAt.Add(ttl)
		if earliest.IsZero() || expiry.Before(earliest) {
			earliest = expiry
		}
	}
	if earliest.IsZero() {
		ap.cancelReclaimedCorrelationLeaseSweepLocked()
		return
	}
	// A newly reclaimed lease always expires later than any existing one, so the
	// earliest expiry never moves earlier on add. If a timer already fires no
	// later than this earliest expiry, keep it and avoid churn.
	if !ap.reclaimedLeaseSweepDeadline.IsZero() && !ap.reclaimedLeaseSweepDeadline.After(earliest) {
		return
	}
	ap.cancelReclaimedCorrelationLeaseSweepLocked()
	ap.reclaimedLeaseSweepGeneration++
	generation := ap.reclaimedLeaseSweepGeneration
	ap.reclaimedLeaseSweepDeadline = earliest
	cancel := make(chan struct{})
	ap.reclaimedLeaseSweepCancel = cancel
	go ap.runReclaimedCorrelationLeaseSweep(earliest, generation, cancel)
}

// cancelReclaimedCorrelationLeaseSweepLocked invalidates any armed sweep timer.
// The generation bump makes an in-flight timer goroutine that already passed
// its select return without sweeping once it reacquires ap.mu. Must be called
// with ap.mu held.
func (ap *Process) cancelReclaimedCorrelationLeaseSweepLocked() {
	ap.reclaimedLeaseSweepGeneration++
	if ap.reclaimedLeaseSweepCancel != nil {
		close(ap.reclaimedLeaseSweepCancel)
		ap.reclaimedLeaseSweepCancel = nil
	}
	ap.reclaimedLeaseSweepDeadline = time.Time{}
}

func (ap *Process) runReclaimedCorrelationLeaseSweep(deadline time.Time, generation uint64, cancel <-chan struct{}) {
	duration := time.Until(deadline)
	if duration < 0 {
		duration = 0
	}
	timer := time.NewTimer(duration)
	defer timer.Stop()
	select {
	case <-timer.C:
	case <-cancel:
		return
	case <-ap.processDone():
		return
	}
	ap.mu.Lock()
	defer ap.mu.Unlock()
	if ap.reclaimedLeaseSweepGeneration != generation {
		return
	}
	if ap.stopping || ap.state == processStateStopped {
		return
	}
	now := time.Now().UTC()
	ap.sweepReclaimedCorrelationLeasesLocked(now)
	ap.cancelReclaimedCorrelationLeaseSweepLocked()
	ap.scheduleReclaimedCorrelationLeaseSweepLocked(now)
}

func deliveryOwnerCreatedBefore(left, right time.Time) bool {
	if left.IsZero() {
		return false
	}
	return right.IsZero() || left.Before(right)
}

func inboxMessageBytes(msg types.InboxMessage) int {
	author := msg.AuthorIdentity()
	total := len(msg.MessageID) + len(msg.TargetKey) + len(msg.Target) + len(msg.ChannelType) + len(msg.ChannelName) +
		len(author.ActorType) + len(author.DisplayName) + len(msg.Content) + len(msg.Timestamp) +
		len(msg.MessageType) + len(msg.AttachmentIDs) + len(msg.AttachmentNames)
	for k, v := range msg.Metadata {
		total += len(k) + len(fmt.Sprint(v))
	}
	return total
}

func (ap *Process) rejectInboxBackpressure(msg types.InboxMessage) error {
	detail := fmt.Sprintf(
		"local inbox backpressure: queued messages exceed %d entries or %d bytes",
		processInboxMaxMessages,
		processInboxMaxBytes,
	)
	ap.emitRuntimeEvent(RuntimeEvent{
		MessageID:  strings.TrimSpace(msg.MessageID),
		EventClass: consts.RuntimeEventClassWakeFailed,
		Target:     strings.TrimSpace(msg.Target),
		Detail:     detail,
	})
	return errors.New(detail)
}

func requeueInboxMessage(inbox []types.InboxMessage, msg types.InboxMessage, fromQueue bool) []types.InboxMessage {
	if fromQueue {
		return append([]types.InboxMessage{msg}, inbox...)
	}
	return append(inbox, msg)
}

func (ap *Process) popNextDeliverableInboxMessageLocked(reason string) (types.InboxMessage, bool, error) {
	for len(ap.inbox) > 0 {
		msg := ap.inbox[0]
		if _, reserved := ap.deliveryReconcileReservations[strings.TrimSpace(msg.MessageID)]; reserved {
			// Reconciliation must retain FIFO ownership until its acked tombstone
			// is durable. Do not skip the reserved head and let q2 overtake it.
			return types.InboxMessage{}, false, nil
		}
		stale, err := ap.durablySuppressStaleDeliveryLocked(msg, reason)
		if err != nil {
			ap.noteInboxRetryFailureLocked(msg, err)
			return types.InboxMessage{}, false, err
		}
		ap.inbox = ap.inbox[1:]
		if stale {
			continue
		}
		return msg, true, nil
	}
	return types.InboxMessage{}, false, nil
}

func (ap *Process) suppressStaleDeliveryLocked(msg types.InboxMessage, reason string) bool {
	targetKey := strings.TrimSpace(msg.TargetKey)
	if targetKey == "" || msg.TargetSeq <= 0 || strings.TrimSpace(msg.MessageType) == "reaction" {
		return false
	}
	if len(ap.freshness.ByTarget) == 0 {
		return false
	}
	frontier, ok := ap.freshness.ByTarget[targetKey]
	if !ok || frontier.ObservedSeq <= 0 || msg.TargetSeq > frontier.ObservedSeq {
		return false
	}
	slog.Debug("suppressing stale AgentTeam delivery",
		"agent_id", ap.AgentID,
		"delivery_id", strings.TrimSpace(msg.MessageID),
		"message_id", strings.TrimSpace(msg.MessageID),
		"target", strings.TrimSpace(msg.Target),
		"target_key", targetKey,
		"target_seq", msg.TargetSeq,
		"observed_seq", frontier.ObservedSeq,
		"latest_known_seq", frontier.LatestKnownSeq,
		"reason", strings.TrimSpace(reason),
	)
	return true
}

// durablySuppressStaleDeliveryLocked turns freshness evidence into an acked
// terminal owner before the caller removes a queued body. It must be called
// with ap.mu held. A persistence error reports stale=true so callers retain the
// local owner rather than replaying or consuming it without a durable terminal.
func (ap *Process) durablySuppressStaleDeliveryLocked(msg types.InboxMessage, reason string) (stale bool, err error) {
	if !ap.suppressStaleDeliveryLocked(msg, reason) {
		return false, nil
	}
	if err := ap.persistObservedDeliveryAsAckedLocked(msg); err != nil {
		return true, err
	}
	return true, nil
}

// persistObservedDeliveryAsAckedLocked is the synchronous durability
// boundary used by local freshness/read/reconcile slots. Holding ap.mu
// keeps the in-memory FIFO owner in place until Put returns; the dispatcher
// store does not call back into Process.
func (ap *Process) persistObservedDeliveryAsAckedLocked(msg types.InboxMessage) error {
	messageID := strings.TrimSpace(msg.MessageID)
	if messageID == "" {
		return fmt.Errorf("observed runtime delivery requires message id")
	}
	record := NormalizeDeliveryOwnershipRecordForPersistence(DeliveryOwnershipRecord{
		Message:    msg,
		Input:      drivercontract.RuntimeInput{ID: messageID},
		SessionID:  strings.TrimSpace(ap.SessionID),
		State:      DeliveryOwnershipTerminal,
		Outcome:    DeliveryOutcomeAcked,
		AcceptedAt: time.Now().UTC(),
		UpdatedAt:  time.Now().UTC(),
	})
	if err := ap.persistDeliveryOwnershipLocked(record); err != nil {
		return fmt.Errorf("persist observed runtime delivery %s as acked: %w", messageID, err)
	}
	ap.rememberRuntimeAckedMessageLocked(messageID)
	ap.freshness.MarkObserved(msg.TargetKey, msg.TargetSeq, messageID)
	return nil
}

// directMessageDriverLocked returns the drivercontract.BusyStdinMessageDriver if the current
// driver exposes protocol-level routed-message delivery support.
// Must be called with ap.mu held.
func (ap *Process) directMessageDriverLocked() (drivercontract.BusyStdinMessageDriver, bool) {
	if ap.driver == nil || !ap.driver.SupportsStdinNotification() {
		return nil, false
	}
	if strings.TrimSpace(ap.SessionID) == "" {
		return nil, false
	}
	driver, ok := ap.driver.(drivercontract.BusyStdinMessageDriver)
	return driver, ok
}

// ackedBusyRuntimeInputDriverLocked returns a driver that can accept active-turn
// direct input with a later model acknowledgement. Must be called with
// ap.mu held.
func (ap *Process) ackedBusyRuntimeInputDriverLocked() (drivercontract.AckedBusyRuntimeInputDriver, bool) {
	if ap.driver == nil || !ap.driver.SupportsStdinNotification() {
		return nil, false
	}
	if strings.TrimSpace(ap.SessionID) == "" {
		return nil, false
	}
	driver, ok := ap.driver.(drivercontract.AckedBusyRuntimeInputDriver)
	if !ok || !driver.SupportsAckedBusyRuntimeInputDelivery() {
		return nil, false
	}
	return driver, true
}

// InboxMessagesForFreshnessPage returns a bounded internal refresh page for one
// target key and consumes only returned target deliveries.
func (ap *Process) InboxMessagesForFreshnessPage(targetKey string, throughSeq int64, limit int) types.InboxReadPage {
	return ap.InboxMessagesForFreshnessPageForRuntime(targetKey, throughSeq, limit, "")
}

// InboxMessagesForFreshnessPageForRuntime is the final-hop-aware freshness
// reader. Returned bodies move from queued ownership to a durable uncertain
// final-hop claim; only the transport that writes the enclosing MCP result
// may later commit the receipt as acked.
func (ap *Process) InboxMessagesForFreshnessPageForRuntime(targetKey string, throughSeq int64, limit int, runtimeSessionID string) types.InboxReadPage {
	_ = throughSeq
	targetKey = strings.TrimSpace(targetKey)
	if targetKey == "" {
		return types.InboxReadPage{}
	}
	if !ap.beginLifecycleMutation() {
		return types.InboxReadPage{}
	}
	lifecycleMutationTransferred := false
	defer func() {
		if !lifecycleMutationTransferred {
			ap.endLifecycleMutation()
		}
	}()

	// Freshness paging is invoked from a runtime MCP/server request and therefore
	// must not acquire deliveryGate: a runtime can issue message_send while an
	// outer protocol delivery is still waiting for its response, and the
	// non-reentrant gate would deadlock that request cycle. lifecycleMutationGate
	// is its reentrant-safe admission barrier instead; this read lease spans every
	// ownership fsync and post-mutation persistence callback. Stop first marks the
	// Process non-accepting, then drains this lease before session cleanup.
	ap.mu.Lock()
	if len(ap.inbox) == 0 {
		ap.mu.Unlock()
		return types.InboxReadPage{}
	}
	if ap.hasLivePendingRuntimeInputLocked() {
		// A prior runtime/pending/tool owner is the FIFO head. In particular, a
		// deferred q1 final-hop lease must prevent a concurrent freshness callback
		// from claiming and surfacing q2 first. A reclaimed inflight entry is
		// correlation-only and no longer holds the FIFO head.
		remaining := 0
		for _, msg := range ap.inbox {
			if strings.TrimSpace(msg.TargetKey) == targetKey {
				remaining++
			}
		}
		ap.mu.Unlock()
		return types.InboxReadPage{HasMore: remaining > 0, RemainingCount: remaining}
	}
	if limit <= 0 {
		limit = len(ap.inbox)
	}
	runtimeSessionID = strings.TrimSpace(runtimeSessionID)
	if runtimeSessionID == "" {
		runtimeSessionID = strings.TrimSpace(ap.SessionID)
	}
	result := make([]types.InboxMessage, 0, min(limit, len(ap.inbox)))
	claims := make([]types.RuntimeDeliveryClaim, 0, min(limit, len(ap.inbox)))
	claimLeases := make([]pendingRuntimeInput, 0, min(limit, len(ap.inbox)))
	next := make([]types.InboxMessage, 0, len(ap.inbox))
	remaining := 0
	persistBlocked := false
	claimBlocked := false
	leaseDuration := ap.runtimeDeliveryDeferredLease
	if leaseDuration <= 0 {
		leaseDuration = processRuntimeDeliveryDeferredLease
	}
	for _, msg := range ap.inbox {
		if claimBlocked {
			next = append(next, msg)
			if strings.TrimSpace(msg.TargetKey) == targetKey {
				remaining++
			}
			continue
		}
		if stale, err := ap.durablySuppressStaleDeliveryLocked(msg, "already_observed_before_freshness_refresh_page"); stale {
			if err != nil {
				persistBlocked = true
				claimBlocked = true
				next = append(next, msg)
				if strings.TrimSpace(msg.TargetKey) == targetKey {
					remaining++
				}
			}
			continue
		}
		if strings.TrimSpace(msg.TargetKey) != targetKey {
			// The inbox is one global ownership FIFO, not one queue per target.
			// Once a live older owner belongs to another target, this refresh may
			// not return any later target match ahead of it. Stale owners above
			// were safe to suppress durably before reaching this boundary.
			claimBlocked = true
			next = append(next, msg)
			continue
		}
		if len(result) >= limit || !ap.canAcceptWakeLifecycleOwnerLocked(msg.MessageID) {
			// Capacity is reserved when each pending claim is inserted below, so
			// same-page q1 counts before q2 is considered. Keep the blocked body
			// and every later global FIFO owner daemon-owned.
			claimBlocked = true
			remaining++
			next = append(next, msg)
			continue
		}
		{
			receipt := ""
			var receiptErr error
			for attempt := 0; attempt < 4; attempt++ {
				receipt, receiptErr = ap.runtimeDeliveryReceiptLocked()
				if receiptErr != nil {
					break
				}
				if _, collision := ap.runtimeDeliveryReceiptSet[receipt]; collision {
					receipt = ""
					continue
				}
				break
			}
			if receiptErr == nil && receipt == "" {
				receiptErr = errors.New("runtime delivery receipt collision budget exhausted")
			}
			if receiptErr != nil {
				// Receipt identity is the only crash-safe correlation between the
				// queued owner and a later final tool write. Random-source failure
				// therefore fails closed: keep q1 queued and expose no body.
				persistBlocked = true
				claimBlocked = true
				next = append(next, msg)
				remaining++
				continue
			}
			marker := types.RuntimeDeliveryReceiptMarker(receipt)
			markerCarrier, markerErr := ap.compileRuntimeMarkerPrompt(msg, marker, runtimeSessionID)
			if markerErr != nil {
				persistBlocked = true
				claimBlocked = true
				next = append(next, msg)
				remaining++
				continue
			}
			createdAt := time.Now().UTC()
			pending := pendingRuntimeInput{
				Message: msg,
				Input: drivercontract.RuntimeInput{
					ID:             receipt,
					CompiledPrompt: &markerCarrier,
				},
				SessionID:      runtimeSessionID,
				CreatedAt:      createdAt,
				InflightReason: DeliveryInflightOutcomeUnknown,
				// Claiming the body is already the beginning of the final-hop
				// interval. Establish the lock and its bounded orphan lease before
				// the same critical section removes q1 from inbox; otherwise a
				// concurrent terminal/retry callback can reconcile an idle runtime
				// during the gap before the MCP HTTP Deferred acknowledgement.
				FinalHopDeferred:        true,
				finalHopLeaseGeneration: 1,
				finalHopLeaseCancel:     make(chan struct{}),
				ackEpoch:                ap.ackEpoch,
				sendScopeGen:            ap.sendScope.generation,
			}
			// The tool result has not crossed its final runtime boundary yet. Fsync
			// uncertain ownership before removing the queued FIFO owner so a crash
			// can reconcile the marker without either losing or blindly replaying
			// the exact user body.
			if err := ap.persistDeliveryOwnershipLocked(ap.ownershipRecord(pending, deliveryOwnershipOutcomeUnknown, 0, nil)); err != nil {
				persistBlocked = true
				claimBlocked = true
				next = append(next, msg)
				remaining++
				continue
			}
			if ap.pendingRuntimeInputs == nil {
				ap.pendingRuntimeInputs = make(map[string]pendingRuntimeInput)
			}
			// Start the full lease after the ownership fsync rather than before a
			// potentially slow store write; production's six-minute interval must
			// remain longer than Codex's five-minute request timeout.
			pending.finalHopLeaseDeadline = time.Now().UTC().Add(leaseDuration)
			ap.pendingRuntimeInputs[receipt] = pending
			claimLeases = append(claimLeases, pending)
			ap.rememberRuntimeDeliveryReceiptLocked(receipt)
			result = append(result, msg)
			claims = append(claims, types.RuntimeDeliveryClaim{
				MessageID: strings.TrimSpace(msg.MessageID),
				Receipt:   receipt,
			})
			continue
		}
	}
	ap.inbox = next
	var runtimeDeliveryCommitOwner types.MCPToolResultDeliveryOwner
	if len(claims) > 0 {
		receipts := make([]string, 0, len(claims))
		for _, claim := range claims {
			receipts = append(receipts, claim.Receipt)
		}
		lease := newRuntimeDeliveryLifecycleLease(ap, receipts)
		for _, receipt := range receipts {
			pending := ap.pendingRuntimeInputs[receipt]
			pending.finalHopLifecycleLease = lease
			ap.pendingRuntimeInputs[receipt] = pending
		}
		for i := range claimLeases {
			claimLeases[i].finalHopLifecycleLease = lease
		}
		runtimeDeliveryCommitOwner = lease
	}
	ap.mu.Unlock()
	// Timers are installed before this claim call returns. The state they guard
	// was published while ap.mu was still held, so any concurrent scheduler in
	// this narrow post-unlock interval already observes FinalHopDeferred=true.
	// A crash before the goroutine starts restores the durable uncertain record
	// and reconciles it immediately rather than replaying it blindly.
	if persistBlocked {
		ap.scheduleInboxRetry(false)
	}
	if len(claims) > 0 {
		ap.markResumeSafety(false)
	} else {
		ap.markResumeSafetyIfQuiescent()
		ap.scheduleOldestPendingRuntimeInputReconcile(true)
	}
	if runtimeDeliveryCommitOwner != nil {
		// Transfer the read-side lifecycle owner only after all synchronous claim
		// bookkeeping is complete; from this point the exact MCP result or its
		// orphan timers are solely responsible for releasing it.
		lifecycleMutationTransferred = true
	}
	for _, pending := range claimLeases {
		ap.startRuntimeMessageFinalHopDeferredLease(pending)
	}
	return types.InboxReadPage{
		Messages:                   result,
		RuntimeDeliveryClaims:      claims,
		RuntimeDeliveryCommitOwner: runtimeDeliveryCommitOwner,
		HasMore:                    remaining > 0,
		RemainingCount:             remaining,
	}
}

// AcknowledgeRuntimeMessageToolDeliveriesForSession is the runtime event variant
// with an exact session fence. Direct one-hop MCP acknowledgements omit the
// fence because they execute against the same live handler/Process instance.
func (ap *Process) AcknowledgeRuntimeMessageToolDeliveriesForSession(receipts []string, outcome types.MCPToolDeliveryOutcome, expectedSessionID string) {
	ap.acknowledgeRuntimeMessageToolDeliveriesForOwner(receipts, outcome, expectedSessionID, nil)
}

// acknowledgeRuntimeMessageToolDeliveriesForOwner is the exact-owner variant
// used by the in-process MCP result capability. owner=nil is reserved for
// generation/session-fenced runtime events and the legacy handler fallback.
func (ap *Process) acknowledgeRuntimeMessageToolDeliveriesForOwner(
	receipts []string,
	outcome types.MCPToolDeliveryOutcome,
	expectedSessionID string,
	owner *runtimeDeliveryLifecycleLease,
) {
	if ap == nil || len(receipts) == 0 {
		return
	}
	expectedSessionID = strings.TrimSpace(expectedSessionID)
	validated := make([]string, 0, len(receipts))
	seen := make(map[string]struct{}, len(receipts))
	for _, receipt := range receipts {
		receipt = strings.TrimSpace(receipt)
		if !types.IsRuntimeDeliveryReceipt(receipt) {
			continue
		}
		if _, duplicate := seen[receipt]; duplicate {
			continue
		}
		seen[receipt] = struct{}{}
		validated = append(validated, receipt)
	}
	if len(validated) == 0 {
		return
	}

	switch outcome {
	case types.MCPToolDeliveryDelivered:
		for _, receipt := range validated {
			ap.mu.Lock()
			pending, ok := ap.pendingRuntimeInputs[receipt]
			ap.mu.Unlock()
			if !ok {
				if owner != nil {
					owner.resolve(receipt)
				}
				continue
			}
			if (owner != nil && pending.finalHopLifecycleLease != owner) ||
				(expectedSessionID != "" && strings.TrimSpace(pending.SessionID) != "" && strings.TrimSpace(pending.SessionID) != expectedSessionID) {
				continue
			}
			ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
				Kind:           consts.ParsedEventKindRuntimeInputAcked,
				RuntimeInputID: receipt,
				SessionID:      strings.TrimSpace(pending.SessionID),
			})
		}
	case types.MCPToolDeliveryDeferred:
		// The first-hop MCP response has succeeded, but the runtime-owned final
		// response write is still in progress. Reconciliation here could observe
		// an idle thread before the marker is installed and replay the exact body
		// concurrently with that second hop. The Codex driver must report either
		// Delivered or Uncertain when it releases this lease.
		var leases []pendingRuntimeInput
		ap.mu.Lock()
		leaseDuration := ap.runtimeDeliveryDeferredLease
		if leaseDuration <= 0 {
			leaseDuration = processRuntimeDeliveryDeferredLease
		}
		now := time.Now().UTC()
		for _, receipt := range validated {
			pending, ok := ap.pendingRuntimeInputs[receipt]
			if !ok || (owner != nil && pending.finalHopLifecycleLease != owner) ||
				(expectedSessionID != "" && strings.TrimSpace(pending.SessionID) != "" && strings.TrimSpace(pending.SessionID) != expectedSessionID) {
				continue
			}
			// Claim creation, not this callback, acquires the lock. A late
			// Deferred event after Delivered/Uncertain must never resurrect it;
			// while it is live, this acknowledgement only renews the bounded
			// orphan lease for the runtime-owned second hop.
			if !pending.FinalHopDeferred {
				continue
			}
			cancelRuntimeDeliveryLeaseTimerLocked(&pending)
			pending.finalHopLeaseGeneration++
			pending.finalHopLeaseDeadline = now.Add(leaseDuration)
			pending.finalHopLeaseCancel = make(chan struct{})
			pending.FinalHopRuntimeTransferred = true
			ap.pendingRuntimeInputs[receipt] = pending
			leases = append(leases, pending)
		}
		ap.mu.Unlock()
		for _, pending := range leases {
			ap.startRuntimeMessageFinalHopDeferredLease(pending)
		}
		return
	case types.MCPToolDeliveryUncertain:
		// Preserve FIFO by starting only the oldest matching claim. Every
		// terminal reconciliation path schedules the next owner.
		var oldest pendingRuntimeInput
		oldestID := ""
		leasesToResolve := make(map[string]*runtimeDeliveryLifecycleLease, len(validated))
		ap.mu.Lock()
		for _, receipt := range validated {
			pending, ok := ap.pendingRuntimeInputs[receipt]
			if !ok {
				if owner != nil {
					leasesToResolve[receipt] = owner
				}
				continue
			}
			if (owner != nil && pending.finalHopLifecycleLease != owner) ||
				(expectedSessionID != "" && strings.TrimSpace(pending.SessionID) != "" && strings.TrimSpace(pending.SessionID) != expectedSessionID) {
				continue
			}
			leasesToResolve[receipt] = pending.finalHopLifecycleLease
			cancelRuntimeDeliveryLeaseTimerLocked(&pending)
			pending.FinalHopDeferred = false
			pending.FinalHopRuntimeTransferred = false
			pending.finalHopLeaseGeneration++
			pending.finalHopLeaseDeadline = time.Time{}
			pending.finalHopLifecycleLease = nil
			ap.pendingRuntimeInputs[receipt] = pending
		}
		reconcileAlreadyRunning := false
		for _, pending := range ap.pendingRuntimeInputs {
			if pending.reconcileRunning {
				reconcileAlreadyRunning = true
				break
			}
		}
		if !reconcileAlreadyRunning {
			for inputID, pending := range ap.pendingRuntimeInputs {
				if strings.TrimSpace(pending.Input.ID) == "" {
					continue
				}
				if oldestID == "" || pending.CreatedAt.Before(oldest.CreatedAt) ||
					(pending.CreatedAt.Equal(oldest.CreatedAt) && inputID < oldestID) {
					oldest = pending
					oldestID = inputID
				}
			}
		}
		_, requestedOldest := seen[oldestID]
		if oldest.FinalHopDeferred || !requestedOldest {
			oldestID = ""
		}
		ap.mu.Unlock()
		if oldestID != "" {
			ap.schedulePendingRuntimeInputReconcile(oldestID, oldest.CreatedAt, true)
		}
		// Resolve only after all synchronous state/scheduling side effects are
		// complete. An admitted Uncertain callback must not continue mutating the
		// retired Process after Stop is allowed to publish replacement.
		for receipt, lease := range leasesToResolve {
			if lease != nil {
				lease.resolve(receipt)
			}
		}
	}
}

func (ap *Process) startRuntimeMessageFinalHopDeferredLease(pending pendingRuntimeInput) {
	receipt := strings.TrimSpace(pending.Input.ID)
	deadline := pending.finalHopLeaseDeadline
	generation := pending.finalHopLeaseGeneration
	canceled := pending.finalHopLeaseCancel
	if receipt == "" || deadline.IsZero() || generation == 0 || canceled == nil {
		return
	}
	duration := time.Until(deadline)
	if duration < 0 {
		duration = 0
	}
	go func() {
		timer := time.NewTimer(duration)
		defer timer.Stop()
		// Never select on processDone: exited closes only after
		// drainLifecycleMutations, and this receipt owns that read lock. Likewise,
		// admission cancellation occurs before the child transport is retired and
		// is not proof that a concurrent final write cannot still succeed. The
		// six-minute deadline intentionally exceeds Codex's five-minute request
		// timeout, making Deferred expiry safe to reconcile.
		select {
		case <-timer.C:
		case <-canceled:
			return
		}
		ap.mu.Lock()
		current, ok := ap.pendingRuntimeInputs[receipt]
		if !ok || !current.CreatedAt.Equal(pending.CreatedAt) || !current.FinalHopDeferred ||
			current.finalHopLeaseGeneration != generation || !current.finalHopLeaseDeadline.Equal(deadline) ||
			current.finalHopLifecycleLease != pending.finalHopLifecycleLease {
			ap.mu.Unlock()
			return
		}
		if !current.FinalHopRuntimeTransferred {
			// A direct HTTP ResponseWriter may still complete after the deadline;
			// treating an idle/absent runtime as replayable Uncertain would duplicate
			// q1 when that late write succeeds. Fail closed durably instead. If a
			// a late complete write reports Delivered after retirement may have
			// replaced the session/store generation, so the safe terminal Failed record
			// is intentionally not rewritten through this old Process.
			current.reconcileGeneration++
			current.reconcileRunning = true
			ap.pendingRuntimeInputs[receipt] = current
			token := pendingRuntimeInputMutationToken{
				inputID:    receipt,
				createdAt:  current.CreatedAt,
				generation: current.reconcileGeneration,
			}
			ap.mu.Unlock()

			cause := errors.New("direct MCP final response exceeded its bounded owner lease")
			record := ap.ownershipRecord(current, deliveryOwnershipFailed, current.ReconcileAttempts, cause)
			record.FailureCode = "direct_mcp_final_response_lease_expired"
			committed, err := ap.commitPendingRuntimeInputTransition(token, record, true, func(claim pendingRuntimeInput) {
				cancelRuntimeDeliveryLeaseTimerLocked(&claim)
				delete(ap.pendingRuntimeInputs, receipt)
				if len(ap.pendingRuntimeInputs) == 0 {
					ap.pendingRuntimeInputs = nil
				}
				ap.rememberFailedMessageLocked(strings.TrimSpace(claim.Message.MessageID))
			})
			if err != nil || !committed {
				return
			}
			if current.finalHopLifecycleLease != nil {
				// The Failed tombstone is durable, but Stop/replacement remains
				// fenced until the diagnostic and retry/resume bookkeeping below
				// have also completed.
				defer current.finalHopLifecycleLease.resolve(receipt)
			}
			ap.markResumeSafetyIfQuiescent()
			ap.emitRuntimeEvent(RuntimeEvent{
				MessageID:  strings.TrimSpace(current.Message.MessageID),
				WakeToken:  strings.TrimSpace(current.WakeToken),
				EventClass: consts.RuntimeEventClassWakeFailed,
				Target:     strings.TrimSpace(current.Message.Target),
				Detail: boundDeliveryOwnershipErrorDetail(
					"direct MCP final response exceeded its bounded owner lease; delivery failed closed without replay",
				),
			})
			ap.scheduleInboxRetry(true)
			return
		}
		// Release this exact lease token while still holding the state lock. A
		// concurrent Deferred renewal either wins first (making this token stale)
		// or observes false and cannot resurrect a lease after expiry.
		cancelRuntimeDeliveryLeaseTimerLocked(&current)
		current.FinalHopDeferred = false
		current.FinalHopRuntimeTransferred = false
		current.finalHopLeaseGeneration++
		current.finalHopLeaseDeadline = time.Time{}
		lifecycleLease := current.finalHopLifecycleLease
		current.finalHopLifecycleLease = nil
		ap.pendingRuntimeInputs[receipt] = current
		stopping := ap.stopping || ap.state == processStateStopped || ap.deliveryAdmissionLifetimeContext().Err() != nil
		ap.mu.Unlock()
		if lifecycleLease != nil {
			lifecycleLease.resolve(receipt)
		}
		if !stopping {
			ap.scheduleOldestPendingRuntimeInputReconcile(true)
		}
	}()
}

func (ap *Process) rememberRuntimeDeliveryReceiptLocked(receipt string) {
	receipt = strings.TrimSpace(receipt)
	if !types.IsRuntimeDeliveryReceipt(receipt) {
		return
	}
	if ap.runtimeDeliveryReceiptSet == nil {
		ap.runtimeDeliveryReceiptSet = make(map[string]struct{})
	}
	if _, exists := ap.runtimeDeliveryReceiptSet[receipt]; exists {
		return
	}
	ap.runtimeDeliveryReceiptSet[receipt] = struct{}{}
	ap.runtimeDeliveryReceipts = append(ap.runtimeDeliveryReceipts, receipt)
	if len(ap.runtimeDeliveryReceipts) <= maxDeliveryReceipts {
		return
	}
	evicted := ap.runtimeDeliveryReceipts[0]
	ap.runtimeDeliveryReceipts = ap.runtimeDeliveryReceipts[1:]
	delete(ap.runtimeDeliveryReceiptSet, evicted)
}

// RememberRuntimeAckedMessage records a delivery message ID that the runtime has
// already consumed or acknowledged. Dispatcher later reconciles these IDs against
// daemon-local queued deliveries to avoid replaying the same routed message.
func (ap *Process) RememberRuntimeAckedMessage(messageID string) {
	trimmedMessageID := strings.TrimSpace(messageID)
	if trimmedMessageID == "" {
		return
	}

	ap.mu.Lock()
	defer ap.mu.Unlock()
	ap.rememberRuntimeAckedMessageLocked(trimmedMessageID)
}

func (ap *Process) rememberRuntimeAckedMessageLocked(messageID string) {
	if messageID == "" {
		return
	}
	if ap.ackedSet == nil {
		ap.ackedSet = make(map[string]struct{}, 1)
	}
	if _, exists := ap.ackedSet[messageID]; exists {
		return
	}

	ap.ackedMessages = append(ap.ackedMessages, messageID)
	ap.ackedSet[messageID] = struct{}{}
	if len(ap.ackedMessages) <= maxTerminalMessages {
		return
	}

	evicted := ap.ackedMessages[0]
	ap.ackedMessages = ap.ackedMessages[1:]
	delete(ap.ackedSet, evicted)
}

// ReconcileRuntimeAckedMessages removes messageIDs that were previously recorded
// with RememberRuntimeAckedMessage and returns the suppressed IDs. Unknown IDs are
// left tracked so later daemon-local delivery reconciliation can still suppress them.
func (ap *Process) ReconcileRuntimeAckedMessages(messageIDs []string) []string {
	if len(messageIDs) == 0 {
		return nil
	}

	seen := make(map[string]struct{}, len(messageIDs))
	for _, messageID := range messageIDs {
		trimmed := strings.TrimSpace(messageID)
		if trimmed == "" {
			continue
		}
		seen[trimmed] = struct{}{}
	}
	if len(seen) == 0 {
		return nil
	}
	if !ap.beginLifecycleMutation() {
		return nil
	}
	defer ap.endLifecycleMutation()

	ap.mu.Lock()
	if len(ap.ackedMessages) == 0 {
		ap.mu.Unlock()
		return nil
	}

	suppressed := make([]string, 0, len(seen))
	next := make([]string, 0, len(ap.ackedMessages))
	for _, messageID := range ap.ackedMessages {
		if _, ok := seen[messageID]; ok {
			suppressed = append(suppressed, messageID)
			delete(ap.ackedSet, messageID)
			continue
		}
		next = append(next, messageID)
	}
	ap.ackedMessages = next
	if len(ap.ackedMessages) == 0 {
		ap.ackedSet = nil
	}
	ap.mu.Unlock()
	if len(suppressed) == 0 {
		return nil
	}
	// Keep the bounded durable terminal tombstone. Deleting it here would let a
	// stale Process generation recreate queued/pending ownership after the server
	// had already reconciled this acked ID.
	return suppressed
}

// ReconcileInboxMessages removes daemon-owned inbox entries that the runtime
// already consumed through another local delivery.
func (ap *Process) ReconcileInboxMessages(messageIDs []string) int {
	if len(messageIDs) == 0 {
		return 0
	}

	seen := make(map[string]struct{}, len(messageIDs))
	for _, messageID := range messageIDs {
		trimmed := strings.TrimSpace(messageID)
		if trimmed == "" {
			continue
		}
		seen[trimmed] = struct{}{}
	}
	if len(seen) == 0 {
		return 0
	}
	if !ap.beginLifecycleMutation() {
		return 0
	}
	defer ap.endLifecycleMutation()

	ap.mu.Lock()
	if len(ap.inbox) == 0 {
		ap.mu.Unlock()
		return 0
	}

	matched := make([]types.InboxMessage, 0, len(seen))
	reservations := make(map[string]uint64, len(seen))
	sessionID := strings.TrimSpace(ap.SessionID)
	for _, msg := range ap.inbox {
		messageID := strings.TrimSpace(msg.MessageID)
		if _, ok := seen[messageID]; ok {
			if ap.deliveryReconcileReservations == nil {
				ap.deliveryReconcileReservations = make(map[string]uint64)
			}
			ap.deliveryReconcileGeneration++
			generation := ap.deliveryReconcileGeneration
			ap.deliveryReconcileReservations[messageID] = generation
			reservations[messageID] = generation
			matched = append(matched, msg)
		}
	}
	ap.mu.Unlock()
	if len(matched) == 0 {
		return 0
	}

	durable := make(map[string]struct{}, len(matched))
	for _, msg := range matched {
		messageID := strings.TrimSpace(msg.MessageID)
		record := DeliveryOwnershipRecord{
			Message:    msg,
			Input:      drivercontract.RuntimeInput{ID: messageID},
			SessionID:  sessionID,
			State:      DeliveryOwnershipTerminal,
			Outcome:    DeliveryOutcomeAcked,
			AcceptedAt: time.Now().UTC(),
			UpdatedAt:  time.Now().UTC(),
		}
		if err := ap.persistTerminalDeliveryOwnership(record); err != nil {
			slog.Debug("persist reconciled inbox delivery as acked failed; retaining local owner",
				"agent_id", ap.AgentID, "message_id", messageID, "error", err)
			continue
		}
		durable[messageID] = struct{}{}
	}

	ap.mu.Lock()
	next := make([]types.InboxMessage, 0, len(ap.inbox))
	removed := 0
	for _, msg := range ap.inbox {
		messageID := strings.TrimSpace(msg.MessageID)
		generation, reserved := reservations[messageID]
		currentGeneration, stillReserved := ap.deliveryReconcileReservations[messageID]
		if _, ok := durable[messageID]; ok && reserved && stillReserved && currentGeneration == generation {
			removed++
			ap.rememberRuntimeAckedMessageLocked(messageID)
			ap.freshness.MarkObserved(msg.TargetKey, msg.TargetSeq, messageID)
			delete(ap.deliveryReconcileReservations, messageID)
			continue
		}
		next = append(next, msg)
	}
	for messageID, generation := range reservations {
		if ap.deliveryReconcileReservations[messageID] == generation {
			delete(ap.deliveryReconcileReservations, messageID)
		}
	}
	if len(ap.deliveryReconcileReservations) == 0 {
		ap.deliveryReconcileReservations = nil
	}
	ap.inbox = next
	ap.mu.Unlock()
	ap.markResumeSafetyIfQuiescent()
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
	return removed
}
func (ap *Process) injectFinalVisibleReplyCheck() bool {
	ap.mu.Lock()
	if ap.stopping ||
		ap.pendingWake == nil ||
		!ap.pendingWakeNeedsFinalCheckLocked() {
		ap.mu.Unlock()
		return false
	}
	sessionID := strings.TrimSpace(ap.SessionID)
	stdinPipe := ap.stdinPipe
	structuredDriver, hasStructuredDriver := ap.driver.(drivercontract.RuntimeInputDeliveryDriver)
	ap.mu.Unlock()

	if sessionID == "" || (!hasStructuredDriver && stdinPipe == nil) {
		return false
	}
	if err := ap.markResumeSafetyUnsafe(); err != nil {
		slog.Error("agent final visible reply check blocked by resume-safety persistence", "agent_id", ap.AgentID, "error", err)
		return false
	}
	ap.mu.Lock()
	if ap.stopping || ap.pendingWake == nil || !ap.pendingWakeNeedsFinalCheckLocked() {
		ap.mu.Unlock()
		return false
	}
	deliveries := ap.pendingDeliveriesLocked()
	ap.markPendingDeliveriesCheckedLocked()
	ap.isIdle = false
	ap.mu.Unlock()
	carrier, compileErr := ap.compileFinalVisibleReplyPromptForDeliveries(deliveries, ap.driver.Name())
	if compileErr != nil {
		return false
	}
	text := carrier.SlotText("turnInput")
	if hasStructuredDriver {
		deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(context.Background())
		err := structuredDriver.DeliverRuntimeInput(deliveryCtx, drivercontract.RuntimeInput{
			ID:             promptcompiler.NewEventID(promptcompiler.EventFinalVisibleReplyCheck),
			CompiledPrompt: &carrier,
		}, sessionID)
		cancel()
		if err != nil {
			slog.Error("agent final visible reply check protocol delivery error", "agent_id", ap.AgentID, "error", err)
			return false
		}
	} else {
		data := ap.driver.EncodeStdinMessage(text, sessionID)
		if len(data) == 0 || strings.TrimSpace(string(data)) == "{}" {
			return false
		}
		if _, err := stdinPipe.Write(data); err != nil {
			slog.Error("agent final visible reply check stdin write error", "agent_id", ap.AgentID, "error", err)
			return false
		}
	}
	ap.markRuntimeHandoffSucceeded()
	ap.recordGeneratedPromptCompileMetrics(
		promptcompiler.EventFinalVisibleReplyCheck,
		promptcompiler.GeneratedFinalReplyCheck,
	)
	ap.emitParsedEvent(newPromptInputParsedEvent(
		text,
		consts.PromptKindFinalVisibleReplyCheck,
		ap.driver.Name(),
		promptcompiler.EventFinalVisibleReplyCheck,
		sessionID,
	))
	return true
}

func (ap *Process) finalVisibleReplyCheckPrompt(target string) string {
	return ap.finalVisibleReplyCheckPromptForDeliveries([]pendingDelivery{{Wake: WakeInfo{Target: target, DeliveryClass: DeliveryClassVisibleReply}, Status: pendingDeliveryChecked}})
}

func (ap *Process) finalVisibleReplyCheckPromptForDeliveries(deliveries []pendingDelivery) string {
	return promptcompiler.FinalVisibleReplyCheckPrompt(ap.finalVisibleReplyTargetsForDeliveries(deliveries), ap.finalVisibleReplySendTool())
}

func (ap *Process) finalVisibleReplyTargetsForDeliveries(deliveries []pendingDelivery) []promptcompiler.FinalVisibleReplyTarget {
	type targetGroup struct {
		target       string
		messageID    string
		hasTargetKey bool
		count        int
	}
	groups := make([]targetGroup, 0, len(deliveries))
	unresolved := make([]promptcompiler.FinalVisibleReplyTarget, 0, len(deliveries))
	groupIndexes := make(map[string]int, len(deliveries))
	for _, delivery := range deliveries {
		if delivery.Wake.DeliveryClass == DeliveryClassSignalOnly {
			continue
		}
		target := strings.TrimSpace(delivery.Wake.Target)
		messageID := strings.TrimSpace(delivery.Wake.MessageID)
		if target == "" {
			unresolved = append(unresolved, promptcompiler.FinalVisibleReplyTarget{MessageID: messageID})
			continue
		}
		targetKey := strings.TrimSpace(delivery.Wake.TargetKey)
		groupKey := "target:" + target
		if targetKey != "" {
			groupKey = "target-key:" + targetKey
		}
		if index, ok := groupIndexes[groupKey]; ok {
			groups[index].count++
			continue
		}
		groupIndexes[groupKey] = len(groups)
		groups = append(groups, targetGroup{target: target, messageID: messageID, hasTargetKey: targetKey != "", count: 1})
	}
	targets := make([]promptcompiler.FinalVisibleReplyTarget, 0, len(unresolved)+len(groups))
	targets = append(targets, unresolved...)
	for _, group := range groups {
		targets = append(targets, promptcompiler.FinalVisibleReplyTarget{Target: group.target, MessageID: group.messageID, HasTargetKey: group.hasTargetKey, Count: group.count})
	}
	return targets
}

func (ap *Process) compileFinalVisibleReplyPromptForDeliveries(deliveries []pendingDelivery, runtime string) (promptcompiler.CompiledPrompt, error) {
	compiler := promptcompiler.New(promptcompiler.DefaultPolicy())
	eventType := promptcompiler.EventFinalVisibleReplyCheck
	return compiler.CompileGeneratedPrompt(context.Background(), eventType, promptcompiler.NewEventID(eventType), promptcompiler.GeneratedPromptInput{
		Kind: promptcompiler.GeneratedFinalReplyCheck,
		FinalReply: promptcompiler.FinalVisibleReplyCheckTemplateData{
			Wakes:           promptcompiler.FinalVisibleReplyCheckWakes(ap.finalVisibleReplyTargetsForDeliveries(deliveries)),
			SendMessageTool: ap.finalVisibleReplySendTool(),
		},
	}, strings.TrimSpace(runtime))
}

func (ap *Process) finalVisibleReplySendTool() string {
	sendTool := ap.driver.MCPToolPrefix() + "message_send"
	if strings.TrimSpace(sendTool) == "" {
		sendTool = "message_send"
	}
	return sendTool
}

// deliverMessageViaStdin formats and writes a message directly to the agent's
// stdin pipe. Used when the agent is idle and a new message arrives.
func (ap *Process) deliverMessageViaStdin(msg types.InboxMessage, deliveryMode string, wakeToken string) error {
	_, err := ap.deliverMessageViaStdinWithContext(context.Background(), msg, deliveryMode, wakeToken)
	return err
}

func (ap *Process) deliverMessageViaStdinWithContext(ctx context.Context, msg types.InboxMessage, deliveryMode string, wakeToken string) (pendingRuntimeInput, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	ap.mu.Lock()
	sessionID := ap.SessionID
	stdinPipe := ap.stdinPipe
	ap.mu.Unlock()

	if stdinPipe == nil || strings.TrimSpace(sessionID) == "" {
		return pendingRuntimeInput{}, errors.New("stdin delivery unavailable")
	}

	compiled, err := ap.compileDeliveryPromptWithContext(ctx, msg, deliveryMode, wakeToken)
	if err != nil {
		return pendingRuntimeInput{}, err
	}
	deliveryCtx, cancel := ap.newProtocolDeliveryContextFor(ctx)
	input := ap.runtimeInputForDelivery(deliveryCtx, msg, compiled)
	cancel()
	if input.Empty() {
		return pendingRuntimeInput{}, errors.New("stdin delivery produced empty input")
	}
	if strings.TrimSpace(input.ID) == "" {
		return pendingRuntimeInput{}, errors.New("stdin delivery requires a runtime input id")
	}

	// Attachment preparation and compiler creation may complete after the
	// Process is fenced, but raw stdin is still a runtime mutation. Leave the
	// durable queued owner for replacement rather than writing to an exited
	// generation.
	if !ap.AcceptingDeliveries() {
		return pendingRuntimeInput{}, ErrProcessNotAcceptingDelivery
	}
	data := ap.driver.EncodeStdinMessage(input.PromptText(), sessionID)
	if data == nil {
		// The driver cannot encode the message right now (e.g. a compaction
		// turn is active). The caller re-queues the message so it is picked up
		// when the next turn_end / runtime_idle drains the inbox.
		return pendingRuntimeInput{}, errors.New("stdin encoder returned nil")
	}
	if len(data) == 0 {
		return pendingRuntimeInput{}, errors.New("stdin encoder returned empty input")
	}

	// Raw stdin has no transport acknowledgement or correlated replay query. Its
	// only safe crash boundary is therefore a durable pending owner immediately
	// before Write: queued is replayable, while pending/uncertain is fail-closed.
	// Persisting after Write would leave a process-crash window in which the old
	// queued record is replayed even though the runtime already consumed bytes.
	pending := pendingRuntimeInput{
		Message:        msg,
		Input:          input,
		SessionID:      sessionID,
		WakeToken:      strings.TrimSpace(wakeToken),
		CreatedAt:      time.Now().UTC(),
		InflightReason: DeliveryInflightDispatching,
	}
	if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipDispatching, 0, nil)); err != nil {
		return pendingRuntimeInput{}, err
	}
	ap.rememberPendingRuntimeInput(pending)

	if !ap.beginLifecycleMutation() {
		ap.forgetPendingRuntimeInput(pending.Input.ID, pending.CreatedAt)
		return pendingRuntimeInput{}, ErrProcessNotAcceptingDelivery
	}
	n, writeErr := stdinPipe.Write(data)
	ap.endLifecycleMutation()
	if n == len(data) && writeErr == nil {
		ap.rememberTransientRuntimeRetryInput(msg.MessageID, input, sessionID)
		ap.emitParsedEvent(newPromptInputParsedEvent(
			input.PromptText(),
			"delivery_text",
			ap.driver.Name(),
			deliveryPromptEventType(msg),
			sessionID,
		))
		return pending, nil
	}

	if writeErr == nil {
		writeErr = io.ErrShortWrite
	}
	if n < 0 || n > len(data) {
		writeErr = fmt.Errorf("invalid stdin write count %d for %d bytes: %w", n, len(data), writeErr)
	}
	writeFailure := fmt.Errorf("stdin write accepted %d of %d bytes: %w", n, len(data), writeErr)
	slog.Error("agent stdin write error", "agent_id", ap.AgentID, "written", n, "total", len(data), "error", writeErr)

	if n == 0 {
		// No bytes crossed the transport boundary, so exact replay is safe—but
		// only after queued ownership is itself durable. If that fsync fails, the
		// prior pending owner remains authoritative and we must fail closed.
		if err := ap.persistQueuedDelivery(msg, pending.CreatedAt, 0, writeFailure); err == nil {
			ap.forgetPendingRuntimeInput(input.ID, pending.CreatedAt)
			return pendingRuntimeInput{}, writeFailure
		} else {
			writeFailure = errors.Join(writeFailure, fmt.Errorf("restore queued stdin ownership: %w", err))
		}
	}

	// Any positive/invalid write count is ambiguous: the runtime may have parsed
	// a complete message even when Write reported a short count or error. Record
	// uncertain when possible; a failed uncertain fsync still leaves the already
	// durable pending record live, which is equally non-replayable.
	pending.InflightReason = DeliveryInflightOutcomeUnknown
	if err := ap.persistDeliveryOwnership(ap.ownershipRecord(pending, deliveryOwnershipOutcomeUnknown, 0, writeFailure)); err != nil {
		slog.Error("persist uncertain stdin ownership failed; retaining durable pending owner",
			"agent_id", ap.AgentID, "message_id", input.ID, "error", err)
		writeFailure = errors.Join(writeFailure, err)
	}
	ap.mu.Lock()
	if current, ok := ap.pendingRuntimeInputs[input.ID]; ok && current.CreatedAt.Equal(pending.CreatedAt) {
		current.InflightReason = DeliveryInflightOutcomeUnknown
		ap.pendingRuntimeInputs[input.ID] = current
	}
	ap.mu.Unlock()
	return pending, errors.Join(errStdinDeliveryOutcomeUncertain, writeFailure)
}

// retireAmbiguousStdinTransport permanently fences a raw byte stream after a
// partial/ambiguous frame. It intentionally does not set stopping: waitProcess
// must expose this as an unexpected runtime failure so Dispatcher can
// auto-continue accepted work in a replacement Process. Marking state stopped
// immediately prevents a newer delivery from appending bytes to the damaged
// stream before the child exits. The terminal failed transition is committed
// synchronously while the caller still owns deliveryGate; waitProcess cannot
// publish a replacement until this function releases that gate.
func (ap *Process) retireAmbiguousStdinTransport(pending pendingRuntimeInput, cause error) {
	inputID := strings.TrimSpace(pending.Input.ID)
	ap.commitDeliveryPromptCompileMetrics(inputID)
	ap.mu.Lock()
	if ap.runtimeErrorDetail == "" {
		ap.runtimeErrorDetail = boundDeliveryOwnershipErrorDetail(
			fmt.Sprintf("raw stdin delivery transport failed unexpectedly: %v", cause),
		)
	}
	current, transitionToken, claimed := ap.claimPendingRuntimeInputTransitionLocked(inputID, pending.CreatedAt, nil)
	ap.state = processStateStopped
	ap.isIdle = false
	ap.stopInboxRetryLocked()
	stdinPipe := ap.stdinPipe
	ap.stdinPipe = nil
	proc := ap.cmd
	terminate := ap.terminateFunc
	ap.mu.Unlock()

	if stdinPipe != nil {
		logIgnoredCleanupError("failed to close ambiguous runtime stdin pipe", stdinPipe.Close(), "agent_id", ap.AgentID)
	}
	terminateAgentRuntime(proc, terminate)

	if !claimed {
		return
	}
	record := ap.ownershipRecord(current, deliveryOwnershipFailed, current.ReconcileAttempts, cause)
	record.FailureCode = "raw_stdin_transport_ambiguous"
	if committed, err := ap.commitAmbiguousRawStdinFailure(transitionToken, record); err != nil {
		slog.Error("persist terminal ambiguous stdin ownership failed; retaining live owner",
			"agent_id", ap.AgentID, "message_id", inputID, "error", err)
	} else if committed {
		ap.markResumeSafetyIfQuiescent()
	}
}

// commitAmbiguousRawStdinFailure is the one terminal mutation allowed after
// state=stopped for an unexpected raw-transport retirement. It still rejects
// explicit Stop/Sleep (stopping=true), and it holds ap.mu across each Put so a
// concurrent public Stop is ordered before or after the durable transition.
// Unexpected waitProcess may set state=stopped again, but cannot run onExit
// until the caller releases deliveryGate.
func (ap *Process) commitAmbiguousRawStdinFailure(token pendingRuntimeInputMutationToken, record DeliveryOwnershipRecord) (bool, error) {
	var lastErr error
	for attempt := 1; ; attempt++ {
		ap.mu.Lock()
		current, ok := ap.currentPendingRuntimeInputForTokenLocked(token)
		if !ok || ap.stopping {
			ap.mu.Unlock()
			return false, lastErr
		}
		err := ap.persistDeliveryOwnershipLocked(record)
		if err == nil {
			delete(ap.pendingRuntimeInputs, token.inputID)
			if len(ap.pendingRuntimeInputs) == 0 {
				ap.pendingRuntimeInputs = nil
			}
			ap.rememberFailedMessageLocked(strings.TrimSpace(current.Message.MessageID))
			ap.mu.Unlock()
			return true, nil
		}
		lastErr = err
		delay := ap.inboxRetryDelayForAttemptLocked(attempt)
		ap.mu.Unlock()

		if delay <= 0 {
			delay = processInboxRetryInitialDelay
		}
		timer := time.NewTimer(delay)
		poll := time.NewTicker(10 * time.Millisecond)
		aborted := false
		for !aborted {
			select {
			case <-timer.C:
				aborted = true
			case <-poll.C:
				ap.mu.Lock()
				stopping := ap.stopping
				ap.mu.Unlock()
				if stopping {
					timer.Stop()
					poll.Stop()
					return false, lastErr
				}
			}
		}
		poll.Stop()
	}
}

func (ap *Process) compileDeliveryPromptWithContext(ctx context.Context, msg types.InboxMessage, deliveryMode string, wakeToken string) (promptcompiler.CompiledPrompt, error) {
	if ctx == nil {
		ctx = context.Background()
	}
	compiler := promptcompiler.New(promptcompiler.DefaultPolicy())
	eventType := deliveryPromptEventType(msg)
	desc := promptcompiler.CompileDescriptor{
		EventID:   promptcompiler.NewEventID(eventType),
		EventKind: promptcompiler.EventDeliveryEnvelope,
		EventType: eventType,
	}
	ap.mu.Lock()
	sessionID := ap.SessionID
	workflowContext := msg.ResolvedWorkflowContext()
	if strings.TrimSpace(wakeToken) == "" {
		wakeToken = ap.pendingWakeToken
	}
	ap.mu.Unlock()
	activeTurnID := ""
	if provider, ok := ap.driver.(drivercontract.ActiveTurnIDProvider); ok {
		activeTurnID = strings.TrimSpace(provider.CurrentActiveTurnID())
	}
	threadID := ""
	if strings.TrimSpace(ap.driver.Name()) == "codex" {
		threadID = strings.TrimSpace(sessionID)
	}
	author := msg.AuthorIdentity()
	attachmentResolution := promptcompiler.AttachmentResolution{}
	if messageAttachments := msg.Attachments(); len(messageAttachments) > 0 {
		ap.mu.Lock()
		resolver := ap.attachmentResolver
		ap.mu.Unlock()
		if resolver != nil {
			resolved, resolveErr := resolver.ResolveRuntimeAttachments(ctx, messageAttachments)
			if resolveErr != nil {
				attachmentResolution = promptcompiler.AttachmentResolutionError(messageAttachments, resolveErr)
			} else {
				attachmentResolution = promptcompiler.AttachmentResolution{Prepared: resolved}
			}
		}
	}
	input := promptcompiler.CompileInput{
		AgentID: ap.AgentID,
		Runtime: ap.driver.Name(),
		WorkDir: ap.workDir,
		Delivery: promptcompiler.DeliveryCompileInput{
			MessageID:           strings.TrimSpace(msg.MessageID),
			TargetKey:           strings.TrimSpace(msg.TargetKey),
			TargetSeq:           msg.TargetSeq,
			Target:              strings.TrimSpace(msg.Target),
			ChannelType:         strings.TrimSpace(msg.ChannelType),
			ChannelName:         strings.TrimSpace(msg.ChannelName),
			AuthorIdentity:      author,
			Content:             msg.Content,
			Timestamp:           strings.TrimSpace(msg.Timestamp),
			MessageType:         strings.TrimSpace(msg.MessageType),
			Attachments:         msg.Attachments(),
			ResolvedAttachments: append([]types.RuntimeAttachment(nil), attachmentResolution.Prepared...),
			AttachmentFailure:   attachmentResolution.Failure,
			Metadata:            msg.Metadata,
			TeamContexts:        msg.TeamContexts,
			WorkflowContext:     workflowContext,
			DeliveryMode:        strings.TrimSpace(deliveryMode),
			RuntimeSessionID:    strings.TrimSpace(sessionID),
			ThreadID:            threadID,
			ActiveTurnID:        activeTurnID,
			WakeToken:           strings.TrimSpace(wakeToken),
			ActiveTaskTarget:    msg.ResolvedActiveTaskTarget(),
		},
	}
	desc.TurnID = activeTurnID
	// This is the compiler/attachment phase of an already admitted delivery.
	// Its bounded caller/local timeout remains authoritative; the process
	// retirement lifetime only fences gate acquisition and runtime transport so
	// an unexpected exit cannot abort before the first durable owner is written.
	ctx, cancel := newDeliveryPromptContextFor(ctx)
	defer cancel()

	compiled, err := compiler.CompileDeliveryPrompt(ctx, eventType, desc.EventID, input, ap.driver.Name())
	if err != nil {
		ap.emitRuntimeEvent(RuntimeEvent{
			MessageID:  strings.TrimSpace(msg.MessageID),
			WakeToken:  strings.TrimSpace(wakeToken),
			EventClass: consts.RuntimeEventClassWakeFailed,
			Target:     strings.TrimSpace(msg.Target),
			Detail:     err.Error(),
		})
		return promptcompiler.CompiledPrompt{}, err
	}
	ap.stageDeliveryPromptCompileMetrics(strings.TrimSpace(msg.MessageID), compiled.Envelope(), compiled.Rendered())
	return compiled, nil
}

func (ap *Process) renderDeliveryTextWithContext(ctx context.Context, msg types.InboxMessage, deliveryMode string, wakeToken string) (string, error) {
	compiled, err := ap.compileDeliveryPromptWithContext(ctx, msg, deliveryMode, wakeToken)
	if err != nil {
		return "", err
	}
	return compiled.SlotText("deliveryText"), nil
}

func (ap *Process) runtimeInputForDelivery(_ context.Context, msg types.InboxMessage, compiled promptcompiler.CompiledPrompt) drivercontract.RuntimeInput {
	input := drivercontract.RuntimeInput{
		ID:             strings.TrimSpace(msg.MessageID),
		CompiledPrompt: &compiled,
	}
	if delivery := compiled.Envelope().Delivery; delivery != nil && len(delivery.ResolvedAttachments) > 0 {
		if _, ok := ap.driver.(drivercontract.RuntimeInputDeliveryDriver); ok {
			input.Attachments = append([]types.RuntimeAttachment(nil), delivery.ResolvedAttachments...)
		}
	}
	return input
}

func (ap *Process) deliverProtocolRuntimeInput(ctx context.Context, driver drivercontract.BusyStdinMessageDriver, input drivercontract.RuntimeInput, sessionID string) error {
	if structured, ok := ap.driver.(drivercontract.RuntimeInputDeliveryDriver); ok {
		return structured.DeliverRuntimeInput(ctx, input, sessionID)
	}
	return driver.DeliverBusyStdinInput(ctx, input, sessionID)
}

func formatStdinDeliveredMessage(workDir string, msg types.InboxMessage) string {
	compiler := promptcompiler.New(promptcompiler.DefaultPolicy())
	eventType := deliveryPromptEventType(msg)
	ctx, cancel := newDeliveryPromptContext()
	defer cancel()

	compiled, err := compiler.CompileRuntimeInput(ctx, promptcompiler.CompileDescriptor{
		EventID:   promptcompiler.NewEventID(eventType),
		EventKind: promptcompiler.EventDeliveryEnvelope,
		EventType: eventType,
	}, promptcompiler.CompileInput{
		WorkDir: workDir,
		Delivery: promptcompiler.DeliveryCompileInput{
			MessageID:        strings.TrimSpace(msg.MessageID),
			TargetKey:        strings.TrimSpace(msg.TargetKey),
			TargetSeq:        msg.TargetSeq,
			Target:           strings.TrimSpace(msg.Target),
			ChannelType:      strings.TrimSpace(msg.ChannelType),
			ChannelName:      strings.TrimSpace(msg.ChannelName),
			AuthorIdentity:   msg.AuthorIdentity(),
			Content:          msg.Content,
			Timestamp:        strings.TrimSpace(msg.Timestamp),
			MessageType:      strings.TrimSpace(msg.MessageType),
			Attachments:      msg.Attachments(),
			Metadata:         msg.Metadata,
			TeamContexts:     msg.TeamContexts,
			WorkflowContext:  msg.ResolvedWorkflowContext(),
			DeliveryMode:     "stdin_delivery",
			ActiveTaskTarget: msg.ResolvedActiveTaskTarget(),
		},
	})
	if err != nil {
		return ""
	}
	rendered, err := compiler.RenderRuntimeInput(ctx, compiled, "stdin", promptcompiler.RenderOptions{})
	if err != nil {
		return ""
	}
	return rendered.SlotText("deliveryText")
}

func deliveryPromptEventType(msg types.InboxMessage) promptcompiler.CompileEventType {
	if strings.TrimSpace(msg.MessageType) == "reaction" {
		return promptcompiler.EventReactionDelivery
	}
	return promptcompiler.EventRoutedMessageDelivery
}

func newPromptInputParsedEvent(
	text string,
	promptKind string,
	runtime string,
	eventType promptcompiler.CompileEventType,
	sessionID string,
) drivercontract.ParsedEvent {
	return drivercontract.ParsedEvent{
		Kind:       consts.ParsedEventKindPromptInput,
		Text:       strings.TrimSpace(text),
		PromptKind: strings.TrimSpace(promptKind),
		Runtime:    strings.TrimSpace(runtime),
		EventType:  string(eventType),
		SessionID:  strings.TrimSpace(sessionID),
	}
}

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