package agent

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"strings"
	"sync"

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

type lockedDriverWriter struct {
	driver     *CodexAppServerDriver
	generation uint64
	writer     io.WriteCloser
}

var errCodexAppServerRequestDispatchRejected = errors.New("codex app-server request dispatch rejected")

var errCodexAppServerTransportSuperseded = errors.New("codex app-server transport produced output after it was superseded")

// errCodexAppServerTransportAmbiguous classifies requests whose response was
// lost because the app-server transport failed. In particular, callers must
// not assume a mutating request such as turn/start or turn/steer was rejected:
// Codex may have accepted it before stdout disconnected. Delivery ownership is
// reconciled above the transport layer by the correlated client message ID.
var errCodexAppServerTransportAmbiguous = errors.New("codex app-server transport outcome is ambiguous")

const codexAppServerTransportFailureCode = -32098

func codexAppServerTransportRequestError(method string, err error) error {
	return codexAppServerAmbiguousOutcomeError(method, "codex app-server transport write failed", err)
}

func codexAppServerAmbiguousOutcomeError(method string, message string, err error) error {
	data, _ := json.Marshal(map[string]any{
		"kind":      "transport_ambiguous",
		"retryable": true,
	})
	message = strings.TrimSpace(message)
	if message == "" {
		message = "codex app-server request outcome is ambiguous"
	}
	if err != nil {
		message += ": " + err.Error()
	}
	return &appServerRequestError{
		Method:  strings.TrimSpace(method),
		Code:    codexAppServerTransportFailureCode,
		Message: message,
		Data:    data,
		Cause:   err,
	}
}

type codexAppServerTransportFailure struct {
	mu        sync.Mutex
	abortOnce sync.Once
	abort     func()
	reason    string
	closing   bool
	shutdown  chan struct{}
}

// codexAppServerRetiredTransport owns every generation-scoped resource that
// must be drained before a driver instance can safely start another app-server.
// Keeping this snapshot separate from d.codexRPCState prevents a late cleanup
// or reader from closing the replacement generation's pipes and channels.
type codexAppServerRetiredTransport struct {
	pending          map[int]pendingAppServerRequest
	serverRequests   map[string]inFlightServerRequest
	stdin            io.WriteCloser
	translatedLines  chan string
	translatedOutput *codexTranslatedOutput
	transportFailure *codexAppServerTransportFailure
}

func (d *CodexAppServerDriver) snapshotTransportForRetirementLocked() codexAppServerRetiredTransport {
	return codexAppServerRetiredTransport{
		pending:          d.pending,
		serverRequests:   d.serverRequests,
		stdin:            d.rawStdin,
		translatedLines:  d.translatedLines,
		translatedOutput: d.translatedOutput,
		transportFailure: d.transportFailure,
	}
}

func (retired codexAppServerRetiredTransport) retire(reason string) {
	reason = strings.TrimSpace(reason)
	if reason == "" {
		reason = "codex app-server transport was superseded"
	}
	failure := codexAppServerTransportFailureEnvelope(reason)
	for _, pending := range retired.pending {
		pending.responseCh <- failure
	}
	for _, request := range retired.serverRequests {
		if request.cancel != nil {
			request.cancel()
		}
	}
	if retired.stdin != nil {
		logIgnoredCleanupError("failed to close superseded codex stdin", retired.stdin.Close())
	}
	if retired.translatedOutput != nil {
		retired.translatedOutput.close()
	}
	if retired.translatedLines != nil &&
		(retired.translatedOutput == nil || retired.translatedOutput.lines != retired.translatedLines) {
		close(retired.translatedLines)
	}
	retired.transportFailure.fail(reason)
}

func codexAppServerTransportFailureEnvelope(message string) appServerEnvelope {
	failureData, _ := json.Marshal(map[string]any{
		"kind":      "transport_ambiguous",
		"retryable": true,
	})
	return appServerEnvelope{
		Error: &appServerResponseError{
			Code:    codexAppServerTransportFailureCode,
			Message: strings.TrimSpace(message),
			Data:    failureData,
		},
	}
}

func (failure *codexAppServerTransportFailure) fail(reason string) {
	if failure == nil {
		return
	}
	reason = strings.TrimSpace(reason)
	failure.mu.Lock()
	if failure.closing {
		failure.mu.Unlock()
		return
	}
	if failure.reason == "" {
		failure.reason = reason
	}
	failure.mu.Unlock()
	failure.terminate()
}

func (failure *codexAppServerTransportFailure) terminate() {
	if failure == nil {
		return
	}
	failure.abortOnce.Do(func() {
		if failure.abort != nil {
			failure.abort()
		}
	})
}

func (failure *codexAppServerTransportFailure) beginShutdown() {
	if failure == nil {
		return
	}
	failure.mu.Lock()
	if failure.closing {
		failure.mu.Unlock()
		return
	}
	failure.closing = true
	if failure.shutdown == nil {
		failure.shutdown = make(chan struct{})
	}
	shutdown := failure.shutdown
	failure.mu.Unlock()
	close(shutdown)
}

