package agent

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/google/uuid"

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

const (
	kimiWireProtocolVersion     = "1.10"
	kimiWireRequestTimeout      = 300 * time.Second
	kimiWireTransportClosedCode = -32098
)

// KimiWireDriver drives one long-lived Kimi Code Wire session over JSON-RPC
// stdio. Kimi has one protocol path rather than a legacy compatibility branch.
type KimiWireDriver struct {
	mu      sync.Mutex
	writeMu sync.Mutex

	stdin  io.WriteCloser
	lines  chan string
	output *kimiTranslatedOutput

	nextID  int
	pending map[string]chan kimiWireEnvelope

	sessionID string
	launchCfg drivercontract.LaunchConfig
	toolNames map[string]string
	// countedStatusUpdates dedupes StatusUpdate token usage by session+message_id;
	// kimi-cli emits one per step with a stable message_id (result.id), so Wire/SSE
	// replay would otherwise double-count.
	countedStatusUpdates map[string]struct{}
	contextCompacted     bool
	completedTurns       int
	turnActive           bool
	turnTerminal         chan struct{}
	promptInFlight       bool
	promptOutcome        kimiPromptOutcome
	wireTurnEnded        bool
	turnEndEmitted       bool
	transportClosed      bool
	runtimeCtx           context.Context
	promptInputID        string
	promptAccepted       chan error
	steerWaiters         map[string]chan error

	requestFuncForTest func(context.Context, string, any) (json.RawMessage, error)
}

type kimiWireEnvelope struct {
	JSONRPC string                 `json:"jsonrpc,omitempty"`
	ID      json.RawMessage        `json:"id,omitempty"`
	Method  string                 `json:"method,omitempty"`
	Params  json.RawMessage        `json:"params,omitempty"`
	Result  json.RawMessage        `json:"result,omitempty"`
	Error   *kimiWireResponseError `json:"error,omitempty"`
}

type kimiWireResponseError struct {
	Code    int             `json:"code"`
	Message string          `json:"message"`
	Data    json.RawMessage `json:"data,omitempty"`
}

type kimiPromptOutcome struct {
	ready      bool
	errMessage string
	stopReason string
}

var errKimiWireTransportAmbiguous = errors.New("kimi Wire transport outcome is ambiguous")

func (e *kimiWireResponseError) Error() string {
	if e == nil {
		return "Kimi Wire error"
	}
	if message := strings.TrimSpace(e.Message); message != "" {
		return message
	}
	return fmt.Sprintf("JSON-RPC error %d", e.Code)
}

func (d *KimiWireDriver) Name() string { return "kimi" }

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

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

func (d *KimiWireDriver) UsesExplicitTurnLifecycle() bool { return true }

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

