package agent

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"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 blockingAckedRuntimeInputDriver struct {
	busyDeliveryStubDriver

	mu       sync.Mutex
	calls    []string
	firstErr error
	started  chan string
	release  chan struct{}
}

type deliveryOrderRecorder struct {
	mu      sync.Mutex
	entries []string
}

func (r *deliveryOrderRecorder) add(entry string) {
	r.mu.Lock()
	r.entries = append(r.entries, entry)
	r.mu.Unlock()
}

func (r *deliveryOrderRecorder) snapshot() []string {
	r.mu.Lock()
	defer r.mu.Unlock()
	return append([]string(nil), r.entries...)
}

type boundaryBlockingAckedDriver struct {
	busyDeliveryStubDriver

	blockOnce       sync.Once
	supportsEntered chan struct{}
	supportsRelease chan struct{}
	order           *deliveryOrderRecorder
}

func (d *boundaryBlockingAckedDriver) SupportsStdinNotification() bool {
	d.blockOnce.Do(func() {
		close(d.supportsEntered)
		<-d.supportsRelease
	})
	return true
}

func (d *boundaryBlockingAckedDriver) SupportsAckedBusyRuntimeInputDelivery() bool {
	return true
}

func (d *boundaryBlockingAckedDriver) DeliverAckedBusyRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.order.add(input.ID)
	return nil
}

type orderedRecordingWriteCloser struct {
	order *deliveryOrderRecorder
}

func (w *orderedRecordingWriteCloser) Write(p []byte) (int, error) {
	w.order.add("msg-1")
	return len(p), nil
}

func (w *orderedRecordingWriteCloser) Close() error { return nil }

type blockingRecoveryDeliveryWriter struct {
	entered chan struct{}
	release chan struct{}
	once    sync.Once
}

func (w *blockingRecoveryDeliveryWriter) Write(p []byte) (int, error) {
	w.once.Do(func() { close(w.entered) })
	<-w.release
	return len(p), nil
}

func (w *blockingRecoveryDeliveryWriter) Close() error { return nil }

type blockingRuntimeAttachmentResolver struct {
	entered chan struct{}
	release chan struct{}
	once    sync.Once
}

func (r *blockingRuntimeAttachmentResolver) ResolveRuntimeAttachments(context.Context, []types.MessageAttachment) ([]types.RuntimeAttachment, error) {
	r.once.Do(func() { close(r.entered) })
	<-r.release
	return []types.RuntimeAttachment{{
		ID:       "attachment-1",
		Filename: "handoff.txt",
		MIME:     "text/plain",
		Path:     "/tmp/handoff.txt",
		URI:      "file:///tmp/handoff.txt",
	}}, nil
}

type contextAwareRuntimeAttachmentResolver struct {
	entered chan struct{}
	once    sync.Once
}

func (r *contextAwareRuntimeAttachmentResolver) ResolveRuntimeAttachments(ctx context.Context, _ []types.MessageAttachment) ([]types.RuntimeAttachment, error) {
	r.once.Do(func() { close(r.entered) })
	<-ctx.Done()
	return nil, ctx.Err()
}

type blockingLiveRecoveryDriver struct {
	busyDeliveryStubDriver

	restoreStarted chan struct{}
	restoreRelease chan struct{}
	order          *deliveryOrderRecorder
}

func (d *blockingLiveRecoveryDriver) Name() string { return "codex" }

func (d *blockingLiveRecoveryDriver) PrepareRecoveryPoint() (string, string, bool, error) {
	return "codex_app_server_turns", "recovery-ref", true, nil
}

func (d *blockingLiveRecoveryDriver) RestoreRecoveryPoint(context.Context, string, string) error {
	close(d.restoreStarted)
	<-d.restoreRelease
	return nil
}

func (d *blockingLiveRecoveryDriver) DeliverRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.order.add(input.ID)
	return nil
}

func (d *blockingLiveRecoveryDriver) SupportsAckedBusyRuntimeInputDelivery() bool { return true }

func (d *blockingLiveRecoveryDriver) DeliverAckedBusyRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.order.add(input.ID)
	return nil
}

func (d *blockingAckedRuntimeInputDriver) SupportsAckedBusyRuntimeInputDelivery() bool {
	return true
}

func (d *blockingAckedRuntimeInputDriver) DeliverAckedBusyRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.mu.Lock()
	callIndex := len(d.calls)
	d.calls = append(d.calls, input.ID)
	d.mu.Unlock()
	d.started <- input.ID
	if callIndex == 0 {
		<-d.release
		return d.firstErr
	}
	return nil
}

func (d *blockingAckedRuntimeInputDriver) callIDs() []string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return append([]string(nil), d.calls...)
}

