package dispatcher

import (
	"context"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"reflect"
	"sync"
	"testing"
	"time"
	"unsafe"

	"github.com/stretchr/testify/require"

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

// twoPhaseOwnershipDriver is shared by dispatcher launch and signal tests.
type twoPhaseOwnershipDriver struct {
	mu        sync.Mutex
	lastInput drivercontract.RuntimeInput
	calls     int
}

// replacementOwnershipDriver exercises the real Process ownership restore and
// reconciliation state machine without introducing a second dispatcher seam.
// Spawn is deliberately no-prompt; only the durable queued successor may reach
// DeliverRuntimeInput after the restored owner is terminalized.
type replacementOwnershipDriver struct {
	mu             sync.Mutex
	spawnCfg       drivercontract.LaunchConfig
	delivered      []drivercontract.RuntimeInput
	reconcileCalls map[string]int
	firstResult    agent.RuntimeInputReconciliation
	firstErr       error
	secondResult   agent.RuntimeInputReconciliation
	secondErr      error
	firstSeen      chan struct{}
	secondEntered  chan struct{}
	releaseSecond  chan struct{}
	firstOnce      sync.Once
	secondOnce     sync.Once
	releaseOnce    sync.Once
	stopOnce       sync.Once
	done           chan struct{}
	stdout         *io.PipeReader
	stdoutWriter   *io.PipeWriter
}

func newReplacementOwnershipDriver(first agent.RuntimeInputReconciliation, firstErr error, second agent.RuntimeInputReconciliation, secondErr error) *replacementOwnershipDriver {
	reader, writer := io.Pipe()
	return &replacementOwnershipDriver{
		reconcileCalls: make(map[string]int),
		firstResult:    first,
		firstErr:       firstErr,
		secondResult:   second,
		secondErr:      secondErr,
		firstSeen:      make(chan struct{}),
		secondEntered:  make(chan struct{}),
		releaseSecond:  make(chan struct{}),
		done:           make(chan struct{}),
		stdout:         reader,
		stdoutWriter:   writer,
	}
}

func (d *replacementOwnershipDriver) Name() string                    { return "codex" }
func (d *replacementOwnershipDriver) SupportsStdinNotification() bool { return true }
func (d *replacementOwnershipDriver) MCPToolPrefix() string           { return "mcp_agentteam_" }
func (d *replacementOwnershipDriver) DeliverBusyStdinInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	return d.DeliverRuntimeInput(ctx, input, sessionID)
}
func (d *replacementOwnershipDriver) Spawn(_ context.Context, cfg drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	d.mu.Lock()
	d.spawnCfg = cfg
	d.mu.Unlock()
	cleanup := func() {
		d.stopOnce.Do(func() {
			close(d.done)
			_ = d.stdoutWriter.Close()
		})
	}
	return drivercontract.SpawnResult{
		Stdin:          noopWriteCloser{},
		Stdout:         d.stdout,
		SessionID:      "thread-active",
		SessionResumed: true,
		Wait:           func() error { <-d.done; return nil },
		Terminate:      cleanup,
		Cleanup:        cleanup,
	}, nil
}
func (*replacementOwnershipDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }
func (*replacementOwnershipDriver) EncodeStdinMessage(text, _ string) []byte      { return []byte(text) }
func (d *replacementOwnershipDriver) DeliverRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.mu.Lock()
	d.delivered = append(d.delivered, input)
	d.mu.Unlock()
	return nil
}
func (d *replacementOwnershipDriver) ReconcileRuntimeInput(_ context.Context, inputID, _ string) (agent.RuntimeInputReconciliation, error) {
	d.mu.Lock()
	d.reconcileCalls[inputID]++
	call := d.reconcileCalls[inputID]
	d.mu.Unlock()
	if inputID == "active-inflight" {
		if call == 1 {
			d.firstOnce.Do(func() { close(d.firstSeen) })
			return d.firstResult, d.firstErr
		}
		if call == 2 {
			d.secondOnce.Do(func() { close(d.secondEntered) })
			<-d.releaseSecond
			return d.secondResult, d.secondErr
		}
	}
	if inputID == "queued-successor" {
		return agent.RuntimeInputReconciliationAcked, nil
	}
	return agent.RuntimeInputReconciliationUnknown, nil
}

