package agent

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"io"
	"net/http"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

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

type codexTransportCloseRecorder struct {
	mu     sync.Mutex
	closed int
}

type codexTransportFailingWriter struct {
	err error
}

type codexTransportShortWriter struct{}

func (codexTransportShortWriter) Write(p []byte) (int, error) {
	if len(p) == 0 {
		return 0, nil
	}
	return len(p) - 1, nil
}

func (codexTransportShortWriter) Close() error { return nil }

func (writer codexTransportFailingWriter) Write([]byte) (int, error) { return 0, writer.err }
func (writer codexTransportFailingWriter) Close() error              { return nil }

type codexTransportSignalWriter struct {
	wrote chan struct{}
	once  sync.Once
}

type codexTransportBlockingWriter struct {
	entered   chan struct{}
	release   chan struct{}
	once      sync.Once
	closeOnce sync.Once
	mu        sync.Mutex
	closed    bool
}

func (writer *codexTransportBlockingWriter) Write(p []byte) (int, error) {
	writer.once.Do(func() { close(writer.entered) })
	<-writer.release
	writer.mu.Lock()
	closed := writer.closed
	writer.mu.Unlock()
	if closed {
		return 0, io.ErrClosedPipe
	}
	return len(p), nil
}

func (writer *codexTransportBlockingWriter) Close() error {
	writer.mu.Lock()
	writer.closed = true
	writer.mu.Unlock()
	writer.closeOnce.Do(func() { close(writer.release) })
	return nil
}

type codexTransportBlockingRoundTripper struct {
	started chan struct{}
	release chan struct{}
	once    sync.Once
}

func (transport *codexTransportBlockingRoundTripper) RoundTrip(*http.Request) (*http.Response, error) {
	transport.once.Do(func() { close(transport.started) })
	<-transport.release
	return &http.Response{
		StatusCode: http.StatusOK,
		Header:     make(http.Header),
		Body: io.NopCloser(strings.NewReader(
			`{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"old tool output"}]}}`,
		)),
	}, nil
}

func (writer *codexTransportSignalWriter) Write(p []byte) (int, error) {
	writer.once.Do(func() { close(writer.wrote) })
	return len(p), nil
}

func (writer *codexTransportSignalWriter) Close() error { return nil }

func (recorder *codexTransportCloseRecorder) Write(p []byte) (int, error) {
	return len(p), nil
}

func (recorder *codexTransportCloseRecorder) Close() error {
	recorder.mu.Lock()
	recorder.closed++
	recorder.mu.Unlock()
	return nil
}

func (recorder *codexTransportCloseRecorder) closeCount() int {
	recorder.mu.Lock()
	defer recorder.mu.Unlock()
	return recorder.closed
}

func newCodexTransportFailureTestDriver() (*CodexAppServerDriver, chan appServerEnvelope, *codexTransportCloseRecorder, <-chan struct{}, chan string) {
	pending := make(chan appServerEnvelope, 1)
	stdin := &codexTransportCloseRecorder{}
	aborted := make(chan struct{}, 1)
	lines := make(chan string, 1)
	driver := &CodexAppServerDriver{
		launchGeneration: 7,
		codexRPCState: codexRPCState{
			pending:         map[int]pendingAppServerRequest{1: {generation: 7, responseCh: pending}},
			serverRequests:  make(map[string]inFlightServerRequest),
			rawStdin:        stdin,
			translatedLines: lines,
			transportFailure: &codexAppServerTransportFailure{abort: func() {
				aborted <- struct{}{}
			}},
		},
	}
	return driver, pending, stdin, aborted, lines
}

func requireCodexTransportFailed(t *testing.T, driver *CodexAppServerDriver, pending <-chan appServerEnvelope, stdin *codexTransportCloseRecorder, aborted <-chan struct{}, lines <-chan string) {
	t.Helper()

	response := <-pending
	require.NotNil(t, response.Error)
	requestErr := &appServerRequestError{
		Method:  "turn/start",
		Code:    response.Error.Code,
		Message: response.Error.Message,
		Data:    response.Error.Data,
	}
	require.ErrorIs(t, requestErr, errCodexAppServerTransportAmbiguous)
	require.Equal(t, codexAppServerTransportFailureCode, response.Error.Code)
	require.JSONEq(t, `{"kind":"transport_ambiguous","retryable":true}`, string(response.Error.Data))
	require.Equal(t, 1, stdin.closeCount())
	select {
	case <-aborted:
	default:
		t.Fatal("live app-server child abort hook was not invoked")
	}
	_, ok := <-lines
	require.False(t, ok, "translated output must close so Process.parseStdout exits")
	_, err := driver.writeRaw([]byte("after failure"))
	require.ErrorIs(t, err, io.ErrClosedPipe)
}

