package agent

import (
	"context"
	"errors"
	"fmt"
	"io"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

type memoryDeliveryOwnershipStore struct {
	mu      sync.Mutex
	records map[string]DeliveryOwnershipRecord
}

type reclaimingDeliveryOwnershipStore struct {
	*memoryDeliveryOwnershipStore
	reclaimed []DeliveryOwnershipRecord
}

func (s *reclaimingDeliveryOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	if _, err := s.memoryDeliveryOwnershipStore.PutDeliveryOwnership(record); err != nil {
		return nil, err
	}
	reclaimed := s.reclaimed
	s.reclaimed = nil
	return reclaimed, nil
}

type rejectAckedDeliveryOwnershipStore struct {
	base *memoryDeliveryOwnershipStore
}

type rejectQueuedDeliveryOwnershipStore struct {
	base *memoryDeliveryOwnershipStore
}

type blockingRuntimeInputReconciler struct {
	busyDeliveryStubDriver
	started  chan struct{}
	canceled chan struct{}
	once     sync.Once
}

func (d *blockingRuntimeInputReconciler) ReconcileRuntimeInput(ctx context.Context, _, _ string) (RuntimeInputReconciliation, error) {
	d.once.Do(func() { close(d.started) })
	<-ctx.Done()
	close(d.canceled)
	return RuntimeInputReconciliationUnknown, ctx.Err()
}

type writeCloserFunc func([]byte) (int, error)

func (fn writeCloserFunc) Write(p []byte) (int, error) { return fn(p) }
func (writeCloserFunc) Close() error                   { return nil }

type retiringShortWriteCloser struct {
	closed chan struct{}
	once   sync.Once
}

func (w *retiringShortWriteCloser) Write([]byte) (int, error) { return 1, nil }
func (w *retiringShortWriteCloser) Close() error {
	w.once.Do(func() { close(w.closed) })
	return nil
}

type blockingDeliveryOwnershipStore struct {
	base      *memoryDeliveryOwnershipStore
	messageID string
	state     DeliveryOwnershipState
	outcome   DeliveryOwnershipOutcome
	entered   chan struct{}
	release   chan struct{}
	once      sync.Once
}

type failingBlockedTerminalOwnershipStore struct {
	base     *memoryDeliveryOwnershipStore
	entered  chan struct{}
	release  chan struct{}
	once     sync.Once
	mu       sync.Mutex
	attempts int
}

func (s *failingBlockedTerminalOwnershipStore) LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error) {
	return s.base.LoadDeliveryOwnership()
}

func (s *failingBlockedTerminalOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	if record.IsFailed() {
		s.mu.Lock()
		s.attempts++
		s.mu.Unlock()
		s.once.Do(func() { close(s.entered) })
		<-s.release
		return nil, errors.New("simulated terminal fsync failure")
	}
	return s.base.PutDeliveryOwnership(record)
}

func (s *failingBlockedTerminalOwnershipStore) DeleteDeliveryOwnership(messageID string) error {
	return s.base.DeleteDeliveryOwnership(messageID)
}

func (s *failingBlockedTerminalOwnershipStore) attemptCount() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.attempts
}

func (s *blockingDeliveryOwnershipStore) LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error) {
	return s.base.LoadDeliveryOwnership()
}

func (s *blockingDeliveryOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	if record.MessageID() == s.messageID && record.State == s.state && (s.outcome == "" || record.Outcome == s.outcome) {
		s.once.Do(func() { close(s.entered) })
		<-s.release
	}
	return s.base.PutDeliveryOwnership(record)
}

func (s *blockingDeliveryOwnershipStore) DeleteDeliveryOwnership(messageID string) error {
	return s.base.DeleteDeliveryOwnership(messageID)
}

func (s *rejectAckedDeliveryOwnershipStore) LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error) {
	return s.base.LoadDeliveryOwnership()
}

func (s *rejectAckedDeliveryOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	if record.IsAcked() || record.IsInflight(DeliveryInflightOutcomeUnknown) {
		return nil, errors.New("simulated final-hop ownership fsync failure")
	}
	return s.base.PutDeliveryOwnership(record)
}

func (s *rejectAckedDeliveryOwnershipStore) DeleteDeliveryOwnership(messageID string) error {
	return s.base.DeleteDeliveryOwnership(messageID)
}

func (s *rejectQueuedDeliveryOwnershipStore) LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error) {
	return s.base.LoadDeliveryOwnership()
}

func (s *rejectQueuedDeliveryOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	if record.IsQueued() {
		return nil, errors.New("simulated queued ownership fsync failure")
	}
	return s.base.PutDeliveryOwnership(record)
}

func (s *rejectQueuedDeliveryOwnershipStore) DeleteDeliveryOwnership(messageID string) error {
	return s.base.DeleteDeliveryOwnership(messageID)
}

func (s *memoryDeliveryOwnershipStore) LoadDeliveryOwnership() ([]DeliveryOwnershipRecord, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	result := make([]DeliveryOwnershipRecord, 0, len(s.records))
	for _, record := range s.records {
		result = append(result, record)
	}
	return result, nil
}

func (s *memoryDeliveryOwnershipStore) PutDeliveryOwnership(record DeliveryOwnershipRecord) ([]DeliveryOwnershipRecord, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.records == nil {
		s.records = map[string]DeliveryOwnershipRecord{}
	}
	if prior, ok := s.records[record.MessageID()]; ok && !prior.AcceptedAt.IsZero() {
		record.AcceptedAt = prior.AcceptedAt
	}
	s.records[record.MessageID()] = record
	return nil, nil
}

func (s *memoryDeliveryOwnershipStore) DeleteDeliveryOwnership(messageID string) error {
	s.mu.Lock()
	delete(s.records, messageID)
	s.mu.Unlock()
	return nil
}

func (s *memoryDeliveryOwnershipStore) record(messageID string) (DeliveryOwnershipRecord, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	record, ok := s.records[messageID]
	return record, ok
}

func newCodexDeliveryOwnershipTestProcess(t *testing.T, driver *CodexAppServerDriver, store DeliveryOwnershipStore, sessionID string) *Process {
	t.Helper()
	done := make(chan struct{})
	driver.launchPermissionMode = types.AgentPermissionModeFullAccess
	driver.launchWorkDir = t.TempDir()
	ap := &Process{
		AgentID:   "agent-codex-delivery-ownership",
		SessionID: sessionID,
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
			cancel: func() {},
		},
		processLifecycleState: processLifecycleState{
			state:    processStateRunning,
			isIdle:   true,
			exitCode: -1,
			exited:   done,
		},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore:       store,
			inboxRetryDelay:              func(int) time.Duration { return time.Hour },
			pendingReconcileSuccessGrace: 5 * time.Millisecond,
			pendingReconcileMaxElapsed:   time.Hour,
		},
		processRecoveryState: processRecoveryState{
			workDir:         t.TempDir(),
			disableRecovery: true,
		},
	}
	t.Cleanup(func() {
		stopInboxRetryTestProcess(ap)
		select {
		case <-done:
		default:
			close(done)
		}
	})
	return ap
}

func countCapturedRPCMethod(requests []capturedRPCRequest, method string) int {
	count := 0
	for _, request := range requests {
		if request.Method == method {
			count++
		}
	}
	return count
}

func TestRuntimeInputMutationNotAttemptedOverridesAmbiguousCause(t *testing.T) {
	for _, cause := range []error{
		context.DeadlineExceeded,
		errors.New("read timeout"),
		fmt.Errorf("%w: transport disconnected", errCodexAppServerTransportAmbiguous),
	} {
		err := fmt.Errorf("%w: delivery preflight failed: %w", errRuntimeInputMutationNotAttempted, cause)
		require.False(t, isAmbiguousRuntimeDeliveryError(err))
		require.Equal(t, deliveryRetryTransient, classifyDeliveryRetryError(err))
	}
}

func TestKimiWireTransportAmbiguityIsClassifiedAsUncertain(t *testing.T) {
	require.True(t, isAmbiguousRuntimeDeliveryError(
		fmt.Errorf("kimi prompt failed: %w", errors.Join(errKimiWireTransportAmbiguous, context.Canceled)),
	))
}

func newRawStdinOwnershipTestProcess(store DeliveryOwnershipStore, stdin io.WriteCloser) *Process {
	return &Process{
		AgentID:   "agent-raw-stdin-ownership",
		SessionID: "claude-session",
		processRuntimeBinding: processRuntimeBinding{
			driver:    &busyDeliveryStubDriver{},
			stdinPipe: stdin,
			cancel:    func() {},
		},
		processLifecycleState: processLifecycleState{
			state:    processStateRunning,
			isIdle:   true,
			exitCode: -1,
			exited:   make(chan struct{}),
		},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			inboxRetryDelay:        func(int) time.Duration { return time.Hour },
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
}

func TestRawStdinDeliveryPersistsPendingBeforePhysicalWrite(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	entered := make(chan struct{})
	release := make(chan struct{})
	var once sync.Once
	stdin := writeCloserFunc(func(p []byte) (int, error) {
		once.Do(func() { close(entered) })
		<-release
		return len(p), nil
	})
	ap := newRawStdinOwnershipTestProcess(store, stdin)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	done := make(chan error, 1)
	go func() {
		done <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "stdin-pending-before-write",
			Target:    "#ops",
			Content:   "crash-window body",
		})
	}()
	<-entered

	record, ok := store.record("stdin-pending-before-write")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightDispatching), "the replayable queued owner must be replaced before stdin can consume bytes")
	ap.mu.Lock()
	pending, pendingOK := ap.pendingRuntimeInputs["stdin-pending-before-write"]
	ap.mu.Unlock()
	require.True(t, pendingOK)
	require.Equal(t, DeliveryInflightDispatching, pending.InflightReason)

	close(release)
	require.NoError(t, <-done)
	record, ok = store.record("stdin-pending-before-write")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	_, pendingOK = ap.pendingRuntimeInputs["stdin-pending-before-write"]
	ap.mu.Unlock()
	require.False(t, pendingOK)
}

func TestRawStdinShortWriteIsUncertainAndNeverReplayed(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{}
	failedTransitionEntered := make(chan struct{})
	failedTransitionRelease := make(chan struct{})
	store := &blockingDeliveryOwnershipStore{
		base:      base,
		messageID: "stdin-short-write",
		state:     DeliveryOwnershipTerminal, outcome: DeliveryOutcomeFailed,
		entered: failedTransitionEntered,
		release: failedTransitionRelease,
	}
	stdin := &retiringShortWriteCloser{closed: make(chan struct{})}
	first := newRawStdinOwnershipTestProcess(store, stdin)

	deliveryDone := make(chan error, 1)
	go func() {
		deliveryDone <- first.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "stdin-short-write",
			Target:    "#ops",
			Content:   "must not be replayed after a partial write",
		})
	}()
	<-failedTransitionEntered
	record, ok := base.record("stdin-short-write")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
	select {
	case <-stdin.closed:
	case <-time.After(time.Second):
		t.Fatal("ambiguous raw stdin stream was not retired")
	}
	q2Done := make(chan error, 1)
	go func() {
		q2Done <- first.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "stdin-q2-before-retirement",
			Target:    "#ops",
			Content:   "must wait behind terminal q1",
		})
	}()
	select {
	case err := <-q2Done:
		t.Fatalf("q2 crossed the ambiguous q1 terminal durability boundary: %v", err)
	case <-time.After(20 * time.Millisecond):
	}

	close(failedTransitionRelease)
	require.NoError(t, <-deliveryDone)
	require.ErrorIs(t, <-q2Done, ErrProcessNotAcceptingDelivery)
	require.Eventually(t, func() bool {
		record, ok := base.record("stdin-short-write")
		return ok && record.IsFailed()
	}, time.Second, 5*time.Millisecond)
	first.mu.Lock()
	state := first.state
	_, pending := first.pendingRuntimeInputs["stdin-short-write"]
	inbox := append([]types.InboxMessage(nil), first.inbox...)
	first.mu.Unlock()
	require.Equal(t, processStateStopped, state)
	require.True(t, first.ShouldAutoContinueAcceptedDeliveries(),
		"ambiguous stdin retirement must remain an unexpected-exit continuation")
	require.False(t, pending)
	require.Empty(t, inbox, "ambiguous bytes must never regain a replayable FIFO owner")

	// Raw stdin has no correlated reconciliation API. A replacement therefore
	// observes the terminal failed owner rather than writing the exact q1 body a
	// second time, then can continue q2 on a clean stream.
	var replacementMu sync.Mutex
	var replacementPayloads []string
	replacement := newRawStdinOwnershipTestProcess(base, writeCloserFunc(func(p []byte) (int, error) {
		replacementMu.Lock()
		replacementPayloads = append(replacementPayloads, string(p))
		replacementMu.Unlock()
		return len(p), nil
	}))
	t.Cleanup(func() { stopInboxRetryTestProcess(replacement) })
	require.NoError(t, replacement.SetDeliveryOwnershipStore(base))
	require.NoError(t, replacement.DeliverForHandoff(context.Background(), types.InboxMessage{
		MessageID: "stdin-q2-after-replacement",
		Target:    "#ops",
		Content:   "q2 continues on clean replacement",
	}))
	replacementMu.Lock()
	payloads := append([]string(nil), replacementPayloads...)
	replacementMu.Unlock()
	require.Len(t, payloads, 1)
	require.Contains(t, payloads[0], "q2 continues on clean replacement")
	require.NotContains(t, payloads[0], "must not be replayed after a partial write")
	record, ok = base.record("stdin-short-write")
	require.True(t, ok)
	require.True(t, record.IsFailed())
	q2Record, ok := base.record("stdin-q2-after-replacement")
	require.True(t, ok)
	require.True(t, q2Record.IsAcked())
}