func TestFreshAckedInputBypassesDeferredBacklog(t *testing.T) {
	stdin := &recordingWriteCloser{}
	driver := &ackedRuntimeInputStubDriver{
		supports:   true,
		deliverErr: errors.New("active turn temporarily cannot accept input"),
	}
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver:    driver,
			stdinPipe: stdin,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
		processDeliveryState: processDeliveryState{
			// This test drives the authoritative TurnEnd dequeue itself. Keep the
			// independent liveness worker parked so its immediate post-handoff
			// kick cannot consume q2 before the boundary assertions inspect it.
			inboxRetryRunning: true,
		},
	}

	q1 := types.InboxMessage{MessageID: "msg-1", TargetKey: "scope", TargetSeq: 1, Content: "first"}
	q2 := types.InboxMessage{MessageID: "msg-2", TargetKey: "scope", TargetSeq: 2, Content: "second"}

	require.NoError(t, ap.deliverMessage(q1, false))
	require.Equal(t, 1, driver.deliverCalls)

	driver.deliverErr = nil
	require.NoError(t, ap.deliverMessage(q2, false))
	require.Equal(t, 2, driver.deliverCalls, "fresh q2 must not be blocked by deferred q1")

	// A duplicate upstream delivery must not create another queue owner.
	require.NoError(t, ap.deliverMessage(q2, false))
	ap.mu.Lock()
	queuedIDs := []string{ap.inbox[0].MessageID}
	ap.mu.Unlock()
	require.Equal(t, []string{"msg-1"}, queuedIDs)

	// The matching root turn boundary claims exactly the oldest original body.
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd, SessionID: "thread-root"})
	require.Contains(t, stdin.String(), "first")
	require.NotContains(t, stdin.String(), "second")
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	ap.mu.Unlock()
}

func TestConcurrentAckedInputAdmissionIsSerialized(t *testing.T) {
	t.Run("newer input proceeds after serialized older failure", func(t *testing.T) {
		driver := &blockingAckedRuntimeInputDriver{
			firstErr: errors.New("temporarily non-steerable"),
			started:  make(chan string, 2),
			release:  make(chan struct{}),
		}
		ap := &Process{
			AgentID:   "agent-codex",
			SessionID: "thread-root",
			processRuntimeBinding: processRuntimeBinding{
				driver: driver,
			},
			processLifecycleState: processLifecycleState{
				state:  processStateRunning,
				isIdle: false,
			},
			processRecoveryState: processRecoveryState{disableRecovery: true},
		}

		q1Done := make(chan error, 1)
		go func() {
			q1Done <- ap.deliverMessage(types.InboxMessage{MessageID: "msg-1", Content: "first"}, false)
		}()
		require.Equal(t, "msg-1", <-driver.started)

		q2Done := make(chan error, 1)
		go func() {
			q2Done <- ap.deliverMessage(types.InboxMessage{MessageID: "msg-2", Content: "second"}, false)
		}()
		select {
		case inputID := <-driver.started:
			t.Fatalf("newer input bypassed blocked admission via runtime call %q", inputID)
		case <-time.After(20 * time.Millisecond):
		}

		close(driver.release)
		require.NoError(t, <-q1Done)
		require.NoError(t, <-q2Done)
		calls := driver.callIDs()
		require.GreaterOrEqual(t, len(calls), 2)
		require.Equal(t, []string{"msg-1", "msg-2"}, calls[:2], "q2 must get the next serialized foreground mutation before q1 background retry")
		ap.mu.Lock()
		queuedQ1 := inboxContainsMessageID(ap.inbox, "msg-1")
		_, q2Pending := ap.pendingRuntimeInputs["msg-2"]
		ap.mu.Unlock()
		require.True(t, queuedQ1 || q2Pending, "serialized delivery must retain each non-terminal owner without forcing q2 behind q1")
	})

	t.Run("concurrent duplicate has one runtime owner", func(t *testing.T) {
		driver := &blockingAckedRuntimeInputDriver{
			started: make(chan string, 2),
			release: make(chan struct{}),
		}
		ap := &Process{
			AgentID:   "agent-codex",
			SessionID: "thread-root",
			processRuntimeBinding: processRuntimeBinding{
				driver: driver,
			},
			processLifecycleState: processLifecycleState{
				state:  processStateRunning,
				isIdle: false,
			},
			processRecoveryState: processRecoveryState{disableRecovery: true},
		}
		msg := types.InboxMessage{MessageID: "msg-duplicate", Content: "same body"}

		firstDone := make(chan error, 1)
		go func() { firstDone <- ap.deliverMessage(msg, false) }()
		require.Equal(t, "msg-duplicate", <-driver.started)

		secondDone := make(chan error, 1)
		go func() { secondDone <- ap.deliverMessage(msg, false) }()
		select {
		case inputID := <-driver.started:
			t.Fatalf("duplicate input issued a second runtime call %q", inputID)
		case <-time.After(20 * time.Millisecond):
		}

		close(driver.release)
		require.NoError(t, <-firstDone)
		require.NoError(t, <-secondDone)
		require.Equal(t, []string{"msg-duplicate"}, driver.callIDs())
		ap.mu.Lock()
		require.Empty(t, ap.inbox)
		require.Contains(t, ap.pendingRuntimeInputs, "msg-duplicate")
		ap.mu.Unlock()
	})
}