func (d *KimiWireDriver) Spawn(ctx context.Context, cfg drivercontract.LaunchConfig) (drivercontract.SpawnResult, error) {
	carrier, err := cfg.RequireCompiledPrompt()
	if err != nil {
		return drivercontract.SpawnResult{}, err
	}
	cfg.CompiledPrompt = carrier
	if err := materializeContractCarrier(cfg); err != nil {
		return drivercontract.SpawnResult{}, fmt.Errorf("kimi: materialize contract carrier: %w", err)
	}

	sessionID, resumed, err := resolveKimiLaunchSession(cfg)
	if err != nil {
		return drivercontract.SpawnResult{}, err
	}
	cfg.SessionID = sessionID
	prepared, err := prepareKimiLaunch(cfg, sessionID)
	if err != nil {
		return drivercontract.SpawnResult{}, err
	}

	translatedReader, translatedWriter := io.Pipe()
	output := newKimiTranslatedOutput(512)
	output.attachWriter(translatedWriter)
	lines := output.lines
	readerDone := make(chan struct{})
	go func() {
		defer close(readerDone)
		defer translatedWriter.Close()
		for line := range lines {
			if _, err := io.WriteString(translatedWriter, line+"\n"); err != nil {
				return
			}
		}
	}()

	cmd, stdin, stdout, stderr, terminate, wait, err := startKimiWireProcess(ctx, cfg, prepared)
	if err != nil {
		prepared.Cleanup()
		output.close()
		_ = translatedReader.Close()
		return drivercontract.SpawnResult{}, err
	}

	d.mu.Lock()
	d.stdin = stdin
	d.lines = lines
	d.output = output
	d.pending = make(map[string]chan kimiWireEnvelope)
	d.sessionID = sessionID
	if prepared.ShareDir != "" {
		if cfg.Env == nil {
			cfg.Env = make(map[string]string)
		}
		cfg.Env["KIMI_SHARE_DIR"] = prepared.ShareDir
	}
	d.launchCfg = cloneKimiLaunchConfig(cfg)
	d.toolNames = make(map[string]string)
	d.countedStatusUpdates = make(map[string]struct{})
	d.contextCompacted = false
	d.completedTurns = 0
	d.turnActive = false
	d.turnTerminal = closedKimiSignal()
	d.promptInFlight = false
	d.promptOutcome = kimiPromptOutcome{}
	d.wireTurnEnded = false
	d.turnEndEmitted = false
	d.transportClosed = false
	d.runtimeCtx = ctx
	d.promptInputID = ""
	d.promptAccepted = nil
	d.steerWaiters = make(map[string]chan error)
	d.mu.Unlock()
	if completedTurns, countErr := kimiCompletedTurnCount(cfg, sessionID); countErr == nil {
		d.mu.Lock()
		d.completedTurns = completedTurns
		d.mu.Unlock()
	} else {
		_ = stdin.Close()
		_ = stdout.Close()
		terminate()
		prepared.Cleanup()
		output.close()
		_ = translatedReader.Close()
		return drivercontract.SpawnResult{}, fmt.Errorf("kimi: inspect session history: %w", countErr)
	}

	cleanupOnce := sync.Once{}
	cleanup := func() {
		cleanupOnce.Do(func() {
			// Abort translated-output backpressure before transport teardown.
			// Process.waitProcess/onExit is authoritative for an unexpected
			// process exit; an intentional Stop already rejects ordinary parsed
			// events. Therefore closeTransport's synthetic terminal projection
			// may be dropped here rather than deadlocking retirement.
			output.close()
			d.closeTransport()
			terminate()
		})
	}
	preparedCleanupOnce := sync.Once{}
	cleanupPrepared := func() {
		preparedCleanupOnce.Do(prepared.Cleanup)
	}
	go d.readLoop(stdout, cleanup)

	initializeResult, err := d.request(ctx, "initialize", map[string]any{
		"protocol_version": kimiWireProtocolVersion,
		"client":           map[string]any{"name": "agentteam", "version": "0.0.0"},
		"capabilities": map[string]any{
			"supports_question":  false,
			"supports_plan_mode": false,
		},
	})
	if err != nil {
		cleanup()
		_ = wait()
		cleanupPrepared()
		return drivercontract.SpawnResult{}, fmt.Errorf("kimi: initialize Wire session: %w", err)
	}
	if err := validateKimiInitializeResult(initializeResult); err != nil {
		cleanup()
		_ = wait()
		cleanupPrepared()
		return drivercontract.SpawnResult{}, err
	}

	d.emit(translatedEvent{Kind: consts.ParsedEventKindSessionInit, SessionID: sessionID})
	preparedRecoveryPoint := prepareSpawnRecoveryPoint(d, cfg.InitialDeliveryMessageID, sessionID)
	initialPrompt := kimiInitialPrompt(cfg, resumed)
	if strings.TrimSpace(initialPrompt) != "" || len(cfg.PromptAttachments) > 0 {
		go func() {
			input := drivercontract.RuntimeInput{
				ID:             cfg.InitialDeliveryMessageID,
				Resume:         resumed,
				CompiledPrompt: carrier,
				Attachments:    cfg.PromptAttachments,
			}
			if err := d.DeliverRuntimeInput(ctx, input, sessionID); err != nil {
				slog.Error("kimi: initial Wire prompt failed", "agent_id", cfg.AgentID, "error", err)
			}
		}()
	}

	return drivercontract.SpawnResult{
		Cmd:        cmd,
		Stdin:      stdin,
		Stdout:     translatedReader,
		StderrTail: stderr.Tail,
		Terminate:  cleanup,
		Wait: func() error {
			err := wait()
			cleanup()
			<-readerDone
			cleanupPrepared()
			return err
		},
		Cleanup:               cleanup,
		PreparedRecoveryPoint: preparedRecoveryPoint,
	}, nil
}