func TestCodexAppServerReadLoopEOFAbortsLiveTransportAndFailsPending(t *testing.T) {
	driver, pending, stdin, aborted, lines := newCodexTransportFailureTestDriver()

	driver.readLoopForGeneration(strings.NewReader(""), 7, driver.transportFailure)

	requireCodexTransportFailed(t, driver, pending, stdin, aborted, lines)
}

func TestCodexAppServerReadLoopInvalidJSONIsTransportFatal(t *testing.T) {
	driver, pending, stdin, aborted, lines := newCodexTransportFailureTestDriver()

	driver.readLoopForGeneration(strings.NewReader("{not-json}\n"), 7, driver.transportFailure)

	requireCodexTransportFailed(t, driver, pending, stdin, aborted, lines)
}

func TestCodexAppServerReadLoopOversizedFrameIsTransportFatal(t *testing.T) {
	driver, pending, stdin, aborted, lines := newCodexTransportFailureTestDriver()
	oversized := strings.NewReader(strings.Repeat("x", codexAppServerMaxMessageBytes+1))

	driver.readLoopForGeneration(oversized, 7, driver.transportFailure)

	requireCodexTransportFailed(t, driver, pending, stdin, aborted, lines)
}

func TestCodexAppServerStaleReadLoopCannotAbortNewLaunch(t *testing.T) {
	driver, pending, stdin, aborted, lines := newCodexTransportFailureTestDriver()
	staleAborted := make(chan struct{}, 1)
	staleFailure := &codexAppServerTransportFailure{abort: func() {
		staleAborted <- struct{}{}
	}}

	driver.readLoopForGeneration(strings.NewReader(""), 6, staleFailure)

	require.Equal(t, 0, stdin.closeCount())
	select {
	case <-staleAborted:
	default:
		t.Fatal("stale reader did not terminate its own old app-server child")
	}
	select {
	case <-aborted:
		t.Fatal("stale reader aborted the current app-server child")
	default:
	}
	select {
	case <-pending:
		t.Fatal("stale reader failed a current-generation pending request")
	default:
	}
	select {
	case _, ok := <-lines:
		if !ok {
			t.Fatal("stale reader closed current-generation translated output")
		}
	default:
	}
}

func TestCodexAppServerStaleFrameCannotSatisfyReusedRequestID(t *testing.T) {
	driver, pending, stdin, aborted, lines := newCodexTransportFailureTestDriver()
	staleAborted := make(chan struct{}, 1)
	staleFailure := &codexAppServerTransportFailure{abort: func() {
		staleAborted <- struct{}{}
	}}

	driver.readLoopForGeneration(strings.NewReader(`{"jsonrpc":"2.0","id":1,"result":{"turn":{"id":"old-turn"}}}`+"\n"), 6, staleFailure)

	select {
	case response := <-pending:
		t.Fatalf("stale response crossed the generation fence: %+v", response)
	default:
	}
	select {
	case <-staleAborted:
	default:
		t.Fatal("stale frame did not terminate its own old app-server child")
	}
	require.Equal(t, 0, stdin.closeCount())
	select {
	case <-aborted:
		t.Fatal("stale frame aborted the current app-server child")
	default:
	}
	select {
	case _, ok := <-lines:
		if !ok {
			t.Fatal("stale frame closed current-generation translated output")
		}
	default:
	}
}