func TestRawStdinTerminalFailureRetainsLiveOwnerAndStopHasNoLateWrite(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{}
	store := &failingBlockedTerminalOwnershipStore{
		base:    base,
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	stdin := &retiringShortWriteCloser{closed: make(chan struct{})}
	ap := newRawStdinOwnershipTestProcess(store, stdin)

	deliveryDone := make(chan error, 1)
	go func() {
		deliveryDone <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "stdin-terminal-fsync-fails",
			Target:    "#ops",
			Content:   "must remain fail-closed",
		})
	}()
	<-store.entered
	select {
	case <-stdin.closed:
	case <-time.After(time.Second):
		t.Fatal("ambiguous stdin transport was not physically retired before terminal retry")
	}

	stopDone := make(chan struct{})
	go func() { ap.Stop(); close(stopDone) }()
	select {
	case <-stopDone:
		t.Fatal("Stop returned while terminal ownership Put was still in flight")
	case <-time.After(20 * time.Millisecond):
	}
	close(store.release)
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not cancel terminal retry and drain deliveryGate")
	}
	require.NoError(t, <-deliveryDone)

	record, ok := base.record("stdin-terminal-fsync-fails")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown), "failed terminal fsync must retain the non-replayable live owner")
	attempts := store.attemptCount()
	time.Sleep(30 * time.Millisecond)
	require.Equal(t, attempts, store.attemptCount(), "no ownership write may begin after Stop completes")
}

func TestRawStdinZeroByteFailureRestoresQueuedOwnership(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	writeErr := errors.New("stdin closed before accepting bytes")
	stdin := writeCloserFunc(func([]byte) (int, error) { return 0, writeErr })
	ap := newRawStdinOwnershipTestProcess(store, stdin)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{
		MessageID: "stdin-zero-write",
		Target:    "#ops",
		Content:   "safe to replay",
	}))
	record, ok := store.record("stdin-zero-write")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	ap.mu.Lock()
	_, pending := ap.pendingRuntimeInputs["stdin-zero-write"]
	queued := append([]types.InboxMessage(nil), ap.inbox...)
	idle := ap.isIdle
	ap.mu.Unlock()
	require.False(t, pending)
	require.Len(t, queued, 1)
	require.Equal(t, "stdin-zero-write", queued[0].MessageID)
	require.True(t, idle, "a definite no-write failure may restore the prior idle state")
}

func TestIdleProtocolDeliveryFailureAtomicallyRestoresIdleAndQueues(t *testing.T) {
	driver := &protocolDeliveryStubDriver{deliverErr: errors.New("runtime temporarily rejected turn/start")}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-idle-rollback",
		SessionID:             "session-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			inboxRetryDelay:        func(int) time.Duration { return time.Hour },
		},
		processRecoveryState: processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	err := ap.deliverMessage(types.InboxMessage{MessageID: "msg-1", Target: "#ops", Content: "hello"}, false)

	require.NoError(t, err)
	require.True(t, ap.IsIdle(), "failed idle turn/start must restore the prior runtime phase")
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "msg-1", ap.inbox[0].MessageID)
	ap.mu.Unlock()
	record, ok := store.record("msg-1")
	require.True(t, ok)
	require.True(t, record.IsQueued())
}

func TestEmptyMessageIDIsRejectedBeforeDurableOrRuntimeAcceptance(t *testing.T) {
	driver := &inboxRetryAckedDriver{failures: map[string]int{}}
	store := &memoryDeliveryOwnershipStore{}
	var eventsMu sync.Mutex
	var events []RuntimeEvent
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.deliveryOwnershipStore = store
	ap.onRuntimeEvent = func(ev RuntimeEvent) {
		eventsMu.Lock()
		events = append(events, ev)
		eventsMu.Unlock()
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	err := ap.DeliverForHandoff(context.Background(), types.InboxMessage{Target: "#ops", Content: "missing id"})
	require.ErrorContains(t, err, "non-empty message id")
	ap.Deliver(types.InboxMessage{Target: "#ops", Content: "missing id again"})

	calls, _, _, _, _ := driver.snapshot()
	require.Empty(t, calls)
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.Empty(t, ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Empty(t, records)
	eventsMu.Lock()
	require.Len(t, events, 1)
	require.Equal(t, consts.RuntimeEventClassWakeFailed, events[0].EventClass)
	eventsMu.Unlock()
}

type ackedReconcileDriver struct{ busyDeliveryStubDriver }

func (d *ackedReconcileDriver) ReconcileRuntimeInput(context.Context, string, string) (RuntimeInputReconciliation, error) {
	return RuntimeInputReconciliationAcked, nil
}

type correlatedOwnershipDriver struct {
	protocolDeliveryStubDriver
	mu               sync.Mutex
	runtimeErr       error
	interruptErr     error
	lastInput        drivercontract.RuntimeInput
	runtimeCalls     int
	interruptCalls   int
	reconcileCalls   int
	reconcileResult  RuntimeInputReconciliation
	reconcileErr     error
	reconcileResults map[string]RuntimeInputReconciliation
	reconcileBlocks  map[string]<-chan struct{}
	reconcileStarted chan string
	reconcileIDs     []string
	onRuntimeDeliver func(drivercontract.RuntimeInput)
	onInterrupt      func(drivercontract.RuntimeInput)
}

func (d *correlatedOwnershipDriver) UsesExplicitTurnLifecycle() bool { return true }

func (d *correlatedOwnershipDriver) DeliverRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.mu.Lock()
	d.lastInput = input
	d.runtimeCalls++
	err := d.runtimeErr
	onDeliver := d.onRuntimeDeliver
	d.mu.Unlock()
	if onDeliver != nil {
		onDeliver(input)
	}
	return err
}

func (d *correlatedOwnershipDriver) DeliverInterruptingRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.mu.Lock()
	d.lastInput = input
	d.interruptCalls++
	err := d.interruptErr
	onInterrupt := d.onInterrupt
	d.mu.Unlock()
	if onInterrupt != nil {
		onInterrupt(input)
	}
	return err
}

func (d *correlatedOwnershipDriver) ReconcileRuntimeInput(_ context.Context, inputID string, _ string) (RuntimeInputReconciliation, error) {
	d.mu.Lock()
	d.reconcileCalls++
	result, err := d.reconcileResult, d.reconcileErr
	if candidate, ok := d.reconcileResults[inputID]; ok {
		result = candidate
	}
	block := d.reconcileBlocks[inputID]
	started := d.reconcileStarted
	d.reconcileIDs = append(d.reconcileIDs, inputID)
	d.mu.Unlock()
	if started != nil {
		started <- inputID
	}
	if block != nil {
		<-block
	}
	return result, err
}

func (d *correlatedOwnershipDriver) snapshot() (drivercontract.RuntimeInput, int, int, int) {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.lastInput, d.runtimeCalls, d.interruptCalls, d.reconcileCalls
}

func (d *correlatedOwnershipDriver) reconcileSnapshot() []string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return append([]string(nil), d.reconcileIDs...)
}

func TestRestoredAckedInputDoesNotMakeFreshProcessBusy(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"msg-old": {
			Message:   types.InboxMessage{MessageID: "msg-old", TargetKey: "thread-1", TargetSeq: 7, Target: "#ops:root", Content: "old body"},
			Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-old"}, "old body"),
			SessionID: "thread-old",
			State:     DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching,
			AcceptedAt: time.Now().Add(-time.Minute),
		},
	}}
	var events []RuntimeEvent
	var eventsMu sync.Mutex
	ap := &Process{
		AgentID:   "agent-reconcile",
		SessionID: "thread-fresh",
		processRuntimeBinding: processRuntimeBinding{
			driver: &ackedReconcileDriver{}, cancel: func() {},
			onRuntimeEvent: func(ev RuntimeEvent) {
				eventsMu.Lock()
				events = append(events, ev)
				eventsMu.Unlock()
			},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		ap.mu.Lock()
		_, pending := ap.pendingRuntimeInputs["msg-old"]
		idle := ap.isIdle
		ap.mu.Unlock()
		eventsMu.Lock()
		defer eventsMu.Unlock()
		return !pending && idle && len(events) == 1 && events[0].EventClass == consts.RuntimeEventClassWakeFailed
	}, time.Second, 5*time.Millisecond)
	record, ok := store.record("msg-old")
	require.True(t, ok)
	require.True(t, record.IsAcked())
}

func TestCorrelatedDirectSuccessStaysPendingUntilGraceReconcile(t *testing.T) {
	driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationAcked}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-direct-correlated",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore:       store,
			pendingReconcileSuccessGrace: 75 * time.Millisecond,
		},
		processRecoveryState: processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{
		MessageID: "msg-direct", Target: "#ops", Content: "direct body",
	}))
	input, runtimeCalls, _, reconcileCalls := driver.snapshot()
	require.Equal(t, "msg-direct", input.ID)
	require.Equal(t, 1, runtimeCalls)
	require.Zero(t, reconcileCalls)
	ap.mu.Lock()
	require.Contains(t, ap.pendingRuntimeInputs, "msg-direct")
	_, acked := ap.ackedSet["msg-direct"]
	ap.mu.Unlock()
	require.False(t, acked, "RPC success alone is not model-visible")
	record, ok := store.record("msg-direct")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightDispatching))

	require.Eventually(t, func() bool {
		ap.mu.Lock()
		_, pending := ap.pendingRuntimeInputs["msg-direct"]
		_, acked := ap.ackedSet["msg-direct"]
		ap.mu.Unlock()
		_, _, _, reconcileCalls := driver.snapshot()
		return !pending && acked && reconcileCalls == 1
	}, time.Second, 5*time.Millisecond)
	record, ok = store.record("msg-direct")
	require.True(t, ok)
	require.True(t, record.IsAcked())
}

func TestAmbiguousCorrelatedDirectRestartsWithoutDuplicateReplay(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	firstDriver := &correlatedOwnershipDriver{
		runtimeErr:      context.DeadlineExceeded,
		reconcileResult: RuntimeInputReconciliationPending,
	}
	first := &Process{
		AgentID:               "agent-direct-restart",
		SessionID:             "thread-old",
		processRuntimeBinding: processRuntimeBinding{driver: firstDriver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store, pendingReconcileMaxElapsed: time.Hour},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}

	require.NoError(t, first.DeliverForHandoff(context.Background(), types.InboxMessage{MessageID: "msg-ambiguous", Target: "#ops", Content: "once"}))
	input, runtimeCalls, _, _ := firstDriver.snapshot()
	require.Equal(t, "msg-ambiguous", input.ID)
	require.Equal(t, 1, runtimeCalls)
	record, ok := store.record("msg-ambiguous")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
	stopInboxRetryTestProcess(first)

	secondDriver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationAcked}
	second := &Process{
		AgentID:               "agent-direct-restart",
		SessionID:             "thread-fresh",
		processRuntimeBinding: processRuntimeBinding{driver: secondDriver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(second) })
	require.NoError(t, second.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		record, ok := store.record("msg-ambiguous")
		return ok && record.IsAcked()
	}, time.Second, 5*time.Millisecond)
	_, secondRuntimeCalls, _, reconcileCalls := secondDriver.snapshot()
	require.Zero(t, secondRuntimeCalls, "restart reconciliation must not replay an ambiguous request")
	require.Equal(t, 1, reconcileCalls)
}