func (d *twoPhaseOwnershipDriver) Name() string                    { return "codex" }
func (d *twoPhaseOwnershipDriver) SupportsStdinNotification() bool { return true }
func (d *twoPhaseOwnershipDriver) MCPToolPrefix() string           { return "mcp_agentteam_" }
func (d *twoPhaseOwnershipDriver) Spawn(context.Context, drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	return drivercontract.SpawnResult{}, nil
}
func (d *twoPhaseOwnershipDriver) ParseLine(string) []drivercontract.ParsedEvent { return nil }
func (d *twoPhaseOwnershipDriver) EncodeStdinMessage(text, _ string) []byte      { return []byte(text) }
func (d *twoPhaseOwnershipDriver) DeliverBusyStdinInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	return d.DeliverRuntimeInput(context.Background(), input, "")
}
func (d *twoPhaseOwnershipDriver) DeliverRuntimeInput(_ context.Context, input drivercontract.RuntimeInput, _ string) error {
	d.mu.Lock()
	d.lastInput = input
	d.calls++
	d.mu.Unlock()
	return nil
}
func (d *twoPhaseOwnershipDriver) ReconcileRuntimeInput(context.Context, string, string) (agent.RuntimeInputReconciliation, error) {
	return agent.RuntimeInputReconciliationPending, nil
}

func setUnexportedValueForOwnershipTest(t *testing.T, field reflect.Value, value any) {
	t.Helper()
	require.True(t, field.CanAddr())
	writable := reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem()
	writable.Set(reflect.ValueOf(value))
}

func TestPrepareRuntimeDeliveryOwnershipLaunchPersistsCurrentWake(t *testing.T) {
	workDir := t.TempDir()
	msg := types.InboxMessage{MessageID: "msg-current", Target: "#ops", Content: "current thread mention"}

	restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, msg, true)

	require.NoError(t, err)
	require.Equal(t, msg, restored.Message)
	require.True(t, restored.HandoffPersisted)
	require.False(t, restored.WaitForReconcile)
	require.Equal(t, []string{"msg-current"}, restored.NewlyPersistedIDs)
	records, err := newSessionDeliveryOwnershipStore(workDir).LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, 1)
	require.True(t, records[0].IsQueued())
}

func TestCurrentThreadWakeIsNeverReplacedByHistoricalInflightOwner(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	acceptedAt := time.Now().UTC().Add(-24 * time.Hour)
	_, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:        types.InboxMessage{MessageID: "old-unknown", Target: "#old", Content: "historical"},
		Input:          drivercontract.RuntimeInput{ID: "old-unknown"},
		SessionID:      "old-thread",
		State:          agent.DeliveryOwnershipInflight,
		InflightReason: agent.DeliveryInflightOutcomeUnknown,
		AcceptedAt:     acceptedAt,
	})
	require.NoError(t, err)
	current := types.InboxMessage{MessageID: "new-thread-mention", Target: "#thread", Content: "@agent handle this now"}

	restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, current, true)

	require.NoError(t, err)
	require.Equal(t, current, restored.Message)
	require.True(t, restored.HandoffPersisted)
	require.False(t, restored.WaitForReconcile, "historical uncertainty must not create an empty thread_start")
	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, 2)
}

