package agent

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

	"github.com/stretchr/testify/require"

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

type busyDeliveryStubDriver struct{}

func (d *busyDeliveryStubDriver) Name() string { return "claude" }

func (d *busyDeliveryStubDriver) SupportsStdinNotification() bool { return true }

func (d *busyDeliveryStubDriver) MCPToolPrefix() string { return "mcp__agentteam__" }

func (d *busyDeliveryStubDriver) Spawn(context.Context, drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	return drivercontract.SpawnResult{}, nil
}

func (d *busyDeliveryStubDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }

func (d *busyDeliveryStubDriver) EncodeStdinMessage(text, _ string) []byte {
	return []byte(text)
}

type recordingWriteCloser struct {
	mu   sync.Mutex
	data []byte
}

func (w *recordingWriteCloser) Write(p []byte) (int, error) {
	w.mu.Lock()
	defer w.mu.Unlock()
	w.data = append(w.data, p...)
	return len(p), nil
}

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

func (w *recordingWriteCloser) String() string {
	w.mu.Lock()
	defer w.mu.Unlock()
	return string(w.data)
}

type busyStdinProcessStubDriver struct {
	busyDeliveryStubDriver
	encodeNil bool
}

func (d *busyStdinProcessStubDriver) SupportsBusyStdinDelivery() bool { return true }

func (d *busyStdinProcessStubDriver) EncodeStdinMessage(text, _ string) []byte {
	if d.encodeNil {
		return nil
	}
	return []byte(text)
}

type noStdinNotificationStubDriver struct {
	busyDeliveryStubDriver
}

func (d *noStdinNotificationStubDriver) SupportsStdinNotification() bool { return false }

type testRuntimeAttachmentResolver struct {
	attachments []types.RuntimeAttachment
	err         error
}

func (r testRuntimeAttachmentResolver) ResolveRuntimeAttachments(context.Context, []types.MessageAttachment) ([]types.RuntimeAttachment, error) {
	return r.attachments, r.err
}

type protocolRecoveryStubDriver struct {
	runtimeName        string
	deliverErr         error
	recoveryRef        string
	activeTurnID       string
	supportsBusy       bool
	onDeliver          func()
	prepareCalls       int
	deliverCalls       int
	lastText           string
	contextDeadline    time.Time
	contextHasDeadline bool
}

type protocolDeliveryStubDriver struct {
	runtimeName        string
	deliverErr         error
	supportsBusy       bool
	onDeliver          func()
	deliverCalls       int
	lastText           string
	contextDeadline    time.Time
	contextHasDeadline bool
}

type runtimeInputDeliveryStubDriver struct {
	busyDeliveryStubDriver
	lastInput   drivercontract.RuntimeInput
	lastSession string
}

func (d *runtimeInputDeliveryStubDriver) DeliverRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	d.lastInput = input
	d.lastSession = sessionID
	return nil
}

type ackedRuntimeInputStubDriver struct {
	busyDeliveryStubDriver
	mu           sync.Mutex
	supports     bool
	deliverErr   error
	onDeliver    func(input drivercontract.RuntimeInput, sessionID string)
	deliverCalls int
	lastInput    drivercontract.RuntimeInput
	lastSession  string
}

func (d *ackedRuntimeInputStubDriver) SupportsAckedBusyRuntimeInputDelivery() bool {
	return d.supports
}

func (d *ackedRuntimeInputStubDriver) DeliverAckedBusyRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	d.mu.Lock()
	d.deliverCalls++
	d.lastInput = input
	d.lastSession = sessionID
	onDeliver := d.onDeliver
	deliverErr := d.deliverErr
	d.mu.Unlock()
	if onDeliver != nil {
		onDeliver(input, sessionID)
	}
	return deliverErr
}

func (d *ackedRuntimeInputStubDriver) snapshot() (int, drivercontract.RuntimeInput, string) {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.deliverCalls, d.lastInput, d.lastSession
}

type ackedTurnCountRecoveryStubDriver struct {
	ackedRuntimeInputStubDriver
	prepareCalls int
}

func (d *ackedTurnCountRecoveryStubDriver) PrepareRecoveryPoint() (string, string, bool, error) {
	d.prepareCalls++
	return "codex_app_server_turns", `{"threadId":"sess-1","turnCount":1}`, true, nil
}

func (d *ackedTurnCountRecoveryStubDriver) RestoreRecoveryPoint(context.Context, string, string) error {
	return nil
}

type interruptingRuntimeInputStubDriver struct {
	busyDeliveryStubDriver
	calls       int
	lastInput   drivercontract.RuntimeInput
	lastSession string
	onDeliver   func()
}

func (d *interruptingRuntimeInputStubDriver) DeliverInterruptingRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	d.calls++
	d.lastInput = input
	d.lastSession = sessionID
	if d.onDeliver != nil {
		d.onDeliver()
	}
	return nil
}

func (d *protocolRecoveryStubDriver) Name() string {
	if d.runtimeName != "" {
		return d.runtimeName
	}
	return "codex"
}

func (d *protocolRecoveryStubDriver) SupportsStdinNotification() bool { return true }

func (d *protocolRecoveryStubDriver) SupportsBusyStdinDelivery() bool { return d.supportsBusy }

func (d *protocolRecoveryStubDriver) MCPToolPrefix() string { return "mcp_agentteam_" }

func (d *protocolRecoveryStubDriver) Spawn(context.Context, drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	return drivercontract.SpawnResult{}, nil
}

func (d *protocolRecoveryStubDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }

func (d *protocolRecoveryStubDriver) EncodeStdinMessage(string, string) []byte { return nil }

func (d *protocolRecoveryStubDriver) DeliverBusyStdinInput(ctx context.Context, input drivercontract.RuntimeInput, _ string) error {
	text := input.PromptText()
	d.deliverCalls++
	d.lastText = text
	d.contextDeadline, d.contextHasDeadline = ctx.Deadline()
	if d.onDeliver != nil {
		d.onDeliver()
	}
	if d.deliverErr == nil {
		d.recoveryRef = "fresh-ref"
	}
	return d.deliverErr
}

func (d *protocolRecoveryStubDriver) PrepareRecoveryPoint() (string, string, bool, error) {
	d.prepareCalls++
	return "codex_app_server_turns", d.recoveryRef, true, nil
}

func (d *protocolRecoveryStubDriver) RestoreRecoveryPoint(context.Context, string, string) error {
	return nil
}

func (d *protocolRecoveryStubDriver) CurrentActiveTurnID() string {
	return d.activeTurnID
}

func (d *protocolDeliveryStubDriver) Name() string {
	if d.runtimeName != "" {
		return d.runtimeName
	}
	return "claude"
}

func (d *protocolDeliveryStubDriver) SupportsStdinNotification() bool { return true }

func (d *protocolDeliveryStubDriver) SupportsBusyStdinDelivery() bool { return d.supportsBusy }

func (d *protocolDeliveryStubDriver) MCPToolPrefix() string { return "mcp_agentteam_" }

func (d *protocolDeliveryStubDriver) Spawn(context.Context, drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	return drivercontract.SpawnResult{}, nil
}

func (d *protocolDeliveryStubDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }

func (d *protocolDeliveryStubDriver) EncodeStdinMessage(string, string) []byte { return nil }

func (d *protocolDeliveryStubDriver) DeliverBusyStdinInput(ctx context.Context, input drivercontract.RuntimeInput, _ string) error {
	text := input.PromptText()
	d.deliverCalls++
	d.lastText = text
	d.contextDeadline, d.contextHasDeadline = ctx.Deadline()
	if d.onDeliver != nil {
		d.onDeliver()
	}
	return d.deliverErr
}

func TestLaunchConfigStartsTurnWhenPromptAttachmentsPresent(t *testing.T) {
	require.True(t, launchConfigStartsTurn(drivercontract.LaunchConfig{
		PromptAttachments: []types.RuntimeAttachment{{
			Filename: "context.txt",
		}},
	}))
}

func TestLaunchConfigStartsTurnForClaudeEmptyPrompt(t *testing.T) {
	// Claude enforces the authoritative no-prompt gate: a true no-text/no-attachment
	// launch is idle and does not start a turn, so the Process waits for the
	// daemon's first durable handoff to write the first real frame.
	require.False(t, launchConfigStartsTurn(drivercontract.LaunchConfig{
		RuntimeType: runtimeconfig.TypeClaudeCode,
	}))
	require.False(t, launchConfigStartsTurn(drivercontract.LaunchConfig{
		RuntimeType: runtimeconfig.TypeOpenCode,
	}))
	require.True(t, launchConfigStartsTurn(testLaunchConfig(drivercontract.LaunchConfig{RuntimeType: runtimeconfig.TypeClaudeCode}, testLaunchSurfaces{Prompt: "hello"})))
	require.True(t, launchConfigStartsTurn(drivercontract.LaunchConfig{
		RuntimeType: runtimeconfig.TypeClaudeCode,
		PromptAttachments: []types.RuntimeAttachment{{
			Filename: "context.txt",
		}},
	}))
}

func TestDeliveryContextsHaveDeadlines(t *testing.T) {
	protocolCtx, protocolCancel := newProtocolDeliveryContextFor(context.Background())
	defer protocolCancel()
	protocolDeadline, ok := protocolCtx.Deadline()
	require.True(t, ok)
	require.WithinDuration(t, time.Now().Add(processProtocolDeliveryTimeout), protocolDeadline, time.Second)

	promptCtx, promptCancel := newDeliveryPromptContext()
	defer promptCancel()
	promptDeadline, ok := promptCtx.Deadline()
	require.True(t, ok)
	require.WithinDuration(t, time.Now().Add(processDeliveryPromptTimeout), promptDeadline, time.Second)
}