func TestStopAndSleepWaitForInFlightDeliveryAdmission(t *testing.T) {
	for _, tt := range []struct {
		name     string
		shutdown func(*Process)
		sleeping bool
	}{
		{name: "stop", shutdown: (*Process).Stop},
		{name: "sleep", shutdown: (*Process).Sleep, sleeping: true},
	} {
		t.Run(tt.name, func(t *testing.T) {
			driver := &blockingAckedRuntimeInputDriver{
				started: make(chan string, 1),
				release: make(chan struct{}),
			}
			store := &memoryDeliveryOwnershipStore{}
			cancelled := make(chan struct{})
			ap := &Process{
				AgentID:   "agent-stop-order",
				SessionID: "thread-root",
				processRuntimeBinding: processRuntimeBinding{
					driver: driver,
					cancel: func() {
						close(cancelled)
					},
				},
				processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: false},
				processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
				processRecoveryState:  processRecoveryState{disableRecovery: true},
			}

			deliveryDone := make(chan error, 1)
			go func() {
				deliveryDone <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
					MessageID: "accepted-before-stop",
					Target:    "#ops:thread",
					Content:   "must finish durable admission before retirement",
				})
			}()
			require.Equal(t, "accepted-before-stop", <-driver.started)

			shutdownStarted := make(chan struct{})
			shutdownDone := make(chan struct{})
			go func() {
				close(shutdownStarted)
				tt.shutdown(ap)
				close(shutdownDone)
			}()
			<-shutdownStarted
			select {
			case <-shutdownDone:
				t.Fatal("retirement returned before the in-flight delivery released admission ownership")
			case <-time.After(20 * time.Millisecond):
			}
			select {
			case <-cancelled:
			case <-time.After(time.Second):
				t.Fatal("retirement did not cancel the runtime to unblock in-flight delivery")
			}
			require.False(t, ap.AcceptingDeliveries(), "retirement must fence new admissions while it drains the older owner")

			close(driver.release)
			require.NoError(t, <-deliveryDone)
			select {
			case <-shutdownDone:
			case <-time.After(time.Second):
				t.Fatal("retirement did not finish after delivery released the gate")
			}
			select {
			case <-cancelled:
			default:
				t.Fatal("retirement did not terminate the runtime")
			}
			require.False(t, ap.AcceptingDeliveries())
			require.Equal(t, tt.sleeping, ap.IsSleeping())
			record, ok := store.record("accepted-before-stop")
			require.True(t, ok, "accepted body must be durable before retirement returns")
			require.True(t, record.IsInflight(DeliveryInflightDispatching))
		})
	}
}