func TestUnsafeProviderResumeKeepsThreadButRestoredInflightOnlyReconciles(t *testing.T) {
	cases := []struct {
		name         string
		firstResult  agent.RuntimeInputReconciliation
		firstErr     error
		secondResult agent.RuntimeInputReconciliation
		secondErr    error
	}{
		{
			name:         "authoritative pending then failure",
			firstResult:  agent.RuntimeInputReconciliationPending,
			secondResult: agent.RuntimeInputReconciliationUnknown,
			secondErr:    errors.New("reconciliation budget exhausted"),
		},
		{
			name:         "unknown then unknown",
			firstResult:  agent.RuntimeInputReconciliationUnknown,
			secondResult: agent.RuntimeInputReconciliationUnknown,
		},
		{
			name:         "reconciliation error then failure",
			firstResult:  agent.RuntimeInputReconciliationUnknown,
			firstErr:     errors.New("read-side reconciliation failed"),
			secondResult: agent.RuntimeInputReconciliationUnknown,
			secondErr:    errors.New("read-side reconciliation failed again"),
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			workDir := t.TempDir()
			acceptedAt := time.Now().UTC().Add(-time.Minute)
			require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
				LastSessionID:     "thread-active",
				SessionResumeSafe: false,
				LastStartConfig: &types.AgentStartConfig{
					Runtime:   string(runtimeconfig.TypeCodex),
					SessionID: "thread-active",
				},
				RuntimeDeliveryOwnershipVersion: runtimeDeliveryOwnershipVersion,
				RuntimeDeliveryOwnership: []agent.DeliveryOwnershipRecord{
					{
						Message:        types.InboxMessage{MessageID: "active-inflight", Target: "#ops", Content: "old body must never replay"},
						Input:          drivercontract.RuntimeInput{ID: "active-inflight"},
						SessionID:      "thread-active",
						State:          agent.DeliveryOwnershipInflight,
						InflightReason: agent.DeliveryInflightOutcomeUnknown,
						AcceptedAt:     acceptedAt,
					},
					{
						Message:    types.InboxMessage{MessageID: "queued-successor", Target: "#ops", Content: "new queued body"},
						Input:      drivercontract.RuntimeInput{ID: "queued-successor"},
						SessionID:  "thread-active",
						State:      agent.DeliveryOwnershipQueued,
						AcceptedAt: acceptedAt.Add(time.Second),
					},
				},
			}))

			dispatcher := &Dispatcher{}
			// SessionResumeSafe=false is deliberately not a provider-session guard.
			// The replacement launch still attempts the exact persisted provider thread.
			require.Equal(t, "thread-active", dispatcher.validateSessionResume(
				workDir,
				"thread-active",
				string(runtimeconfig.TypeCodex),
				"http://127.0.0.1:44000",
				"",
			))

			driver := newReplacementOwnershipDriver(tc.firstResult, tc.firstErr, tc.secondResult, tc.secondErr)
			process, err := agent.NewProcess(context.Background(), driver, drivercontract.LaunchConfig{
				AgentID:     "replacement-agent",
				RuntimeType: runtimeconfig.TypeCodex,
				SessionID:   "thread-active",
				WorkDir:     workDir,
			}, agent.ProcessOptions{})
			require.NoError(t, err)
			t.Cleanup(func() {
				driver.releaseOnce.Do(func() { close(driver.releaseSecond) })
				process.Stop()
				require.True(t, process.WaitForExit(time.Second))
			})
			// Keep the real scheduler/reconciler on a short deterministic budget while
			// still exercising its durable progress and terminal-failure transitions.
			processValue := reflect.ValueOf(process).Elem()
			setUnexportedValueForOwnershipTest(t, processValue.FieldByName("pendingReconcileMaxAttempts"), 2)
			setUnexportedValueForOwnershipTest(t, processValue.FieldByName("pendingReconcileMaxElapsed"), time.Hour)
			setUnexportedValueForOwnershipTest(t, processValue.FieldByName("inboxRetryDelay"), func(int) time.Duration { return time.Millisecond })

			store := newSessionDeliveryOwnershipStore(workDir)
			require.NoError(t, process.SetDeliveryOwnershipStore(store))
			require.Eventually(t, func() bool {
				select {
				case <-driver.firstSeen:
					return true
				default:
					return false
				}
			}, time.Second, time.Millisecond)
			require.Eventually(t, func() bool {
				select {
				case <-driver.secondEntered:
					return true
				default:
					return false
				}
			}, 2*time.Second, time.Millisecond)

			// Pending/unknown/error progress never puts the old body in the inbox; it
			// only advances the same durable reconciliation journal entry.
			records, err := store.LoadDeliveryOwnership()
			require.NoError(t, err)
			var q1 agent.DeliveryOwnershipRecord
			for _, record := range records {
				if record.MessageID() == "active-inflight" {
					q1 = record
				}
			}
			require.True(t, q1.IsInflight(agent.DeliveryInflightOutcomeUnknown))
			require.Equal(t, 1, q1.Attempts)
			driver.releaseOnce.Do(func() { close(driver.releaseSecond) })
			require.Eventually(t, func() bool {
				records, loadErr := store.LoadDeliveryOwnership()
				if loadErr != nil {
					return false
				}
				var old, successor agent.DeliveryOwnershipRecord
				for _, record := range records {
					switch record.MessageID() {
					case "active-inflight":
						old = record
					case "queued-successor":
						successor = record
					}
				}
				return old.IsFailed() && old.FailureCode == "runtime_input_reconciliation_exhausted" && successor.IsAcked()
			}, 4*time.Second, 5*time.Millisecond)

			driver.mu.Lock()
			spawnCfg := driver.spawnCfg
			delivered := append([]drivercontract.RuntimeInput(nil), driver.delivered...)
			driver.mu.Unlock()
			require.Equal(t, "thread-active", spawnCfg.SessionID)
			require.Empty(t, testLaunchSurface(spawnCfg, "freshPrompt"), "restored inflight body must never become a fresh Spawn prompt")
			require.Empty(t, testLaunchSurface(spawnCfg, "resumePrompt"))
			require.Empty(t, testLaunchSurface(spawnCfg, "prompt"))
			require.Len(t, delivered, 1, "only the queued successor may reach the replacement runtime")
			require.Equal(t, "queued-successor", delivered[0].ID)
			require.Contains(t, delivered[0].PromptText(), "new queued body")
			require.NotContains(t, delivered[0].PromptText(), "old body must never replay")
		})
	}
}