func TestCodexPreflightTransportFailureRequeuesThenStartsExactlyOnce(t *testing.T) {
	const (
		threadID  = "thread-fresh-preflight"
		messageID = "msg-preflight-retry"
	)
	idleResult := rawRPCResult(t, appServerThreadResult{Thread: appServerThread{
		ID: threadID, Status: appServerThreadStatus{Type: "idle"},
	}})
	startResult := rawRPCResult(t, appServerTurnResult{Turn: appServerTurn{
		ID: "turn-preflight-retry", Status: appServerTurnStatus{Type: "inProgress"},
	}})
	ackedResult := rawRPCResult(t, appServerThreadResult{Thread: appServerThread{
		ID: threadID, Status: appServerThreadStatus{Type: "idle"},
		Turns: []appServerTurn{{
			ID:        "turn-preflight-retry",
			ItemsView: "full",
			Status:    appServerTurnStatus{Type: "completed"},
			Items: []appServerItem{{
				ID: "item-user", Type: "userMessage", ClientID: messageID,
			}},
		}},
	}})

	readCalls := 0
	startClientID := ""
	handlerError := ""
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		switch req.Method {
		case "thread/read":
			readCalls++
			switch readCalls {
			case 1:
				if req.Params["includeTurns"] != true {
					handlerError = "first lifecycle read did not request turns"
				}
				return appServerEnvelope{Error: &appServerResponseError{
					Code:    -32600,
					Message: "thread " + threadID + " is not materialized yet; includeTurns is unavailable before first user message",
				}}
			case 2:
				if req.Params["includeTurns"] != false {
					handlerError = "fresh-thread fallback did not request metadata only"
				}
				return appServerEnvelope{Error: &appServerResponseError{
					Code:    codexAppServerTransportFailureCode,
					Message: "metadata fallback response lost",
				}}
			case 3:
				return appServerEnvelope{Result: idleResult}
			case 4:
				return appServerEnvelope{Result: ackedResult}
			default:
				handlerError = fmt.Sprintf("unexpected thread/read call %d", readCalls)
				return appServerEnvelope{Error: &appServerResponseError{Code: -32603, Message: handlerError}}
			}
		case "turn/start":
			startClientID = fmt.Sprint(req.Params["clientUserMessageId"])
			return appServerEnvelope{Result: startResult}
		default:
			handlerError = "unexpected RPC method " + req.Method
			return appServerEnvelope{Error: &appServerResponseError{Code: -32603, Message: handlerError}}
		}
	})
	driver.threadID = threadID
	driver.threadStatus = appServerThreadStatus{Type: "idle"}
	store := &memoryDeliveryOwnershipStore{}
	ap := newCodexDeliveryOwnershipTestProcess(t, driver, store, threadID)
	msg := types.InboxMessage{MessageID: messageID, Target: "#ops", Content: "deliver exactly once"}

	require.NoError(t, ap.DeliverForHandoff(context.Background(), msg))
	record, ok := store.record(messageID)
	require.True(t, ok)
	require.True(t, record.IsQueued())
	require.Contains(t, record.LastError, "runtime input mutation was not attempted")
	ap.mu.Lock()
	queued := append([]types.InboxMessage(nil), ap.inbox...)
	_, pending := ap.pendingRuntimeInputs[messageID]
	idle := ap.isIdle
	ap.mu.Unlock()
	require.False(t, pending)
	require.Len(t, queued, 1)
	require.Equal(t, messageID, queued[0].MessageID)
	require.True(t, idle)
	require.Equal(t, []string{"thread/read", "thread/read"}, capturedRPCMethods(writer.requests))

	ap.scheduleInboxRetry(true)
	require.Eventually(t, func() bool {
		record, ok := store.record(messageID)
		return ok && record.IsAcked()
	}, time.Second, 5*time.Millisecond)

	require.Empty(t, handlerError)
	require.Equal(t, messageID, startClientID)
	methods := capturedRPCMethods(writer.requests)
	require.Equal(t, []string{
		"thread/read", "thread/read", "thread/read", "turn/start", "thread/read",
	}, methods)
	require.Equal(t, 1, countCapturedRPCMethod(writer.requests, "turn/start"))
	require.Zero(t, countCapturedRPCMethod(writer.requests, "turn/steer"))
	ap.mu.Lock()
	_, pending = ap.pendingRuntimeInputs[messageID]
	queued = append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()
	require.False(t, pending)
	require.Empty(t, queued)
}

func TestCodexTurnStartTransportAmbiguityStaysUncertainWithoutReplay(t *testing.T) {
	const (
		threadID  = "thread-turn-start-ambiguous"
		messageID = "msg-turn-start-ambiguous"
	)
	idleResult := rawRPCResult(t, appServerThreadResult{Thread: appServerThread{
		ID: threadID, Status: appServerThreadStatus{Type: "idle"},
	}})
	ackedResult := rawRPCResult(t, appServerThreadResult{Thread: appServerThread{
		ID: threadID, Status: appServerThreadStatus{Type: "idle"},
		Turns: []appServerTurn{{
			ID:        "turn-ambiguous",
			ItemsView: "full",
			Status:    appServerTurnStatus{Type: "completed"},
			Items: []appServerItem{{
				ID: "item-user", Type: "userMessage", ClientID: messageID,
			}},
		}},
	}})
	reconcileStarted := make(chan struct{})
	releaseReconcile := make(chan struct{})
	var reconcileStartedOnce sync.Once
	t.Cleanup(func() {
		select {
		case <-releaseReconcile:
		default:
			close(releaseReconcile)
		}
	})

	readCalls := 0
	handlerError := ""
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		switch req.Method {
		case "thread/read":
			readCalls++
			if readCalls == 1 {
				return appServerEnvelope{Result: idleResult}
			}
			if readCalls == 2 {
				reconcileStartedOnce.Do(func() { close(reconcileStarted) })
				<-releaseReconcile
				return appServerEnvelope{Result: ackedResult}
			}
			handlerError = fmt.Sprintf("unexpected thread/read call %d", readCalls)
			return appServerEnvelope{Error: &appServerResponseError{Code: -32603, Message: handlerError}}
		case "turn/start":
			return appServerEnvelope{Error: &appServerResponseError{
				Code:    codexAppServerTransportFailureCode,
				Message: "response lost after turn/start dispatch",
			}}
		default:
			handlerError = "unexpected RPC method " + req.Method
			return appServerEnvelope{Error: &appServerResponseError{Code: -32603, Message: handlerError}}
		}
	})
	driver.threadID = threadID
	driver.threadStatus = appServerThreadStatus{Type: "idle"}
	store := &memoryDeliveryOwnershipStore{}
	ap := newCodexDeliveryOwnershipTestProcess(t, driver, store, threadID)
	msg := types.InboxMessage{MessageID: messageID, Target: "#ops", Content: "must not replay"}

	require.NoError(t, ap.DeliverForHandoff(context.Background(), msg))
	select {
	case <-reconcileStarted:
	case <-time.After(time.Second):
		t.Fatal("correlated reconciliation did not start")
	}
	record, ok := store.record(messageID)
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
	ap.mu.Lock()
	pending, pendingOK := ap.pendingRuntimeInputs[messageID]
	queued := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()
	require.True(t, pendingOK)
	require.Equal(t, DeliveryInflightOutcomeUnknown, pending.InflightReason)
	require.Empty(t, queued)

	// An upstream replay of the same durable identity must be absorbed by the
	// uncertain owner while only read-only correlated reconciliation is allowed.
	require.NoError(t, ap.DeliverForHandoff(context.Background(), msg))
	require.Equal(t, []string{"thread/read", "turn/start", "thread/read"}, capturedRPCMethods(writer.requests))
	require.Equal(t, 1, countCapturedRPCMethod(writer.requests, "turn/start"))
	require.Zero(t, countCapturedRPCMethod(writer.requests, "turn/steer"))

	close(releaseReconcile)
	require.Eventually(t, func() bool {
		record, ok := store.record(messageID)
		return ok && record.IsAcked()
	}, time.Second, 5*time.Millisecond)
	require.Empty(t, handlerError)
	require.Equal(t, 1, countCapturedRPCMethod(writer.requests, "turn/start"))
}

func TestRestoredPendingOwnershipReconcilesStrictlyInFIFOOrder(t *testing.T) {
	acceptedAt := time.Now().Add(-time.Minute)
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"p1": {
			Message: types.InboxMessage{MessageID: "p1", Content: "first"}, Input: testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "p1"}, "first"),
			SessionID: "thread-old", State: DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching, AcceptedAt: acceptedAt,
		},
		"p2": {
			Message: types.InboxMessage{MessageID: "p2", Content: "second"}, Input: testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "p2"}, "second"),
			SessionID: "thread-old", State: DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching, AcceptedAt: acceptedAt.Add(time.Second),
		},
	}}
	releaseP1 := make(chan struct{})
	driver := &correlatedOwnershipDriver{
		reconcileResults: map[string]RuntimeInputReconciliation{
			"p1": RuntimeInputReconciliationAcked,
			"p2": RuntimeInputReconciliationAcked,
		},
		reconcileBlocks:  map[string]<-chan struct{}{"p1": releaseP1},
		reconcileStarted: make(chan string, 2),
	}
	ap := &Process{
		AgentID:               "agent-restored-fifo",
		SessionID:             "thread-fresh",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() {
		select {
		case <-releaseP1:
		default:
			close(releaseP1)
		}
		stopInboxRetryTestProcess(ap)
	})

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Equal(t, "p1", <-driver.reconcileStarted)
	time.Sleep(20 * time.Millisecond)
	require.Equal(t, []string{"p1"}, driver.reconcileSnapshot(), "p2 must not reconcile while p1 is unresolved")

	close(releaseP1)
	require.Equal(t, "p2", <-driver.reconcileStarted)
	require.Eventually(t, func() bool {
		p1, p1OK := store.record("p1")
		p2, p2OK := store.record("p2")
		return p1OK && p2OK && p1.IsAcked() && p2.IsAcked()
	}, time.Second, 5*time.Millisecond)
	require.Equal(t, []string{"p1", "p2"}, driver.reconcileSnapshot())
}

func TestRestoredMissingPiSessionFailsClosedBeforeAdvancingFIFO(t *testing.T) {
	acceptedAt := time.Now().Add(-time.Minute)
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {
			Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Target: "#ops", Content: "must never replay"},
			Input:   drivercontract.RuntimeInput{ID: "q1"}, SessionID: "session-before-marker",
			State: DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching, AcceptedAt: acceptedAt,
		},
		"q2": {
			Message: types.InboxMessage{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Target: "#ops", Content: "advance after q1 terminal"},
			Input:   drivercontract.RuntimeInput{ID: "q2"}, SessionID: "session-before-marker",
			State: DeliveryOwnershipQueued, AcceptedAt: acceptedAt.Add(time.Second),
		},
	}}
	driver := &correlatedOwnershipDriver{
		reconcileResults: map[string]RuntimeInputReconciliation{
			"q1": RuntimeInputReconciliationUnknown,
			"q2": RuntimeInputReconciliationAcked,
		},
	}
	ap := &Process{
		AgentID:               "agent-pi-before-marker-restart",
		SessionID:             "session-replacement",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore:       store,
			inboxRetryDelay:              func(int) time.Duration { return time.Millisecond },
			pendingReconcileSuccessGrace: 5 * time.Millisecond,
			pendingReconcileMaxAttempts:  2,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	// The replacement has no old driver correlation and no session marker. Its
	// read-only unknown result must fail q1 closed; q2 may not cross that FIFO
	// head until the durable failed tombstone is committed.
	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		q1, q1OK := store.record("q1")
		q2, q2OK := store.record("q2")
		return q1OK && q1.IsFailed() && q1.FailureCode == "runtime_input_reconciliation_exhausted" && q2OK && q2.IsAcked()
	}, 2*time.Second, 5*time.Millisecond)

	calls := driver.reconcileSnapshot()
	require.GreaterOrEqual(t, len(calls), 2)
	require.Equal(t, "q1", calls[0], "q1 remains the only reconciliation head until terminal")
	lastInput, runtimeCalls, _, _ := driver.snapshot()
	require.Equal(t, 1, runtimeCalls, "only q2 may mutate the replacement after q1 fails closed")
	require.Equal(t, "q2", lastInput.ID, "q1's exact body must never be replayed")
}