func TestCodexAppServerBidirectionalRequestIDCollisionPreservesClientPending(t *testing.T) {
	const generation = uint64(7)
	pending := make(chan appServerEnvelope, 1)
	writer := &rawCaptureWriteCloser{}
	driver := &CodexAppServerDriver{
		launchGeneration: generation,
		codexRPCState: codexRPCState{
			pending: map[int]pendingAppServerRequest{
				1: {generation: generation, responseCh: pending},
			},
			serverRequests: make(map[string]inFlightServerRequest),
			rawStdin:       writer,
		},
	}

	// Codex's independently allocated server-request ID collides with our live
	// client request ID. The method-bearing frame must execute as a server
	// request without consuming or satisfying the pending client response.
	require.NoError(t, driver.handleAppServerLineForGeneration(
		`{"jsonrpc":"2.0","id":1,"method":"test/colliding-server-request","params":{}}`,
		generation,
	))
	select {
	case response := <-pending:
		t.Fatalf("server request collision consumed client pending response: %+v", response)
	default:
	}
	driver.mu.Lock()
	_, stillPending := driver.pending[1]
	driver.mu.Unlock()
	require.True(t, stillPending)
	require.Eventually(t, func() bool { return len(writer.Writes()) == 1 }, time.Second, time.Millisecond,
		"colliding server request must still execute and receive a response")
	var serverResponse appServerEnvelope
	require.NoError(t, json.Unmarshal(writer.Writes()[0], &serverResponse))
	require.JSONEq(t, `1`, string(serverResponse.ID))
	require.NotNil(t, serverResponse.Error)
	require.Equal(t, -32601, serverResponse.Error.Code)

	// A later method-free frame with the same ID is the actual response in the
	// opposite direction and only now completes/removes our pending request.
	require.NoError(t, driver.handleAppServerLineForGeneration(
		`{"jsonrpc":"2.0","id":1,"result":{"turn":{"id":"turn-1"}}}`,
		generation,
	))
	clientResponse := <-pending
	require.Nil(t, clientResponse.Error)
	require.JSONEq(t, `{"turn":{"id":"turn-1"}}`, string(clientResponse.Result))
	driver.mu.Lock()
	_, stillPending = driver.pending[1]
	driver.mu.Unlock()
	require.False(t, stillPending)
}

func TestCodexAppServerOldServerRequestCannotRemoveReusedRequestID(t *testing.T) {
	driver := &CodexAppServerDriver{
		launchGeneration: 8,
		codexRPCState: codexRPCState{serverRequests: map[string]inFlightServerRequest{
			"1": {generation: 8},
		}},
	}

	_, ok := driver.finishServerRequestForGeneration("1", 7)
	require.False(t, ok)
	require.Contains(t, driver.serverRequests, "1")

	_, ok = driver.finishServerRequestForGeneration("1", 8)
	require.True(t, ok)
	require.NotContains(t, driver.serverRequests, "1")
}

func TestCodexAppServerTranslatedOutputFlushIsGenerationBound(t *testing.T) {
	driver := &CodexAppServerDriver{}
	oldLines := make(chan string, 1)
	newLines := make(chan string, 1)
	oldLines <- "old generation"
	close(oldLines)
	newLines <- "new generation"
	driver.translatedLines = newLines
	var output bytes.Buffer

	driver.flushTranslatedLines(oldLines, &output)

	require.Equal(t, "old generation\n", output.String())
	require.Equal(t, "new generation", <-newLines, "old flusher must not drain the current launch")
}

func TestCodexAppServerTransportFailureWaitReportsDisconnectReason(t *testing.T) {
	failure := &codexAppServerTransportFailure{}
	failure.fail("stdout reader disconnected")

	err := failure.wait(func() error { return errors.New("signal: terminated") })

	require.Error(t, err)
	require.Contains(t, err.Error(), "stdout reader disconnected")
	require.Contains(t, err.Error(), "signal: terminated")
}

func TestCodexAppServerIntentionalShutdownDoesNotBecomeTransportFailure(t *testing.T) {
	aborts := 0
	failure := &codexAppServerTransportFailure{abort: func() { aborts++ }}
	failure.beginShutdown()
	failure.fail("expected EOF after Stop")
	failure.terminate()
	failure.terminate()

	require.NoError(t, failure.wait(func() error { return nil }))
	require.Equal(t, 1, aborts, "explicit stop and reader EOF must share one termination")
}

func TestCodexAppServerRequestWriteFailureIsTransportAmbiguous(t *testing.T) {
	driver := &CodexAppServerDriver{codexRPCState: codexRPCState{
		pending:  make(map[int]pendingAppServerRequest),
		rawStdin: codexTransportFailingWriter{err: io.ErrClosedPipe},
	}}

	_, err := driver.request(t.Context(), "turn/start", map[string]any{"threadId": "thread-1"})

	require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	require.Contains(t, err.Error(), "turn/start")
	require.Empty(t, driver.pending)
}

func TestCodexAppServerRequestShortWriteIsTransportAmbiguous(t *testing.T) {
	for _, guarded := range []bool{false, true} {
		t.Run(map[bool]string{false: "request", true: "guarded_request"}[guarded], func(t *testing.T) {
			driver := &CodexAppServerDriver{codexRPCState: codexRPCState{
				pending:  make(map[int]pendingAppServerRequest),
				rawStdin: codexTransportShortWriter{},
			}}
			var err error
			if guarded {
				_, err = driver.requestWithDispatchGuard(t.Context(), "turn/steer", map[string]any{"threadId": "thread-1"}, nil)
			} else {
				_, err = driver.request(t.Context(), "turn/start", map[string]any{"threadId": "thread-1"})
			}

			require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
			require.ErrorIs(t, err, io.ErrShortWrite)
			require.Empty(t, driver.pending)
		})
	}
}