func TestUnexpectedExitDrainsInFlightDeliveryBeforeOnExitReplacement(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	oldDriver := &correlatedOwnershipDriver{
		protocolDeliveryStubDriver: protocolDeliveryStubDriver{},
		reconcileResult:            RuntimeInputReconciliationPending,
	}
	resolver := &blockingRuntimeAttachmentResolver{
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	onExit := make(chan struct{})
	oldProcess := &Process{
		AgentID:   "agent-unexpected-exit-order",
		SessionID: "thread-old",
		processRuntimeBinding: processRuntimeBinding{
			driver:   oldDriver,
			waitFunc: func() error { return nil },
			cancel:   func() {},
			onExit: func(*Process) {
				close(onExit)
			},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			attachmentResolver:     resolver,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	deliveryDone := make(chan error, 1)
	go func() {
		deliveryDone <- oldProcess.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "owner-before-unexpected-exit",
			Target:    "#ops:thread",
			Content:   "persist before replacement attaches the journal",
			AttachmentList: []types.MessageAttachment{{
				ID:       "attachment-1",
				Filename: "handoff.txt",
			}},
		})
	}()
	<-resolver.entered

	waitDone := make(chan struct{})
	go func() {
		oldProcess.waitProcess()
		close(waitDone)
	}()
	require.Eventually(t, func() bool { return !oldProcess.AcceptingDeliveries() }, time.Second, time.Millisecond)
	select {
	case <-onExit:
		t.Fatal("onExit exposed replacement before the in-flight owner was durable")
	case <-time.After(20 * time.Millisecond):
	}
	_, exists := store.record("owner-before-unexpected-exit")
	require.False(t, exists, "test must hold the delivery before its first ownership Put")

	close(resolver.release)
	require.NoError(t, <-deliveryDone)
	select {
	case <-onExit:
	case <-time.After(time.Second):
		t.Fatal("onExit did not run after the delivery admission drained")
	}
	select {
	case <-waitDone:
	case <-time.After(time.Second):
		t.Fatal("waitProcess did not finish after admission drained")
	}
	record, exists := store.record("owner-before-unexpected-exit")
	require.True(t, exists)
	require.True(t, record.IsInflight(DeliveryInflightDispatching))
	_, runtimeCalls, _, _ := oldDriver.snapshot()
	require.Zero(t, runtimeCalls, "an admitted delivery must not mutate the exited runtime")

	// A replacement may attach only after onExit. It restores the correlated
	// owner and continues with reconciliation rather than waiting for another
	// external wake.
	reconcileStarted := make(chan string, 1)
	newDriver := &correlatedOwnershipDriver{
		protocolDeliveryStubDriver: protocolDeliveryStubDriver{},
		reconcileResult:            RuntimeInputReconciliationAcked,
		reconcileStarted:           reconcileStarted,
	}
	newProcess := &Process{
		AgentID:   oldProcess.AgentID,
		SessionID: "thread-new",
		processRuntimeBinding: processRuntimeBinding{
			driver: newDriver,
			cancel: func() {},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}
	require.NoError(t, newProcess.SetDeliveryOwnershipStore(store))
	t.Cleanup(newProcess.Stop)
	select {
	case messageID := <-reconcileStarted:
		require.Equal(t, "owner-before-unexpected-exit", messageID)
	case <-time.After(time.Second):
		t.Fatal("replacement did not continue restored ownership reconciliation")
	}
	require.Eventually(t, func() bool {
		restored, ok := store.record("owner-before-unexpected-exit")
		return ok && restored.IsAcked()
	}, time.Second, time.Millisecond)
}

func TestAdmittedDeliveryCallerCancellationPersistsQueuedOwnershipWithoutRuntimeMutation(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	driver := &correlatedOwnershipDriver{protocolDeliveryStubDriver: protocolDeliveryStubDriver{}}
	resolver := &contextAwareRuntimeAttachmentResolver{entered: make(chan struct{})}
	ap := &Process{
		AgentID:   "agent-admitted-cancel",
		SessionID: "thread-old",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			attachmentResolver:     resolver,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	err := ap.DeliverForHandoff(ctx, types.InboxMessage{
		MessageID: "owner-caller-timeout",
		Target:    "#ops:thread",
		Content:   "caller cancellation must retain an exact queued owner",
		AttachmentList: []types.MessageAttachment{{
			ID:       "attachment-1",
			Filename: "handoff.txt",
		}},
	})
	require.NoError(t, err, "a bounded caller timeout is accepted as durable queue ownership")
	record, ok := store.record("owner-caller-timeout")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	_, runtimeCalls, _, _ := driver.snapshot()
	require.Zero(t, runtimeCalls)
}

func TestAdmittedDeliveryCompilerFailurePersistsQueuedOwnershipBeforeUnexpectedExit(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{}
	driver := &correlatedOwnershipDriver{protocolDeliveryStubDriver: protocolDeliveryStubDriver{}}
	resolver := &blockingRuntimeAttachmentResolver{
		entered: make(chan struct{}),
		release: make(chan struct{}),
	}
	onExit := make(chan struct{})
	ap := &Process{
		AgentID:   "agent-admitted-compiler-failure",
		SessionID: "thread-old",
		processRuntimeBinding: processRuntimeBinding{
			driver:   driver,
			waitFunc: func() error { return nil },
			onExit: func(*Process) {
				close(onExit)
			},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore: store,
			attachmentResolver:     resolver,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	deliveryCtx, cancelDelivery := context.WithCancel(context.Background())
	defer cancelDelivery()
	deliveryDone := make(chan error, 1)
	go func() {
		deliveryDone <- ap.DeliverForHandoff(deliveryCtx, types.InboxMessage{
			MessageID: "owner-compiler-failure",
			Target:    "#ops:thread",
			Content:   "compiler failure must be durable before replacement",
			AttachmentList: []types.MessageAttachment{{
				ID:       "attachment-1",
				Filename: "handoff.txt",
			}},
		})
	}()
	<-resolver.entered

	waitDone := make(chan struct{})
	go func() {
		ap.waitProcess()
		close(waitDone)
	}()
	require.Eventually(t, func() bool { return !ap.AcceptingDeliveries() }, time.Second, time.Millisecond)
	select {
	case <-onExit:
		t.Fatal("onExit exposed replacement before compiler failure was durably queued")
	case <-time.After(20 * time.Millisecond):
	}
	// Cancel after admission but before the compiler starts. The compiler error
	// must become a durable queued owner before the retired Process exposes its
	// replacement.
	cancelDelivery()
	close(resolver.release)
	require.NoError(t, <-deliveryDone)
	select {
	case <-onExit:
	case <-time.After(time.Second):
		t.Fatal("onExit did not run after admitted compiler failure drained")
	}
	select {
	case <-waitDone:
	case <-time.After(time.Second):
		t.Fatal("waitProcess did not finish after compiler failure drained")
	}
	record, ok := store.record("owner-compiler-failure")
	require.True(t, ok)
	require.True(t, record.IsQueued())
	_, runtimeCalls, _, _ := driver.snapshot()
	require.Zero(t, runtimeCalls)
}

func TestRuntimeInputAckedAtomicallyOwnsMessageBeforeWakeCallback(t *testing.T) {
	driver := &blockingAckedRuntimeInputDriver{
		started: make(chan string, 2),
		release: make(chan struct{}),
	}
	close(driver.release)
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	msg := types.InboxMessage{
		MessageID: "msg-acked",
		Target:    "#agentteam:thread",
		TargetKey: "scope",
		TargetSeq: 1,
		Content:   "single runtime body",
	}
	require.NoError(t, ap.deliverMessage(msg, false))
	require.Equal(t, "msg-acked", <-driver.started)

	wakeEntered := make(chan struct{})
	wakeRelease := make(chan struct{})
	var wakeOnce sync.Once
	ap.onWakeStatus = func(WakeStatusEvent) {
		wakeOnce.Do(func() { close(wakeEntered) })
		<-wakeRelease
	}

	ackDone := make(chan struct{})
	go func() {
		ap.handleParsedEvent(drivercontract.ParsedEvent{
			Kind:           consts.ParsedEventKindRuntimeInputAcked,
			RuntimeInputID: "msg-acked",
			SessionID:      "thread-root",
		})
		close(ackDone)
	}()
	<-wakeEntered

	// The acked owner must already be visible even though the callback has
	// not returned. A concurrent duplicate therefore completes without a
	// second transport call.
	duplicateDone := make(chan error, 1)
	go func() { duplicateDone <- ap.deliverMessage(msg, false) }()
	require.NoError(t, <-duplicateDone)
	select {
	case inputID := <-driver.started:
		t.Fatalf("duplicate input issued a second runtime call %q", inputID)
	case <-time.After(20 * time.Millisecond):
	}

	close(wakeRelease)
	<-ackDone
	require.Equal(t, []string{"msg-acked"}, driver.callIDs())
	ap.mu.Lock()
	require.Empty(t, ap.pendingRuntimeInputs)
	require.Contains(t, ap.ackedSet, "msg-acked")
	ap.mu.Unlock()
}

func TestTurnEndClaimsQueuedBodyBeforeConcurrentFreshDelivery(t *testing.T) {
	order := &deliveryOrderRecorder{}
	driver := &boundaryBlockingAckedDriver{
		supportsEntered: make(chan struct{}),
		supportsRelease: make(chan struct{}),
		order:           order,
	}
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver:    driver,
			stdinPipe: &orderedRecordingWriteCloser{order: order},
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-1",
			Content:   "older queued body",
		}}},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	boundaryDone := make(chan struct{})
	go func() {
		ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd, SessionID: "thread-root"})
		close(boundaryDone)
	}()
	<-driver.supportsEntered

	newerDone := make(chan error, 1)
	go func() {
		newerDone <- ap.deliverMessage(types.InboxMessage{MessageID: "msg-2", Content: "newer body"}, false)
	}()
	select {
	case err := <-newerDone:
		t.Fatalf("newer delivery escaped while the older boundary claim was in progress: %v", err)
	case <-time.After(20 * time.Millisecond):
	}

	close(driver.supportsRelease)
	<-boundaryDone
	require.NoError(t, <-newerDone)
	require.Equal(t, []string{"msg-1", "msg-2"}, order.snapshot())
}