func startKimiWireProcess(
	ctx context.Context,
	cfg drivercontract.LaunchConfig,
	prepared kimiPreparedLaunch,
) (*exec.Cmd, io.WriteCloser, io.ReadCloser, *capturingPrefixWriter, func(), func() error, error) {
	binary := strings.TrimSpace(cfg.Binary)
	if binary == "" {
		binary = "kimi"
	}
	command, args := runtimeconfig.RuntimeCommand(runtimeconfig.TypeKimi, binary, prepared.Args)
	cmd := exec.CommandContext(ctx, command, args...)
	cmd.Dir = prepared.WorkDir
	cmd.Env = prepared.Env
	stderr := newCapturingPrefixWriter(fmt.Sprintf("[kimi %s] ", cfg.AgentID), 32*1024)
	cmd.Stderr = stderr
	configureRuntimeCommand(cmd)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, nil, nil, nil, nil, nil, fmt.Errorf("kimi: get stdout pipe: %w", err)
	}
	stdin, err := cmd.StdinPipe()
	if err != nil {
		_ = stdout.Close()
		return nil, nil, nil, nil, nil, nil, fmt.Errorf("kimi: get stdin pipe: %w", err)
	}
	if err := cmd.Start(); err != nil {
		_ = stdout.Close()
		_ = stdin.Close()
		return nil, nil, nil, nil, nil, nil, fmt.Errorf("kimi: start Wire process: %w", err)
	}
	terminateOnce := sync.Once{}
	terminate := func() {
		terminateOnce.Do(func() {
			_ = stdin.Close()
			_ = stdout.Close()
			terminateRuntimeCommand(cmd)
		})
	}
	return cmd, stdin, stdout, stderr, terminate, cmd.Wait, nil
}

func resolveKimiLaunchSession(cfg drivercontract.LaunchConfig) (string, bool, error) {
	sessionID := strings.TrimSpace(cfg.SessionID)
	if sessionID == "" {
		return uuid.NewString(), false, nil
	}
	if _, err := uuid.Parse(sessionID); err != nil {
		return "", false, fmt.Errorf("kimi: invalid session id %q", sessionID)
	}
	lookupCfg, err := kimiSessionLookupConfig(cfg)
	if err != nil {
		return "", false, err
	}
	sessionDir, err := kimiSessionDir(lookupCfg, sessionID, false)
	if err != nil {
		return "", false, err
	}
	info, statErr := os.Lstat(sessionDir)
	if statErr == nil {
		if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
			return "", false, fmt.Errorf("kimi: session path %q is not a safe directory", sessionDir)
		}
		if _, err := kimiSessionDir(lookupCfg, sessionID, true); err != nil {
			return "", false, err
		}
		if err := recoverKimiRecoveryTransaction(sessionDir); err != nil {
			return "", false, fmt.Errorf("kimi: recover interrupted rewind: %w", err)
		}
		if _, err := kimiCompletedTurnCount(lookupCfg, sessionID); errors.Is(err, errKimiIncompleteTurn) {
			return uuid.NewString(), false, nil
		} else if err != nil {
			return "", false, fmt.Errorf("kimi: inspect session history: %w", err)
		}
		return sessionID, true, nil
	}
	if !errors.Is(statErr, os.ErrNotExist) {
		return "", false, fmt.Errorf("kimi: inspect session %q: %w", sessionID, statErr)
	}
	return uuid.NewString(), false, nil
}