func TestCodexAppServerGuardedRequestOnClosedTransportIsAmbiguous(t *testing.T) {
	driver := &CodexAppServerDriver{}

	_, err := driver.requestWithDispatchGuard(
		t.Context(),
		"turn/steer",
		map[string]any{"threadId": "thread-1"},
		nil,
	)

	require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
}

func TestCodexAppServerRequestCanceledAfterWriteIsAmbiguous(t *testing.T) {
	writer := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{codexRPCState: codexRPCState{
		pending:  make(map[int]pendingAppServerRequest),
		rawStdin: writer,
	}}
	ctx, cancel := context.WithCancel(t.Context())
	result := make(chan error, 1)
	go func() {
		_, err := driver.request(ctx, "turn/start", map[string]any{"threadId": "thread-1"})
		result <- err
	}()
	<-writer.wrote
	cancel()

	err := <-result
	require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	require.ErrorIs(t, err, context.Canceled)
	require.Empty(t, driver.pending)
}

func TestCodexAppServerRequestCanceledBeforeWriteIsDefinite(t *testing.T) {
	writer := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{codexRPCState: codexRPCState{
		pending:  make(map[int]pendingAppServerRequest),
		rawStdin: writer,
	}}
	ctx, cancel := context.WithCancel(t.Context())
	cancel()

	_, err := driver.request(ctx, "turn/start", map[string]any{"threadId": "thread-1"})

	require.ErrorIs(t, err, context.Canceled)
	require.NotErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	select {
	case <-writer.wrote:
		t.Fatal("pre-canceled request was written to app-server")
	default:
	}
}

func TestCodexIdleCompactionCanceledBeforeWriteRestoresQueuedIdleState(t *testing.T) {
	writer := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{
		codexRPCState: codexRPCState{
			pending:  make(map[int]pendingAppServerRequest),
			rawStdin: writer,
		},
		codexThreadState: codexThreadState{
			threadID:      "thread-1",
			threadStatus:  appServerThreadStatus{Type: "idle"},
			compactQueued: true,
		},
	}
	ctx, cancel := context.WithCancel(t.Context())
	cancel()

	started, err := driver.StartIdleCompaction(ctx, "thread-1")

	require.False(t, started)
	require.ErrorIs(t, err, context.Canceled)
	require.NotErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	select {
	case <-writer.wrote:
		t.Fatal("pre-canceled compaction request was written to app-server")
	default:
	}
	driver.mu.Lock()
	require.True(t, driver.compactQueued)
	require.False(t, driver.compactionInProgress)
	require.Equal(t, appServerThreadStatus{Type: "idle"}, driver.threadStatus)
	driver.mu.Unlock()
}

func TestCodexAppServerOldRequestCancelCannotDeleteReusedPendingID(t *testing.T) {
	writer := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			pending:  make(map[int]pendingAppServerRequest),
			rawStdin: writer,
		},
	}
	ctx, cancel := context.WithCancel(t.Context())
	oldResult := make(chan error, 1)
	go func() {
		_, err := driver.request(ctx, "turn/start", map[string]any{"threadId": "thread-old"})
		oldResult <- err
	}()
	<-writer.wrote

	newResponse := make(chan appServerEnvelope, 1)
	driver.mu.Lock()
	driver.launchGeneration = 2
	driver.requestID = 1 // replacement launch reuses JSON-RPC id 1
	driver.pending = map[int]pendingAppServerRequest{
		1: {generation: 2, responseCh: newResponse},
	}
	driver.rawStdin = &codexTransportCloseRecorder{}
	driver.mu.Unlock()

	cancel()
	err := <-oldResult
	require.ErrorIs(t, err, context.Canceled)
	require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)

	driver.mu.Lock()
	pending, ok := driver.pending[1]
	driver.mu.Unlock()
	require.True(t, ok, "old context cancellation must not unregister a reused id in the replacement generation")
	require.Equal(t, uint64(2), pending.generation)
	require.Equal(t, newResponse, pending.responseCh)
}