func TestHandoffDeliveryGateWaitHonorsContext(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	require.NoError(t, ap.acquireDeliveryGate(nil))
	defer ap.releaseDeliveryGate()

	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	err := ap.DeliverForHandoff(ctx, types.InboxMessage{MessageID: "msg-timeout", Content: "body"})
	require.ErrorIs(t, err, context.DeadlineExceeded)
	require.Zero(t, driver.deliverCalls)
}

func TestStaleCodexTurnEndCannotMarkNewlyStartedTurnIdle(t *testing.T) {
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		switch req.Method {
		case "thread/read":
			return appServerEnvelope{Result: rawRPCResult(t, appServerThreadResult{
				Thread: appServerThread{ID: "thread-root", Status: appServerThreadStatus{Type: "idle"}},
			})}
		case "turn/start":
			require.Equal(t, "msg-2", req.Params["clientUserMessageId"])
			return appServerEnvelope{Result: rawRPCResult(t, appServerTurnResult{Turn: appServerTurn{
				ID:     "turn-new",
				Status: appServerTurnStatus{Type: "inProgress"},
			}})}
		case "turn/steer":
			require.Equal(t, "turn-new", req.Params["expectedTurnId"])
			require.Equal(t, "msg-3", req.Params["clientUserMessageId"])
			return appServerEnvelope{Result: rawRPCResult(t, appServerTurnIDResult{TurnID: "turn-new"})}
		default:
			t.Fatalf("unexpected request %s", req.Method)
			return appServerEnvelope{}
		}
	})
	driver.translatedLines = make(chan string, 4)
	driver.updateThreadSnapshot(appServerThread{
		ID:     "thread-root",
		Status: appServerThreadStatus{Type: "active"},
		Turns: []appServerTurn{{
			ID:     "turn-old",
			Status: appServerTurnStatus{Type: "inProgress"},
		}},
	})
	driver.handleNotification(appServerEnvelope{
		Method: "turn/completed",
		Params: rawRPCResult(t, appServerTurnNotification{
			ThreadID: "thread-root",
			Turn:     appServerTurn{ID: "turn-old", Status: appServerTurnStatus{Type: "completed"}},
		}),
	})
	turnEndEvents := driver.ParseLine(<-driver.translatedLines)
	require.Len(t, turnEndEvents, 1)
	turnEnd := turnEndEvents[0]
	require.Equal(t, "turn-old", turnEnd.Metadata[runtimeCompletedTurnIDKey])

	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	var emittedKinds []string
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		emittedKinds = append(emittedKinds, ev.Kind)
	})
	require.NoError(t, ap.acquireDeliveryGate(nil))
	turnEndDone := make(chan struct{})
	go func() {
		ap.handleParsedEvent(turnEnd)
		close(turnEndDone)
	}()

	// While the older TurnEnd is waiting for the gate, the authoritative driver
	// has already become idle. Delivering q2 must refresh and start the original
	// body, advancing the Process handoff epoch before the old boundary resumes.
	require.NoError(t, ap.deliverMessageWithContextSerialized(nil, types.InboxMessage{
		MessageID: "msg-2",
		Content:   "start after old completion",
	}, false))
	require.Equal(t, "turn-new", driver.currentActiveTurnID())
	ap.releaseDeliveryGate()
	<-turnEndDone

	ap.mu.Lock()
	require.False(t, ap.isIdle, "old TurnEnd must not mark the newly started turn idle")
	require.Equal(t, uint64(1), ap.runtimeHandoffEpoch)
	ap.mu.Unlock()
	require.NotContains(t, emittedKinds, consts.ParsedEventKindTurnEnd, "stale TurnEnd must not terminate external consumers")

	// A late RuntimeIdle from the old turn is stale for the same reason.
	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:      consts.ParsedEventKindRuntimeIdle,
		SessionID: "thread-root",
	})
	ap.mu.Lock()
	require.False(t, ap.isIdle)
	ap.mu.Unlock()
	require.NotContains(t, emittedKinds, consts.ParsedEventKindRuntimeIdle, "stale RuntimeIdle must not publish external idle state")

	// The next body must still take the correlated acked/steer path rather than
	// the idle protocol path with prematurely acked freshness.
	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-3",
		Content:   "steer while new turn is active",
	}, false))
	require.Equal(t, []string{"thread/read", "turn/start", "turn/steer"}, capturedRPCMethods(writer.requests))
	ap.mu.Lock()
	require.False(t, ap.isIdle)
	require.Equal(t, uint64(2), ap.runtimeHandoffEpoch)
	require.Contains(t, ap.pendingRuntimeInputs, "msg-2")
	require.Contains(t, ap.pendingRuntimeInputs, "msg-3")
	ap.mu.Unlock()
}