func TestRestoredPendingOwnershipWithoutReconcilerFailsClosed(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"restored": {
			Message:        types.InboxMessage{MessageID: "restored", Target: "#ops", Content: "uncertain"},
			Input:          drivercontract.RuntimeInput{ID: "restored"},
			SessionID:      "session-old",
			State:          DeliveryOwnershipInflight,
			InflightReason: DeliveryInflightDispatching,
			AcceptedAt:     time.Now().Add(-time.Minute).UTC(),
		},
	}}
	ap := newInboxRetryTestProcess(&ackedRuntimeInputStubDriver{supports: true}, time.Hour)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		record, ok := store.record("restored")
		return ok &&
			record.IsFailed() &&
			record.FailureCode == "runtime_input_reconciliation_exhausted"
	}, time.Second, 5*time.Millisecond)

	ap.mu.Lock()
	_, pending := ap.pendingRuntimeInputs["restored"]
	_, failed := ap.failedSet["restored"]
	ap.mu.Unlock()
	require.False(t, pending)
	require.True(t, failed)
}

func TestKnownAckOnlyMutationFailureWithQueuedFsyncFailureFailsClosed(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{
		supports:   true,
		deliverErr: errors.New("definitive runtime rejection"),
	}
	base := &memoryDeliveryOwnershipStore{}
	store := &rejectQueuedDeliveryOwnershipStore{base: base}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	ap.deliveryOwnershipStore = store
	ap.workDir = t.TempDir()
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "known-failure",
		TargetKey: "thread-1",
		TargetSeq: 1,
		Target:    "#agentteam:thread",
		Content:   "must converge after the failed rollback fsync",
	}, false))

	require.Eventually(t, func() bool {
		record, ok := base.record("known-failure")
		return ok &&
			record.IsFailed() &&
			record.FailureCode == "runtime_input_reconciliation_exhausted"
	}, time.Second, 5*time.Millisecond)

	ap.mu.Lock()
	_, pending := ap.pendingRuntimeInputs["known-failure"]
	_, failed := ap.failedSet["known-failure"]
	ap.mu.Unlock()
	require.False(t, pending)
	require.True(t, failed)
}

func TestLiveAckedEventInvalidatesInflightReconcileMutation(t *testing.T) {
	tests := []struct {
		name         string
		result       RuntimeInputReconciliation
		reconcileErr error
		maxAttempts  int
	}{
		{name: "unknown progress", result: RuntimeInputReconciliationUnknown, maxAttempts: 3},
		{name: "provider error progress", reconcileErr: errors.New("temporary reconcile failure"), maxAttempts: 3},
		{name: "authoritative absent", result: RuntimeInputReconciliationAbsent, maxAttempts: 3},
		{name: "budget exhausted failure", result: RuntimeInputReconciliationUnknown, maxAttempts: 1},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			createdAt := time.Now().Add(-time.Minute).UTC()
			msg := types.InboxMessage{MessageID: "q1", Target: "#ops", Content: "exactly once"}
			pending := pendingRuntimeInput{
				Message:             msg,
				Input:               testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "q1"}, "exactly once"),
				SessionID:           "thread-1",
				CreatedAt:           createdAt,
				InflightReason:      DeliveryInflightDispatching,
				reconcileRunning:    true,
				reconcileGeneration: 1,
			}
			store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
				"q1": {
					Message: msg, Input: pending.Input, SessionID: pending.SessionID,
					State: DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching, AcceptedAt: createdAt,
				},
			}}
			release := make(chan struct{})
			driver := &correlatedOwnershipDriver{
				reconcileResult:  tt.result,
				reconcileErr:     tt.reconcileErr,
				reconcileBlocks:  map[string]<-chan struct{}{"q1": release},
				reconcileStarted: make(chan string, 1),
			}
			var eventsMu sync.Mutex
			var events []RuntimeEvent
			ap := &Process{
				AgentID:   "agent-reconcile-acked-race",
				SessionID: "thread-1",
				processRuntimeBinding: processRuntimeBinding{
					driver: driver, cancel: func() {},
					onRuntimeEvent: func(ev RuntimeEvent) {
						eventsMu.Lock()
						events = append(events, ev)
						eventsMu.Unlock()
					},
				},
				processLifecycleState: processLifecycleState{state: processStateRunning, exited: make(chan struct{})},
				processDeliveryState: processDeliveryState{
					deliveryOwnershipStore:      store,
					pendingRuntimeInputs:        map[string]pendingRuntimeInput{"q1": pending},
					pendingReconcileMaxAttempts: tt.maxAttempts,
					pendingReconcileMaxElapsed:  time.Hour,
				},
				processRecoveryState: processRecoveryState{disableRecovery: true},
			}
			t.Cleanup(func() {
				select {
				case <-release:
				default:
					close(release)
				}
				stopInboxRetryTestProcess(ap)
			})

			reconcileDone := make(chan struct{})
			go func() {
				ap.runPendingRuntimeInputReconcile("q1", createdAt, 1, true)
				close(reconcileDone)
			}()
			require.Equal(t, "q1", <-driver.reconcileStarted)

			require.True(t, ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
				Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: "q1", SessionID: "thread-1",
			}))
			close(release)
			select {
			case <-reconcileDone:
			case <-time.After(time.Second):
				t.Fatal("stale reconciler did not stop after acked generation won")
			}

			record, ok := store.record("q1")
			require.True(t, ok)
			require.True(t, record.IsAcked())
			_, runtimeCalls, _, _ := driver.snapshot()
			require.Zero(t, runtimeCalls, "a stale absent result must not replay a acked input")
			ap.mu.Lock()
			require.Empty(t, ap.inbox)
			require.NotContains(t, ap.pendingRuntimeInputs, "q1")
			_, acked := ap.ackedSet["q1"]
			_, failed := ap.failedSet["q1"]
			ap.mu.Unlock()
			require.True(t, acked)
			require.False(t, failed)
			eventsMu.Lock()
			require.Empty(t, events, "stale budget exhaustion must not emit WakeFailed")
			eventsMu.Unlock()
		})
	}
}

func TestMismatchedSessionAckInvalidatesHistoricalReconciler(t *testing.T) {
	createdAt := time.Now().Add(-time.Minute).UTC()
	msg := types.InboxMessage{MessageID: "q1", Target: "#ops", Content: "replay only after mismatch"}
	pending := pendingRuntimeInput{
		Message: msg, Input: testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "q1"}, msg.Content),
		SessionID: "thread-old", CreatedAt: createdAt, InflightReason: DeliveryInflightDispatching,
		Restored: true, reconcileRunning: true, reconcileGeneration: 1,
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: msg, Input: pending.Input, SessionID: pending.SessionID, State: DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching, AcceptedAt: createdAt},
	}}
	release := make(chan struct{})
	driver := &correlatedOwnershipDriver{
		reconcileResult: RuntimeInputReconciliationAcked,
		reconcileBlocks: map[string]<-chan struct{}{"q1": release}, reconcileStarted: make(chan string, 1),
	}
	ap := &Process{
		AgentID: "agent-mismatched-session-race", SessionID: "thread-new",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: false, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			pendingRuntimeInputs:   map[string]pendingRuntimeInput{"q1": pending},
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() {
		select {
		case <-release:
		default:
			close(release)
		}
		stopInboxRetryTestProcess(ap)
	})

	reconcileDone := make(chan struct{})
	go func() {
		ap.runPendingRuntimeInputReconcile("q1", createdAt, 1, true)
		close(reconcileDone)
	}()
	require.Equal(t, "q1", <-driver.reconcileStarted)
	require.True(t, ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
		Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: "q1", SessionID: "thread-new",
	}))
	close(release)
	select {
	case <-reconcileDone:
	case <-time.After(time.Second):
		t.Fatal("historical reconciler did not stop after mismatch transfer won")
	}

	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	_, runtimeCalls, _, _ := driver.snapshot()
	require.Zero(t, runtimeCalls)
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "q1", ap.inbox[0].MessageID)
	require.NotContains(t, ap.pendingRuntimeInputs, "q1")
	_, acked := ap.ackedSet["q1"]
	_, failed := ap.failedSet["q1"]
	ap.mu.Unlock()
	require.False(t, acked)
	require.False(t, failed)
}

func TestCorrelatedInterruptUsesDurablePendingAndGraceReconcile(t *testing.T) {
	driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationAcked}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-interrupt-correlated",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore:       store,
			pendingReconcileSuccessGrace: 25 * time.Millisecond,
		},
		processRecoveryState: processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{MessageID: "msg-urgent", Target: "#ops", Content: "urgent"})
	require.NoError(t, err)
	require.True(t, handled)
	input, _, interruptCalls, reconcileCalls := driver.snapshot()
	require.Equal(t, "msg-urgent", input.ID)
	require.Equal(t, 1, interruptCalls)
	require.Zero(t, reconcileCalls)
	record, ok := store.record("msg-urgent")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightDispatching))

	require.Eventually(t, func() bool {
		record, ok := store.record("msg-urgent")
		return ok && record.IsAcked()
	}, time.Second, 5*time.Millisecond)
}

func TestAuthoritativePendingUsesElapsedBudgetNotShortAttemptBudget(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"msg-pending": {
			Message:   types.InboxMessage{MessageID: "msg-pending", Content: "waiting on approval"},
			Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-pending"}, "waiting on approval"),
			SessionID: "thread-old",
			State:     DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching,
			AcceptedAt: time.Now(),
		},
	}}
	driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationPending}
	ap := &Process{
		AgentID:               "agent-long-pending",
		SessionID:             "thread-fresh",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			pendingReconcileMaxAttempts: 1,
			pendingReconcileMaxElapsed:  time.Hour,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		_, _, _, calls := driver.snapshot()
		return calls >= 2
	}, time.Second, 5*time.Millisecond)
	ap.mu.Lock()
	_, pending := ap.pendingRuntimeInputs["msg-pending"]
	_, failed := ap.failedSet["msg-pending"]
	ap.mu.Unlock()
	require.True(t, pending)
	require.False(t, failed)
	record, ok := store.record("msg-pending")
	require.True(t, ok)
	require.False(t, record.IsFailed())
}

func TestAuthoritativePendingEventuallyFailsAfterLongElapsedBudget(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"msg-expired": {
			Message:   types.InboxMessage{MessageID: "msg-expired", Target: "#ops", Content: "stuck"},
			Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-expired"}, "stuck"),
			SessionID: "thread-old",
			State:     DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching,
			AcceptedAt: time.Now().Add(-time.Hour),
		},
	}}
	driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationPending}
	var eventsMu sync.Mutex
	var events []RuntimeEvent
	ap := &Process{
		AgentID:   "agent-expired-pending",
		SessionID: "thread-fresh",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}, onRuntimeEvent: func(ev RuntimeEvent) {
			eventsMu.Lock()
			events = append(events, ev)
			eventsMu.Unlock()
		}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true, exited: make(chan struct{})},
		processDeliveryState:  processDeliveryState{pendingReconcileMaxElapsed: time.Millisecond},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.Eventually(t, func() bool {
		record, ok := store.record("msg-expired")
		eventsMu.Lock()
		defer eventsMu.Unlock()
		return ok && record.IsFailed() && len(events) == 1 && events[0].EventClass == consts.RuntimeEventClassWakeFailed
	}, time.Second, 5*time.Millisecond)
}