func TestDeliverForHandoffPropagatesCallerDeadlineToRuntimeDelivery(t *testing.T) {
	driver := &protocolRecoveryStubDriver{}
	ap := &Process{
		AgentID:               "agent-handoff-deadline",
		SessionID:             "session-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: true},
		processRecoveryState: processRecoveryState{workDir: t.TempDir()}}

	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	wantDeadline, ok := ctx.Deadline()
	require.True(t, ok)

	err := ap.DeliverForHandoff(ctx, types.InboxMessage{
		MessageID:   "msg-handoff-deadline",
		Target:      "#agentteam:ops",
		ChannelType: "channel",
		ChannelName: "ops",
		SenderType:  "user",
		SenderName:  "owner",
		Content:     "bounded handoff",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	})

	require.NoError(t, err)
	require.Equal(t, 1, driver.deliverCalls)
	require.True(t, driver.contextHasDeadline)
	require.WithinDuration(t, wantDeadline, driver.contextDeadline, time.Millisecond)
}

func TestProcessInterruptAndDeliverUsesInterruptingRuntimeInputDriver(t *testing.T) {
	driver := &interruptingRuntimeInputStubDriver{}
	ap := &Process{
		AgentID:               "agent-interrupt",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()}}

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{
		MessageID:   "msg-interrupt",
		TargetKey:   "scp_thread",
		TargetSeq:   7,
		Target:      "#agentteam:abc123ef",
		ChannelType: "thread",
		ChannelName: "thread-abc123ef",
		SenderType:  "user",
		SenderName:  "owner",
		Content:     "new authoritative input",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	})

	require.NoError(t, err)
	require.True(t, handled)
	require.Equal(t, 1, driver.calls)
	require.Equal(t, "thread-1", driver.lastSession)
	require.Contains(t, driver.lastInput.PromptText(), "new authoritative input")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_thread")
	require.True(t, ok)
	require.Equal(t, int64(7), frontier.ObservedSeq)
	require.Equal(t, []string{"msg-interrupt"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-interrupt"}))
}

func TestProcessInterruptAndDeliverBypassesObservedFrontier(t *testing.T) {
	driver := &interruptingRuntimeInputStubDriver{}
	ap := &Process{
		AgentID:               "agent-interrupt",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()}}

	ap.RememberObservedFrontier(types.InboxMessage{
		MessageID: "msg-interrupt",
		TargetKey: "scp_thread",
		TargetSeq: 7,
	})

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{
		MessageID:   "msg-interrupt",
		TargetKey:   "scp_thread",
		TargetSeq:   7,
		Target:      "#agentteam:abc123ef",
		ChannelType: "thread",
		ChannelName: "thread-abc123ef",
		SenderType:  "user",
		SenderName:  "owner",
		Content:     "same visible message replayed urgently",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	})

	require.NoError(t, err)
	require.True(t, handled)
	require.Equal(t, 1, driver.calls)
	require.Contains(t, driver.lastInput.PromptText(), "same visible message replayed urgently")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_thread")
	require.True(t, ok)
	require.Equal(t, int64(7), frontier.ObservedSeq)
}

func TestProcessInterruptAndDeliverMarksFreshnessAfterInterruptHandoff(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-interrupt",
		SessionID:             "thread-1",
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()}}

	driver := &interruptingRuntimeInputStubDriver{
		onDeliver: func() {
			_, ok := ap.MessageFreshnessSnapshot("scp_thread")
			require.False(t, ok)
		},
	}
	ap.driver = driver

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{
		MessageID:   "msg-interrupt-after-handoff",
		TargetKey:   "scp_thread",
		TargetSeq:   8,
		Target:      "#agentteam:abc123ef",
		ChannelType: "thread",
		ChannelName: "thread-abc123ef",
		SenderType:  "user",
		SenderName:  "owner",
		Content:     "interrupt only after old turn is fenced",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	})

	require.NoError(t, err)
	require.True(t, handled)
	require.Equal(t, 1, driver.calls)
	frontier, ok := ap.MessageFreshnessSnapshot("scp_thread")
	require.True(t, ok)
	require.Equal(t, int64(8), frontier.ObservedSeq)
}

func TestRuntimeInputForDeliveryAppendsAttachmentManifestForTextOnlyRuntime(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-a",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}}}

	ap.SetRuntimeAttachmentResolver(testRuntimeAttachmentResolver{
		attachments: []types.RuntimeAttachment{
			{
				ID:       "att_1",
				Filename: "registrations.xlsx",
				MIME:     "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
				Path:     "/tmp/registrations.xlsx",
				URI:      "file:///tmp/registrations.xlsx",
				Size:     123,
			},
		},
	})

	msg := types.InboxMessage{MessageID: "msg-1", Content: "please inspect", AttachmentList: []types.MessageAttachment{{ID: "att_1", Filename: "registrations.xlsx"}}}
	compiled, err := ap.compileDeliveryPromptWithContext(context.Background(), msg, "turn_direct", "")
	require.NoError(t, err)
	input := ap.runtimeInputForDelivery(context.Background(), msg, compiled)

	require.Empty(t, input.Attachments)
	require.Contains(t, input.PromptText(), "please inspect")
	require.Contains(t, input.PromptText(), "AgentTeam has not parsed their contents")
	require.Contains(t, input.PromptText(), "path: /tmp/registrations.xlsx")
}

func TestRenderDeliveryTextAlwaysEmitsStrongReplyInstruction(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-a",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()}}

	msg := types.InboxMessage{
		MessageID:   "msg-1",
		Content:     "please reply",
		Target:      "dm:@alice",
		ChannelType: "dm",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	}

	text, err := ap.renderDeliveryTextWithContext(context.Background(), msg, "stdin_delivery", "wake-strong")
	require.NoError(t, err)
	// The strong participant-delivery reply instruction is always emitted.
	require.Contains(t, text, `Reply to exact target "dm:@alice"`)
	require.Contains(t, text, "Do not write it as plain output, which is not posted.")
	require.Equal(t, 1, strings.Count(text, `mcp__agentteam__message_send(target="dm:@alice", content="...")`))
	require.NotContains(t, text, "routing_kind")
}

func TestDeliverMessagePersistsDurableBodyWithoutEphemeralTurn(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	store := &memoryDeliveryOwnershipStore{}
	ap := &Process{
		AgentID:               "agent-a",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: false},
		processDeliveryState:  processDeliveryState{deliveryOwnershipStore: store},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()},
	}
	msg := types.InboxMessage{
		MessageID:   "msg-1",
		Target:      "dm:@alice",
		ChannelType: "dm",
		Content:     "persist the durable body",
	}

	require.NoError(t, ap.deliverMessageWithContextSerialized(context.Background(), msg, false))
	require.Contains(t, driver.lastInput.PromptText(), `call mcp__agentteam__message_send(target="dm:@alice", content="...")`)

	record, ok := store.record(msg.MessageID)
	require.True(t, ok)
	require.Equal(t, msg, record.Message)
	require.Equal(t, msg.MessageID, record.Input.ID)
	require.Empty(t, record.Input.CompiledPrompt)
	require.Empty(t, record.Input.Attachments)

	stopInboxRetryTestProcess(ap)
	replayDriver := &correlatedOwnershipDriver{
		reconcileResult:  RuntimeInputReconciliationAbsent,
		reconcileStarted: make(chan string, 1),
	}
	replayDriver.onRuntimeDeliver = func(drivercontract.RuntimeInput) {
		replayDriver.mu.Lock()
		replayDriver.reconcileResult = RuntimeInputReconciliationAcked
		replayDriver.mu.Unlock()
	}
	replacement := newInboxRetryTestProcess(replayDriver, time.Millisecond)
	replacement.SessionID = "sess-2"
	replacement.isIdle = true
	replacement.workDir = t.TempDir()
	t.Cleanup(func() { stopInboxRetryTestProcess(replacement) })

	require.NoError(t, replacement.SetDeliveryOwnershipStore(store))
	require.Equal(t, msg.MessageID, <-replayDriver.reconcileStarted)
	require.Eventually(t, func() bool {
		_, runtimeCalls, _, _ := replayDriver.snapshot()
		return runtimeCalls == 1
	}, time.Second, time.Millisecond)

	replayedInput, _, _, _ := replayDriver.snapshot()
	require.Contains(t, replayedInput.PromptText(), msg.Content)
	require.Equal(t, driver.lastInput.PromptText(), replayedInput.PromptText(),
		"regenerating internal delivery state must not perturb model-visible input")
	record, ok = store.record(msg.MessageID)
	require.True(t, ok)
	require.Equal(t, msg, record.Message)
	require.Equal(t, msg.MessageID, record.Input.ID)
	require.Empty(t, record.Input.CompiledPrompt)
	require.Empty(t, record.Input.Attachments)
}

func TestRuntimeInputForDeliveryProducesManifestForAttachmentOnlyMessage(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-a",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}}}

	ap.SetRuntimeAttachmentResolver(testRuntimeAttachmentResolver{
		attachments: []types.RuntimeAttachment{
			{
				ID:       "att_1",
				Filename: "registrations.csv",
				MIME:     "text/csv",
				Path:     "/tmp/registrations.csv",
				URI:      "file:///tmp/registrations.csv",
				Size:     123,
			},
		},
	})

	msg := types.InboxMessage{MessageID: "msg-1", AttachmentList: []types.MessageAttachment{{ID: "att_1", Filename: "registrations.csv"}}}
	compiled, err := ap.compileDeliveryPromptWithContext(context.Background(), msg, "turn_direct", "")
	require.NoError(t, err)
	input := ap.runtimeInputForDelivery(context.Background(), msg, compiled)

	require.Empty(t, input.Attachments)
	require.Contains(t, input.PromptText(), "Attached files have been prepared locally")
	require.Contains(t, input.PromptText(), "path: /tmp/registrations.csv")
}