func TestCodexAppServerRequestRegistrationAndWriteCannotStraddleReset(t *testing.T) {
	oldWriter := &codexTransportBlockingWriter{entered: make(chan struct{}), release: make(chan struct{})}
	newWriter := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			pending:  make(map[int]pendingAppServerRequest),
			rawStdin: oldWriter,
		},
	}
	requestDone := make(chan error, 1)
	go func() {
		_, err := driver.request(t.Context(), "turn/start", map[string]any{"threadId": "thread-old"})
		requestDone <- err
	}()
	<-oldWriter.entered

	resetDone := make(chan struct{})
	go func() {
		driver.transportGenerationMu.Lock()
		driver.tokenUsageMu.Lock()
		driver.mu.Lock()
		retired := driver.snapshotTransportForRetirementLocked()
		driver.resetAppServerStateForLaunchLocked(drivercontract.LaunchConfig{}, newWriter)
		driver.mu.Unlock()
		driver.tokenUsageMu.Unlock()
		retired.retire("superseded")
		driver.transportGenerationMu.Unlock()
		close(resetDone)
	}()

	select {
	case <-resetDone:
	case <-time.After(time.Second):
		t.Fatal("next generation reset deadlocked behind a blocked old stdin Write")
	}
	err := <-requestDone
	require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	select {
	case <-newWriter.wrote:
		t.Fatal("old request was written into the replacement generation stdin")
	default:
	}
}

func TestCodexAppServerLifecycleWaitingForBlockedWriteDoesNotPinGenerationReset(t *testing.T) {
	oldWriter := &codexTransportBlockingWriter{entered: make(chan struct{}), release: make(chan struct{})}
	newWriter := &codexTransportSignalWriter{wrote: make(chan struct{})}
	oldLines := make(chan string, 4)
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			pending:         make(map[int]pendingAppServerRequest),
			rawStdin:        oldWriter,
			translatedLines: oldLines,
		},
		codexThreadState: codexThreadState{
			threadID:         "thread-1",
			threadStatus:     appServerThreadStatus{Type: "idle"},
			completedTurnIDs: make(map[string]struct{}),
		},
	}

	requestDone := make(chan error, 1)
	go func() {
		requestDone <- driver.sendTurnRequest(t.Context(), "thread-1", "turn/start", map[string]any{
			"threadId": "thread-1",
			"input":    []map[string]any{},
		})
	}()
	<-oldWriter.entered

	// The old app-server may emit turn/started while the turn/start stdin write
	// is blocked. That notification must wait for the primary dispatch fence
	// without first taking the generation read fence; replacement needs the
	// generation write fence to detach and close oldWriter.
	params := rawRPCResult(t, appServerTurnNotification{
		ThreadID: "thread-1",
		Turn: appServerTurn{
			ID:     "turn-old",
			Status: appServerTurnStatus{Type: "inProgress"},
		},
	})
	frame, err := json.Marshal(appServerEnvelope{
		Method: "turn/started",
		Params: params,
	})
	require.NoError(t, err)
	readerDone := make(chan struct{})
	go func() {
		driver.readLoopForGeneration(strings.NewReader(string(frame)+"\n"), 1, &codexAppServerTransportFailure{})
		close(readerDone)
	}()
	select {
	case <-readerDone:
		t.Fatal("lifecycle notification crossed the blocked turn dispatch")
	case <-time.After(20 * time.Millisecond):
	}

	resetDone := make(chan struct{})
	go func() {
		driver.transportGenerationMu.Lock()
		driver.tokenUsageMu.Lock()
		driver.mu.Lock()
		retired := driver.snapshotTransportForRetirementLocked()
		driver.resetAppServerStateForLaunchLocked(drivercontract.LaunchConfig{}, newWriter)
		driver.mu.Unlock()
		driver.tokenUsageMu.Unlock()
		retired.retire("superseded")
		driver.transportGenerationMu.Unlock()
		close(resetDone)
	}()

	select {
	case <-resetDone:
	case <-time.After(time.Second):
		t.Fatal("generation reset deadlocked behind lifecycle notification waiting for a blocked stdin write")
	}
	require.ErrorIs(t, <-requestDone, errCodexAppServerTransportAmbiguous)
	select {
	case <-readerDone:
	case <-time.After(time.Second):
		t.Fatal("superseded lifecycle reader did not exit after generation reset")
	}
	status, activeTurnID, _ := driver.currentThreadState("thread-1")
	require.Equal(t, "", normalizeThreadStatus(status).Type, "old turn/started must not pollute the replacement generation")
	require.Empty(t, activeTurnID, "old turn/started must not install an active turn in the replacement generation")
}