func TestContinuationWithoutCurrentWakeReconcilesHistoricalInflight(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	_, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:        types.InboxMessage{MessageID: "old-unknown", Content: "historical"},
		Input:          drivercontract.RuntimeInput{ID: "old-unknown"},
		State:          agent.DeliveryOwnershipInflight,
		InflightReason: agent.DeliveryInflightOutcomeUnknown,
		AcceptedAt:     time.Now().UTC().Add(-time.Hour),
	})
	require.NoError(t, err)

	restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, types.InboxMessage{}, false)

	require.NoError(t, err)
	require.True(t, restored.WaitForReconcile)
	require.Empty(t, restored.Message.MessageID)
}

func TestFullBacklogReclaimsOldestInsteadOfRejectingCurrentWake(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	acceptedAt := time.Now().UTC().Add(-time.Hour)
	for i := 0; i < maxRuntimeDeliveryLiveRecords; i++ {
		id := "queued-" + leftPadDecimal(i, 4)
		_, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
			Message:    types.InboxMessage{MessageID: id, Content: "old backlog"},
			State:      agent.DeliveryOwnershipQueued,
			AcceptedAt: acceptedAt.Add(time.Duration(i) * time.Millisecond),
		})
		require.NoError(t, err)
	}
	current := types.InboxMessage{MessageID: "current-513", Content: "must be accepted"}

	restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, current, true)

	require.NoError(t, err)
	require.True(t, restored.HandoffPersisted)
	require.Equal(t, current, restored.Message)
	require.NotEmpty(t, restored.Reclaimed)
	require.Equal(t, "queued-0000", restored.Reclaimed[0].MessageID())
	require.True(t, restored.Reclaimed[0].IsFailed())
	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	live := 0
	for _, record := range records {
		if agent.IsLiveDeliveryOwnershipState(record.State) {
			live++
		}
	}
	require.Equal(t, maxRuntimeDeliveryLiveRecords, live)
}