func TestLiveTransientRecoverySerializesRollbackAndRedelivery(t *testing.T) {
	order := &deliveryOrderRecorder{}
	driver := &blockingLiveRecoveryDriver{
		restoreStarted: make(chan struct{}),
		restoreRelease: make(chan struct{}),
		order:          order,
	}
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver:    driver,
			stdinPipe: noopWriteCloser{},
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
	}
	require.NoError(t, ap.SetOnResumeSafety(func(uint64, bool, string) error { return nil }))

	retryDone := make(chan error, 1)
	go func() {
		retryDone <- ap.restoreOrRestartForTransientRuntimeRetry(context.Background(), transientRuntimeRetryState{
			MessageID:   "msg-1",
			SessionID:   "thread-root",
			Strategy:    "codex_app_server_turns",
			RecoveryRef: "recovery-ref",
			Input:       testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-1"}, "retry original"),
		})
	}()
	<-driver.restoreStarted

	newerDone := make(chan error, 1)
	go func() {
		newerDone <- ap.deliverMessage(types.InboxMessage{MessageID: "msg-2", Content: "newer body"}, false)
	}()
	select {
	case err := <-newerDone:
		t.Fatalf("newer input bypassed an in-progress live rollback/redelivery: %v", err)
	case <-time.After(20 * time.Millisecond):
	}

	close(driver.restoreRelease)
	require.NoError(t, <-retryDone)
	require.NoError(t, <-newerDone)
	require.Equal(t, []string{"msg-1", "msg-2"}, order.snapshot())
	ap.mu.Lock()
	require.Equal(t, uint64(2), ap.runtimeHandoffEpoch)
	require.Contains(t, ap.pendingRuntimeInputs, "msg-2")
	ap.mu.Unlock()
}

func TestLiveRestoreAndHandoffShareOneExclusiveAdmissionOrder(t *testing.T) {
	t.Run("restore first fences handoff until rollback completes", func(t *testing.T) {
		driver := &blockingLiveRecoveryDriver{
			restoreStarted: make(chan struct{}),
			restoreRelease: make(chan struct{}),
			order:          &deliveryOrderRecorder{},
		}
		writer := &recordingWriteCloser{}
		ap := &Process{
			AgentID:   "agent-restore-first",
			SessionID: "thread-root",
			processRuntimeBinding: processRuntimeBinding{
				driver:    driver,
				stdinPipe: writer,
			},
			processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
			processRecoveryState:  processRecoveryState{disableRecovery: true},
		}

		restoreDone := make(chan error, 1)
		go func() {
			restoreDone <- ap.RestoreRecoveryPoint(context.Background(), "codex_app_server_turns", "recovery-ref")
		}()
		<-driver.restoreStarted

		deliveryDone := make(chan error, 1)
		go func() {
			deliveryDone <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
				MessageID: "wake-after-restore",
				Target:    "#ops:thread",
				Content:   "exact body after rollback",
			})
		}()
		select {
		case err := <-deliveryDone:
			t.Fatalf("handoff escaped while live restore held the admission gate: %v", err)
		case <-time.After(20 * time.Millisecond):
		}

		close(driver.restoreRelease)
		require.NoError(t, <-restoreDone)
		require.NoError(t, <-deliveryDone)
		require.Contains(t, writer.String(), "exact body after rollback")
	})

	t.Run("handoff first makes later restore reject without touching runtime", func(t *testing.T) {
		driver := &blockingLiveRecoveryDriver{
			restoreStarted: make(chan struct{}),
			restoreRelease: make(chan struct{}),
			order:          &deliveryOrderRecorder{},
		}
		writer := &blockingRecoveryDeliveryWriter{
			entered: make(chan struct{}),
			release: make(chan struct{}),
		}
		ap := &Process{
			AgentID:   "agent-handoff-first",
			SessionID: "thread-root",
			processRuntimeBinding: processRuntimeBinding{
				driver:    driver,
				stdinPipe: writer,
			},
			processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
			processRecoveryState:  processRecoveryState{disableRecovery: true},
		}

		deliveryDone := make(chan error, 1)
		go func() {
			deliveryDone <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
				MessageID: "wake-before-restore",
				Target:    "#ops:thread",
				Content:   "exact body before rollback",
			})
		}()
		<-writer.entered

		restoreDone := make(chan error, 1)
		go func() {
			restoreDone <- ap.RestoreRecoveryPoint(context.Background(), "codex_app_server_turns", "recovery-ref")
		}()
		select {
		case <-driver.restoreStarted:
			t.Fatal("restore reached the runtime while an older handoff still owned the admission gate")
		case <-time.After(20 * time.Millisecond):
		}

		close(writer.release)
		require.NoError(t, <-deliveryDone)
		require.ErrorIs(t, <-restoreDone, ErrAgentBusy)
		select {
		case <-driver.restoreStarted:
			t.Fatal("busy restore must not mutate the runtime")
		default:
		}
	})
}