func TestCodexAppServerBlockingRequestWriteDoesNotBlockCleanup(t *testing.T) {
	writer := &codexTransportBlockingWriter{entered: make(chan struct{}), release: make(chan struct{})}
	driver := &CodexAppServerDriver{
		launchGeneration: 3,
		codexRPCState: codexRPCState{
			pending:  make(map[int]pendingAppServerRequest),
			rawStdin: writer,
		},
	}
	requestDone := make(chan error, 1)
	go func() {
		_, err := driver.request(t.Context(), "turn/start", map[string]any{"threadId": "thread-1"})
		requestDone <- err
	}()
	<-writer.entered

	cleanupDone := make(chan error, 1)
	go func() { cleanupDone <- driver.closeRawStdinForGeneration(3, writer) }()
	select {
	case err := <-cleanupDone:
		require.NoError(t, err)
	case <-time.After(time.Second):
		t.Fatal("cleanup deadlocked behind a blocked stdin Write")
	}
	select {
	case err := <-requestDone:
		require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	case <-time.After(time.Second):
		t.Fatal("closing stdin did not unblock the blocked request write")
	}
}

func TestCodexAppServerBlockingRequestWriteDoesNotBlockFailTransport(t *testing.T) {
	writer := &codexTransportBlockingWriter{entered: make(chan struct{}), release: make(chan struct{})}
	failure := &codexAppServerTransportFailure{}
	driver := &CodexAppServerDriver{
		launchGeneration: 4,
		codexRPCState: codexRPCState{
			pending:          make(map[int]pendingAppServerRequest),
			serverRequests:   make(map[string]inFlightServerRequest),
			rawStdin:         writer,
			transportFailure: failure,
		},
	}
	requestDone := make(chan error, 1)
	go func() {
		_, err := driver.request(t.Context(), "turn/start", map[string]any{"threadId": "thread-1"})
		requestDone <- err
	}()
	<-writer.entered

	failDone := make(chan struct{})
	go func() {
		driver.failTransport(4, failure, "forced transport failure")
		close(failDone)
	}()
	select {
	case <-failDone:
	case <-time.After(time.Second):
		t.Fatal("failTransport deadlocked behind a blocked stdin Write")
	}
	select {
	case err := <-requestDone:
		require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
	case <-time.After(time.Second):
		t.Fatal("failTransport did not unblock the blocked request write")
	}
	driver.mu.Lock()
	require.Empty(t, driver.pending)
	require.Nil(t, driver.rawStdin)
	driver.mu.Unlock()
}

func TestCodexAppServerOldProcessWriterCannotTouchReplacementStdin(t *testing.T) {
	oldWriter := &codexTransportCloseRecorder{}
	newWriter := &codexTransportCloseRecorder{}
	driver := &CodexAppServerDriver{
		launchGeneration: 2,
		codexRPCState:    codexRPCState{rawStdin: newWriter},
	}
	oldStream := &lockedDriverWriter{driver: driver, generation: 1, writer: oldWriter}

	require.ErrorIs(t, func() error {
		_, err := oldStream.Write([]byte("stale write"))
		return err
	}(), io.ErrClosedPipe)
	require.NoError(t, oldStream.Close())
	require.Equal(t, 1, oldWriter.closeCount())
	require.Equal(t, 0, newWriter.closeCount())

	driver.mu.Lock()
	require.Same(t, newWriter, driver.rawStdin)
	driver.mu.Unlock()
}

func TestCodexAppServerOldInitializedNotificationCannotReachReplacementStdin(t *testing.T) {
	oldWriter := &codexTransportCloseRecorder{}
	newWriter := &codexTransportSignalWriter{wrote: make(chan struct{})}
	driver := &CodexAppServerDriver{
		launchGeneration: 2,
		codexRPCState:    codexRPCState{rawStdin: newWriter},
	}

	err := driver.notifyForGeneration(1, oldWriter, "initialized", map[string]any{})
	require.ErrorIs(t, err, io.ErrClosedPipe)
	select {
	case <-newWriter.wrote:
		t.Fatal("old initialized notification was written into replacement stdin")
	default:
	}
}