func TestRuntimeInputForDeliveryKeepsFilePartsForStructuredRuntime(t *testing.T) {
	ap := &Process{
		AgentID: "agent-a",
		processRuntimeBinding: processRuntimeBinding{driver: &runtimeInputDeliveryStubDriver{
			busyDeliveryStubDriver: busyDeliveryStubDriver{},
		}}}

	ap.SetRuntimeAttachmentResolver(testRuntimeAttachmentResolver{
		attachments: []types.RuntimeAttachment{
			{ID: "att_1", Filename: "image.png", MIME: "image/png", URI: "file:///tmp/image.png", Path: "/tmp/image.png"},
		},
	})

	msg := types.InboxMessage{MessageID: "msg-1", Content: "describe it", AttachmentList: []types.MessageAttachment{{ID: "att_1", Filename: "image.png"}}}
	compiled, err := ap.compileDeliveryPromptWithContext(context.Background(), msg, "turn_direct", "")
	require.NoError(t, err)
	input := ap.runtimeInputForDelivery(context.Background(), msg, compiled)

	require.Len(t, input.Attachments, 1)
	require.Equal(t, "file:///tmp/image.png", input.Attachments[0].URI)
	require.Contains(t, input.PromptText(), "path: /tmp/image.png")
}

func TestHandleParsedEventTurnEndDeliversQueuedMessageViaStdin(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-busy",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{},
			stdinPipe: stdin},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID:  "00000038abcdefff1111222233334444",
			Target:     "#agentteam:00000037",
			SenderName: "7q8vvi2",
			Content:    "please handle this",
			Timestamp:  "2026-04-12T18:32:55+08:00",
		}}}}

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	ap.mu.Lock()
	inboxLen := len(ap.inbox)
	isIdle := ap.isIdle
	ap.mu.Unlock()

	require.Equal(t, 0, inboxLen)
	require.False(t, isIdle)
	require.Contains(t, stdin.String(), "target=#agentteam:00000037")
	require.Contains(t, stdin.String(), "@7q8vvi2: please handle this")
	require.NotContains(t, stdin.String(), "mode=stdin_delivery")
}

func TestPrepareRuntimeMessageDeliveryCapturesRecoveryForKnownPaths(t *testing.T) {
	paths := []RuntimeDeliveryPath{
		RuntimeDeliveryPathProtocol,
		RuntimeDeliveryPathStdin,
		RuntimeDeliveryPathToolOutput,
	}

	for _, path := range paths {
		t.Run(string(path), func(t *testing.T) {
			driver := &protocolRecoveryStubDriver{recoveryRef: string(path) + "-ref"}
			var prepared []types.RecoveryPointPreparedEvent
			ap := &Process{
				AgentID:               "agent-codex",
				SessionID:             "thread-1",
				processRuntimeBinding: processRuntimeBinding{driver: driver}}

			ap.SetOnRecoveryPoint(func(ev types.RecoveryPointPreparedEvent) {
				prepared = append(prepared, ev)
			})

			require.True(t, ap.PrepareRuntimeMessageDelivery(types.InboxMessage{MessageID: "msg-1"}, path))
			require.Equal(t, 1, driver.prepareCalls)
			require.Len(t, prepared, 1)
			require.Equal(t, "msg-1", prepared[0].MessageID)
			require.Equal(t, string(path)+"-ref", prepared[0].RecoveryRef)
		})
	}
}

func TestPrepareRuntimeMessageDeliveriesCapturesInboxToolMessages(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "tool-output-ref"}
	var prepared []types.RecoveryPointPreparedEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver}}

	ap.SetOnRecoveryPoint(func(ev types.RecoveryPointPreparedEvent) {
		prepared = append(prepared, ev)
	})

	preparedCount := ap.PrepareRuntimeMessageDeliveries([]types.InboxMessage{
		{MessageID: "msg-1", Content: "first"},
		{MessageID: " ", Content: "missing id"},
		{MessageID: "msg-2", Content: "second"},
	}, RuntimeDeliveryPathToolOutput)

	require.Equal(t, 2, preparedCount)
	require.Equal(t, 2, driver.prepareCalls)
	require.Len(t, prepared, 2)
	require.Equal(t, "msg-1", prepared[0].MessageID)
	require.Equal(t, "msg-2", prepared[1].MessageID)
	require.Equal(t, "thread-1", prepared[0].ThreadID)
	require.Equal(t, "tool-output-ref", prepared[0].RecoveryRef)
}