func (failure *codexAppServerTransportFailure) shutdownSignal() <-chan struct{} {
	if failure == nil {
		return nil
	}
	failure.mu.Lock()
	if failure.shutdown == nil {
		failure.shutdown = make(chan struct{})
	}
	shutdown := failure.shutdown
	failure.mu.Unlock()
	return shutdown
}

func (failure *codexAppServerTransportFailure) wait(wait func() error) error {
	err := wait()
	if failure == nil {
		return err
	}
	failure.mu.Lock()
	reason := strings.TrimSpace(failure.reason)
	failure.mu.Unlock()
	if reason == "" {
		return err
	}
	if err == nil {
		return fmt.Errorf("codex app-server transport failed: %s", reason)
	}
	return fmt.Errorf("codex app-server transport failed: %s: %w", reason, err)
}

type codexAppServerRequestDispatchGuard func(*CodexAppServerDriver) bool

func (writer *lockedDriverWriter) Write(p []byte) (int, error) {
	if writer == nil || writer.driver == nil {
		return 0, io.ErrClosedPipe
	}
	return writer.driver.writeRawForGenerationAndWriter(writer.generation, writer.writer, p)
}

func (writer *lockedDriverWriter) Close() error {
	if writer == nil || writer.driver == nil {
		return nil
	}
	return writer.driver.closeRawStdinForGeneration(writer.generation, writer.writer)
}

// ---------------------------------------------------------------------------
// drivercontract.Driver interface
// ---------------------------------------------------------------------------

// Name returns the runtime name.

func (d *CodexAppServerDriver) request(ctx context.Context, method string, params any) (json.RawMessage, error) {
	return d.requestInternal(ctx, method, params, nil, false, nil)
}

// requestWithDispatchGuard is reserved for requests whose dispatch itself has
// a lifecycle side effect. The guard is checked once while registering the
// exact generation/writer and again at the physical write boundary. No state
// mutex is held across Write: cleanup must be able to detach and close that
// exact writer to unblock a wedged pipe.
func (d *CodexAppServerDriver) requestWithDispatchGuard(
	ctx context.Context,
	method string,
	params any,
	guard codexAppServerRequestDispatchGuard,
) (json.RawMessage, error) {
	return d.requestInternal(ctx, method, params, guard, true, nil)
}

// requestWithWriteBoundary invokes release immediately after the physical
// frame write finishes (successfully or not), before waiting for the JSON-RPC
// response. This lets the turn router fence lifecycle state selection through
// dispatch without blocking the stdout reader when Codex emits a lifecycle
// notification before its response.
func (d *CodexAppServerDriver) requestWithWriteBoundary(
	ctx context.Context,
	method string,
	params any,
	release func(),
) (json.RawMessage, error) {
	return d.requestInternal(ctx, method, params, nil, false, release)
}

func (d *CodexAppServerDriver) requestInternal(
	ctx context.Context,
	method string,
	params any,
	guard codexAppServerRequestDispatchGuard,
	guarded bool,
	releaseWriteBoundary func(),
) (json.RawMessage, error) {
	var releaseOnce sync.Once
	releaseBoundary := func() {
		if releaseWriteBoundary != nil {
			releaseOnce.Do(releaseWriteBoundary)
		}
	}
	defer releaseBoundary()

	if err := ctx.Err(); err != nil {
		return nil, err
	}

	// Register against one immutable generation/writer snapshot. Replacement is
	// allowed after this lock is released; the pre-write check below will then
	// reject a known-undispatched guarded request, while a write already in
	// progress is conservatively classified as ambiguous.
	d.mu.Lock()
	if err := ctx.Err(); err != nil {
		d.mu.Unlock()
		return nil, err
	}
	if guard != nil && !guard(d) {
		d.mu.Unlock()
		return nil, errCodexAppServerRequestDispatchRejected
	}
	if d.rawStdin == nil {
		d.mu.Unlock()
		return nil, codexAppServerTransportRequestError(method, io.ErrClosedPipe)
	}
	generation := d.launchGeneration
	writer := d.rawStdin
	requestID := d.nextRequestIDLocked()
	request := map[string]any{
		"jsonrpc": "2.0",
		"id":      requestID,
		"method":  method,
		"params":  params,
	}
	data, err := json.Marshal(request)
	if err != nil {
		d.mu.Unlock()
		return nil, fmt.Errorf("marshal %s request: %w", method, err)
	}
	responseCh := make(chan appServerEnvelope, 1)
	if d.pending == nil {
		d.pending = make(map[int]pendingAppServerRequest)
	}
	d.pending[requestID] = pendingAppServerRequest{generation: generation, responseCh: responseCh}
	payload := append(data, '\n')
	d.mu.Unlock()

	// Serialize frames without serializing cleanup. In particular, neither
	// d.mu nor transportGenerationMu is held while the pipe may block.
	d.transportWriteMu.Lock()
	d.mu.Lock()
	pending, registered := d.pending[requestID]
	registered = registered && pending.generation == generation && pending.responseCh == responseCh
	current := d.launchGeneration == generation && d.rawStdin == writer
	ctxErr := ctx.Err()
	guardAllowed := guard == nil || guard(d)
	if !registered || !current || ctxErr != nil || !guardAllowed {
		if registered {
			delete(d.pending, requestID)
		}
		d.mu.Unlock()
		d.transportWriteMu.Unlock()
		switch {
		case ctxErr != nil:
			return nil, ctxErr
		case guarded:
			return nil, errCodexAppServerRequestDispatchRejected
		default:
			return nil, codexAppServerTransportRequestError(method, io.ErrClosedPipe)
		}
	}
	d.mu.Unlock()

	n, writeErr := writer.Write(payload)
	d.transportWriteMu.Unlock()
	// Release turn lifecycle dispatch fencing before any response wait. The
	// read agentteam may need to apply turn/started or turn/completed before it can
	// reach the response frame that wakes this request.
	releaseBoundary()
	if writeErr != nil {
		d.unregisterPending(generation, requestID)
		return nil, codexAppServerTransportRequestError(method, writeErr)
	}
	if n != len(payload) {
		d.unregisterPending(generation, requestID)
		return nil, codexAppServerTransportRequestError(method, fmt.Errorf("%w: wrote %d of %d bytes", io.ErrShortWrite, n, len(payload)))
	}

	// A generation swap after the physical write cannot prove whether the old
	// app-server accepted the frame. Never turn that race into a definite
	// rejection or retry signal.
	d.mu.Lock()
	current = d.launchGeneration == generation && d.rawStdin == writer
	d.mu.Unlock()
	if !current {
		d.unregisterPending(generation, requestID)
		return nil, codexAppServerAmbiguousOutcomeError(method, "codex app-server transport changed after request dispatch", io.ErrClosedPipe)
	}

	select {
	case <-ctx.Done():
		d.unregisterPending(generation, requestID)
		return nil, codexAppServerAmbiguousOutcomeError(method, "codex app-server response wait ended after request dispatch", ctx.Err())
	case response := <-responseCh:
		if response.Error != nil {
			message := strings.TrimSpace(response.Error.Message)
			if message == "" {
				message = fmt.Sprintf("JSON-RPC error %d", response.Error.Code)
			}
			return nil, &appServerRequestError{
				Method:  method,
				Code:    response.Error.Code,
				Message: message,
				Data:    response.Error.Data,
			}
		}
		return response.Result, nil
	}
}