func TestCodexAppServerStaleAsyncToolResultCannotEnterReplacementGeneration(t *testing.T) {
	transport := &codexTransportBlockingRoundTripper{started: make(chan struct{}), release: make(chan struct{})}
	oldLines := make(chan string, 4)
	newLines := make(chan string, 4)
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			translatedLines: oldLines,
		},
		codexLaunchState: codexLaunchState{
			mcpConfigs: map[string]types.NamedMCPConfig{
				"agentteam": {Name: "agentteam", URL: "http://mcp.invalid"},
			},
			httpClient: &http.Client{Transport: transport},
		},
		codexThreadState: codexThreadState{
			threadID:       "thread-1",
			threadStatus:   appServerThreadStatus{Type: "active"},
			activeTurnID:   "turn-1",
			toolActivities: map[string]toolActivityRef{"call-1": {threadID: "thread-1", turnID: "turn-1", callID: "call-1", toolName: "mcp__agentteam__message_history_read"}},
		},
	}

	done := make(chan *appServerResponseError, 1)
	go func() {
		_, rpcErr := driver.executeServerRequestWithActivityForGeneration(
			t.Context(),
			1,
			"request-1",
			"mcpServer/tool/call",
			json.RawMessage(`{"threadId":"thread-1","turnId":"turn-1","callId":"call-1","server":"agentteam","tool":"message_history_read","arguments":{}}`),
		)
		done <- rpcErr
	}()
	<-transport.started

	driver.transportGenerationMu.Lock()
	driver.mu.Lock()
	driver.launchGeneration = 2
	driver.translatedLines = newLines
	driver.toolActivities = map[string]toolActivityRef{
		"call-1": {threadID: "thread-1", turnID: "turn-1", callID: "call-1", toolName: "mcp__agentteam__message_history_read"},
	}
	driver.mu.Unlock()
	driver.transportGenerationMu.Unlock()
	close(transport.release)

	rpcErr := <-done
	require.NotNil(t, rpcErr)
	require.Contains(t, rpcErr.Message, "superseded")
	select {
	case line := <-newLines:
		t.Fatalf("stale async tool result entered replacement translated output: %s", line)
	default:
	}
	driver.mu.Lock()
	_, stillTracked := driver.toolActivities["call-1"]
	driver.mu.Unlock()
	require.True(t, stillTracked, "stale completion must not forget replacement generation tool activity")
}

func TestCodexAppServerRuntimeDeliveryReceiptCommitsAfterFinalChildWrite(t *testing.T) {
	receipt := types.NewRuntimeDeliveryReceipt("0123456789abcdef0123456789abcdef")
	marker := types.RuntimeDeliveryReceiptMarker(receipt)
	transport := roundTripFunc(func(req *http.Request) (*http.Response, error) {
		require.Equal(t, "true", req.Header.Get(types.MCPDeferredToolDeliveryHeader))
		require.Equal(t, "thread-1", req.Header.Get(types.MCPRuntimeSessionHeader))
		payload, err := json.Marshal(map[string]any{
			"jsonrpc": "2.0",
			"id":      1,
			"result": map[string]any{
				"content": []map[string]any{{"type": "text", "text": marker + "\nq1 body"}},
				"_meta":   map[string]any{types.MCPRuntimeDeliveryReceiptsMetaKey: []string{receipt}},
			},
		})
		require.NoError(t, err)
		return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(bytes.NewReader(payload))}, nil
	})
	rawWriter := &rawCaptureWriteCloser{}
	lines := make(chan string, 8)
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			rawStdin:        rawWriter,
			translatedLines: lines,
			pending:         make(map[int]pendingAppServerRequest),
			serverRequests:  make(map[string]inFlightServerRequest),
		},
		codexLaunchState: codexLaunchState{
			mcpConfigs: map[string]types.NamedMCPConfig{"agentteam": {Name: "agentteam", URL: "http://mcp.local"}},
			httpClient: &http.Client{Transport: transport},
		},
		codexThreadState: codexThreadState{
			threadID:       "thread-1",
			threadStatus:   appServerThreadStatus{Type: "active"},
			activeTurnID:   "turn-1",
			toolActivities: map[string]toolActivityRef{"call-1": {threadID: "thread-1", turnID: "turn-1", callID: "call-1", toolName: "mcp__agentteam__message_send"}},
		},
	}

	driver.handleServerRequest(
		context.Background(),
		func() {},
		1,
		"",
		json.RawMessage(`"server-request-1"`),
		"mcpServer/tool/call",
		json.RawMessage(`{"threadId":"thread-1","turnId":"turn-1","callId":"call-1","server":"agentteam","tool":"message_send","arguments":{}}`),
	)

	require.Len(t, rawWriter.Writes(), 1, "the complete app-server result must be written before Delivered is emitted")
	var acked bool
	for len(lines) > 0 {
		var event translatedEvent
		require.NoError(t, json.Unmarshal([]byte(<-lines), &event))
		if event.Kind == consts.ParsedEventKindRuntimeInputAcked && event.RuntimeInputID == receipt {
			acked = true
			require.Equal(t, "thread-1", event.SessionID)
		}
	}
	require.True(t, acked)
}