func TestLateCorrelatedAckAfterTurnEndRemainsIdle(t *testing.T) {
	driver := &correlatedOwnershipDriver{}
	store := &memoryDeliveryOwnershipStore{}
	createdAt := time.Now().UTC()
	msg := types.InboxMessage{MessageID: "msg-late", TargetKey: "thread-1", TargetSeq: 4, Content: "late echo"}
	ap := &Process{
		AgentID:               "agent-late-ack",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			pendingRuntimeInputs: map[string]pendingRuntimeInput{
				"msg-late": {
					Message: msg, Input: testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-late"}, "late echo"),
					SessionID: "thread-1", CreatedAt: createdAt, InflightReason: DeliveryInflightDispatching,
				},
			},
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	_, err := store.PutDeliveryOwnership(ap.ownershipRecord(ap.pendingRuntimeInputs["msg-late"], deliveryOwnershipDispatching, 0, nil))
	require.NoError(t, err)

	require.True(t, ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
		Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: "msg-late", SessionID: "thread-1",
	}))
	require.True(t, ap.IsIdle(), "a late clientId echo is ownership/freshness evidence, not a new turn-start")
	record, ok := store.record("msg-late")
	require.True(t, ok)
	require.True(t, record.IsAcked())
}

func TestCorrelatedDirectEchoBeforeRPCErrorWinsOverTransportError(t *testing.T) {
	driver := &correlatedOwnershipDriver{runtimeErr: errors.New("late RPC rejection")}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-echo-race-direct",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}
	driver.onRuntimeDeliver = func(input drivercontract.RuntimeInput) {
		ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
			Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: input.ID, SessionID: "thread-1",
		})
	}

	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{MessageID: "msg-race-direct", Content: "once"}))
	record, ok := store.record("msg-race-direct")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.NotContains(t, ap.pendingRuntimeInputs, "msg-race-direct")
	ap.mu.Unlock()
}

func TestCorrelatedInterruptEchoBeforeRPCErrorWinsOverTransportError(t *testing.T) {
	driver := &correlatedOwnershipDriver{interruptErr: errors.New("late interrupt RPC rejection")}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-echo-race-interrupt",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir(), disableRecovery: true},
	}
	driver.onInterrupt = func(input drivercontract.RuntimeInput) {
		ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
			Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: input.ID, SessionID: "thread-1",
		})
	}
	ap.mu.Lock()
	ap.addSendTargetLocked("#old-turn", sendScopeSourceConversation)
	ap.addTurnDeliveryLocked("#old-turn")
	ap.mu.Unlock()

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{
		MessageID: "msg-race-interrupt",
		Target:    "#new-turn:0234da66",
		Content:   "once",
	})
	require.True(t, handled)
	require.NoError(t, err)
	record, ok := store.record("msg-race-interrupt")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.NotContains(t, ap.pendingRuntimeInputs, "msg-race-interrupt")
	ap.mu.Unlock()
	scope, live := ap.SendScope()
	require.True(t, live)
	require.True(t, scope.ContainsTarget("#new-turn:0234da66"))
	require.False(t, scope.ContainsTarget("#old-turn"),
		"an exact ack committed inside the RPC must survive the RPC's later error")
	require.Equal(t, "#new-turn:0234da66", ap.CurrentTurnDeliveryTarget(),
		"the turn target committed by the exact ack must not roll back to the interrupted turn")
}

func TestAckedBusyEchoBeforeRPCErrorWinsOverTransportError(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	driver := &ackedRuntimeInputStubDriver{supports: true, deliverErr: errors.New("late steer rejection")}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.deliveryOwnershipStore = store
	driver.onDeliver = func(input drivercontract.RuntimeInput, sessionID string) {
		ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
			Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: input.ID, SessionID: sessionID,
		})
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{MessageID: "msg-race-acked", Content: "once"}))
	record, ok := store.record("msg-race-acked")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.NotContains(t, ap.pendingRuntimeInputs, "msg-race-acked")
	ap.mu.Unlock()
}

func TestRestoredFailedOwnershipSuppressesDuplicateWithoutAdvancingFreshness(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"msg-failed": {
			Message: types.InboxMessage{MessageID: "msg-failed", TargetKey: "thread-1", TargetSeq: 9, Content: "original"},
			State:   DeliveryOwnershipTerminal, Outcome: DeliveryOutcomeFailed,
			AcceptedAt:  time.Now().Add(-time.Minute),
			FailureCode: "delivery_retry_exhausted",
		},
	}}
	driver := &inboxRetryAckedDriver{failures: map[string]int{}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-failed",
		TargetKey: "thread-1",
		TargetSeq: 9,
		Content:   "upstream duplicate",
	}, false))
	time.Sleep(10 * time.Millisecond)

	calls, _, _, _, _ := driver.snapshot()
	require.Empty(t, calls)
	ap.mu.Lock()
	_, failed := ap.failedSet["msg-failed"]
	require.True(t, failed)
	require.Empty(t, ap.inbox)
	require.Empty(t, ap.pendingRuntimeInputs)
	frontier, observed := ap.freshness.Snapshot("thread-1")
	require.False(t, observed, "failed is terminal dedup evidence, not a acked freshness observation")
	require.Zero(t, frontier.ObservedSeq)
	ap.mu.Unlock()
}

func TestLateAckUpgradesFailedTombstoneWithoutReplaying(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"msg-late-failed": {
			Message:     types.InboxMessage{MessageID: "msg-late-failed", TargetKey: "thread-1", TargetSeq: 10, Content: "original"},
			Input:       drivercontract.RuntimeInput{ID: "msg-late-failed"},
			SessionID:   "thread-1",
			State:       DeliveryOwnershipTerminal,
			Outcome:     DeliveryOutcomeFailed,
			AcceptedAt:  time.Now().Add(-time.Minute),
			FailureCode: "runtime_input_reconciliation_exhausted",
		},
	}}
	ap := newInboxRetryTestProcess(&inboxRetryAckedDriver{failures: map[string]int{}}, time.Millisecond)
	ap.deliveryOwnershipStore = store
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	ap.handleRuntimeInputAcked(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "msg-late-failed",
		SessionID:      "thread-1",
	})
	record, ok := store.record("msg-late-failed")
	require.True(t, ok)
	require.True(t, record.IsAcked(), "late positive evidence upgrades the durable tombstone")
	ap.mu.Lock()
	frontier, observed := ap.freshness.Snapshot("thread-1")
	ap.mu.Unlock()
	require.True(t, observed)
	require.Equal(t, int64(10), frontier.ObservedSeq)
}

func TestRestoredTerminalRingSuppressesOldestDuplicate(t *testing.T) {
	records := make(map[string]DeliveryOwnershipRecord, maxTerminalMessages)
	now := time.Now().Add(-time.Hour)
	for i := 0; i < maxTerminalMessages; i++ {
		messageID := fmt.Sprintf("terminal-%03d", i)
		records[messageID] = DeliveryOwnershipRecord{
			Message: types.InboxMessage{MessageID: messageID},
			State:   DeliveryOwnershipTerminal, Outcome: DeliveryOutcomeFailed,
			AcceptedAt: now.Add(time.Duration(i) * time.Millisecond),
		}
	}
	store := &memoryDeliveryOwnershipStore{records: records}
	driver := &inboxRetryAckedDriver{failures: map[string]int{}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	require.NoError(t, ap.deliverMessage(types.InboxMessage{MessageID: "terminal-000", Content: "duplicate"}, false))
	calls, _, _, _, _ := driver.snapshot()
	require.Empty(t, calls)
}

func TestReclaimedInflightRetainsLateAckCorrelation(t *testing.T) {
	createdAt := time.Now().UTC()
	failed := DeliveryOwnershipRecord{
		Message: types.InboxMessage{MessageID: "reclaimed-inflight", Target: "#ops"},
		Input:   drivercontract.RuntimeInput{ID: "reclaimed-inflight"},
		State:   DeliveryOwnershipTerminal, Outcome: DeliveryOutcomeFailed,
		AcceptedAt: createdAt, FailureCode: "ownership_backlog_reclaimed",
	}
	base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"reclaimed-inflight": failed,
	}}
	store := &reclaimingDeliveryOwnershipStore{memoryDeliveryOwnershipStore: base, reclaimed: []DeliveryOwnershipRecord{failed}}
	driver := &ackedRuntimeInputStubDriver{supports: true}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	ap.deliveryOwnershipStore = store
	ap.pendingRuntimeInputs = map[string]pendingRuntimeInput{
		"reclaimed-inflight": {
			Message: failed.Message, Input: drivercontract.RuntimeInput{ID: "reclaimed-inflight"},
			SessionID: ap.SessionID, CreatedAt: createdAt, InflightReason: DeliveryInflightOutcomeUnknown,
		},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.persistQueuedDelivery(types.InboxMessage{MessageID: "fresh"}, time.Time{}, 0, nil))
	ap.mu.Lock()
	pending := ap.pendingRuntimeInputs["reclaimed-inflight"]
	ap.mu.Unlock()
	require.True(t, pending.reclaimed)
	var wakeEvents []WakeStatusEvent
	ap.onWakeStatus = func(event WakeStatusEvent) { wakeEvents = append(wakeEvents, event) }
	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind: consts.ParsedEventKindRuntimeInputAcked, RuntimeInputID: "reclaimed-inflight", SessionID: ap.SessionID,
	})
	record, ok := base.record("reclaimed-inflight")
	require.True(t, ok)
	require.True(t, record.IsAcked(), "a reclaimed acknowledgement must upgrade the durable record to acked")
	ap.mu.Lock()
	wakeActive := ap.pendingWake != nil
	ap.mu.Unlock()
	require.False(t, wakeActive, "a reclaimed acknowledgement must never begin a wake")
	require.Empty(t, wakeEvents, "a reclaimed acknowledgement must never emit received wake status")
}

func TestReclaimedInflightReleasesWakeCapacityAndLiveOwnershipScans(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	createdAt := time.Now().UTC()
	ap.mu.Lock()
	ap.pendingRuntimeInputs = make(map[string]pendingRuntimeInput, processInboxMaxMessages)
	for i := 0; i < processInboxMaxMessages; i++ {
		id := fmt.Sprintf("pending-%03d", i)
		ap.pendingRuntimeInputs[id] = pendingRuntimeInput{
			Message:        types.InboxMessage{MessageID: id, TargetKey: "#ops", TargetSeq: int64(i + 1)},
			Input:          drivercontract.RuntimeInput{ID: id},
			SessionID:      ap.SessionID,
			CreatedAt:      createdAt.Add(time.Duration(i) * time.Millisecond),
			InflightReason: DeliveryInflightOutcomeUnknown,
		}
	}
	require.Equal(t, processInboxMaxMessages, ap.pendingWakeLifecycleOwnerCountLocked())
	require.False(t, ap.canAcceptWakeLifecycleOwnerLocked("fresh"))
	ap.mu.Unlock()

	// Reclaiming one inflight owner converts it to a correlation-only lease and
	// must free its wake-capacity slot rather than holding the 512-boundary gate.
	failed := DeliveryOwnershipRecord{
		Message:     types.InboxMessage{MessageID: "pending-000", Target: "#ops"},
		Input:       drivercontract.RuntimeInput{ID: "pending-000"},
		State:       DeliveryOwnershipTerminal,
		Outcome:     DeliveryOutcomeFailed,
		AcceptedAt:  createdAt,
		FailureCode: "ownership_backlog_reclaimed",
	}
	ap.applyReclaimedDeliveryOwnership([]DeliveryOwnershipRecord{failed})

	ap.mu.Lock()
	require.True(t, ap.pendingRuntimeInputs["pending-000"].reclaimed)
	require.Equal(t, processInboxMaxMessages-1, ap.pendingWakeLifecycleOwnerCountLocked())
	require.True(t, ap.canAcceptWakeLifecycleOwnerLocked("fresh"))
	require.True(t, ap.hasLocalLiveDeliveryOwnershipLocked())
	ap.mu.Unlock()

	// Once every live owner is reclaimed, only correlation leases remain: the
	// Process must not stay busy, hold recovery, or advertise freshness coverage
	// from terminal-failed owners.
	ap.mu.Lock()
	for id, pending := range ap.pendingRuntimeInputs {
		pending.reclaimed = true
		ap.pendingRuntimeInputs[id] = pending
	}
	require.False(t, ap.hasLocalLiveDeliveryOwnershipLocked())
	require.Zero(t, ap.pendingWakeLifecycleOwnerCountLocked())
	require.True(t, ap.canAcceptWakeLifecycleOwnerLocked("fresh"))
	ap.mu.Unlock()
	require.False(t, ap.PendingRuntimeInputCoversFreshness("#ops", 1))
}