func TestFullBacklogReclaimsOldestForEachFirstLiveAdmission(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	acceptedAt := time.Now().UTC().Add(-time.Hour)
	for i := 0; i < maxRuntimeDeliveryLiveRecords; i++ {
		id := "inflight-" + leftPadDecimal(i, 4)
		_, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
			Message:        types.InboxMessage{MessageID: id, Content: "old accepted input"},
			Input:          drivercontract.RuntimeInput{ID: id},
			State:          agent.DeliveryOwnershipInflight,
			InflightReason: agent.DeliveryInflightDispatching,
			AcceptedAt:     acceptedAt.Add(time.Duration(i) * time.Millisecond),
		})
		require.NoError(t, err)
	}

	queuedReclaimed, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message: types.InboxMessage{MessageID: "queued-513", Content: "capacity-gated input"},
		State:   agent.DeliveryOwnershipQueued,
	})
	require.NoError(t, err)
	require.Len(t, queuedReclaimed, 1)
	require.Equal(t, "inflight-0000", queuedReclaimed[0].MessageID())

	inflightReclaimed, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:        types.InboxMessage{MessageID: "inflight-514", Content: "next direct input"},
		Input:          drivercontract.RuntimeInput{ID: "inflight-514"},
		State:          agent.DeliveryOwnershipInflight,
		InflightReason: agent.DeliveryInflightDispatching,
	})
	require.NoError(t, err)
	require.Len(t, inflightReclaimed, 1)
	require.Equal(t, "inflight-0001", inflightReclaimed[0].MessageID())

	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	live := 0
	for _, record := range records {
		if agent.IsLiveDeliveryOwnershipState(record.State) {
			live++
		}
	}
	require.Equal(t, maxRuntimeDeliveryLiveRecords, live)
}

func TestFullBacklogExistingOwnerTransitionDoesNotReclaim(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	acceptedAt := time.Now().UTC().Add(-time.Hour)
	for i := 0; i < maxRuntimeDeliveryLiveRecords; i++ {
		id := "queued-" + leftPadDecimal(i, 4)
		_, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
			Message:    types.InboxMessage{MessageID: id, Content: "queued input"},
			State:      agent.DeliveryOwnershipQueued,
			AcceptedAt: acceptedAt.Add(time.Duration(i) * time.Millisecond),
		})
		require.NoError(t, err)
	}

	reclaimed, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:        types.InboxMessage{MessageID: "queued-0000", Content: "queued input"},
		Input:          drivercontract.RuntimeInput{ID: "queued-0000"},
		State:          agent.DeliveryOwnershipInflight,
		InflightReason: agent.DeliveryInflightDispatching,
	})
	require.NoError(t, err)
	require.Empty(t, reclaimed)

	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, maxRuntimeDeliveryLiveRecords)
	for _, record := range records {
		require.False(t, record.IsFailed())
	}
}

func TestOnlyOversizedCurrentRecordCanFailAdmission(t *testing.T) {
	workDir := t.TempDir()
	current := types.InboxMessage{MessageID: "too-large", Content: string(make([]byte, maxRuntimeDeliveryRecordBytes))}

	_, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, current, true)

	require.Error(t, err)
	require.ErrorContains(t, err, "exceeds")
}

func TestTerminalOwnershipDropsReplayBodyAndAcceptsLateAckEvidence(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	failed := agent.DeliveryOwnershipRecord{
		Message: types.InboxMessage{
			MessageID: "late-evidence", Target: "#thread", Content: "private replay body",
			Metadata: map[string]any{"large": "metadata"}, AttachmentNames: "secret.txt",
		},
		Input:      testRuntimeInputWithConfig(drivercontract.RuntimeInput{ID: "late-evidence"}, "rendered body"),
		State:      agent.DeliveryOwnershipTerminal,
		Outcome:    agent.DeliveryOutcomeFailed,
		AcceptedAt: time.Now().UTC(),
	}
	_, err := store.PutDeliveryOwnership(failed)
	require.NoError(t, err)

	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, 1)
	require.Empty(t, records[0].Message.Content)
	require.Empty(t, records[0].Message.Metadata)
	require.Empty(t, records[0].Message.AttachmentNames)
	_, err = store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message: failed.Message, Input: failed.Input, State: agent.DeliveryOwnershipTerminal,
		Outcome: agent.DeliveryOutcomeAcked, AcceptedAt: failed.AcceptedAt,
	})
	require.NoError(t, err)
	records, err = store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, 1)
	require.True(t, records[0].IsAcked())
}