func (d *CodexAppServerDriver) notifyForGeneration(generation uint64, writer io.WriteCloser, method string, params any) error {
	request := map[string]any{
		"jsonrpc": "2.0",
		"method":  method,
		"params":  params,
	}
	data, err := json.Marshal(request)
	if err != nil {
		return fmt.Errorf("marshal %s notification: %w", method, err)
	}
	payload := append(data, '\n')
	if n, err := d.writeRawForGenerationAndWriter(generation, writer, payload); err != nil {
		return fmt.Errorf("write %s notification: %w", method, err)
	} else if n != len(payload) {
		return fmt.Errorf("write %s notification: %w: wrote %d of %d bytes", method, io.ErrShortWrite, n, len(payload))
	}
	return nil
}

func (d *CodexAppServerDriver) unregisterPending(generation uint64, requestID int) {
	d.mu.Lock()
	if pending, ok := d.pending[requestID]; ok && pending.generation == generation {
		delete(d.pending, requestID)
	}
	d.mu.Unlock()
}

// ---------------------------------------------------------------------------
// Read agentteam
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) readLoop(r io.Reader) {
	d.mu.Lock()
	generation := d.launchGeneration
	transportFailure := d.transportFailure
	d.mu.Unlock()
	d.readLoopForGeneration(r, generation, transportFailure)
}

func (d *CodexAppServerDriver) readLoopForGeneration(r io.Reader, generation uint64, transportFailure *codexAppServerTransportFailure) {
	reader := bufio.NewReader(r)

	for {
		rawLine, err := readCodexAppServerLine(reader)
		if err != nil {
			if err == io.EOF {
				d.failTransport(generation, transportFailure, "codex app-server connection closed")
			} else {
				d.failTransport(generation, transportFailure, fmt.Sprintf("codex app-server stream error: %v", err))
			}
			return
		}

		line := strings.TrimSpace(rawLine)
		if line == "" {
			continue
		}

		err = d.handleAppServerLineForGeneration(line, generation)
		if err != nil {
			d.failTransport(generation, transportFailure, err.Error())
			return
		}
	}
}

// handleAppServerLine processes one complete JSON-RPC frame. A malformed
// frame is a transport-fatal protocol error, matching Codex's own remote
// app-server client: continuing would leave request ownership and lifecycle
// state unknowable.
func (d *CodexAppServerDriver) handleAppServerLine(line string) error {
	d.mu.Lock()
	generation := d.launchGeneration
	d.mu.Unlock()
	return d.handleAppServerLineForGeneration(line, generation)
}