func TestDeliverMessageCapturesRecoveryPointAfterAcceptedCodexAppServerDelivery(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "stale-ref"}
	var prepared []types.RecoveryPointPreparedEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.SetOnRecoveryPoint(func(ev types.RecoveryPointPreparedEvent) {
		prepared = append(prepared, ev)
	})

	startedAt := time.Now()
	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-1",
		Content:   "please handle this",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
	require.True(t, driver.contextHasDeadline)
	require.WithinDuration(t, startedAt.Add(processProtocolDeliveryTimeout), driver.contextDeadline, 2*time.Second)
	require.Equal(t, 1, driver.prepareCalls)
	require.Len(t, prepared, 1)
	require.Equal(t, "fresh-ref", prepared[0].RecoveryRef)
	require.NotContains(t, driver.lastText, "mode=turn_direct")
	require.NotContains(t, driver.lastText, "session=thread-1")
	require.Equal(t, []string{"msg-1"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-1"}))
}

func TestDeliverMessageUsesDirectDeliveryForNonCodexMessageDriver(t *testing.T) {
	driver := &protocolRecoveryStubDriver{
		runtimeName: "builtin",
		recoveryRef: "stale-ref",
	}

	ap := &Process{
		AgentID:               "agent-builtin",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-1",
		Content:   "please handle this",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
	require.NotContains(t, driver.lastText, "mode=turn_direct")
	require.NotContains(t, driver.lastText, "session=sess-1")
	require.Equal(t, []string{"msg-1"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-1"}))
}

func TestDeliverMessageUsesDirectDeliveryForBusyRuntimeWhenSupported(t *testing.T) {
	stdin := &recordingWriteCloser{}
	driver := &protocolDeliveryStubDriver{
		runtimeName:  "builtin",
		supportsBusy: true,
	}

	ap := &Process{
		AgentID:   "agent-builtin",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver,
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-busy",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:00000037",
		SenderName: "7q8vvi2",
		Content:    "deliver during an active turn",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.NotContains(t, driver.lastText, "mode=turn_direct")
	require.NotContains(t, driver.lastText, "session=sess-1")
	require.Contains(t, driver.lastText, "deliver during an active turn")
	require.Empty(t, stdin.String())
	require.Empty(t, inbox)
	require.Equal(t, []string{"msg-busy"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-busy"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageAckedActiveTurnUsesRuntimeInputForTurnCountRecovery(t *testing.T) {
	driver := &ackedTurnCountRecoveryStubDriver{ackedRuntimeInputStubDriver: ackedRuntimeInputStubDriver{supports: true}}
	var wakeEvents []WakeStatusEvent
	var parsedEvents []drivercontract.ParsedEvent
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processRecoveryState: processRecoveryState{workDir: t.TempDir()}}

	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-2",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:thread",
		SenderName: "owner",
		Content:    "deliver through active-turn acked input",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Empty(t, inbox)
	require.Equal(t, 1, pendingLen)
	require.Equal(t, "msg-2", driver.lastInput.ID)
	require.Contains(t, driver.lastInput.PromptText(), "deliver through active-turn acked input")
	require.Empty(t, wakeEvents)
	require.Len(t, parsedEvents, 1)
	require.Equal(t, consts.ParsedEventKindRuntimeInputPending, parsedEvents[0].Kind)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageAckedActiveTurnWaitsForRuntimeAck(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	var parsedEvents []drivercontract.ParsedEvent
	var wakeEvents []WakeStatusEvent

	ap := &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processRecoveryState: processRecoveryState{workDir: t.TempDir()}}

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-2",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:thread",
		SenderName: "owner",
		Content:    "deliver through acked active input",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pending := ap.pendingRuntimeInputs["msg-2"]
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Equal(t, "sess-1", driver.lastSession)
	require.Equal(t, "msg-2", driver.lastInput.ID)
	require.Contains(t, driver.lastInput.PromptText(), "deliver through acked active input")
	require.Empty(t, inbox)
	require.Equal(t, 1, pendingLen)
	require.Equal(t, "msg-2", pending.Input.ID)
	require.NotEmpty(t, pending.WakeToken)
	require.Len(t, parsedEvents, 1)
	require.Equal(t, consts.ParsedEventKindRuntimeInputPending, parsedEvents[0].Kind)
	require.Equal(t, "msg-2", parsedEvents[0].RuntimeInputID)
	require.Equal(t, "sess-1", parsedEvents[0].SessionID)
	require.Equal(t, pendingRuntimeInputDetail, parsedEvents[0].Message)
	require.Equal(t, consts.PromptKindDeliveryText, parsedEvents[0].PromptKind)
	require.Equal(t, "routed_message_delivery", parsedEvents[0].EventType)
	require.Contains(t, parsedEvents[0].Text, "deliver through acked active input")
	require.Empty(t, wakeEvents)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "msg-2",
		SessionID:      "sess-1",
	})

	ap.mu.Lock()
	pendingLen = len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	require.Zero(t, pendingLen)
	require.Len(t, wakeEvents, 1)
	require.Equal(t, consts.WakeStatusReceived, wakeEvents[0].Status)
	require.Equal(t, "msg-2", wakeEvents[0].MessageID)
	require.Equal(t, pending.WakeToken, wakeEvents[0].WakeToken)
	require.Len(t, parsedEvents, 2)
	require.Equal(t, consts.ParsedEventKindRuntimeInputAcked, parsedEvents[1].Kind)
	require.Equal(t, "msg-2", parsedEvents[1].RuntimeInputID)
	require.Equal(t, "sess-1", parsedEvents[1].SessionID)
	require.Equal(t, runtimeInputAckedDetail, parsedEvents[1].Message)
	require.Empty(t, parsedEvents[1].Text)
	require.Equal(t, []string{"msg-2"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok = ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestStreamAckDriversWithoutReadReconciliation(t *testing.T) {
	claude := &ClaudeDriver{}
	claudeAcked, ok := any(claude).(drivercontract.AckedBusyRuntimeInputDriver)
	require.True(t, ok)
	require.True(t, claudeAcked.SupportsAckedBusyRuntimeInputDelivery())
	_, supportsReconciliation := any(claude).(RuntimeInputReconciler)
	require.False(t, supportsReconciliation)

	openCode := &OpenCodeDriver{}
	openCodeAcked, ok := any(openCode).(drivercontract.AckedBusyRuntimeInputDriver)
	require.True(t, ok)
	require.False(t, openCodeAcked.SupportsAckedBusyRuntimeInputDelivery())
	openCodeCorrelated, ok := any(openCode).(drivercontract.CorrelatedRuntimeInputDeliveryDriver)
	require.True(t, ok)
	require.True(t, openCodeCorrelated.SupportsCorrelatedRuntimeInputDelivery())
	_, supportsReconciliation = any(openCode).(RuntimeInputReconciler)
	require.False(t, supportsReconciliation)
}

func TestAckOnlyBurstAckDoesNotReconcileSibling(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true}
	store := &memoryDeliveryOwnershipStore{}
	ap := newInboxRetryTestProcess(driver, time.Hour)
	ap.deliveryOwnershipStore = store
	ap.workDir = t.TempDir()
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	for seq, messageID := range []string{"q1", "q2"} {
		require.NoError(t, ap.deliverMessage(types.InboxMessage{
			MessageID: messageID,
			TargetKey: "thread-1",
			TargetSeq: int64(seq + 1),
			Target:    "#agentteam:thread",
			Content:   "burst " + messageID,
		}, false))
	}

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "q1",
		SessionID:      ap.SessionID,
	})

	ap.mu.Lock()
	q2, pending := ap.pendingRuntimeInputs["q2"]
	_, failed := ap.failedSet["q2"]
	ap.mu.Unlock()
	require.True(t, pending, "q1 acknowledgement must not fail q2 while q2 awaits its own acknowledgement")
	require.False(t, q2.reconcileRunning)
	require.Equal(t, DeliveryInflightDispatching, q2.InflightReason)
	require.False(t, failed)
	q2Record, ok := store.record("q2")
	require.True(t, ok)
	require.True(t, q2Record.IsInflight(DeliveryInflightDispatching))

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "q2",
		SessionID:      ap.SessionID,
	})

	q1Record, ok := store.record("q1")
	require.True(t, ok)
	require.True(t, q1Record.IsAcked())
	q2Record, ok = store.record("q2")
	require.True(t, ok)
	require.True(t, q2Record.IsAcked())
	ap.mu.Lock()
	require.Empty(t, ap.pendingRuntimeInputs)
	require.NotContains(t, ap.failedSet, "q1")
	require.NotContains(t, ap.failedSet, "q2")
	ap.mu.Unlock()
}

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

	ap.Deliver(types.InboxMessage{
		MessageID: "workflow-1", TargetKey: "thread-1", TargetSeq: 1,
		Content: "workflow work",
		WorkflowContext: types.WorkflowContext{
			WorkspaceID: "workspace-1", RunID: "run-1", EffectID: "step-1", NodeID: "agent-1",
			CallbackExpected: true,
		},
	})
	ap.Deliver(types.InboxMessage{
		MessageID: "ordinary-2", TargetKey: "thread-1", TargetSeq: 2,
		Content: "later ordinary work",
	})

	calls, input, _ := driver.snapshot()
	require.Equal(t, 2, calls)
	require.Equal(t, "ordinary-2", input.ID)
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	_, _, firstPending := ap.pendingRuntimeInputByMessageIDLocked("workflow-1")
	_, _, secondPending := ap.pendingRuntimeInputByMessageIDLocked("ordinary-2")
	ap.mu.Unlock()
	require.True(t, firstPending)
	require.True(t, secondPending)
}

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

	workflowMessage := func(id, runID, stepID string, seq int64) types.InboxMessage {
		return types.InboxMessage{
			MessageID: id, TargetKey: "thread-1", TargetSeq: seq, Content: id,
			WorkflowContext: types.WorkflowContext{
				WorkspaceID: "workspace-1", RunID: runID, EffectID: stepID, NodeID: id,
				CallbackExpected: true,
			},
		}
	}
	ap.Deliver(workflowMessage("workflow-1", "run-1", "step-1", 1))
	ap.Deliver(workflowMessage("workflow-2", "run-2", "step-2", 2))

	calls, input, _ := driver.snapshot()
	require.Equal(t, 2, calls)
	require.Equal(t, "workflow-2", input.ID)
	ap.mu.Lock()
	_, _, firstPending := ap.pendingRuntimeInputByMessageIDLocked("workflow-1")
	_, _, secondPending := ap.pendingRuntimeInputByMessageIDLocked("workflow-2")
	ap.mu.Unlock()
	require.True(t, firstPending)
	require.True(t, secondPending)
}

func TestDeliverMessageAckedActiveTurnShowsPromptBeforeAckWhenAcknowledgedDuringDeliver(t *testing.T) {
	var ap *Process
	driver := &ackedRuntimeInputStubDriver{
		supports: true,
		onDeliver: func(input drivercontract.RuntimeInput, sessionID string) {
			ap.handleParsedEvent(drivercontract.ParsedEvent{
				Kind:           consts.ParsedEventKindRuntimeInputAcked,
				RuntimeInputID: input.ID,
				SessionID:      sessionID,
			})
		},
	}
	var parsedEvents []drivercontract.ParsedEvent
	var wakeEvents []WakeStatusEvent

	ap = &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processRecoveryState: processRecoveryState{workDir: t.TempDir()}}

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-2",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:thread",
		SenderName: "owner",
		Content:    "acked before transport returns",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Empty(t, inbox)
	require.Zero(t, pendingLen)
	require.Len(t, wakeEvents, 1)
	require.Equal(t, consts.WakeStatusReceived, wakeEvents[0].Status)
	require.Len(t, parsedEvents, 2)
	require.Equal(t, consts.ParsedEventKindRuntimeInputPending, parsedEvents[0].Kind)
	require.Contains(t, parsedEvents[0].Text, "acked before transport returns")
	require.Equal(t, consts.ParsedEventKindRuntimeInputAcked, parsedEvents[1].Kind)
	require.Equal(t, runtimeInputAckedDetail, parsedEvents[1].Message)
	require.Empty(t, parsedEvents[1].Text)
	require.Equal(t, []string{"msg-2"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestHandleRuntimeInputAckedSuppressesAlreadyObservedPending(t *testing.T) {
	createdAt := time.Now()
	var parsedEvents []drivercontract.ParsedEvent
	var wakeEvents []WakeStatusEvent
	ap := &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &ackedRuntimeInputStubDriver{supports: true}},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processDeliveryState: processDeliveryState{pendingRuntimeInputs: map[string]pendingRuntimeInput{
			"msg-2": {
				Message:   types.InboxMessage{MessageID: "msg-2", TargetKey: "scp_test", TargetSeq: 2, Target: "#agentteam:thread", Content: "already seen"},
				Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-2"}, "already seen"),
				SessionID: "sess-1",
				WakeToken: "wake-token",
				CreatedAt: createdAt,
			},
		}}}

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-2", TargetKey: "scp_test", TargetSeq: 2})

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "msg-2",
		SessionID:      "sess-1",
	})

	ap.mu.Lock()
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	require.Zero(t, pendingLen)
	require.Empty(t, parsedEvents)
	require.Empty(t, wakeEvents)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
}

func TestPendingRuntimeInputCoversFreshness(t *testing.T) {
	createdAt := time.Now()
	ap := &Process{
		AgentID:   "agent-claude",
		SessionID: "sess-1",
		processDeliveryState: processDeliveryState{pendingRuntimeInputs: map[string]pendingRuntimeInput{
			"msg-4": {
				Message:   types.InboxMessage{MessageID: "msg-4", TargetKey: "scp_test", TargetSeq: 4, Target: "#agentteam:thread"},
				Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-4"}, "pending"),
				SessionID: "sess-1",
				CreatedAt: createdAt,
			},
			"msg-other": {
				Message:   types.InboxMessage{MessageID: "msg-other", TargetKey: "scp_other", TargetSeq: 10, Target: "#agentteam:other"},
				Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-other"}, "other"),
				SessionID: "sess-1",
				CreatedAt: createdAt,
			},
		}}}

	require.True(t, ap.PendingRuntimeInputCoversFreshness("scp_test", 4))
	require.False(t, ap.PendingRuntimeInputCoversFreshness("scp_test", 5))
	require.False(t, ap.PendingRuntimeInputCoversFreshness("scp_other", 11))

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-4", TargetKey: "scp_test", TargetSeq: 4})
	require.False(t, ap.PendingRuntimeInputCoversFreshness("scp_test", 4))
}