func TestReservedDurableHandoffDeliversOldestBeforeRestoredFIFO(t *testing.T) {
	acceptedAt := time.Now().Add(-time.Minute)
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", Content: "first"}, State: DeliveryOwnershipQueued, AcceptedAt: acceptedAt},
		"q2": {Message: types.InboxMessage{MessageID: "q2", Content: "second"}, State: DeliveryOwnershipQueued, AcceptedAt: acceptedAt.Add(time.Second)},
	}}
	driver := &inboxRetryAckedDriver{failures: map[string]int{}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStoreForHandoff(store, "q1"))
	time.Sleep(10 * time.Millisecond)
	calls, _, _, _, _ := driver.snapshot()
	require.Empty(t, calls, "store restore must not race the explicit initial handoff")
	ap.mu.Lock()
	require.Equal(t, "q1", ap.deliveryOwnershipHandoffID)
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "q2", ap.inbox[0].MessageID)
	ap.mu.Unlock()

	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{MessageID: "q1", Content: "first"}))
	require.Eventually(t, func() bool {
		calls, _, _, _, _ := driver.snapshot()
		return len(calls) == 2
	}, time.Second, 2*time.Millisecond)
	calls, _, _, _, _ = driver.snapshot()
	require.Equal(t, []string{"q1", "q2"}, calls)
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightDispatching))
}

func TestStoppedProcessRejectsDurableHandoffWithoutAcceptingOwnership(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	ap := newInboxRetryTestProcess(&inboxRetryAckedDriver{failures: map[string]int{}}, time.Millisecond)
	ap.deliveryOwnershipStore = store
	ap.mu.Lock()
	ap.stopping = true
	ap.state = processStateStopped
	ap.mu.Unlock()

	err := ap.DeliverForHandoff(context.Background(), types.InboxMessage{
		MessageID: "msg-after-stop",
		Target:    "#ops",
		Content:   "must go to a replacement process",
	})
	require.ErrorIs(t, err, ErrProcessNotAcceptingDelivery)
	_, ok := store.record("msg-after-stop")
	require.False(t, ok)
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.Empty(t, ap.pendingRuntimeInputs)
	ap.mu.Unlock()
}

func TestFreshnessPageDurablyClaimsBeforeConsumingQueuedOwners(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"stale": {Message: types.InboxMessage{MessageID: "stale", TargetKey: "thread-1", TargetSeq: 3}, State: DeliveryOwnershipQueued},
		"fresh": {Message: types.InboxMessage{MessageID: "fresh", TargetKey: "thread-1", TargetSeq: 6}, State: DeliveryOwnershipQueued},
	}}
	ap := &Process{
		AgentID:               "agent-freshness-durable",
		SessionID:             "thread-runtime",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateStarting},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			inbox: []types.InboxMessage{
				{MessageID: "stale", TargetKey: "thread-1", TargetSeq: 3, Content: "already covered"},
				{MessageID: "fresh", TargetKey: "thread-1", TargetSeq: 6, Content: "page body"},
			},
		},
	}
	ap.freshness.MarkObserved("thread-1", 5, "frontier")

	page := ap.InboxMessagesForFreshnessPage("thread-1", 6, 10)
	require.Len(t, page.Messages, 1)
	require.Equal(t, "fresh", page.Messages[0].MessageID)
	require.Len(t, page.RuntimeDeliveryClaims, 1)
	require.Equal(t, "fresh", page.RuntimeDeliveryClaims[0].MessageID)
	require.True(t, types.IsRuntimeDeliveryReceipt(page.RuntimeDeliveryClaims[0].Receipt))
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	receiptPending, pendingOK := ap.pendingRuntimeInputs[page.RuntimeDeliveryClaims[0].Receipt]
	_, _, duplicateOwned := ap.pendingRuntimeInputByMessageIDLocked("fresh")
	ap.mu.Unlock()
	require.True(t, pendingOK)
	require.Equal(t, "fresh", receiptPending.Message.MessageID)
	require.True(t, duplicateOwned, "opaque receipt keys must still deduplicate the original AgentTeam MessageID")
	require.NoError(t, ap.DeliverForHandoff(context.Background(), types.InboxMessage{
		MessageID: "fresh", TargetKey: "thread-1", TargetSeq: 6, Content: "duplicate upstream retry",
	}))
	ap.mu.Lock()
	require.Empty(t, ap.inbox, "a duplicate upstream q1 must not create a queued owner beside receipt r1")
	require.Len(t, ap.pendingRuntimeInputs, 1)
	ap.mu.Unlock()
	stale, ok := store.record("stale")
	require.True(t, ok)
	require.True(t, stale.IsAcked())
	fresh, ok := store.record("fresh")
	require.True(t, ok)
	require.True(t, fresh.IsInflight(DeliveryInflightOutcomeUnknown))
	require.Equal(t, page.RuntimeDeliveryClaims[0].Receipt, fresh.Input.ID)
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
		[]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
	)
}

func TestDirectFinalHopOwnerFencesStopAndCommitsAdmittedProcess(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	oldProcess := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
	oldProcess.deliveryOwnershipStore = store
	oldProcess.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "exact body"}}
	page := oldProcess.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 1, "runtime-1")
	require.Len(t, page.RuntimeDeliveryClaims, 1)
	require.NotNil(t, page.RuntimeDeliveryCommitOwner)
	receipt := page.RuntimeDeliveryClaims[0].Receipt

	stopDone := make(chan struct{})
	go func() {
		oldProcess.Stop()
		close(stopDone)
	}()
	select {
	case <-stopDone:
		t.Fatal("Stop returned while the admitted direct final-hop owner still owned q1")
	case <-time.After(30 * time.Millisecond):
	}

	// The owner is an exact in-process capability captured by the page. It may
	// still commit the admitted Process after Stop marks it non-accepting.
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
		[]string{receipt}, types.MCPToolDeliveryDelivered,
	)
	require.Eventually(t, func() bool {
		select {
		case <-stopDone:
			return true
		default:
			return false
		}
	}, time.Second, time.Millisecond)
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsAcked())
}

func TestDirectFinalHopOwnerFencesStopThroughPostCommitCallbacks(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "exact body"}}

	callbackEntered := make(chan struct{})
	releaseCallback := make(chan struct{})
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		if ev.Kind != consts.ParsedEventKindRuntimeInputAcked {
			return
		}
		close(callbackEntered)
		<-releaseCallback
	})
	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 1, "runtime-1")
	require.Len(t, page.RuntimeDeliveryClaims, 1)
	receipt := page.RuntimeDeliveryClaims[0].Receipt

	ackDone := make(chan struct{})
	go func() {
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
			[]string{receipt}, types.MCPToolDeliveryDelivered,
		)
		close(ackDone)
	}()
	select {
	case <-callbackEntered:
	case <-time.After(time.Second):
		t.Fatal("terminal acked callback did not start")
	}
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsAcked(), "terminal fsync must precede user callback")

	stopDone := make(chan struct{})
	go func() {
		ap.Stop()
		close(stopDone)
	}()
	select {
	case <-stopDone:
		t.Fatal("Stop returned while admitted post-commit callback still owned the final-hop lease")
	case <-time.After(30 * time.Millisecond):
	}

	close(releaseCallback)
	select {
	case <-ackDone:
	case <-time.After(time.Second):
		t.Fatal("final-hop callback did not finish")
	}
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not finish after admitted final-hop callback exited")
	}
}

func TestRuntimeMessageFinalHopDeferredWaitsForCommit(t *testing.T) {
	driver := &correlatedOwnershipDriver{
		reconcileResult:  RuntimeInputReconciliationAbsent,
		reconcileStarted: make(chan string, 1),
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	ap.SessionID = "thread-runtime"
	ap.runtimeDeliveryDeferredLease = time.Hour
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "exactly once"}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 10, "thread-runtime")
	require.Len(t, page.RuntimeDeliveryClaims, 1)
	receipt := page.RuntimeDeliveryClaims[0].Receipt
	ap.mu.Lock()
	claimLease := ap.pendingRuntimeInputs[receipt]
	ap.mu.Unlock()
	require.True(t, claimLease.FinalHopDeferred, "claim must acquire the final-hop lock before returning")
	require.NotZero(t, claimLease.finalHopLeaseGeneration)
	require.False(t, claimLease.finalHopLeaseDeadline.IsZero())
	require.NotNil(t, claimLease.finalHopLeaseCancel)
	// This is the exact race window between the page claim returning to the
	// wire layer and MCP reporting its first-hop HTTP outcome. The claim itself
	// must already hold the final-hop lease, so a terminal/retry scheduler cannot
	// observe an idle runtime and requeue q1 while the original result is live.
	ap.scheduleOldestPendingRuntimeInputReconcile(true)
	select {
	case id := <-driver.reconcileStarted:
		t.Fatalf("fresh claim must block reconciliation before its MCP Deferred callback; got %s", id)
	case <-time.After(30 * time.Millisecond):
	}
	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryDeferred, "thread-runtime")
	ap.mu.Lock()
	renewedLease := ap.pendingRuntimeInputs[receipt]
	ap.mu.Unlock()
	require.True(t, renewedLease.FinalHopDeferred)
	require.Greater(t, renewedLease.finalHopLeaseGeneration, claimLease.finalHopLeaseGeneration,
		"Deferred acknowledges and renews the claim-time lease rather than acquiring it for the first time")
	require.True(t, renewedLease.finalHopLeaseDeadline.After(claimLease.finalHopLeaseDeadline))
	require.NotNil(t, renewedLease.finalHopLeaseCancel)
	require.NotEqual(t, claimLease.finalHopLeaseCancel, renewedLease.finalHopLeaseCancel)
	select {
	case <-claimLease.finalHopLeaseCancel:
	default:
		t.Fatal("Deferred renewal left the previous long-lived timer asleep")
	}
	select {
	case id := <-driver.reconcileStarted:
		t.Fatalf("deferred first hop must not reconcile/requeue before final child write; got %s", id)
	case <-time.After(30 * time.Millisecond):
	}
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	_, stillPending := ap.pendingRuntimeInputs[receipt]
	ap.mu.Unlock()
	require.True(t, stillPending)
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))

	// An old-session final-hop event must not mutate the current receipt lease.
	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryUncertain, "thread-old")
	select {
	case id := <-driver.reconcileStarted:
		t.Fatalf("mismatched session uncertainty must be ignored; got %s", id)
	case <-time.After(30 * time.Millisecond):
	}

	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryDelivered, "thread-runtime")
	record, ok = store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	_, stillPending = ap.pendingRuntimeInputs[receipt]
	ap.mu.Unlock()
	require.False(t, stillPending)
	select {
	case <-renewedLease.finalHopLeaseCancel:
	default:
		t.Fatal("Delivered left the renewed long-lived timer asleep")
	}
}

func TestRuntimeMessageFinalHopDeferredOrphanLeaseBecomesUncertain(t *testing.T) {
	driver := &correlatedOwnershipDriver{
		reconcileResult:  RuntimeInputReconciliationAcked,
		reconcileStarted: make(chan string, 1),
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.SessionID = "thread-runtime"
	ap.runtimeDeliveryDeferredLease = 10 * time.Millisecond
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "orphaned second hop"}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 10, "thread-runtime")
	receipt := page.RuntimeDeliveryClaims[0].Receipt
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{receipt}, types.MCPToolDeliveryDeferred)

	require.Equal(t, receipt, <-driver.reconcileStarted, "the bounded Deferred orphan lease must eventually become Uncertain")
	require.Eventually(t, func() bool {
		record, ok := store.record("q1")
		return ok && record.IsAcked()
	}, time.Second, time.Millisecond)
}