func TestCodexAppServerRuntimeDeliveryShortWriteIsUncertainNotDelivered(t *testing.T) {
	receipt := types.NewRuntimeDeliveryReceipt("fedcba9876543210fedcba9876543210")
	marker := types.RuntimeDeliveryReceiptMarker(receipt)
	payload, err := json.Marshal(map[string]any{
		"jsonrpc": "2.0", "id": 1,
		"result": map[string]any{
			"content": []map[string]any{{"type": "text", "text": marker + "\nq1 body"}},
			"_meta":   map[string]any{types.MCPRuntimeDeliveryReceiptsMetaKey: []string{receipt}},
		},
	})
	require.NoError(t, err)
	lines := make(chan string, 8)
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			rawStdin:        codexTransportShortWriter{},
			translatedLines: lines,
			pending:         make(map[int]pendingAppServerRequest),
			serverRequests:  make(map[string]inFlightServerRequest),
		},
		codexLaunchState: codexLaunchState{
			mcpConfigs: map[string]types.NamedMCPConfig{"agentteam": {Name: "agentteam", URL: "http://mcp.local"}},
			httpClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
				return &http.Response{StatusCode: http.StatusOK, Header: make(http.Header), Body: io.NopCloser(bytes.NewReader(payload))}, nil
			})},
		},
		codexThreadState: codexThreadState{
			threadID:       "thread-1",
			threadStatus:   appServerThreadStatus{Type: "active"},
			activeTurnID:   "turn-1",
			toolActivities: map[string]toolActivityRef{"call-1": {threadID: "thread-1", turnID: "turn-1", callID: "call-1", toolName: "mcp__agentteam__message_send"}},
		},
	}

	driver.handleServerRequest(context.Background(), func() {}, 1, "", json.RawMessage(`1`), "mcpServer/tool/call",
		json.RawMessage(`{"threadId":"thread-1","turnId":"turn-1","callId":"call-1","server":"agentteam","tool":"message_send","arguments":{}}`))

	var uncertain, delivered bool
	for len(lines) > 0 {
		var event translatedEvent
		require.NoError(t, json.Unmarshal([]byte(<-lines), &event))
		if event.RuntimeInputID != receipt {
			continue
		}
		uncertain = uncertain || event.Kind == consts.ParsedEventKindRuntimeInputUncertain
		delivered = delivered || event.Kind == consts.ParsedEventKindRuntimeInputAcked
	}
	require.True(t, uncertain)
	require.False(t, delivered)
}

func TestCodexAppServerErrorResponseShortWriteRetiresTransport(t *testing.T) {
	aborted := make(chan struct{}, 1)
	failure := &codexAppServerTransportFailure{abort: func() { aborted <- struct{}{} }}
	driver := &CodexAppServerDriver{
		launchGeneration: 1,
		codexRPCState: codexRPCState{
			rawStdin:         codexTransportShortWriter{},
			pending:          make(map[int]pendingAppServerRequest),
			serverRequests:   make(map[string]inFlightServerRequest),
			transportFailure: failure,
		},
	}

	driver.handleServerRequest(
		context.Background(),
		func() {},
		1,
		"",
		json.RawMessage(`1`),
		"test/unsupported-server-request",
		json.RawMessage(`{}`),
	)

	select {
	case <-aborted:
	case <-time.After(time.Second):
		t.Fatal("partial JSON-RPC error response left the child transport live")
	}
	driver.mu.Lock()
	require.Nil(t, driver.rawStdin)
	driver.mu.Unlock()
}

func TestCodexAppServerAcceptedTurnResponseDecodeFailureIsAmbiguous(t *testing.T) {
	for _, tt := range []struct {
		name   string
		method string
		result []byte
	}{
		{name: "start invalid json", method: "turn/start", result: []byte(`{not-json`)},
		{name: "steer invalid json", method: "turn/steer", result: []byte(`{not-json`)},
		{name: "start missing id", method: "turn/start", result: []byte(`{"turn":{}}`)},
		{name: "steer missing id", method: "turn/steer", result: []byte(`{}`)},
	} {
		t.Run(tt.name, func(t *testing.T) {
			driver, _ := newTestCodexDriver(t, func(capturedRPCRequest) appServerEnvelope {
				return appServerEnvelope{Result: tt.result}
			})

			err := driver.sendTurnRequest(t.Context(), "thread-1", tt.method, map[string]any{"threadId": "thread-1"})

			require.ErrorIs(t, err, errCodexAppServerTransportAmbiguous)
			require.Contains(t, err.Error(), tt.method)
		})
	}
}