func kimiSessionLookupConfig(cfg drivercontract.LaunchConfig) (drivercontract.LaunchConfig, error) {
	if !cfg.DisableNativeTools {
		return cfg, nil
	}
	workDir, err := filepath.Abs(strings.TrimSpace(cfg.WorkDir))
	if err != nil || strings.TrimSpace(cfg.WorkDir) == "" {
		return drivercontract.LaunchConfig{}, fmt.Errorf("kimi: valid workdir is required")
	}
	cfg.Env = runtimeconfig.CloneStringMap(cfg.Env)
	if cfg.Env == nil {
		cfg.Env = make(map[string]string)
	}
	cfg.Env["KIMI_SHARE_DIR"] = filepath.Join(filepath.Dir(filepath.Clean(workDir)), "kimi-share")
	return cfg, nil
}

func (d *KimiWireDriver) request(ctx context.Context, method string, params any) (json.RawMessage, error) {
	if d.requestFuncForTest != nil {
		return d.requestFuncForTest(ctx, method, params)
	}
	if ctx == nil {
		ctx = context.Background()
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	requestCtx, cancel := context.WithTimeout(ctx, kimiWireRequestTimeout)
	defer cancel()

	d.mu.Lock()
	if d.stdin == nil || d.transportClosed {
		d.mu.Unlock()
		return nil, io.ErrClosedPipe
	}
	d.nextID++
	key := "agentteam-" + strconv.Itoa(d.nextID)
	responseCh := make(chan kimiWireEnvelope, 1)
	d.pending[key] = responseCh
	d.mu.Unlock()

	payload := map[string]any{"jsonrpc": "2.0", "id": key, "method": method}
	if params != nil {
		payload["params"] = params
	}
	data, err := json.Marshal(payload)
	if err != nil {
		d.forgetPending(key)
		return nil, err
	}
	data = append(data, '\n')
	if err := d.writeFrame(data); err != nil {
		d.forgetPending(key)
		if kimiMutatingWireMethod(method) {
			return nil, errors.Join(errKimiWireTransportAmbiguous, err)
		}
		return nil, err
	}

	select {
	case <-requestCtx.Done():
		d.forgetPending(key)
		if kimiMutatingWireMethod(method) {
			return nil, errors.Join(errKimiWireTransportAmbiguous, requestCtx.Err())
		}
		return nil, requestCtx.Err()
	case response := <-responseCh:
		if response.Error != nil {
			if kimiMutatingWireMethod(method) && response.Error.Code == kimiWireTransportClosedCode {
				return nil, errors.Join(errKimiWireTransportAmbiguous, response.Error)
			}
			return nil, response.Error
		}
		return response.Result, nil
	}
}

func kimiMutatingWireMethod(method string) bool {
	switch strings.TrimSpace(method) {
	case "prompt", "steer", "cancel":
		return true
	default:
		return false
	}
}

func (d *KimiWireDriver) writeFrame(data []byte) error {
	d.writeMu.Lock()
	defer d.writeMu.Unlock()
	d.mu.Lock()
	stdin := d.stdin
	closed := d.transportClosed
	d.mu.Unlock()
	if stdin == nil || closed {
		return io.ErrClosedPipe
	}
	_, err := stdin.Write(data)
	return err
}

func (d *KimiWireDriver) forgetPending(key string) {
	d.mu.Lock()
	delete(d.pending, key)
	d.mu.Unlock()
}

func (d *KimiWireDriver) readLoop(stdout io.Reader, cleanup func()) {
	defer func() {
		cleanup()
		d.failPending(fmt.Errorf("Kimi Wire transport closed"))
	}()
	scanner := bufio.NewScanner(stdout)
	scanner.Buffer(make([]byte, 64*1024), maxProcessLineBytes)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" {
			continue
		}
		var envelope kimiWireEnvelope
		if err := json.Unmarshal([]byte(line), &envelope); err != nil {
			slog.Warn("kimi: invalid Wire frame", "error", err)
			continue
		}
		d.routeEnvelope(envelope)
	}
	if err := scanner.Err(); err != nil && !isKimiTransportClosed(err) {
		slog.Warn("kimi: Wire read failed", "error", err)
	}
}