func TestLegacyOwnershipSchemaMigratesWithoutDroppingSessionState(t *testing.T) {
	workDir := t.TempDir()
	legacy := map[string]any{
		"lastSessionId": "session-kept",
		"runtimeDeliveryOwnership": []map[string]any{{
			"message": map[string]any{"message_id": "legacy"}, "phase": "uncertain",
		}},
	}
	data, err := json.Marshal(legacy)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(filepath.Join(workDir, sessionResumeStateFilename), data, 0o600))

	state, err := loadSessionResumeState(workDir)

	require.NoError(t, err)
	require.Equal(t, "session-kept", state.LastSessionID)
	require.Equal(t, runtimeDeliveryOwnershipVersion, state.RuntimeDeliveryOwnershipVersion)
	require.Len(t, state.RuntimeDeliveryOwnership, 1)
	require.True(t, state.RuntimeDeliveryOwnership[0].IsInflight(agent.DeliveryInflightOutcomeUnknown))
}

func TestLegacyOwnershipUnsupportedPhaseMigratesToFailedTombstone(t *testing.T) {
	workDir := t.TempDir()
	legacy := map[string]any{
		"lastSessionId": "session-kept",
		"runtimeDeliveryOwnership": []map[string]any{
			{"message": map[string]any{"message_id": "stale-empty"}, "phase": ""},
			{"message": map[string]any{"message_id": "stale-unknown"}, "phase": "accepted"},
			{"message": map[string]any{"message_id": "still-queued"}, "phase": "queued"},
			{"phase": ""},
		},
	}
	data, err := json.Marshal(legacy)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(filepath.Join(workDir, sessionResumeStateFilename), data, 0o600))

	state, err := loadSessionResumeState(workDir)

	require.NoError(t, err)
	require.Equal(t, "session-kept", state.LastSessionID)
	require.Equal(t, runtimeDeliveryOwnershipVersion, state.RuntimeDeliveryOwnershipVersion)
	require.Len(t, state.RuntimeDeliveryOwnership, 3, "record without message id must be dropped")
	byID := map[string]agent.DeliveryOwnershipRecord{}
	for _, record := range state.RuntimeDeliveryOwnership {
		byID[record.MessageID()] = record
	}
	require.True(t, byID["stale-empty"].IsFailed())
	require.Equal(t, "legacy_phase_unsupported", byID["stale-empty"].FailureCode)
	require.True(t, byID["stale-unknown"].IsFailed())
	require.Contains(t, byID["stale-unknown"].LastError, `"accepted"`)
	require.True(t, byID["still-queued"].IsQueued())

	// The migrated tombstones must not block the launch-restore path that
	// previously failed closed on the unsupported phase.
	store := newSessionDeliveryOwnershipStore(workDir)
	records, err := store.LoadDeliveryOwnership()
	require.NoError(t, err)
	require.Len(t, records, 3)
}