func (d *CodexAppServerDriver) handleAppServerLineForGeneration(line string, generation uint64) error {
	var envelope appServerEnvelope
	if err := json.Unmarshal([]byte(line), &envelope); err != nil {
		return fmt.Errorf("codex app-server sent invalid JSON-RPC: %w", err)
	}

	// Primary lifecycle routing and turn request dispatch must be linearized at
	// the physical write boundary. Acquire that fence before the generation read
	// fence: launch replacement needs the generation write fence in order to
	// close a blocked old stdin write, so a reader must never hold generation
	// while it waits for the writer-owned primary fence.
	lifecycleFenceHeld := len(bytes.TrimSpace(envelope.ID)) == 0 &&
		appServerNotificationMutatesPrimaryTurnLifecycle(envelope.Method)
	if lifecycleFenceHeld {
		d.primaryTurnDispatchMu.Lock()
		defer d.primaryTurnDispatchMu.Unlock()
	}

	// The generation check and complete frame dispatch share a read fence with
	// launch-state reset. Without that fence a replacement launch can reset
	// request IDs between the check and pending-map lookup, letting an old
	// response satisfy a new generation's request with the same id.
	d.transportGenerationMu.RLock()
	defer d.transportGenerationMu.RUnlock()
	d.mu.Lock()
	currentGeneration := d.launchGeneration == generation
	d.mu.Unlock()
	if !currentGeneration {
		return errCodexAppServerTransportSuperseded
	}

	if len(bytes.TrimSpace(envelope.ID)) != 0 {
		// JSON-RPC request IDs are scoped independently in each direction. Codex
		// can therefore send a server request whose numeric ID equals one of our
		// outstanding client requests. A method-bearing frame is unambiguously a
		// server request and must be routed before consulting d.pending; otherwise
		// the collision consumes the client waiter and silently drops the tool.
		method := strings.TrimSpace(envelope.Method)
		if method != "" {
			d.startServerRequestForGeneration(generation, envelope.ID, method, envelope.Params)
			return nil
		}

		if requestID, ok := parseJSONRPCID(envelope.ID); ok {
			d.mu.Lock()
			pending, exists := d.pending[requestID]
			if exists && pending.generation == generation {
				delete(d.pending, requestID)
			} else {
				exists = false
			}
			d.mu.Unlock()
			if exists {
				pending.responseCh <- envelope
				return nil
			}
		}

		logUnknownAppServerResponse(envelope.ID, envelope.Error)
		return nil
	}
	if strings.TrimSpace(envelope.Method) == "" {
		return errors.New("codex app-server sent invalid JSON-RPC: frame has neither id nor method")
	}

	d.handleNotificationWithLifecycleFence(envelope, lifecycleFenceHeld)
	if err := d.translatedOutputFailureForGeneration(generation); err != nil {
		return err
	}
	return nil
}

func (d *CodexAppServerDriver) startServerRequestForGeneration(generation uint64, id json.RawMessage, method string, params json.RawMessage) {
	method = strings.TrimSpace(method)
	requestKey := normalizedServerRequestID(id)

	ctx, cancel := context.WithTimeout(context.Background(), codexAppServerRequestTimeout)
	if requestKey != "" {
		logContext := parseServerRequestLogContext(params)
		threadID := logContext.ThreadID
		if strings.TrimSpace(threadID) == "" {
			if ref, ok := d.resolveToolActivity(logContext.CallID, requestKey); ok {
				threadID = ref.threadID
			}
		}
		d.registerServerRequest(requestKey, inFlightServerRequest{
			cancel:     cancel,
			method:     method,
			threadID:   strings.TrimSpace(threadID),
			generation: generation,
		})
	}

	go d.handleServerRequest(ctx, cancel, generation, requestKey, id, method, params)
}