func (d *KimiWireDriver) routeEnvelope(envelope kimiWireEnvelope) {
	if len(envelope.ID) > 0 && envelope.Method == "" {
		key := normalizeKimiRPCID(envelope.ID)
		d.mu.Lock()
		ch := d.pending[key]
		delete(d.pending, key)
		d.mu.Unlock()
		if ch != nil {
			ch <- envelope
		}
		return
	}
	if envelope.Method == "event" {
		d.handleNotification(envelope.Method, envelope.Params)
		return
	}
	if len(envelope.ID) > 0 && envelope.Method != "" {
		d.handleServerRequest(envelope)
	}
}

func normalizeKimiRPCID(raw json.RawMessage) string {
	var stringID string
	if json.Unmarshal(raw, &stringID) == nil {
		return stringID
	}
	var number json.Number
	if json.Unmarshal(raw, &number) == nil {
		return number.String()
	}
	return strings.TrimSpace(string(raw))
}

func validateKimiInitializeResult(result json.RawMessage) error {
	// Wire 1.10 negotiates client interaction capabilities only. Session resume
	// is a CLI launch concern (`--session` plus on-disk history), not an
	// initialize-response capability in Kimi Code 1.49.
	var decoded struct {
		ProtocolVersion string `json:"protocol_version"`
		Server          struct {
			Name    string `json:"name"`
			Version string `json:"version"`
		} `json:"server"`
	}
	if err := json.Unmarshal(result, &decoded); err != nil {
		return fmt.Errorf("kimi: decode Wire initialize response: %w", err)
	}
	if strings.TrimSpace(decoded.ProtocolVersion) != kimiWireProtocolVersion {
		return fmt.Errorf(
			"kimi: unsupported Wire protocol %q from %s %s; require %s",
			decoded.ProtocolVersion,
			strings.TrimSpace(decoded.Server.Name),
			strings.TrimSpace(decoded.Server.Version),
			kimiWireProtocolVersion,
		)
	}
	if strings.TrimSpace(decoded.Server.Name) == "" || strings.TrimSpace(decoded.Server.Version) == "" {
		return fmt.Errorf("kimi: Wire initialize response is missing server identity")
	}
	return nil
}

func (d *KimiWireDriver) failPending(err error) {
	d.mu.Lock()
	pending := d.pending
	d.pending = make(map[string]chan kimiWireEnvelope)
	d.mu.Unlock()
	for _, ch := range pending {
		ch <- kimiWireEnvelope{Error: &kimiWireResponseError{Code: kimiWireTransportClosedCode, Message: err.Error()}}
	}
}

func (d *KimiWireDriver) closeTransport() {
	d.mu.Lock()
	if d.transportClosed {
		d.mu.Unlock()
		return
	}
	d.transportClosed = true
	stdin := d.stdin
	d.stdin = nil
	if d.promptInFlight || d.turnActive {
		d.wireTurnEnded = true
	}
	promptAccepted := d.promptAccepted
	d.promptAccepted = nil
	steerWaiters := d.steerWaiters
	d.steerWaiters = make(map[string]chan error)
	d.markTurnTerminalLocked()
	emitError, emitTurnEnd, sessionID := d.takePromptTerminalEventsLocked()
	d.mu.Unlock()
	if stdin != nil {
		_ = stdin.Close()
	}
	if promptAccepted != nil {
		promptAccepted <- errKimiWireTransportAmbiguous
	}
	for _, waiter := range steerWaiters {
		waiter <- errKimiWireTransportAmbiguous
	}
	if emitError != "" {
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindError,
			Message:   emitError,
			SessionID: sessionID,
			Metadata:  map[string]any{"will_retry": false},
		})
	}
	if emitTurnEnd != "" {
		d.emit(translatedEvent{Kind: consts.ParsedEventKindTurnEnd, SessionID: sessionID, StopReason: emitTurnEnd})
	}
}

func (d *KimiWireDriver) DeliverBusyStdinInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	return d.DeliverRuntimeInput(ctx, input, sessionID)
}