func TestFormat3OwnershipReadsCoreAndNormalizesOnWrite(t *testing.T) {
	workDir := t.TempDir()
	now := time.Now().UTC().Truncate(time.Second)
	format3 := map[string]any{
		"lastSessionId":                   "session-kept",
		"runtimeDeliveryOwnershipVersion": 3,
		"runtimeDeliveryOwnership": []map[string]any{
			{
				"message":    map[string]any{"message_id": "queued-v3", "content": "replay body"},
				"state":      "queued",
				"acceptedAt": now,
				"updatedAt":  now,
			},
			{
				"message":        map[string]any{"message_id": "inflight-v3", "content": "accepted body"},
				"input":          map[string]any{"ID": "inflight-v3"},
				"sessionId":      "provider-session",
				"state":          "inflight",
				"inflightReason": "outcome_unknown",
				"acceptedAt":     now,
				"updatedAt":      now,
			},
			{
				"message":             map[string]any{"message_id": "acked-v3", "target": "#ops"},
				"input":               map[string]any{"ID": "acked-v3"},
				"wakeToken":           "retired-wake",
				"visibleReplyPending": true,
				"state":               "terminal",
				"outcome":             "surfaced",
				"acceptedAt":          now,
				"updatedAt":           now,
			},
			{
				"message":              map[string]any{"message_id": "failed-v3", "target": "#ops"},
				"input":                map[string]any{"ID": "failed-v3"},
				"visibleReplyResolved": true,
				"state":                "terminal",
				"outcome":              "failed",
				"acceptedAt":           now,
				"updatedAt":            now,
			},
		},
	}
	data, err := json.Marshal(format3)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(sessionResumeStatePath(workDir), data, 0o600))

	store := newSessionDeliveryOwnershipStore(workDir)
	records, err := store.LoadDeliveryOwnership()

	require.NoError(t, err)
	require.Len(t, records, 4)
	require.True(t, records[0].IsQueued())
	require.Equal(t, "replay body", records[0].Message.Content)
	require.True(t, records[1].IsInflight(agent.DeliveryInflightOutcomeUnknown))
	require.Equal(t, "provider-session", records[1].SessionID)
	require.True(t, records[2].IsAcked())
	require.Equal(t, "retired-wake", records[2].WakeToken)
	require.True(t, records[3].IsFailed())

	restored, err := prepareRuntimeDeliveryOwnershipLaunch(workDir, types.InboxMessage{}, false)
	require.NoError(t, err)
	require.True(t, restored.HandoffPersisted)
	require.False(t, restored.WaitForReconcile)
	require.Equal(t, "queued-v3", restored.Message.MessageID)

	require.NoError(t, updateSessionResumeState(workDir, func(state *sessionResumeState) error {
		return nil
	}))
	persisted, err := os.ReadFile(sessionResumeStatePath(workDir))
	require.NoError(t, err)
	var saved map[string]any
	require.NoError(t, json.Unmarshal(persisted, &saved))
	require.Equal(t, float64(runtimeDeliveryOwnershipVersion), saved["runtimeDeliveryOwnershipVersion"])
	encoded := string(persisted)
	require.NotContains(t, encoded, "visibleReplyPending")
	require.NotContains(t, encoded, "visibleReplyResolved")
	require.Contains(t, encoded, "replay body")
}

func TestFormat3MalformedCoreFailsClosed(t *testing.T) {
	workDir := t.TempDir()
	data := []byte(`{
		"runtimeDeliveryOwnershipVersion": 3,
		"runtimeDeliveryOwnership": [{
			"message": {"message_id": "malformed-v3"},
			"state": "queued",
			"outcome": "surfaced",
			"acceptedAt": "2026-08-19T00:00:00Z",
			"updatedAt": "2026-08-19T00:00:00Z"
		}]
	}`)
	require.NoError(t, os.WriteFile(sessionResumeStatePath(workDir), data, 0o600))

	_, err := newSessionDeliveryOwnershipStore(workDir).LoadDeliveryOwnership()

	require.ErrorContains(t, err, "queued runtime delivery ownership carries inflight or terminal fields")
	preserved, readErr := os.ReadFile(sessionResumeStatePath(workDir))
	require.NoError(t, readErr)
	require.Equal(t, data, preserved)
}

func TestReclaimRetainsFailedTombstoneWhenTerminalRingIsFull(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)
	now := time.Now().UTC().Add(-time.Hour)
	records := make([]agent.DeliveryOwnershipRecord, 0, maxRuntimeDeliveryTerminalRecords+maxRuntimeDeliveryLiveRecords)
	for i := 0; i < maxRuntimeDeliveryTerminalRecords; i++ {
		records = append(records, agent.DeliveryOwnershipRecord{
			Message: types.InboxMessage{MessageID: "terminal-" + leftPadDecimal(i, 4)},
			State:   agent.DeliveryOwnershipTerminal, Outcome: agent.DeliveryOutcomeAcked,
			AcceptedAt: now.Add(time.Duration(i) * time.Millisecond),
		})
	}
	for i := 0; i < maxRuntimeDeliveryLiveRecords; i++ {
		records = append(records, agent.DeliveryOwnershipRecord{
			Message:    types.InboxMessage{MessageID: "live-" + leftPadDecimal(i, 4), Content: "old"},
			State:      agent.DeliveryOwnershipQueued,
			AcceptedAt: now.Add(time.Hour + time.Duration(i)*time.Millisecond),
		})
	}
	require.NoError(t, saveSessionResumeState(workDir, sessionResumeState{
		RuntimeDeliveryOwnershipVersion: runtimeDeliveryOwnershipVersion,
		RuntimeDeliveryOwnership:        records,
	}))

	reclaimed, err := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message: types.InboxMessage{MessageID: "fresh", Content: "new"},
		State:   agent.DeliveryOwnershipQueued, AcceptedAt: time.Now().UTC(),
	})
	require.NoError(t, err)
	require.NotEmpty(t, reclaimed)
	reclaimedID := reclaimed[0].MessageID()
	records, err = store.LoadDeliveryOwnership()
	require.NoError(t, err)
	for _, record := range records {
		if record.MessageID() == reclaimedID {
			require.True(t, record.IsFailed())
			return
		}
	}
	t.Fatalf("reclaimed tombstone %q was not retained", reclaimedID)
}