func (d *CodexAppServerDriver) handleServerRequest(
	ctx context.Context,
	cancel context.CancelFunc,
	generation uint64,
	requestKey string,
	id json.RawMessage,
	method string,
	params json.RawMessage,
) {
	defer cancel()

	result, rpcErr := d.executeServerRequestWithActivityForGeneration(ctx, generation, requestKey, strings.TrimSpace(method), params)
	runtimeDeliveryReceipts, runtimeDeliverySession := codexRuntimeDeliveryCommit(result)
	if requestKey != "" {
		if _, ok := d.finishServerRequestForGeneration(requestKey, generation); !ok {
			slog.Debug("codex-app-server server request resolved before response",
				"driver", "codex-app-server",
				"method", strings.TrimSpace(method),
				"request_id", requestKey)
			d.emitRuntimeDeliveryUncertainForGeneration(generation, runtimeDeliveryReceipts, runtimeDeliverySession)
			return
		}
	}
	if rpcErr != nil {
		d.emitRuntimeDeliveryUncertainForGeneration(generation, runtimeDeliveryReceipts, runtimeDeliverySession)
		if errors.Is(ctx.Err(), context.Canceled) {
			slog.Debug("codex-app-server server request canceled",
				"driver", "codex-app-server",
				"method", strings.TrimSpace(method),
				"request_id", formatJSONRPCIDForLog(id))
			return
		}
		attrs := []any{
			"driver", "codex-app-server",
			"method", strings.TrimSpace(method),
			"request_id", formatJSONRPCIDForLog(id),
			"error", strings.TrimSpace(rpcErr.Message),
		}
		logParams := parseServerRequestLogContext(params)
		if logParams.ThreadID != "" {
			attrs = append(attrs, "thread_id", logParams.ThreadID)
		}
		if logParams.Server != "" {
			attrs = append(attrs, "server", logParams.Server)
		}
		if logParams.Tool != "" {
			attrs = append(attrs, "tool", logParams.Tool)
		}
		slog.Warn("codex-app-server server request failed", attrs...)
		if err := d.writeJSONRPCError(generation, id, rpcErr); err != nil {
			logCodexAppServerWriteServerRequestFailure(
				ctx,
				"codex-app-server write server request error failed",
				method,
				id,
				err,
			)
			// Once the server request has been removed, a partial/missing error
			// response would leave Codex waiting forever on an otherwise live
			// transport. Retire it exactly like a failed success-result write so
			// the generation cannot continue with unknowable request ownership.
			d.failRuntimeDeliveryResponseTransport(generation, err)
		}
		return
	}
	if err := d.writeJSONRPCResult(generation, id, result); err != nil {
		d.emitRuntimeDeliveryUncertainForGeneration(generation, runtimeDeliveryReceipts, runtimeDeliverySession)
		logCodexAppServerWriteServerRequestFailure(
			ctx,
			"codex-app-server write server request result failed",
			method,
			id,
			err,
		)
		d.failRuntimeDeliveryResponseTransport(generation, err)
		return
	}
	if err := d.emitRuntimeDeliveryDeliveredForGeneration(generation, runtimeDeliveryReceipts, runtimeDeliverySession); err != nil {
		// The child has the complete response, but Process did not receive the
		// commit event. Retire the transport so durable thread/read
		// reconciliation proves the marker rather than losing the claim.
		d.emitRuntimeDeliveryUncertainForGeneration(generation, runtimeDeliveryReceipts, runtimeDeliverySession)
		d.failRuntimeDeliveryResponseTransport(generation, err)
	}
}

func codexRuntimeDeliveryCommit(result any) ([]string, string) {
	switch typed := result.(type) {
	case appServerMCPToolCallResult:
		return append([]string(nil), typed.runtimeDeliveryReceipts...), strings.TrimSpace(typed.runtimeDeliverySession)
	case *appServerMCPToolCallResult:
		if typed != nil {
			return append([]string(nil), typed.runtimeDeliveryReceipts...), strings.TrimSpace(typed.runtimeDeliverySession)
		}
	case appServerDynamicToolCallResponse:
		return append([]string(nil), typed.runtimeDeliveryReceipts...), strings.TrimSpace(typed.runtimeDeliverySession)
	case *appServerDynamicToolCallResponse:
		if typed != nil {
			return append([]string(nil), typed.runtimeDeliveryReceipts...), strings.TrimSpace(typed.runtimeDeliverySession)
		}
	}
	return nil, ""
}

func (d *CodexAppServerDriver) emitRuntimeDeliveryDeliveredForGeneration(generation uint64, receipts []string, sessionID string) error {
	if len(receipts) == 0 {
		return nil
	}
	var emitErr error
	if !d.withLaunchGeneration(generation, func() {
		for _, receipt := range receipts {
			receipt = strings.TrimSpace(receipt)
			if !types.IsRuntimeDeliveryReceipt(receipt) {
				continue
			}
			if err := d.enqueueTranslatedEvent(translatedEvent{
				Kind:           consts.ParsedEventKindRuntimeInputAcked,
				RuntimeInputID: receipt,
				SessionID:      strings.TrimSpace(sessionID),
			}); err != nil {
				emitErr = err
				return
			}
		}
	}) {
		return errCodexAppServerTransportSuperseded
	}
	return emitErr
}

func (d *CodexAppServerDriver) emitRuntimeDeliveryUncertainForGeneration(generation uint64, receipts []string, sessionID string) {
	if len(receipts) == 0 {
		return
	}
	d.withLaunchGeneration(generation, func() {
		for _, receipt := range receipts {
			receipt = strings.TrimSpace(receipt)
			if !types.IsRuntimeDeliveryReceipt(receipt) {
				continue
			}
			if err := d.enqueueTranslatedEvent(translatedEvent{
				Kind:           consts.ParsedEventKindRuntimeInputUncertain,
				RuntimeInputID: receipt,
				SessionID:      strings.TrimSpace(sessionID),
			}); err != nil && !errors.Is(err, errCodexTranslatedOutputSuperseded) {
				slog.Warn("codex-app-server runtime delivery uncertainty enqueue failed",
					"driver", "codex-app-server",
					"runtime_input_id", receipt,
					"error", err)
			}
		}
	})
}

func (d *CodexAppServerDriver) failRuntimeDeliveryResponseTransport(generation uint64, cause error) {
	d.mu.Lock()
	var failure *codexAppServerTransportFailure
	if d.launchGeneration == generation {
		failure = d.transportFailure
	}
	d.mu.Unlock()
	if failure == nil {
		return
	}
	d.failTransport(generation, failure, fmt.Sprintf("codex app-server final tool response delivery failed: %v", cause))
}