func (d *KimiWireDriver) DeliverRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	if input.Empty() {
		return nil
	}
	if err := d.validateSession(sessionID); err != nil {
		return err
	}
	userInput, err := kimiWireUserInput(input)
	if err != nil {
		return err
	}
	d.mu.Lock()
	if d.promptInFlight {
		d.mu.Unlock()
		return fmt.Errorf("kimi: prompt already in flight")
	}
	d.promptInFlight = true
	d.promptOutcome = kimiPromptOutcome{}
	d.wireTurnEnded = false
	d.turnEndEmitted = false
	d.turnTerminal = make(chan struct{})
	d.promptInputID = strings.TrimSpace(input.ID)
	accepted := make(chan error, 1)
	d.promptAccepted = accepted
	runtimeCtx := d.runtimeCtx
	if runtimeCtx == nil {
		runtimeCtx = context.Background()
	}
	d.mu.Unlock()
	go func() {
		result, err := d.request(runtimeCtx, "prompt", map[string]any{"user_input": userInput})
		if err != nil {
			d.finishPrompt(kimiPromptOutcome{ready: true, errMessage: err.Error(), stopReason: "error"})
			return
		}
		stopReason, err := validateKimiPromptResult(result)
		if err != nil {
			d.finishPrompt(kimiPromptOutcome{ready: true, errMessage: err.Error(), stopReason: "error"})
			return
		}
		d.finishPrompt(kimiPromptOutcome{ready: true, stopReason: stopReason})
	}()
	select {
	case <-ctx.Done():
		return errors.Join(errKimiWireTransportAmbiguous, ctx.Err())
	case err := <-accepted:
		return err
	}
}

func (d *KimiWireDriver) DeliverAckedBusyRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	if input.Empty() {
		return nil
	}
	if strings.TrimSpace(input.ID) == "" {
		return fmt.Errorf("kimi: correlated steer input id is required")
	}
	if err := d.validateSession(sessionID); err != nil {
		return err
	}
	userInput, err := kimiWireUserInput(input)
	if err != nil {
		return err
	}
	inputID := strings.TrimSpace(input.ID)
	applied := make(chan error, 1)
	d.mu.Lock()
	if d.steerWaiters == nil {
		d.steerWaiters = make(map[string]chan error)
	}
	if _, exists := d.steerWaiters[inputID]; exists {
		d.mu.Unlock()
		return fmt.Errorf("kimi: correlated steer %q is already pending", inputID)
	}
	d.steerWaiters[inputID] = applied
	d.mu.Unlock()
	defer func() {
		d.mu.Lock()
		if d.steerWaiters[inputID] == applied {
			delete(d.steerWaiters, inputID)
		}
		d.mu.Unlock()
	}()
	result, err := d.request(ctx, "steer", map[string]any{"user_input": userInput})
	if err != nil {
		return fmt.Errorf("kimi: steer: %w", err)
	}
	var status struct {
		Status string `json:"status"`
	}
	if json.Unmarshal(result, &status) != nil || status.Status != "steered" {
		return fmt.Errorf("kimi: unexpected steer result %s", strings.TrimSpace(string(result)))
	}
	select {
	case <-ctx.Done():
		return errors.Join(errKimiWireTransportAmbiguous, ctx.Err())
	case err := <-applied:
		return err
	}
}

func (d *KimiWireDriver) InterruptActiveRun(ctx context.Context, sessionID string) error {
	if err := d.validateSession(sessionID); err != nil {
		return err
	}
	d.mu.Lock()
	terminal := d.turnTerminal
	active := d.turnActive || d.promptInFlight
	d.mu.Unlock()
	if !active {
		return nil
	}
	if _, err := d.request(ctx, "cancel", map[string]any{}); err != nil {
		select {
		case <-terminal:
			return nil
		default:
		}
		return fmt.Errorf("kimi: cancel: %w", err)
	}
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-terminal:
		return nil
	}
}

func (d *KimiWireDriver) DeliverInterruptingRuntimeInput(ctx context.Context, input drivercontract.RuntimeInput, sessionID string) error {
	d.mu.Lock()
	terminal := d.turnTerminal
	active := d.turnActive || d.promptInFlight
	d.mu.Unlock()
	if active {
		if err := d.InterruptActiveRun(ctx, sessionID); err != nil {
			return err
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-terminal:
		}
	}
	return d.DeliverRuntimeInput(ctx, input, sessionID)
}