func TestRuntimeMessageFinalHopDirectOrphanLeaseFailsClosedWithoutReplay(t *testing.T) {
	driver := &correlatedOwnershipDriver{
		reconcileResult:  RuntimeInputReconciliationAbsent,
		reconcileStarted: make(chan string, 1),
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	runtimeEvents := make(chan RuntimeEvent, 1)
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.SessionID = "thread-runtime"
	ap.runtimeDeliveryDeferredLease = 10 * time.Millisecond
	ap.deliveryOwnershipStore = store
	ap.onRuntimeEvent = func(event RuntimeEvent) { runtimeEvents <- event }
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "blocked direct writer"}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 10, "thread-runtime")
	receipt := page.RuntimeDeliveryClaims[0].Receipt
	require.Eventually(t, func() bool {
		record, ok := store.record("q1")
		return ok && record.IsFailed() && record.FailureCode == "direct_mcp_final_response_lease_expired"
	}, time.Second, time.Millisecond)
	select {
	case id := <-driver.reconcileStarted:
		t.Fatalf("an unresolved direct ResponseWriter may still complete; it must fail closed, not reconcile/replay %s", id)
	default:
	}
	event := <-runtimeEvents
	require.Equal(t, consts.RuntimeEventClassWakeFailed, event.EventClass)
	require.Equal(t, "q1", event.MessageID)
	require.Contains(t, event.Detail, "failed closed without replay")
	ap.mu.Lock()
	_, pending := ap.pendingRuntimeInputs[receipt]
	_, failed := ap.failedSet["q1"]
	ap.mu.Unlock()
	require.False(t, pending)
	require.True(t, failed)

	// A late complete write through an already-retired owner cannot safely
	// rewrite a replacement session's store generation. Failed remains the
	// authoritative no-replay terminal state.
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{receipt}, types.MCPToolDeliveryDelivered)
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsFailed())
}

func TestDirectFinalHopFailureDiagnosticFencesStop(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Millisecond)
	ap.runtimeDeliveryDeferredLease = 10 * time.Millisecond
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "ambiguous direct write"}}

	diagnosticEntered := make(chan struct{})
	releaseDiagnostic := make(chan struct{})
	ap.SetOnRuntimeEvent(func(event RuntimeEvent) {
		if event.EventClass != consts.RuntimeEventClassWakeFailed {
			return
		}
		close(diagnosticEntered)
		<-releaseDiagnostic
	})
	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 1, "runtime-1")
	require.Len(t, page.RuntimeDeliveryClaims, 1)

	select {
	case <-diagnosticEntered:
	case <-time.After(time.Second):
		t.Fatal("direct final-hop expiry diagnostic did not start")
	}
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsFailed(), "terminal Failed fsync must precede diagnostic callback")

	stopDone := make(chan struct{})
	go func() {
		ap.Stop()
		close(stopDone)
	}()
	select {
	case <-stopDone:
		t.Fatal("Stop returned while admitted fail-closed diagnostic still owned the final-hop lease")
	case <-time.After(30 * time.Millisecond):
	}
	close(releaseDiagnostic)
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not finish after fail-closed diagnostic returned")
	}
}

func TestRuntimeTransportRetirementSettlesOnlyTransferredFinalHopOwners(t *testing.T) {
	newProcess := func(t *testing.T) (*Process, *correlatedOwnershipDriver, *memoryDeliveryOwnershipStore, chan struct{}) {
		t.Helper()
		driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationAbsent, reconcileStarted: make(chan string, 1)}
		store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
			"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
		}}
		exitedCallback := make(chan struct{})
		ap := newInboxRetryTestProcess(driver, time.Hour)
		ap.deliveryOwnershipStore = store
		ap.runtimeDeliveryDeferredLease = time.Hour
		ap.waitFunc = func() error { return nil }
		ap.onExit = func(*Process) { close(exitedCallback) }
		ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}}
		return ap, driver, store, exitedCallback
	}

	t.Run("post-wait proof releases Deferred without old-process reconcile", func(t *testing.T) {
		ap, driver, store, exitedCallback := newProcess(t)
		page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 1, "runtime-1")
		receipt := page.RuntimeDeliveryClaims[0].Receipt
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{receipt}, types.MCPToolDeliveryDeferred)
		go ap.waitProcess()
		require.Eventually(t, func() bool {
			select {
			case <-exitedCallback:
				return true
			default:
				return false
			}
		}, time.Second, time.Millisecond, "waitProcess must settle transferred owners before lifecycle drain/onExit")
		record, ok := store.record("q1")
		require.True(t, ok)
		require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
		select {
		case id := <-driver.reconcileStarted:
			t.Fatalf("retired Process must not reconcile transferred owner %s", id)
		default:
		}
		ap.mu.Lock()
		pending := ap.pendingRuntimeInputs[receipt]
		ap.mu.Unlock()
		require.False(t, pending.FinalHopDeferred)
		require.False(t, pending.FinalHopRuntimeTransferred)
		require.Nil(t, pending.finalHopLeaseCancel)
		require.Nil(t, pending.finalHopLifecycleLease)
	})

	t.Run("post-wait proof does not sweep direct HTTP owner", func(t *testing.T) {
		ap, _, store, exitedCallback := newProcess(t)
		page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 1, "runtime-1")
		receipt := page.RuntimeDeliveryClaims[0].Receipt
		go ap.waitProcess()
		select {
		case <-exitedCallback:
			t.Fatal("child retirement must not sweep an independent direct HTTP final-hop owner")
		case <-time.After(30 * time.Millisecond):
		}
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{receipt}, types.MCPToolDeliveryUncertain)
		require.Eventually(t, func() bool {
			select {
			case <-exitedCallback:
				return true
			default:
				return false
			}
		}, time.Second, time.Millisecond)
		record, ok := store.record("q1")
		require.True(t, ok)
		require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
	})
}

func TestFreshnessPageReceiptEntropyFailureKeepsQueuedOwner(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := &Process{
		AgentID:               "agent-receipt-entropy-failure",
		SessionID:             "thread-runtime",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateStarting},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			inboxRetryDelay:        func(int) time.Duration { return time.Hour },
			runtimeDeliveryReceiptGenerator: func() (string, error) {
				return "", errors.New("entropy unavailable")
			},
			inbox: []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "must remain queued"}},
		},
	}

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 10, "thread-runtime")
	require.Empty(t, page.Messages)
	require.Empty(t, page.RuntimeDeliveryClaims)
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Empty(t, ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	stopInboxRetryTestProcess(ap)
}

func TestFreshnessPageDeferredHeadBlocksSecondPageUntilFinalHop(t *testing.T) {
	driver := &correlatedOwnershipDriver{reconcileResult: RuntimeInputReconciliationAbsent}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	ap.SessionID = "thread-runtime"
	ap.runtimeDeliveryDeferredLease = time.Hour
	ap.inbox = []types.InboxMessage{
		{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "first"},
		{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Content: "second"},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	first := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 1, "thread-runtime")
	require.Equal(t, []string{"q1"}, []string{first.Messages[0].MessageID})
	receipt := first.RuntimeDeliveryClaims[0].Receipt
	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryDeferred, "thread-runtime")

	blocked := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 1, "thread-runtime")
	require.Empty(t, blocked.Messages)
	require.True(t, blocked.HasMore)
	require.Equal(t, 1, blocked.RemainingCount)

	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryDelivered, "thread-runtime")
	second := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 1, "thread-runtime")
	require.Len(t, second.Messages, 1)
	require.Equal(t, "q2", second.Messages[0].MessageID)
	second.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
		[]string{second.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
	)
}

func inboxMessageIDs(messages []types.InboxMessage) []string {
	ids := make([]string, 0, len(messages))
	for _, message := range messages {
		ids = append(ids, message.MessageID)
	}
	return ids
}

func TestFreshnessPageEnforcesWakeLifecycleCapacityPerClaim(t *testing.T) {
	seedWakeOwners := func(ap *Process, count int) {
		if count <= 0 {
			return
		}
		ap.pendingWake = &pendingDelivery{Wake: WakeInfo{MessageID: "wake-0"}, Status: pendingDeliveryAwaitingCheck}
		for i := 1; i < count; i++ {
			ap.pendingDeliveryQueue = append(ap.pendingDeliveryQueue, pendingDelivery{Wake: WakeInfo{MessageID: fmt.Sprintf("wake-%d", i)}})
		}
	}

	t.Run("existing 100 blocks every claim", func(t *testing.T) {
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		seedWakeOwners(ap, processInboxMaxMessages)
		ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 2, "thread-runtime")
		require.Empty(t, page.Messages)
		require.Empty(t, page.RuntimeDeliveryClaims)
		require.Nil(t, page.RuntimeDeliveryCommitOwner)
		require.True(t, page.HasMore)
		require.Equal(t, 1, page.RemainingCount)
	})

	t.Run("existing 99 reserves q1 before considering q2", func(t *testing.T) {
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		seedWakeOwners(ap, processInboxMaxMessages-1)
		ap.inbox = []types.InboxMessage{
			{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1},
			{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2},
		}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 2, "thread-runtime")
		require.Equal(t, []string{"q1"}, []string{page.Messages[0].MessageID})
		require.Len(t, page.RuntimeDeliveryClaims, 1)
		require.NotNil(t, page.RuntimeDeliveryCommitOwner)
		require.True(t, page.HasMore)
		require.Equal(t, 1, page.RemainingCount)
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
			[]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
		)
	})
}

func TestFreshnessPagePreservesGlobalFIFOAcrossTargets(t *testing.T) {
	t.Run("older other target blocks requested target", func(t *testing.T) {
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		ap.inbox = []types.InboxMessage{
			{MessageID: "q1", TargetKey: "target-a", TargetSeq: 1},
			{MessageID: "q2", TargetKey: "target-b", TargetSeq: 1},
		}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("target-b", 1, 10, "thread-runtime")
		require.Empty(t, page.Messages)
		require.True(t, page.HasMore)
		require.Equal(t, 1, page.RemainingCount)
		ap.mu.Lock()
		require.Equal(t, []string{"q1", "q2"}, []string{ap.inbox[0].MessageID, ap.inbox[1].MessageID})
		ap.mu.Unlock()
	})

	t.Run("interleaved other target blocks later requested match", func(t *testing.T) {
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		ap.inbox = []types.InboxMessage{
			{MessageID: "q1", TargetKey: "target-b", TargetSeq: 1},
			{MessageID: "q2", TargetKey: "target-a", TargetSeq: 1},
			{MessageID: "q3", TargetKey: "target-b", TargetSeq: 2},
		}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("target-b", 2, 10, "thread-runtime")
		require.Equal(t, []string{"q1"}, []string{page.Messages[0].MessageID})
		require.Len(t, page.RuntimeDeliveryClaims, 1)
		require.True(t, page.HasMore)
		require.Equal(t, 1, page.RemainingCount)
		ap.mu.Lock()
		require.Equal(t, []string{"q2", "q3"}, []string{ap.inbox[0].MessageID, ap.inbox[1].MessageID})
		ap.mu.Unlock()
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
			[]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
		)
	})
}

func TestFreshnessPageStaleHeadRequiresDurableSuppressionBeforeFIFOAdvances(t *testing.T) {
	t.Run("successful stale tombstone permits next target", func(t *testing.T) {
		store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
			"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "target-a", TargetSeq: 1}, State: DeliveryOwnershipQueued},
			"q2": {Message: types.InboxMessage{MessageID: "q2", TargetKey: "target-b", TargetSeq: 1}, State: DeliveryOwnershipQueued},
		}}
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		ap.deliveryOwnershipStore = store
		ap.freshness.MarkObserved("target-a", 1, "frontier-a")
		ap.inbox = []types.InboxMessage{
			{MessageID: "q1", TargetKey: "target-a", TargetSeq: 1},
			{MessageID: "q2", TargetKey: "target-b", TargetSeq: 1},
		}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("target-b", 1, 10, "thread-runtime")
		require.Equal(t, []string{"q2"}, []string{page.Messages[0].MessageID})
		record, ok := store.record("q1")
		require.True(t, ok)
		require.True(t, record.IsAcked())
		page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
			[]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
		)
	})

	t.Run("failed stale tombstone retains and blocks head", func(t *testing.T) {
		base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
			"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "target-a", TargetSeq: 1}, State: DeliveryOwnershipQueued},
			"q2": {Message: types.InboxMessage{MessageID: "q2", TargetKey: "target-b", TargetSeq: 1}, State: DeliveryOwnershipQueued},
		}}
		ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
		ap.deliveryOwnershipStore = &rejectAckedDeliveryOwnershipStore{base: base}
		ap.freshness.MarkObserved("target-a", 1, "frontier-a")
		ap.inbox = []types.InboxMessage{
			{MessageID: "q1", TargetKey: "target-a", TargetSeq: 1},
			{MessageID: "q2", TargetKey: "target-b", TargetSeq: 1},
		}
		t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

		page := ap.InboxMessagesForFreshnessPageForRuntime("target-b", 1, 10, "thread-runtime")
		require.Empty(t, page.Messages)
		require.True(t, page.HasMore)
		require.Equal(t, 1, page.RemainingCount)
		ap.mu.Lock()
		require.Equal(t, []string{"q1", "q2"}, []string{ap.inbox[0].MessageID, ap.inbox[1].MessageID})
		ap.mu.Unlock()
	})
}