func TestUnknownFutureOwnershipVersionFailsClosedAndPreservesFile(t *testing.T) {
	workDir := t.TempDir()
	store := newSessionDeliveryOwnershipStore(workDir)

	queued := agent.DeliveryOwnershipRecord{
		Message:    types.InboxMessage{MessageID: "q-999", Target: "#ops"},
		Input:      drivercontract.RuntimeInput{ID: "q-999"},
		State:      agent.DeliveryOwnershipQueued,
		AcceptedAt: time.Now().UTC(),
		UpdatedAt:  time.Now().UTC(),
	}
	// Persist a resume-state envelope as if a newer daemon had acknowledged a
	// queued owner under ownership schema v999. saveSessionResumeState rejects
	// unknown versions, so write the opaque envelope directly.
	opaque, err := json.Marshal(sessionResumeState{
		RuntimeDeliveryOwnershipVersion: 999,
		RuntimeDeliveryOwnership:        []agent.DeliveryOwnershipRecord{queued},
	})
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(sessionResumeStatePath(workDir), append(opaque, '\n'), 0o600))

	// Load must fail closed rather than silently dropping the acknowledged owner.
	records, err := store.LoadDeliveryOwnership()
	require.Error(t, err)
	require.ErrorContains(t, err, "unsupported runtime delivery ownership version 999")
	require.ErrorContains(t, err, "supported read versions: 0, 2, 3")
	require.Empty(t, records)

	// The opaque file must be preserved verbatim: re-decoding still shows v999
	// and the acknowledged queued owner.
	raw, err := os.ReadFile(sessionResumeStatePath(workDir))
	require.NoError(t, err)
	var preserved sessionResumeState
	require.NoError(t, json.Unmarshal(raw, &preserved))
	require.Equal(t, 999, preserved.RuntimeDeliveryOwnershipVersion)
	require.Len(t, preserved.RuntimeDeliveryOwnership, 1)
	require.True(t, preserved.RuntimeDeliveryOwnership[0].IsQueued())

	// A later Put must also fail closed and must not overwrite the opaque file.
	_, putErr := store.PutDeliveryOwnership(agent.DeliveryOwnershipRecord{
		Message:    types.InboxMessage{MessageID: "fresh", Target: "#ops"},
		Input:      drivercontract.RuntimeInput{ID: "fresh"},
		State:      agent.DeliveryOwnershipQueued,
		AcceptedAt: time.Now().UTC(),
		UpdatedAt:  time.Now().UTC(),
	})
	require.Error(t, putErr)
	require.ErrorContains(t, putErr, "unsupported runtime delivery ownership version 999")

	raw2, err := os.ReadFile(sessionResumeStatePath(workDir))
	require.NoError(t, err)
	require.Equal(t, string(raw), string(raw2), "opaque ownership file must not be mutated by failed Put")
}

func leftPadDecimal(value, width int) string {
	text := ""
	for value > 0 {
		text = string(rune('0'+value%10)) + text
		value /= 10
	}
	if text == "" {
		text = "0"
	}
	for len(text) < width {
		text = "0" + text
	}
	return text
}