func TestHandleRuntimeInputAckedSessionMismatchRequeuesWithoutAcknowledging(t *testing.T) {
	createdAt := time.Now()
	var parsedEvents []drivercontract.ParsedEvent
	var wakeEvents []WakeStatusEvent
	ap := &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &ackedRuntimeInputStubDriver{supports: true}},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processDeliveryState: processDeliveryState{
			inbox: []types.InboxMessage{{MessageID: "msg-3", TargetKey: "scp_test", TargetSeq: 3, Target: "#agentteam:thread", Content: "newer queued body"}},
			pendingRuntimeInputs: map[string]pendingRuntimeInput{
				"msg-2": {
					Message:   types.InboxMessage{MessageID: "msg-2", TargetKey: "scp_test", TargetSeq: 2, Target: "#agentteam:thread", Content: "wrong session"},
					Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-2"}, "wrong session"),
					SessionID: "sess-1",
					WakeToken: "wake-token",
					CreatedAt: createdAt,
				},
			},
		}}

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})
	ap.RememberUnobservedFrontierHint(types.InboxMessage{MessageID: "msg-2", TargetKey: "scp_test", TargetSeq: 2})

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "msg-2",
		SessionID:      "sess-2",
	})

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	require.Zero(t, pendingLen)
	require.Len(t, inbox, 2)
	require.Equal(t, "msg-2", inbox[0].MessageID)
	require.Equal(t, "msg-3", inbox[1].MessageID)
	require.Empty(t, parsedEvents)
	require.Empty(t, wakeEvents)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageAckedActiveTurnFailureRequeuesWithoutSurfacing(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{supports: true, deliverErr: errors.New("transport failed")}
	ap := &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-2",
		TargetKey: "scp_test",
		TargetSeq: 2,
		Content:   "retry after transport failure",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Len(t, inbox, 1)
	require.Equal(t, "msg-2", inbox[0].MessageID)
	require.Zero(t, pendingLen)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageAckedActiveTurnAmbiguousWithoutReconcilerStaysFailClosed(t *testing.T) {
	driver := &ackedRuntimeInputStubDriver{
		supports:   true,
		deliverErr: errors.Join(errOpenCodePromptTransportAmbiguous, io.ErrUnexpectedEOF),
	}
	store := &memoryDeliveryOwnershipStore{}
	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: "msg-ambiguous",
		TargetKey: "scp_test",
		TargetSeq: 2,
		Target:    "#agentteam:thread",
		Content:   "possibly accepted once",
	}, false))

	require.Equal(t, 1, driver.deliverCalls)
	record, ok := store.record("msg-ambiguous")
	require.True(t, ok)
	require.True(t, record.IsInflight(DeliveryInflightOutcomeUnknown))
	ap.mu.Lock()
	pending, pendingOK := ap.pendingRuntimeInputs["msg-ambiguous"]
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()
	require.True(t, pendingOK)
	require.Equal(t, DeliveryInflightOutcomeUnknown, pending.InflightReason)
	require.False(t, pending.reconcileRunning)
	require.Empty(t, inbox, "an ambiguous mutation must not be replayed without authoritative absence")
}

func TestPendingRuntimeInputPersistsAcrossTurnEndWithoutAck(t *testing.T) {
	createdAt := time.Now()
	ap := &Process{
		AgentID:               "agent-claude",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &ackedRuntimeInputStubDriver{supports: true}},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processDeliveryState: processDeliveryState{pendingRuntimeInputs: map[string]pendingRuntimeInput{
			"msg-pending": {
				Message:   types.InboxMessage{MessageID: "msg-pending", TargetKey: "scp_test", TargetSeq: 2, Content: "still pending"},
				Input:     testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "msg-pending"}, "still pending"),
				SessionID: "sess-1",
				WakeToken: "wake-token",
				CreatedAt: createdAt,
			},
		}}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})
	ap.RememberUnobservedFrontierHint(types.InboxMessage{MessageID: "msg-pending", TargetKey: "scp_test", TargetSeq: 2})

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:      consts.ParsedEventKindTurnEnd,
		SessionID: "sess-1",
	})

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()

	require.Equal(t, 1, pendingLen)
	require.Empty(t, inbox)
	require.True(t, ap.PendingRuntimeInputCoversFreshness("scp_test", 2))
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-pending"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageBusyDirectFailureRequeuesWithoutSurfacing(t *testing.T) {
	driver := &protocolDeliveryStubDriver{
		deliverErr:   errors.New("busy delivery failed"),
		runtimeName:  "builtin",
		supportsBusy: true,
	}

	ap := &Process{
		AgentID:               "agent-builtin",
		SessionID:             "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false}}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-busy",
		TargetKey: "scp_test",
		TargetSeq: 9,
		Content:   "please handle this",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Len(t, inbox, 1)
	require.Equal(t, "msg-busy", inbox[0].MessageID)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-busy"}))
	_, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.False(t, ok)
}

func TestDeliverMessageDoesNotCaptureRecoveryPointWhenCodexAppServerDeliveryFails(t *testing.T) {
	driver := &protocolRecoveryStubDriver{
		deliverErr:  errors.New("cannot accept direct delivery"),
		recoveryRef: "stale-ref",
	}
	var prepared []types.RecoveryPointPreparedEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.SetOnRecoveryPoint(func(ev types.RecoveryPointPreparedEvent) {
		prepared = append(prepared, ev)
	})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-1",
		TargetKey: "scp_test",
		TargetSeq: 1,
		Content:   "please handle this",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Equal(t, 1, driver.deliverCalls)
	require.Equal(t, 0, driver.prepareCalls)
	require.Empty(t, prepared)
	require.Len(t, inbox, 1)
	require.Equal(t, "msg-1", inbox[0].MessageID)
	_, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.False(t, ok)
}

func TestIdleProtocolDeliveryFailureRestoresPreviousAuthorityScopes(t *testing.T) {
	driver := &protocolRecoveryStubDriver{
		deliverErr:  errors.New("definitive runtime rejection"),
		recoveryRef: "stale-ref",
	}
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning, isIdle: true},
	}
	ap.mu.Lock()
	ap.addSendTargetLocked("#old", sendScopeSourceConversation)
	ap.addTurnDeliveryLocked("#old")
	ap.mu.Unlock()

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-rejected",
		TargetKey: "scp_test",
		TargetSeq: 2,
		Target:    "#new:thread",
		Content:   "definitively rejected replacement",
	}, false)

	scope, live := ap.SendScope()
	require.True(t, live)
	require.True(t, scope.ContainsTarget("#old"),
		"a deterministic idle replacement failure must restore the previous send scope")
	require.False(t, scope.ContainsTarget("#new:thread"))
	require.Equal(t, "#old", ap.CurrentTurnDeliveryTarget(),
		"a deterministic idle replacement failure must restore the previous turn target")
}

func TestDeliverMessage_ProtocolIdleDeliveryEmitsPromptInput(t *testing.T) {
	driver := &protocolRecoveryStubDriver{
		recoveryRef:  "stale-ref",
		activeTurnID: "turn-42",
	}
	var parsedEvents []drivercontract.ParsedEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-1",
		Target:    "#ops",
		Content:   "please handle this",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
	require.NotContains(t, driver.lastText, "thread=thread-1")
	require.NotContains(t, driver.lastText, "providerTurn=turn-42")
	require.Len(t, parsedEvents, 1)
	require.Equal(t, consts.ParsedEventKindPromptInput, parsedEvents[0].Kind)
	require.Equal(t, "delivery_text", parsedEvents[0].PromptKind)
	require.Equal(t, "codex", parsedEvents[0].Runtime)
	require.Equal(t, "routed_message_delivery", parsedEvents[0].EventType)
	require.Equal(t, "thread-1", parsedEvents[0].SessionID)
	require.NotContains(t, parsedEvents[0].Text, "thread=thread-1")
	require.NotContains(t, parsedEvents[0].Text, "providerTurn=turn-42")
}

func TestDeliverMessage_ProtocolActiveTurnQueuesWithoutDirectDelivery(t *testing.T) {
	driver := &protocolRecoveryStubDriver{
		recoveryRef:  "stale-ref",
		activeTurnID: "turn-42",
	}
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: false}}

	ap.RememberObservedFrontier(types.InboxMessage{
		MessageID: "msg-1",
		TargetKey: "scp_test",
		TargetSeq: 1,
	})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-2",
		Target:    "#test",
		TargetKey: "scp_test",
		TargetSeq: 2,
		Content:   "new message during active turn",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Equal(t, 0, driver.deliverCalls)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	require.Len(t, inbox, 1)
	require.Equal(t, "msg-2", inbox[0].MessageID)
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestInboxMessagesForFreshnessReadableAtCheckpoint(t *testing.T) {
	driver := &ackedTurnCountRecoveryStubDriver{ackedRuntimeInputStubDriver: ackedRuntimeInputStubDriver{supports: true}}
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: false},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{
			{MessageID: "msg-2", Target: "#test", TargetKey: "scp_test", TargetSeq: 2, Content: "active queued"},
			{MessageID: "msg-3", Target: "#other", TargetKey: "scp_other", TargetSeq: 1, Content: "other queued"},
		}}}

	first := ap.InboxMessagesForFreshnessPage("scp_test", 2, 1)
	require.Len(t, first.Messages, 1)
	require.Equal(t, "msg-2", first.Messages[0].MessageID)
	require.False(t, first.HasMore)

	blocked := ap.InboxMessagesForFreshnessPage("scp_other", 1, 1)
	require.Empty(t, blocked.Messages, "a later target must not overtake the outstanding final-hop owner")
	require.True(t, blocked.HasMore)
	require.Equal(t, 1, blocked.RemainingCount)
	require.Len(t, first.RuntimeDeliveryClaims, 1)
	first.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{first.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered)

	fresh := ap.InboxMessagesForFreshnessPage("scp_other", 1, 1)
	require.Len(t, fresh.Messages, 1)
	require.Equal(t, "msg-3", fresh.Messages[0].MessageID)
	ap.mu.Lock()
	require.Empty(t, ap.inbox)
	ap.mu.Unlock()
	fresh.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{fresh.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered)
}