func TestFreshnessPageReceiptCollisionNeverOverwritesOwner(t *testing.T) {
	firstReceipt := types.NewRuntimeDeliveryReceipt("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
	secondReceipt := types.NewRuntimeDeliveryReceipt("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
	calls := 0
	ap := newInboxRetryTestProcess(&correlatedOwnershipDriver{}, time.Hour)
	ap.SessionID = "thread-runtime"
	ap.runtimeDeliveryReceiptGenerator = func() (string, error) {
		calls++
		if calls <= 5 { // q1 plus q2's four bounded collision attempts
			return firstReceipt, nil
		}
		return secondReceipt, nil
	}
	ap.inbox = []types.InboxMessage{
		{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "first"},
		{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Content: "second"},
	}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 2, "thread-runtime")
	require.Len(t, page.Messages, 1)
	require.Equal(t, "q1", page.Messages[0].MessageID)
	require.Equal(t, firstReceipt, page.RuntimeDeliveryClaims[0].Receipt)
	require.True(t, page.HasMore)
	require.Equal(t, 1, page.RemainingCount)
	ap.mu.Lock()
	require.Len(t, ap.pendingRuntimeInputs, 1)
	require.Equal(t, "q1", ap.pendingRuntimeInputs[firstReceipt].Message.MessageID)
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "q2", ap.inbox[0].MessageID)
	ap.mu.Unlock()

	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{firstReceipt}, types.MCPToolDeliveryDelivered, "thread-runtime")
	next := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 2, 1, "thread-runtime")
	require.Len(t, next.Messages, 1)
	require.Equal(t, "q2", next.Messages[0].MessageID)
	require.Equal(t, secondReceipt, next.RuntimeDeliveryClaims[0].Receipt)
	next.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{secondReceipt}, types.MCPToolDeliveryDelivered)
}

func TestRuntimeMessageFinalHopUncertainStartsReceiptReconciliation(t *testing.T) {
	driver := &correlatedOwnershipDriver{
		reconcileResult:  RuntimeInputReconciliationAcked,
		reconcileStarted: make(chan string, 1),
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.SessionID = "thread-runtime"
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "recover me"}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	page := ap.InboxMessagesForFreshnessPageForRuntime("thread-1", 1, 10, "thread-runtime")
	receipt := page.RuntimeDeliveryClaims[0].Receipt
	ap.AcknowledgeRuntimeMessageToolDeliveriesForSession([]string{receipt}, types.MCPToolDeliveryUncertain, "thread-runtime")
	require.Equal(t, receipt, <-driver.reconcileStarted)
	require.Eventually(t, func() bool {
		record, ok := store.record("q1")
		return ok && record.IsAcked()
	}, time.Second, time.Millisecond)
}

func TestFreshnessOwnershipMutationDrainsBeforeStopAndRejectsLateWrites(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
		"q2": {Message: types.InboxMessage{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2}, State: DeliveryOwnershipQueued},
	}}
	entered := make(chan struct{})
	release := make(chan struct{})
	store := &blockingDeliveryOwnershipStore{
		base:      base,
		messageID: "q1",
		state:     DeliveryOwnershipInflight,
		outcome:   "",
		entered:   entered,
		release:   release,
	}
	ap := &Process{
		AgentID:               "agent-freshness-stop-barrier",
		SessionID:             "thread-runtime",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateRunning, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			inbox: []types.InboxMessage{
				{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "first page"},
				{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Content: "must remain after stop"},
			},
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	pageDone := make(chan types.InboxReadPage, 1)
	go func() { pageDone <- ap.InboxMessagesForFreshnessPage("thread-1", 2, 1) }()
	<-entered

	stopDone := make(chan struct{})
	go func() {
		ap.Stop()
		close(stopDone)
	}()
	select {
	case <-stopDone:
		t.Fatal("Stop returned while an admitted freshness ownership fsync was still in flight")
	case <-time.After(20 * time.Millisecond):
	}

	close(release)
	page := <-pageDone
	require.Len(t, page.Messages, 1)
	require.Equal(t, "q1", page.Messages[0].MessageID)
	require.True(t, page.HasMore)
	require.Equal(t, 1, page.RemainingCount)
	select {
	case <-stopDone:
		t.Fatal("Stop returned after claim fsync but before the transferred final-hop owner settled")
	case <-time.After(20 * time.Millisecond):
	}
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery(
		[]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered,
	)
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not finish after the admitted final-hop owner committed")
	}

	late := ap.InboxMessagesForFreshnessPage("thread-1", 2, 10)
	require.Empty(t, late.Messages, "a stopped Process must reject late freshness mutation admission")
	record, ok := base.record("q2")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	ap.mu.Lock()
	queued := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()
	require.Len(t, queued, 1)
	require.Equal(t, "q2", queued[0].MessageID)
}

func TestFreshnessPageRetainsOwnerWhenClaimPersistenceFails(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := &Process{
		AgentID:               "agent-freshness-blocked",
		SessionID:             "thread-runtime",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}, cancel: func() {}},
		processLifecycleState: processLifecycleState{state: processStateStarting},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: &rejectAckedDeliveryOwnershipStore{base: base},
			inbox:                  []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "retain"}},
		},
	}

	page := ap.InboxMessagesForFreshnessPage("thread-1", 1, 10)
	require.Empty(t, page.Messages)
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	ap.mu.Unlock()
	record, ok := base.record("q1")
	require.True(t, ok)
	require.True(t, record.IsQueued())
}

func TestReconcileInboxMessagesDurablyAcksBeforeRemoval(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1}, State: DeliveryOwnershipQueued},
	}}
	ap := newInboxRetryTestProcess(&inboxRetryAckedDriver{failures: map[string]int{}}, time.Millisecond)
	ap.deliveryOwnershipStore = store
	ap.inbox = []types.InboxMessage{{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "already consumed"}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.Equal(t, 1, ap.ReconcileInboxMessages([]string{"q1"}))
	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	ap.mu.Unlock()
}

func TestReconcileInboxMessagesDrainsBeforeStopAndRejectsLateMutation(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1"}, State: DeliveryOwnershipQueued},
		"q2": {Message: types.InboxMessage{MessageID: "q2"}, State: DeliveryOwnershipQueued},
	}}
	blocking := &blockingDeliveryOwnershipStore{
		base:      base,
		messageID: "q1",
		state:     DeliveryOwnershipTerminal, outcome: DeliveryOutcomeAcked,
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	ap := newInboxRetryTestProcess(&inboxRetryAckedDriver{failures: map[string]int{}}, time.Hour)
	ap.deliveryOwnershipStore = blocking
	ap.inbox = []types.InboxMessage{
		{MessageID: "q1", TargetKey: "thread-1", TargetSeq: 1, Content: "ack before stop"},
		{MessageID: "q2", TargetKey: "thread-1", TargetSeq: 2, Content: "must remain queued"},
	}

	reconcileDone := make(chan int, 1)
	go func() { reconcileDone <- ap.ReconcileInboxMessages([]string{"q1"}) }()
	<-blocking.entered

	stopDone := make(chan struct{})
	go func() { ap.Stop(); close(stopDone) }()
	select {
	case <-stopDone:
		t.Fatal("Stop returned before admitted inbox reconciliation drained")
	case <-time.After(20 * time.Millisecond):
	}
	close(blocking.release)
	require.Equal(t, 1, <-reconcileDone)
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not finish after inbox reconciliation drained")
	}

	require.Equal(t, 0, ap.ReconcileInboxMessages([]string{"q2"}), "stopped Process must reject late reconciliation admission")
	q1, ok := base.record("q1")
	require.True(t, ok)
	require.True(t, q1.IsAcked())
	q2, ok := base.record("q2")
	require.True(t, ok)
	require.True(t, q2.IsQueued())
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "q2", ap.inbox[0].MessageID)
	ap.mu.Unlock()
}

func TestReconcileInboxReservationPreventsConcurrentRetryDelivery(t *testing.T) {
	base := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {Message: types.InboxMessage{MessageID: "q1"}, State: DeliveryOwnershipQueued},
	}}
	blocking := &blockingDeliveryOwnershipStore{
		base:      base,
		messageID: "q1",
		state:     DeliveryOwnershipTerminal, outcome: DeliveryOutcomeAcked,
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	driver := &inboxRetryAckedDriver{failures: map[string]int{}}
	ap := newInboxRetryTestProcess(driver, time.Millisecond)
	ap.deliveryOwnershipStore = blocking
	ap.inbox = []types.InboxMessage{{MessageID: "q1", Target: "#ops", Content: "exact body"}}
	t.Cleanup(func() { ap.Stop() })

	reconcileDone := make(chan int, 1)
	go func() { reconcileDone <- ap.ReconcileInboxMessages([]string{"q1"}) }()
	<-blocking.entered
	ap.scheduleInboxRetry(true)
	require.Never(t, func() bool {
		calls, _, _, _, _ := driver.snapshot()
		return len(calls) != 0
	}, 30*time.Millisecond, time.Millisecond, "reserved inbox owner must not enter runtime while acked fsync is pending")

	close(blocking.release)
	require.Equal(t, 1, <-reconcileDone)
	time.Sleep(10 * time.Millisecond)
	calls, _, _, _, _ := driver.snapshot()
	require.Empty(t, calls)
	record, ok := base.record("q1")
	require.True(t, ok)
	require.True(t, record.IsAcked())
}

func TestPendingRuntimeInputReconcilerCancellationDoesNotWriteAfterStop(t *testing.T) {
	createdAt := time.Now().UTC()
	driver := &blockingRuntimeInputReconciler{
		started:  make(chan struct{}),
		canceled: make(chan struct{}),
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q1": {
			Message:   types.InboxMessage{MessageID: "q1", Content: "retained owner"},
			Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "q1"}, "retained owner"),
			SessionID: "thread-1",
			State:     DeliveryOwnershipInflight, InflightReason: DeliveryInflightDispatching,
			AcceptedAt: createdAt,
		},
	}}
	ap := &Process{
		AgentID:   "agent-blocked-reconcile-stop",
		SessionID: "thread-1",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
			cancel: func() {},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			pendingRuntimeInputs: map[string]pendingRuntimeInput{
				"q1": {
					Message:        types.InboxMessage{MessageID: "q1", Content: "retained owner"},
					Input:          testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "q1"}, "retained owner"),
					SessionID:      "thread-1",
					CreatedAt:      createdAt,
					InflightReason: DeliveryInflightDispatching,
				},
			},
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	ap.schedulePendingRuntimeInputReconcile("q1", createdAt, true)
	<-driver.started

	stopDone := make(chan struct{})
	go func() { ap.Stop(); close(stopDone) }()
	select {
	case <-driver.canceled:
	case <-time.After(time.Second):
		t.Fatal("reconciler RPC did not observe Process admission cancellation")
	}
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("Stop did not drain canceled reconciler mutation")
	}

	record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightDispatching))
	require.Zero(t, record.Attempts, "Stop cancellation must not persist reconcile progress or a terminal phase")
	ap.mu.Lock()
	pending, ok := ap.pendingRuntimeInputs["q1"]
	ap.mu.Unlock()
	require.True(t, ok)
	require.Equal(t, DeliveryInflightDispatching, pending.InflightReason)
}

func TestOwnershipRestoreDurablyAcksFreshnessStaleQueuedRecord(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"q-old": {
			Message: types.InboxMessage{MessageID: "q-old", TargetKey: "thread-1", TargetSeq: 4, Content: "covered"},
			State:   DeliveryOwnershipQueued, AcceptedAt: time.Now().Add(-time.Minute),
		},
	}}
	ap := newInboxRetryTestProcess(&inboxRetryAckedDriver{failures: map[string]int{}}, time.Millisecond)
	ap.freshness.MarkObserved("thread-1", 5, "newer")
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	require.NoError(t, ap.SetDeliveryOwnershipStore(store))
	record, ok := store.record("q-old")
	require.True(t, ok)
	require.True(t, record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	ap.mu.Unlock()
}