func logCodexAppServerWriteServerRequestFailure(ctx context.Context, message, method string, id json.RawMessage, err error) {
	attrs := []any{
		"driver", "codex-app-server",
		"method", strings.TrimSpace(method),
		"request_id", formatJSONRPCIDForLog(id),
		"error", err,
	}
	if errors.Is(err, io.ErrClosedPipe) {
		slog.DebugContext(ctx, message, attrs...)
		return
	}
	slog.WarnContext(ctx, message, attrs...)
}

func (d *CodexAppServerDriver) writeJSONRPCResult(generation uint64, id json.RawMessage, result any) error {
	payload := map[string]any{
		"jsonrpc": "2.0",
		"id":      json.RawMessage(bytes.TrimSpace(id)),
		"result":  result,
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal codex-app-server result: %w", err)
	}
	wirePayload := append(data, '\n')
	n, err := d.writeRawForGeneration(generation, wirePayload)
	if err != nil {
		return fmt.Errorf("write codex-app-server result: %w", err)
	}
	if n != len(wirePayload) {
		return fmt.Errorf("write codex-app-server result: %w: wrote %d of %d bytes", io.ErrShortWrite, n, len(wirePayload))
	}
	return nil
}

func (d *CodexAppServerDriver) writeJSONRPCError(generation uint64, id json.RawMessage, rpcErr *appServerResponseError) error {
	payload := map[string]any{
		"jsonrpc": "2.0",
		"id":      json.RawMessage(bytes.TrimSpace(id)),
		"error":   rpcErr,
	}
	data, err := json.Marshal(payload)
	if err != nil {
		return fmt.Errorf("marshal codex-app-server error: %w", err)
	}
	wirePayload := append(data, '\n')
	n, err := d.writeRawForGeneration(generation, wirePayload)
	if err != nil {
		return fmt.Errorf("write codex-app-server error: %w", err)
	}
	if n != len(wirePayload) {
		return fmt.Errorf("write codex-app-server error: %w: wrote %d of %d bytes", io.ErrShortWrite, n, len(wirePayload))
	}
	return nil
}

func logUnknownAppServerResponse(id json.RawMessage, rpcErr *appServerResponseError) {
	requestID := formatJSONRPCIDForLog(id)
	if rpcErr != nil {
		slog.Warn("response for unknown request",
			"driver", "codex-app-server",
			"request_id", requestID,
			"error", strings.TrimSpace(rpcErr.Message))
		return
	}
	slog.Debug("response for unknown request",
		"driver", "codex-app-server",
		"request_id", requestID)
}

func formatJSONRPCIDForLog(raw json.RawMessage) string {
	trimmed := strings.TrimSpace(string(bytes.TrimSpace(raw)))
	if trimmed == "" {
		return ""
	}
	var asString string
	if err := json.Unmarshal([]byte(trimmed), &asString); err == nil {
		return strings.TrimSpace(asString)
	}
	return trimmed
}

func normalizedServerRequestID(raw json.RawMessage) string {
	return strings.TrimSpace(formatJSONRPCIDForLog(raw))
}

// ---------------------------------------------------------------------------
// Notification handler
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) flushTranslatedLines(lines <-chan string, writer io.Writer) {
	for line := range lines {
		if _, err := io.WriteString(writer, line+"\n"); err != nil {
			break
		}
	}
	if closer, ok := writer.(io.Closer); ok {
		logIgnoredCleanupError("failed to close codex translated output writer", closer.Close())
	}
}

func (d *CodexAppServerDriver) flushTranslatedOutput(output *codexTranslatedOutput, writer io.Writer) {
	if output == nil {
		if closer, ok := writer.(io.Closer); ok {
			logIgnoredCleanupError("failed to close codex translated output writer", closer.Close())
		}
		return
	}
	defer output.finishFlush()
	for line := range output.lines {
		wireBytes := len(line) + 1
		n, err := io.WriteString(writer, line+"\n")
		if err == nil && n != wireBytes {
			err = fmt.Errorf("%w: wrote %d of %d translated output bytes", io.ErrShortWrite, n, wireBytes)
		}
		if err != nil {
			// Replacement and cleanup deliberately close the generation's writer
			// to release a no-consumer io.Pipe Write. That is cancellation, not a
			// second transport failure.
			if !output.isAborted() {
				output.fail(fmt.Errorf("codex translated output write failed: %w", err))
			}
			output.release(wireBytes)
			break
		}
		output.release(wireBytes)
		// A source-side EOF aborts producers before sealing the queue, but every
		// line admitted before that boundary still has to drain. Only explicit
		// cancellation (replacement, cleanup, or output failure) cuts the drain
		// short after a successful write.
		if output.isCanceled() {
			break
		}
	}
	output.closeWriter()
}