func TestDeliverMessage_ProtocolIdleDeliveryAdvancesFreshnessObserved(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "stale-ref"}
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-2",
		Target:    "#test",
		TargetKey: "scp_test",
		TargetSeq: 2,
		Content:   "new idle-turn message",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
	require.Equal(t, []string{"msg-1", "msg-2"}, frontier.SourceMessageIDs)
}

func TestDeliverMessage_ProtocolDeliveryMarksContiguousFreshnessBeforeRuntimeCanSend(t *testing.T) {
	var ap *Process
	driver := &protocolRecoveryStubDriver{recoveryRef: "stale-ref"}
	driver.onDeliver = func() {
		frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
		require.True(t, ok)
		require.Equal(t, int64(5), frontier.ObservedSeq)
		require.Equal(t, int64(5), frontier.LatestKnownSeq)
		require.Zero(t, frontier.NeedsRefreshUntilSeq)
	}
	ap = &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-4", TargetKey: "scp_test", TargetSeq: 4})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-5",
		Target:    "dm:@owner",
		TargetKey: "scp_test",
		TargetSeq: 5,
		Content:   "message that may trigger an immediate message_send call",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
}

func TestDeliverMessage_ProtocolDeliveryWithTargetSeqGapCoversCurrentMessage(t *testing.T) {
	var ap *Process
	driver := &protocolRecoveryStubDriver{recoveryRef: "stale-ref"}
	driver.onDeliver = func() {
		frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
		require.True(t, ok)
		require.Equal(t, int64(6), frontier.ObservedSeq)
		require.Equal(t, int64(6), frontier.LatestKnownSeq)
		require.Zero(t, frontier.NeedsRefreshUntilSeq)
	}
	ap = &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-2", TargetKey: "scp_test", TargetSeq: 2})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-6",
		Target:    "dm:@owner",
		TargetKey: "scp_test",
		TargetSeq: 6,
		Content:   "message after unseen intervening target messages",
	}, false)

	require.Equal(t, 1, driver.deliverCalls)
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(6), frontier.ObservedSeq)
	require.Equal(t, int64(6), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessage_StaleDirectDeliveryIsSuppressed(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "stale-ref"}
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-43", TargetKey: "scp_thread", TargetSeq: 43})

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-36",
		Target:    "#test:thread",
		TargetKey: "scp_thread",
		TargetSeq: 36,
		Content:   "stale replay",
	}, false)

	ap.mu.Lock()
	inboxLen := len(ap.inbox)
	isIdle := ap.isIdle
	ap.mu.Unlock()

	require.Zero(t, driver.deliverCalls)
	require.Zero(t, inboxLen)
	require.True(t, isIdle)
	require.Equal(t, []string{"msg-36"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-36"}),
		"freshness suppression is authoritative acked evidence and keeps a terminal tombstone until reconciliation")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_thread")
	require.True(t, ok)
	require.Equal(t, int64(43), frontier.ObservedSeq)
}

func TestInboxMessagesForFreshnessSuppressesStaleEntriesBeforeReturningToRuntime(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-codex",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{
			{MessageID: "msg-36", Target: "#test:thread", TargetKey: "scp_thread", TargetSeq: 36, Content: "stale"},
			{MessageID: "msg-44", Target: "#test:thread", TargetKey: "scp_thread", TargetSeq: 44, Content: "fresh"},
		}}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-43", TargetKey: "scp_thread", TargetSeq: 43})

	page := ap.InboxMessagesForFreshnessPage("scp_thread", 44, 50)

	require.Len(t, page.Messages, 1)
	require.Equal(t, "msg-44", page.Messages[0].MessageID)
	ap.mu.Lock()
	inboxLen := len(ap.inbox)
	ap.mu.Unlock()
	require.Zero(t, inboxLen)
	page.RuntimeDeliveryCommitOwner.AcknowledgeMCPToolResultDelivery([]string{page.RuntimeDeliveryClaims[0].Receipt}, types.MCPToolDeliveryDelivered)
}

func TestHandleParsedEventTurnEndSkipsStaleQueuedDelivery(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-busy",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{},
			stdinPipe: stdin},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{
			{
				MessageID:  "msg-36",
				Target:     "#agentteam:00000037",
				TargetKey:  "scp_thread",
				TargetSeq:  36,
				SenderName: "alice",
				Content:    "stale replay",
				Timestamp:  "2026-04-12T18:32:55+08:00",
			},
			{
				MessageID:  "msg-44",
				Target:     "#agentteam:00000037",
				TargetKey:  "scp_thread",
				TargetSeq:  44,
				SenderName: "alice",
				Content:    "fresh follow-up",
				Timestamp:  "2026-04-12T18:33:55+08:00",
			},
		}}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-43", TargetKey: "scp_thread", TargetSeq: 43})

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	ap.mu.Lock()
	inboxLen := len(ap.inbox)
	isIdle := ap.isIdle
	ap.mu.Unlock()

	require.Zero(t, inboxLen)
	require.False(t, isIdle)
	require.NotContains(t, stdin.String(), "stale replay")
	require.Contains(t, stdin.String(), "fresh follow-up")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_thread")
	require.True(t, ok)
	require.Equal(t, int64(44), frontier.ObservedSeq)
}

func TestHandleParsedEventTurnEndEmitsWakeCompletedWithoutVisibleReply(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "fresh-ref"}
	var wakeEvents []WakeStatusEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-no-reply",
		Target:     "#agentteam:00000014",
		SenderName: "owner",
		Content:    "please handle this",
	}, false)
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.Len(t, wakeEvents, 2)
	require.Equal(t, consts.WakeStatusReceived, wakeEvents[0].Status)
	require.Equal(t, "msg-no-reply", wakeEvents[0].MessageID)
	require.Equal(t, "#agentteam:00000014", wakeEvents[0].Target)
	require.Equal(t, consts.WakeStatusCompletedWithoutVisibleReply, wakeEvents[1].Status)
	require.Equal(t, wakeEvents[0].WakeToken, wakeEvents[1].WakeToken)
	require.Equal(t, "msg-no-reply", wakeEvents[1].MessageID)
	require.Equal(t, "reason=no_reply", wakeEvents[1].Detail)
}

func TestHandleParsedEventTurnEndInjectsFinalDeliveryCheckOnce(t *testing.T) {
	stdin := &recordingWriteCloser{}
	var wakeEvents []WakeStatusEvent
	var parsedEvents []drivercontract.ParsedEvent

	ap := &Process{
		AgentID:   "agent-visible-contract",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{isIdle: true},
		processWakeState:      processWakeState{}}

	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}
	ap.onEvent = func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-final-check",
		Target:    "#agentteam:00000014",
		Content:   "please reply visibly",
	}, false)
	wakeToken := ap.PrepareVisibleReply()
	require.NotEmpty(t, wakeToken)

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.Len(t, wakeEvents, 1)
	require.Equal(t, consts.WakeStatusReceived, wakeEvents[0].Status)
	require.Equal(t, wakeToken, ap.PrepareVisibleReply())
	require.Contains(t, stdin.String(), "Final delivery check")
	require.Contains(t, stdin.String(), `exact target "#agentteam:00000014"`)
	require.NotContains(t, stdin.String(), "routing_kind")
	var finalCheckEvents []drivercontract.ParsedEvent
	for _, ev := range parsedEvents {
		if ev.Kind == consts.ParsedEventKindPromptInput && ev.PromptKind == consts.PromptKindFinalVisibleReplyCheck {
			finalCheckEvents = append(finalCheckEvents, ev)
		}
	}
	require.Len(t, finalCheckEvents, 1)

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.Len(t, wakeEvents, 2)
	require.Equal(t, consts.WakeStatusCompletedWithoutVisibleReply, wakeEvents[1].Status)
	require.Equal(t, wakeToken, wakeEvents[1].WakeToken)
	require.Contains(t, wakeEvents[1].Detail, "reason=no_reply")
	require.Equal(t, 1, strings.Count(stdin.String(), "Final delivery check"))
	require.Empty(t, ap.PrepareVisibleReply())

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})
	require.Len(t, wakeEvents, 2)
}

func TestHandleParsedEventTurnEndFinalDeliveryCheckUsesStructuredDelivery(t *testing.T) {
	driver := &runtimeInputDeliveryStubDriver{}
	var parsedEvents []drivercontract.ParsedEvent
	ap := &Process{
		AgentID:   "agent-visible-contract-structured",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{
			driver: driver,
		},
		processLifecycleState: processLifecycleState{isIdle: true},
		processWakeState: processWakeState{
			pendingWake:      &pendingDelivery{Wake: WakeInfo{MessageID: "msg-final-check-structured", Target: "#agentteam:00000014", DeliveryClass: DeliveryClassVisibleReply}, Status: pendingDeliveryAwaitingCheck},
			pendingWakeToken: "wake-final-check-structured",
		},
	}
	ap.onEvent = func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	}

	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.NotEmpty(t, driver.lastInput.ID)
	require.Contains(t, driver.lastInput.PromptText(), "Final delivery check")
	require.Contains(t, driver.lastInput.PromptText(), `exact target "#agentteam:00000014"`)
	require.NotContains(t, driver.lastInput.PromptText(), "routing_kind")
	require.Equal(t, "sess-1", driver.lastSession)
	var finalCheckEvents []drivercontract.ParsedEvent
	for _, ev := range parsedEvents {
		if ev.Kind == consts.ParsedEventKindPromptInput && ev.PromptKind == consts.PromptKindFinalVisibleReplyCheck {
			finalCheckEvents = append(finalCheckEvents, ev)
		}
	}
	require.Len(t, finalCheckEvents, 1)
}