func (d *KimiWireDriver) validateSession(sessionID string) error {
	d.mu.Lock()
	current := d.sessionID
	d.mu.Unlock()
	sessionID = strings.TrimSpace(sessionID)
	if sessionID == "" {
		sessionID = current
	}
	if current == "" {
		return fmt.Errorf("kimi: session is not initialized")
	}
	if sessionID != current {
		return fmt.Errorf("kimi: session mismatch: got %q, want %q", sessionID, current)
	}
	return nil
}

func validateKimiPromptResult(result json.RawMessage) (string, error) {
	// Kimi Code 1.49's Wire server awaits run_soul before responding to
	// `prompt`, and returns exactly these three terminal statuses. TurnEnd is a
	// separately streamed event, so takePromptTerminalEventsLocked intentionally
	// joins both ordered signals before publishing AgentTeam's terminal event.
	var decoded struct {
		Status string `json:"status"`
	}
	if err := json.Unmarshal(result, &decoded); err != nil {
		return "", fmt.Errorf("kimi: decode prompt result: %w", err)
	}
	switch strings.TrimSpace(decoded.Status) {
	case "finished":
		return consts.ParsedEventStopReasonEndTurn, nil
	case "cancelled":
		return "cancelled", nil
	case "max_steps_reached":
		return "max_steps_reached", nil
	default:
		return "", fmt.Errorf("kimi: unexpected prompt result status %q", decoded.Status)
	}
}

func (d *KimiWireDriver) finishPrompt(outcome kimiPromptOutcome) {
	d.mu.Lock()
	promptAccepted := d.promptAccepted
	d.promptOutcome = outcome
	d.promptInFlight = false
	d.promptInputID = ""
	d.promptAccepted = nil
	d.markTurnTerminalLocked()
	emitError, emitTurnEnd, sessionID := d.takePromptTerminalEventsLocked()
	d.mu.Unlock()
	if promptAccepted != nil {
		promptAccepted <- errors.Join(errKimiWireTransportAmbiguous, errors.New(firstNonEmptyString(outcome.errMessage, "Kimi prompt ended before TurnBegin")))
	}
	if emitError != "" {
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindError,
			Message:   emitError,
			SessionID: sessionID,
			Metadata:  map[string]any{"will_retry": false},
		})
	}
	if emitTurnEnd != "" {
		d.emit(translatedEvent{Kind: consts.ParsedEventKindTurnEnd, SessionID: sessionID, StopReason: emitTurnEnd})
	}
}

func (d *KimiWireDriver) takePromptTerminalEventsLocked() (string, string, string) {
	if !d.wireTurnEnded || !d.promptOutcome.ready || d.turnEndEmitted {
		return "", "", ""
	}
	d.turnEndEmitted = true
	return d.promptOutcome.errMessage, d.promptOutcome.stopReason, d.sessionID
}

func (d *KimiWireDriver) markTurnTerminalLocked() {
	d.turnActive = false
	select {
	case <-d.turnTerminal:
	default:
		close(d.turnTerminal)
	}
}

func closedKimiSignal() chan struct{} {
	ch := make(chan struct{})
	close(ch)
	return ch
}

func (d *KimiWireDriver) EncodeStdinMessage(string, string) []byte { return []byte("{}\n") }

func isKimiTransportClosed(err error) bool {
	return errors.Is(err, io.ErrClosedPipe) || errors.Is(err, os.ErrClosed)
}

var _ drivercontract.Driver = (*KimiWireDriver)(nil)
var _ drivercontract.RuntimeInputDeliveryDriver = (*KimiWireDriver)(nil)
var _ drivercontract.AckedBusyRuntimeInputDriver = (*KimiWireDriver)(nil)
var _ drivercontract.InterruptingRuntimeInputDriver = (*KimiWireDriver)(nil)
var _ drivercontract.ActiveRunInterruptDriver = (*KimiWireDriver)(nil)
var _ drivercontract.ExplicitTurnLifecycleDriver = (*KimiWireDriver)(nil)