func (d *CodexAppServerDriver) enqueueTranslatedEvent(event translatedEvent) error {
	d.mu.Lock()
	lines := d.translatedLines
	output := d.translatedOutput
	d.mu.Unlock()
	if lines == nil {
		return errCodexTranslatedOutputSuperseded
	}

	line, wireBytes, err := encodeCodexTranslatedEvent(event)
	if err != nil {
		if output != nil && output.lines == lines {
			return output.fail(fmt.Errorf("encode codex translated output: %w", err))
		}
		return fmt.Errorf("encode codex translated output: %w", err)
	}
	if output != nil && output.lines == lines {
		if err := output.enqueue(line, wireBytes); err != nil {
			if !errors.Is(err, errCodexTranslatedOutputSuperseded) {
				slog.Warn("codex translated output enqueue failed", "driver", "codex-app-server", "error", err)
			}
			return err
		}
		return nil
	}
	if !sendCodexAppServerTranslatedLine(lines, line) {
		slog.Debug("dropping translated event after translated output closed", "driver", "codex-app-server")
		return errCodexTranslatedOutputSuperseded
	}
	return nil
}

func sendCodexAppServerTranslatedLine(lines chan string, line string) (sent bool) {
	if lines == nil {
		return false
	}
	defer func() {
		if recover() != nil {
			sent = false
		}
	}()
	lines <- line
	return true
}

func (d *CodexAppServerDriver) closeTranslatedLinesForGeneration(generation uint64, lines chan string) {
	if lines == nil {
		return
	}
	d.mu.Lock()
	var output *codexTranslatedOutput
	if d.launchGeneration == generation && d.translatedLines == lines &&
		d.translatedOutput != nil && d.translatedOutput.lines == lines {
		output = d.translatedOutput
	}
	d.mu.Unlock()
	if output != nil {
		output.abort()
	}

	d.transportGenerationMu.Lock()
	d.mu.Lock()
	if d.launchGeneration != generation || d.translatedLines != lines {
		d.mu.Unlock()
		d.transportGenerationMu.Unlock()
		return
	}
	d.translatedLines = nil
	d.translatedOutput = nil
	d.mu.Unlock()
	if output != nil {
		output.close()
	}
	if output == nil || output.lines != lines {
		close(lines)
	}
	d.transportGenerationMu.Unlock()
}

// failTransport atomically retires the current app-server transport. Pending
// client requests receive a distinct ambiguous-outcome error, server requests
// are canceled, and both stdio streams are closed before the bound child is
// terminated. The generation fence prevents a stale reader from tearing down a
// newer launch that reused the same driver instance.
func (d *CodexAppServerDriver) failTransport(generation uint64, sourceFailure *codexAppServerTransportFailure, message string) {
	message = strings.TrimSpace(message)
	if message == "" {
		message = "codex app-server transport failed"
	}

	// Abort before taking the generation write fence. A lossless enqueue may be
	// waiting on a full old-generation queue while holding the read fence.
	d.mu.Lock()
	var outputToAbort *codexTranslatedOutput
	if d.launchGeneration == generation {
		outputToAbort = d.translatedOutput
	}
	d.mu.Unlock()
	if outputToAbort != nil {
		outputToAbort.abort()
	}

	d.transportGenerationMu.Lock()
	d.mu.Lock()
	if d.launchGeneration != generation {
		d.mu.Unlock()
		d.transportGenerationMu.Unlock()
		// The old reader still owns its old child. Retire that child without
		// touching the newer generation's pending map or stdio streams.
		sourceFailure.fail(message)
		return
	}
	pending := d.pending
	d.pending = make(map[int]pendingAppServerRequest)
	serverRequests := d.serverRequests
	d.serverRequests = make(map[string]inFlightServerRequest)
	stdin := d.rawStdin
	d.rawStdin = nil
	lines := d.translatedLines
	d.translatedLines = nil
	output := d.translatedOutput
	d.translatedOutput = nil
	transportFailure := d.transportFailure
	d.transportFailure = nil
	d.mu.Unlock()

	failure := codexAppServerTransportFailureEnvelope(message)
	for _, request := range pending {
		if request.generation == generation {
			request.responseCh <- failure
		}
	}
	for _, request := range serverRequests {
		if request.cancel != nil {
			request.cancel()
		}
	}
	if stdin != nil {
		logIgnoredCleanupError("failed to close codex stdin after transport failure", stdin.Close())
	}
	if output != nil {
		// Source EOF/protocol failure must not cut off terminal or transcript
		// lines already admitted before the failing frame. Stop producers, seal
		// the queue, and let the flusher drain in order. Replacement and explicit
		// cleanup call output.cancel separately when they must break a no-consumer
		// pipe immediately.
		output.seal()
	}
	if lines != nil && (output == nil || output.lines != lines) {
		close(lines)
	}
	if transportFailure != nil {
		transportFailure.fail(message)
	} else {
		sourceFailure.fail(message)
	}
	d.transportGenerationMu.Unlock()
}

func (d *CodexAppServerDriver) translatedOutputFailureForGeneration(generation uint64) error {
	d.mu.Lock()
	if d.launchGeneration != generation {
		d.mu.Unlock()
		return errCodexAppServerTransportSuperseded
	}
	output := d.translatedOutput
	lines := d.translatedLines
	d.mu.Unlock()
	if output == nil || output.lines != lines {
		return nil
	}
	return output.fatalError()
}