func TestConfirmReactionReplyCompletesWakeWithoutNoReplyStatus(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "fresh-ref"}
	var wakeEvents []WakeStatusEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-reacted",
		Target:    "#agentteam:00000014",
		Content:   "please acknowledge",
	}, false)
	wakeToken := ap.PrepareVisibleReply()
	require.NotEmpty(t, wakeToken)

	require.True(t, ap.ConfirmReactionReply(wakeToken, "msg-reacted"))
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.Len(t, wakeEvents, 2)
	require.Equal(t, consts.WakeStatusReceived, wakeEvents[0].Status)
	require.Equal(t, consts.WakeStatusReplySent, wakeEvents[1].Status)
	require.Equal(t, wakeToken, wakeEvents[1].WakeToken)
	require.Equal(t, "msg-reacted", wakeEvents[1].MessageID)
	require.Empty(t, wakeEvents[1].ReplyMessageID)
}

func TestConfirmReactionReplyRejectsUnrelatedMessage(t *testing.T) {
	driver := &protocolRecoveryStubDriver{recoveryRef: "fresh-ref"}
	var wakeEvents []WakeStatusEvent

	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.onWakeStatus = func(ev WakeStatusEvent) {
		wakeEvents = append(wakeEvents, ev)
	}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-trigger",
		Target:    "#agentteam:00000014",
		Content:   "please acknowledge",
	}, false)
	wakeToken := ap.PrepareVisibleReply()
	require.NotEmpty(t, wakeToken)

	require.False(t, ap.ConfirmReactionReply(wakeToken, "msg-other"))
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	require.Len(t, wakeEvents, 2)
	require.Equal(t, consts.WakeStatusCompletedWithoutVisibleReply, wakeEvents[1].Status)
	require.Equal(t, "msg-trigger", wakeEvents[1].MessageID)
}

func TestReconcileInboxMessagesRemovesMatchingInboxEntries(t *testing.T) {
	ap := &Process{processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{
		{MessageID: "msg-1", Content: "keep"},
		{MessageID: "msg-2", Content: "drop"},
	}}}

	removed := ap.ReconcileInboxMessages([]string{"msg-2", "msg-3"})

	ap.mu.Lock()
	remaining := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Equal(t, 1, removed)
	require.Len(t, remaining, 1)
	require.Equal(t, "msg-1", remaining[0].MessageID)
}

func TestHandleParsedEventTurnEndSkipsAlreadyReconciledQueuedMessage(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-busy",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{},
			stdinPipe: stdin},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID:  "msg-1",
			Target:     "#agentteam:00000037",
			SenderName: "7q8vvi2",
			Content:    "already consumed via receive",
			Timestamp:  "2026-04-12T18:32:55+08:00",
		}}}}

	removed := ap.ReconcileInboxMessages([]string{"msg-1"})
	ap.handleParsedEvent(drivercontract.ParsedEvent{Kind: consts.ParsedEventKindTurnEnd})

	ap.mu.Lock()
	inboxLen := len(ap.inbox)
	isIdle := ap.isIdle
	ap.mu.Unlock()

	require.Equal(t, 1, removed)
	require.Equal(t, 0, inboxLen)
	require.True(t, isIdle)
	require.Empty(t, stdin.String())
}

func TestDeliverMessageIdleStdinMarksMessageAsAcked(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-idle",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{isIdle: true}}

	ap.RememberObservedFrontier(types.InboxMessage{MessageID: "msg-1", TargetKey: "scp_test", TargetSeq: 1})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-stdin",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:00000037",
		SenderName: "7q8vvi2",
		Content:    "please handle this",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	require.NotContains(t, stdin.String(), "mode=stdin_delivery")
	require.Contains(t, stdin.String(), "target=#agentteam:00000037")
	require.Equal(t, []string{"msg-stdin"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-stdin"}))
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageBusyStdinDuringFirstTurnQueues(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-busy",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyStdinProcessStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{isIdle: false}}

	ap.deliverMessage(types.InboxMessage{
		MessageID: "msg-first-turn",
		Content:   "queue until the runtime finishes its first turn",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Empty(t, stdin.String())
	require.Len(t, inbox, 1)
	require.Equal(t, "msg-first-turn", inbox[0].MessageID)
}

func TestSessionInitDrainsQueuedInboxWhenProcessStartsIdle(t *testing.T) {
	driver := &protocolRecoveryStubDriver{}
	ap := &Process{
		AgentID:               "agent-idle",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: true},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-queued",
			Content:   "deliver this queued message",
			Timestamp: time.Now().UTC().Format(time.RFC3339),
		}}}}

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

	require.Equal(t, "thread-1", ap.GetSessionID())
	require.Equal(t, 1, driver.deliverCalls)
	require.Contains(t, driver.lastText, "deliver this queued message")
	ap.mu.Lock()
	defer ap.mu.Unlock()
	require.Empty(t, ap.inbox)
}

func TestSessionInitEmitsBeforeQueuedDeliveryEvents(t *testing.T) {
	driver := &protocolRecoveryStubDriver{}
	ap := &Process{
		AgentID:               "agent-idle",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: true},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-queued",
			Content:   "deliver this queued message",
			Timestamp: time.Now().UTC().Format(time.RFC3339),
		}}}}

	var events []string
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		label := string(ev.Kind)
		if ev.Kind == consts.ParsedEventKindPromptInput {
			label += ":" + ev.PromptKind
		}
		events = append(events, label)
	})

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

	require.GreaterOrEqual(t, len(events), 2)
	require.Equal(t, consts.ParsedEventKindSessionInit, events[0])
	require.Equal(t, string(consts.ParsedEventKindPromptInput)+":"+consts.PromptKindDeliveryText, events[1])
}

func TestSessionInitQueuesCallbackDeliveryUntilAfterSessionInitEvent(t *testing.T) {
	driver := &protocolRecoveryStubDriver{}
	ap := &Process{
		AgentID:               "agent-idle",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: true}}

	var events []string
	delivered := false
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		if ev.Kind == consts.ParsedEventKindSessionInit && !delivered {
			delivered = true
			ap.Deliver(types.InboxMessage{
				MessageID: "msg-during-session-init",
				Content:   "deliver from callback",
				Timestamp: time.Now().UTC().Format(time.RFC3339),
			})
		}
		label := string(ev.Kind)
		if ev.Kind == consts.ParsedEventKindPromptInput {
			label += ":" + ev.PromptKind
		}
		events = append(events, label)
	})

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

	require.True(t, delivered)
	require.GreaterOrEqual(t, len(events), 2)
	require.Equal(t, consts.ParsedEventKindSessionInit, events[0])
	require.Equal(t, string(consts.ParsedEventKindPromptInput)+":"+consts.PromptKindDeliveryText, events[1])
}

func TestSessionInitKeepsQueuedInboxDaemonOwnedDuringInitialTurn(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID: "agent-busy",
		processRuntimeBinding: processRuntimeBinding{driver: &busyStdinProcessStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-queued",
			Content:   "queued before session init",
		}}}}

	t.Cleanup(func() {
		ap.mu.Lock()
		ap.stopInboxRetryLocked()
		ap.mu.Unlock()
	})

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

	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "msg-queued", ap.inbox[0].MessageID)
	ap.mu.Unlock()
	require.Empty(t, stdin.String(), "daemon retry must not inject a generic queue notification")
}

func TestSessionInitCallbackDeliveryQueuesWithoutGenericNotification(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-busy",
		processRuntimeBinding: processRuntimeBinding{driver: &busyStdinProcessStubDriver{}},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false}}

	t.Cleanup(func() {
		ap.mu.Lock()
		ap.stopInboxRetryLocked()
		ap.mu.Unlock()
	})
	delivered := false
	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		if ev.Kind == consts.ParsedEventKindSessionInit && !delivered {
			delivered = true
			ap.Deliver(types.InboxMessage{
				MessageID: "msg-during-session-init",
				Content:   "deliver from callback",
			})
		}
	})

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

	require.True(t, delivered)
	ap.mu.Lock()
	require.Len(t, ap.inbox, 1)
	require.Equal(t, "msg-during-session-init", ap.inbox[0].MessageID)
	ap.mu.Unlock()
}

func TestSessionInitKeepsInboxWhenRuntimeDoesNotSupportStdinDelivery(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID: "agent-busy",
		processRuntimeBinding: processRuntimeBinding{driver: &noStdinNotificationStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false},
		processDeliveryState: processDeliveryState{inbox: []types.InboxMessage{{
			MessageID: "msg-queued",
			Content:   "queued before session init",
		}}}}

	t.Cleanup(func() {
		ap.mu.Lock()
		ap.stopInboxRetryLocked()
		ap.mu.Unlock()
	})

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

	require.Never(t, func() bool {
		return stdin.String() != ""
	}, 50*time.Millisecond, 5*time.Millisecond)
}