func TestLiveRestoreRejectsDurableOwnershipMissingFromLocalContainers(t *testing.T) {
	driver := &blockingLiveRecoveryDriver{
		restoreStarted: make(chan struct{}),
		restoreRelease: make(chan struct{}),
		order:          &deliveryOrderRecorder{},
	}
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"accepted-wake": {
			Message: types.InboxMessage{MessageID: "accepted-wake", Content: "journal-only body"},
			State:   DeliveryOwnershipQueued,
		},
	}}
	ap := &Process{
		AgentID:               "agent-journal-proof",
		SessionID:             "thread-root",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}

	err := ap.RestoreRecoveryPoint(context.Background(), "codex_app_server_turns", "recovery-ref")

	require.ErrorIs(t, err, ErrAgentBusy)
	select {
	case <-driver.restoreStarted:
		t.Fatal("restore must not touch the runtime while a durable live owner exists")
	default:
	}
}

func TestDetachedRecoverySleepRefusesToRetireDurableOwner(t *testing.T) {
	store := &memoryDeliveryOwnershipStore{records: map[string]DeliveryOwnershipRecord{
		"accepted-wake": {
			Message: types.InboxMessage{MessageID: "accepted-wake", Content: "journal-only body"},
			State:   DeliveryOwnershipInflight, InflightReason: DeliveryInflightOutcomeUnknown,
		},
	}}
	ap := &Process{
		AgentID: "agent-detached-proof",
		processRuntimeBinding: processRuntimeBinding{
			driver: &busyDeliveryStubDriver{},
			cancel: func() {},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}

	err := ap.SleepForRecovery(context.Background())

	require.ErrorIs(t, err, ErrAgentBusy)
	require.True(t, ap.AcceptingDeliveries(), "recovery must not retire the only Process while its journal still owns a wake")
}

func TestRestoreFinalizationKeepsDeliveryFencedUntilProcessRetired(t *testing.T) {
	driver := &blockingLiveRecoveryDriver{
		restoreStarted: make(chan struct{}),
		restoreRelease: make(chan struct{}),
		order:          &deliveryOrderRecorder{},
	}
	finalizeStarted := make(chan struct{})
	finalizeRelease := make(chan struct{})
	ap := &Process{
		AgentID:   "agent-finalize-fence",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver:    driver,
			stdinPipe: &recordingWriteCloser{},
			cancel:    func() {},
		},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
		processRecoveryState:  processRecoveryState{disableRecovery: true},
	}

	restoreDone := make(chan error, 1)
	go func() {
		restoreDone <- ap.RestoreRecoveryPointAndFinalize(
			context.Background(),
			"codex_app_server_turns",
			"recovery-ref",
			func() error {
				close(finalizeStarted)
				<-finalizeRelease
				return nil
			},
		)
	}()
	<-driver.restoreStarted
	close(driver.restoreRelease)
	<-finalizeStarted

	deliveryDone := make(chan error, 1)
	go func() {
		deliveryDone <- ap.DeliverForHandoff(context.Background(), types.InboxMessage{
			MessageID: "wake-during-finalize",
			Content:   "must not enter retiring process",
		})
	}()
	require.ErrorIs(t, <-deliveryDone, ErrProcessNotAcceptingDelivery)

	// A concurrent external Stop is a strong completion boundary. It observes
	// the recovery-owned transition and waits for both its finalizer and the
	// outer deliveryGate lease instead of returning early or recursively trying
	// to become the stop owner.
	stopDone := make(chan struct{})
	go func() {
		ap.Stop()
		close(stopDone)
	}()
	select {
	case <-stopDone:
		t.Fatal("concurrent Stop returned before recovery finalization completed")
	case <-time.After(20 * time.Millisecond):
	}

	close(finalizeRelease)
	require.NoError(t, <-restoreDone)
	select {
	case <-stopDone:
	case <-time.After(time.Second):
		t.Fatal("concurrent Stop did not complete after recovery owner released delivery gate")
	}
}

func TestTransientRetryPendingQueuesNewBodyUntilRecoveryOwnsRuntime(t *testing.T) {
	t.Cleanup(setTransientRuntimeRetryDelayFuncForTest(func(int) time.Duration { return 0 }))

	order := &deliveryOrderRecorder{}
	driver := &blockingLiveRecoveryDriver{
		restoreStarted: make(chan struct{}),
		restoreRelease: make(chan struct{}),
		order:          order,
	}
	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver:    driver,
			stdinPipe: noopWriteCloser{},
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{workDir: t.TempDir()},
	}
	require.NoError(t, ap.SetOnResumeSafety(func(uint64, bool, string) error { return nil }))
	ap.pendingWake = &pendingDelivery{Wake: WakeInfo{MessageID: "msg-1", Target: "#agentteam:thread"}, Status: pendingDeliveryAwaitingCheck}
	ap.rememberTransientRuntimeRetryInput("msg-1", testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-1"}, "retry original"), "thread-root")

	// Error A reserves the runtime for its rollback/redelivery, even if the
	// Codex driver has already observed the failed turn as idle.
	ap.handleParsedEvent(transient529Event())
	ap.mu.Lock()
	require.True(t, ap.transientRuntimeRetryPending)
	require.True(t, ap.transientRuntimeRetryAwaitingTurnEnd)
	ap.mu.Unlock()

	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-2",
		Target:    "#agentteam:thread",
		TargetKey: "scope",
		TargetSeq: 2,
		Content:   "newer body",
	}, false))
	require.Empty(t, order.snapshot(), "q2 must not start or steer while q1 recovery is pending")
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "msg-2", ap.inbox[0].MessageID)
	ap.mu.Unlock()

	// The failed turn boundary releases the retry goroutine, not the newer
	// queued body. Keep recovery blocked to make the ordering observable.
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd, SessionID: "thread-root"})
	<-driver.restoreStarted
	require.Empty(t, order.snapshot())
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "msg-2", ap.inbox[0].MessageID)
	ap.mu.Unlock()

	close(driver.restoreRelease)
	require.Eventually(t, func() bool {
		return len(order.snapshot()) == 2
	}, time.Second, 10*time.Millisecond)
	require.Equal(t, []string{"msg-1", "msg-2"}, order.snapshot(), "q1 recovery must own the first runtime handoff before q2 auto-retries")
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.Contains(t, ap.pendingRuntimeInputs, "msg-2")
	ap.mu.Unlock()
}