func (d *CodexAppServerDriver) registerServerRequest(requestKey string, request inFlightServerRequest) {
	if strings.TrimSpace(requestKey) == "" {
		return
	}
	d.mu.Lock()
	if d.serverRequests == nil {
		d.serverRequests = make(map[string]inFlightServerRequest)
	}
	if existing, ok := d.serverRequests[requestKey]; ok && existing.cancel != nil {
		existing.cancel()
	}
	d.serverRequests[requestKey] = request
	d.mu.Unlock()
}

func (d *CodexAppServerDriver) finishServerRequest(requestKey string) (inFlightServerRequest, bool) {
	d.mu.Lock()
	generation := d.launchGeneration
	d.mu.Unlock()
	return d.finishServerRequestForGeneration(requestKey, generation)
}

func (d *CodexAppServerDriver) finishServerRequestForGeneration(requestKey string, generation uint64) (inFlightServerRequest, bool) {
	if strings.TrimSpace(requestKey) == "" {
		return inFlightServerRequest{}, false
	}
	d.mu.Lock()
	request, ok := d.serverRequests[requestKey]
	if ok && request.generation == generation {
		delete(d.serverRequests, requestKey)
	} else {
		ok = false
	}
	d.mu.Unlock()
	return request, ok
}

func (d *CodexAppServerDriver) resolveServerRequest(requestKey string) bool {
	request, ok := d.finishServerRequest(requestKey)
	if ok && request.cancel != nil {
		request.cancel()
	}
	return ok
}

// ---------------------------------------------------------------------------
// Raw I/O helpers
// ---------------------------------------------------------------------------

func (d *CodexAppServerDriver) writeRaw(data []byte) (int, error) {
	d.mu.Lock()
	generation := d.launchGeneration
	writer := d.rawStdin
	d.mu.Unlock()
	if writer == nil {
		return 0, io.ErrClosedPipe
	}
	return d.writeRawForGenerationAndWriter(generation, writer, data)
}

func (d *CodexAppServerDriver) writeRawForGeneration(generation uint64, data []byte) (int, error) {
	return d.writeRawForGenerationAndWriter(generation, nil, data)
}

func (d *CodexAppServerDriver) writeRawForGenerationAndWriter(generation uint64, writer io.WriteCloser, data []byte) (int, error) {
	// Snapshot and revalidate on both sides of the serialized physical write,
	// but never hold d.mu while Write may block. Cleanup intentionally does not
	// take transportWriteMu, so closing the exact writer can break a wedged pipe.
	d.mu.Lock()
	if d.launchGeneration != generation || d.rawStdin == nil || (writer != nil && d.rawStdin != writer) {
		d.mu.Unlock()
		return 0, io.ErrClosedPipe
	}
	if writer == nil {
		writer = d.rawStdin
	}
	d.mu.Unlock()

	d.transportWriteMu.Lock()
	d.mu.Lock()
	current := d.launchGeneration == generation && d.rawStdin == writer
	d.mu.Unlock()
	if !current {
		d.transportWriteMu.Unlock()
		return 0, io.ErrClosedPipe
	}
	n, err := writer.Write(data)
	d.transportWriteMu.Unlock()

	d.mu.Lock()
	current = d.launchGeneration == generation && d.rawStdin == writer
	d.mu.Unlock()
	if !current && err == nil {
		return n, io.ErrClosedPipe
	}
	return n, err
}

// withLaunchGeneration executes one bounded state transition while launch
// replacement is excluded. Generation zero is reserved for direct unit-level
// server-request execution that is not attached to a transport reader.
func (d *CodexAppServerDriver) withLaunchGeneration(generation uint64, fn func()) bool {
	if generation == 0 {
		fn()
		return true
	}
	d.transportGenerationMu.RLock()
	d.mu.Lock()
	current := d.launchGeneration == generation
	d.mu.Unlock()
	if current {
		fn()
	}
	d.transportGenerationMu.RUnlock()
	return current
}

func (d *CodexAppServerDriver) closeRawStdin() error {
	d.mu.Lock()
	writer := d.rawStdin
	d.rawStdin = nil
	d.mu.Unlock()
	if writer == nil {
		return nil
	}
	return writer.Close()
}

func (d *CodexAppServerDriver) closeRawStdinForGeneration(generation uint64, writer io.WriteCloser) error {
	if writer == nil {
		return nil
	}
	d.mu.Lock()
	if d.launchGeneration == generation && d.rawStdin == writer {
		d.rawStdin = nil
	}
	d.mu.Unlock()
	return writer.Close()
}

func (d *CodexAppServerDriver) nextRequestIDLocked() int {
	d.requestID++
	return d.requestID
}

func readCodexAppServerLine(reader *bufio.Reader) (string, error) {
	return readLineWithLimit(reader, codexAppServerMaxMessageBytes)
}

func unmarshalAppServerStatusType(data []byte) (string, bool, error) {
	trimmed := bytes.TrimSpace(data)
	if len(trimmed) == 0 || bytes.Equal(trimmed, []byte("null")) {
		return "", true, nil
	}

	var statusType string
	if err := json.Unmarshal(trimmed, &statusType); err == nil {
		return strings.TrimSpace(statusType), true, nil
	}

	return "", false, nil
}