func TestDeliverMessageBusyStdinAfterFirstTurnQueuesWithoutGenericNotification(t *testing.T) {
	stdin := &recordingWriteCloser{}
	ap := &Process{
		AgentID:   "agent-busy",
		SessionID: "sess-1",
		processRuntimeBinding: processRuntimeBinding{driver: &busyStdinProcessStubDriver{},
			stdinPipe: stdin},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false}}
	t.Cleanup(func() {
		ap.mu.Lock()
		ap.stopInboxRetryLocked()
		ap.mu.Unlock()
	})

	ap.RememberObservedFrontier(types.InboxMessage{
		MessageID: "msg-1",
		TargetKey: "scp_test",
		TargetSeq: 1,
	})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-busy",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:00000037",
		SenderName: "7q8vvi2",
		Content:    "deliver during an active turn",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	ap.mu.Unlock()

	require.Len(t, inbox, 1)
	require.Equal(t, "msg-busy", inbox[0].MessageID)
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-busy"}))
	require.Empty(t, stdin.String(), "daemon retry must not replace the original body with a generic notification")
	require.NotContains(t, stdin.String(), "mode=stdin_delivery")
	require.NotContains(t, stdin.String(), "deliver during an active turn")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)
}

func TestDeliverMessageCodexActiveTurnUsesAckedSteer(t *testing.T) {
	driver, writer := newTestCodexDriver(t, func(req capturedRPCRequest) appServerEnvelope {
		require.Equal(t, "turn/steer", req.Method)
		require.Equal(t, "thread-1", req.Params["threadId"])
		require.Equal(t, "turn-42", req.Params["expectedTurnId"])
		require.Equal(t, "msg-busy", req.Params["clientUserMessageId"])
		return appServerEnvelope{
			Result: rawRPCResult(t, appServerTurnIDResult{TurnID: "turn-42"}),
		}
	})
	driver.threadID = "thread-1"
	driver.threadStatus = appServerThreadStatus{Type: "active"}
	driver.activeTurnID = "turn-42"

	var parsedEvents []drivercontract.ParsedEvent
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning,
			isIdle: false, exited: make(chan struct{})},
		processDeliveryState: processDeliveryState{
			deliveryOwnershipStore:       &memoryDeliveryOwnershipStore{},
			pendingReconcileSuccessGrace: 5 * time.Millisecond,
		}}
	t.Cleanup(func() { stopInboxRetryTestProcess(ap) })

	ap.SetOnEvent(func(ev drivercontract.ParsedEvent) {
		parsedEvents = append(parsedEvents, ev)
	})
	ap.RememberObservedFrontier(types.InboxMessage{
		MessageID: "msg-1",
		TargetKey: "scp_test",
		TargetSeq: 1,
	})

	ap.deliverMessage(types.InboxMessage{
		MessageID:  "msg-busy",
		TargetKey:  "scp_test",
		TargetSeq:  2,
		Target:     "#agentteam:00000037",
		SenderName: "7q8vvi2",
		Content:    "deliver during an active turn",
		Timestamp:  "2026-04-12T18:32:55+08:00",
	}, false)

	require.Len(t, writer.requests, 1)
	req := writer.requests[0]
	require.Equal(t, "turn/steer", req.Method)
	input, ok := req.Params["input"].([]any)
	require.True(t, ok)
	require.Len(t, input, 1)
	item, ok := input[0].(map[string]any)
	require.True(t, ok)
	inputText, ok := item["text"].(string)
	require.True(t, ok)
	require.NotContains(t, inputText, "mode=turn_direct")
	require.Contains(t, inputText, "deliver during an active turn")
	require.NotContains(t, inputText, "Call mcp__agentteam__queued_messages")

	ap.mu.Lock()
	inbox := append([]types.InboxMessage(nil), ap.inbox...)
	pending := ap.pendingRuntimeInputs["msg-busy"]
	pendingLen := len(ap.pendingRuntimeInputs)
	ap.mu.Unlock()
	require.Empty(t, inbox)
	require.Equal(t, 1, pendingLen)
	require.Equal(t, "msg-busy", pending.Input.ID)
	require.False(t, pending.reconcileRunning, "accepted Codex steer must remain runtime-owned until its correlated echo")
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-busy"}))
	require.Len(t, parsedEvents, 1)
	require.Equal(t, consts.ParsedEventKindRuntimeInputPending, parsedEvents[0].Kind)
	require.Equal(t, "msg-busy", parsedEvents[0].RuntimeInputID)
	require.Equal(t, "thread-1", parsedEvents[0].SessionID)
	require.Equal(t, consts.PromptKindDeliveryText, parsedEvents[0].PromptKind)
	require.Contains(t, parsedEvents[0].Text, "deliver during an active turn")
	frontier, ok := ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(1), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Equal(t, int64(2), frontier.NeedsRefreshUntilSeq)

	ap.handleParsedEvent(drivercontract.ParsedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: "msg-busy",
		SessionID:      "thread-1",
	})

	require.Len(t, parsedEvents, 2)
	require.Equal(t, consts.ParsedEventKindRuntimeInputAcked, parsedEvents[1].Kind)
	require.Equal(t, "msg-busy", parsedEvents[1].RuntimeInputID)
	require.Equal(t, runtimeInputAckedDetail, parsedEvents[1].Message)
	require.Empty(t, parsedEvents[1].Text)
	require.Equal(t, []string{"msg-busy"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-busy"}))
	frontier, ok = ap.MessageFreshnessSnapshot("scp_test")
	require.True(t, ok)
	require.Equal(t, int64(2), frontier.ObservedSeq)
	require.Equal(t, int64(2), frontier.LatestKnownSeq)
	require.Zero(t, frontier.NeedsRefreshUntilSeq)
}

func TestReconcileAckedMessagesRemovesMatchingEntries(t *testing.T) {
	ap := &Process{}
	ap.RememberRuntimeAckedMessage("msg-1")
	ap.RememberRuntimeAckedMessage("msg-2")

	require.Equal(t, []string{"msg-2"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-2", "msg-3"}))
	require.Nil(t, ap.ReconcileRuntimeAckedMessages([]string{"msg-2"}))
	require.Equal(t, []string{"msg-1"}, ap.ReconcileRuntimeAckedMessages([]string{"msg-1"}))
}

func TestUsageMetricDimensionsFallbackToCodexAgentSource(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-codex",
		processRuntimeBinding: processRuntimeBinding{driver: &CodexAppServerDriver{}},
		processUsageState:     processUsageState{model: "gpt-5.2"}}

	apiKey, model, ok := ap.usageMetricDimensions(drivercontract.ParsedEvent{})
	require.True(t, ok)
	require.Equal(t, "codex:agent-codex", apiKey)
	require.Equal(t, "gpt-5.2", model)
}

func TestUsageMetricDimensionsDoesNotAttributeUnknownForeignThreadToPrimaryModel(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-codex",
		SessionID:             "thread-primary",
		processRuntimeBinding: processRuntimeBinding{driver: &CodexAppServerDriver{}},
		processUsageState:     processUsageState{model: "gpt-parent"},
	}

	_, _, ok := ap.usageMetricDimensions(drivercontract.ParsedEvent{
		SessionID: "thread-child",
	})
	require.False(t, ok)

	apiKey, model, ok := ap.usageMetricDimensions(drivercontract.ParsedEvent{
		SessionID: "thread-primary",
	})
	require.True(t, ok)
	require.Equal(t, "codex:agent-codex", apiKey)
	require.Equal(t, "gpt-parent", model)

	apiKey, model, ok = ap.usageMetricDimensions(drivercontract.ParsedEvent{
		SessionID: "thread-child",
		Model:     "gpt-child",
	})
	require.True(t, ok)
	require.Equal(t, "codex:agent-codex", apiKey)
	require.Equal(t, "gpt-child", model)
}

func TestUsageMetricDimensionsRequiresSourceForNonCodexRuntime(t *testing.T) {
	ap := &Process{
		AgentID:               "agent-claude",
		processRuntimeBinding: processRuntimeBinding{driver: &busyDeliveryStubDriver{}},
		processUsageState:     processUsageState{model: "claude-sonnet"}}

	_, _, ok := ap.usageMetricDimensions(drivercontract.ParsedEvent{})
	require.False(t, ok)
}

// TestProcessInterruptAndDeliverCommitsTurnDeliveryTarget locks the
// uncorrelated-interrupt production path: a successful interrupt must end the
// replace handover and expose the interrupting message's target in the same
// critical section that commits the send scope, so first-turn task-create
// defaults work immediately after the turn switch.
func TestProcessInterruptAndDeliverCommitsTurnDeliveryTarget(t *testing.T) {
	driver := &interruptingRuntimeInputStubDriver{}
	ap := &Process{
		AgentID:               "agent-interrupt",
		SessionID:             "thread-1",
		processRuntimeBinding: processRuntimeBinding{driver: driver},
		processLifecycleState: processLifecycleState{state: processStateRunning},
		processRecoveryState:  processRecoveryState{workDir: t.TempDir()}}

	handled, err := ap.InterruptAndDeliver(context.Background(), types.InboxMessage{
		MessageID:   "msg-interrupt",
		TargetKey:   "scp_thread",
		TargetSeq:   7,
		Target:      "#agentteam:abc123ef",
		ChannelType: "thread",
		ChannelName: "thread-abc123ef",
		SenderType:  "user",
		SenderName:  "owner",
		Content:     "new authoritative input",
		Timestamp:   time.Now().UTC().Format(time.RFC3339),
	})

	require.NoError(t, err)
	require.True(t, handled)
	require.Equal(t, "#agentteam:abc123ef", ap.CurrentTurnDeliveryTarget(),
		"uncorrelated interrupt success must commit the new turn's target")
}