func TestSessionInitRetriesQueuedOriginalBodyThroughAckedRuntimeInput(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	ap := &Process{
		AgentID:               "agent-codex",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-before-session",
			TargetKey: "scope",
			TargetSeq: 1,
			Content:   "original queued body",
		}}},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:      consts.ParsedEventKindSessionInit,
		SessionID: "thread-root",
	})

	require.Equal(t, 1, driver.deliverCalls)
	require.Equal(t, "msg-before-session", driver.lastInput.ID)
	require.Contains(t, driver.lastInput.PromptText(), "original queued body")
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.Contains(t, ap.pendingRuntimeInputs, "msg-before-session")
	ap.mu.Unlock()
}

func TestAckedIdleStartFailureRetriesOriginalBodyWithoutRuntimeBoundary(t *testing.T) {
	var startCalls atomic.Int32
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		switch req.Method {
		case "thread/read":
			return appServerEnvelope{Result: rawRPCResult(t, appServerThreadResult{
				Thread: appServerThread{
					ID:     "thread-root",
					Status: appServerThreadStatus{Type: "idle"},
				},
			})}
		case "turn/start":
			require.Equal(t, "msg-1", req.Params["clientUserMessageId"])
			require.Contains(t, ackedDeliveryInputText(t, req.Params), "original body")
			if startCalls.Add(1) == 1 {
				return appServerEnvelope{Error: &appServerResponseError{
					Code:    -32000,
					Message: "temporary turn/start failure",
				}}
			}
			return appServerEnvelope{Result: rawRPCResult(t, appServerTurnIDResult{TurnID: "turn-retry"})}
		default:
			t.Fatalf("unexpected request %s", req.Method)
			return appServerEnvelope{}
		}
	})
	// Process still believes a turn is busy, while the driver cache cannot
	// steer and must refresh. The refresh then proves the thread is idle.
	driver.updateThreadSnapshot(appServerThread{
		ID:     "thread-root",
		Status: appServerThreadStatus{Type: "active"},
	})

	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processDeliveryState: processDeliveryState{
			inboxRetryDelay: func(int) time.Duration { return 5 * time.Millisecond },
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}
	t.Cleanup(func() {
		ap.mu.Lock()
		ap.stopInboxRetryLocked()
		ap.mu.Unlock()
	})

	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-1",
		Target:    "#agentteam:thread",
		TargetKey: "scope",
		TargetSeq: 1,
		Content:   "original body",
	}, false))
	require.Equal(t, []string{"thread/read", "turn/start"}, capturedRPCMethods(writer.requests))
	require.Eventually(t, func() bool {
		ap.mu.Lock()
		defer ap.mu.Unlock()
		_, pending := ap.pendingRuntimeInputs["msg-1"]
		return startCalls.Load() >= 2 && pending && len(ap.inbox) == 0
	}, time.Second, 5*time.Millisecond, "daemon retry must recover without TurnEnd/RuntimeIdle/SessionInit")
}

func TestCodexChildCompletionKeepsRootBusyAndSteersOriginalBody(t *testing.T) {
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		require.Equal(t, "turn/steer", req.Method)
		require.Equal(t, "thread-root", req.Params["threadId"])
		require.Equal(t, "turn-root", req.Params["expectedTurnId"])
		require.Equal(t, "msg-after-child", req.Params["clientUserMessageId"])
		require.Contains(t, ackedDeliveryInputText(t, req.Params), "body after child completed")
		return appServerEnvelope{Result: rawRPCResult(t, appServerTurnIDResult{TurnID: "turn-root"})}
	})
	driver.updateThreadSnapshot(appServerThread{
		ID:     "thread-root",
		Status: appServerThreadStatus{Type: "active"},
		Turns: []appServerTurn{{
			ID:     "turn-root",
			Status: appServerTurnStatus{Type: "inProgress"},
		}},
	})

	ap := &Process{
		AgentID:   "agent-codex",
		SessionID: "thread-root",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{
			state:  processStateRunning,
			isIdle: false,
		},
		processRecoveryState: processRecoveryState{disableRecovery: true},
	}

	// App-server multiplexes root and subagent notifications on one stream.
	// A child completion must not make either the driver or Process treat the
	// still-running root turn as idle.
	driver.handleNotification(appServerEnvelope{
		Method: "turn/completed",
		Params: rawRPCResult(t, appServerTurnNotification{
			ThreadID: "thread-child",
			Turn: appServerTurn{
				ID:     "turn-child",
				Status: appServerTurnStatus{Type: "completed"},
			},
		}),
	})

	require.NoError(t, ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-after-child",
		TargetKey: "scope",
		TargetSeq: 1,
		Content:   "body after child completed",
	}, false))

	require.Equal(t, []string{"turn/steer"}, capturedRPCMethods(writer.requests))
	require.Equal(t, "turn-root", driver.currentActiveTurnID())
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	require.Contains(t, ap.pendingRuntimeInputs, "msg-after-child")
	ap.mu.Unlock()
}