package agent

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"strings"
	"sync"
	"time"

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

// Spawn starts an `opencode serve` process, waits for it to become healthy,
// creates a session, subscribes to SSE events, and sends the initial prompt.
func (d *OpenCodeDriver) 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("opencode: materialize contract carrier: %w", err)
	}
	permissionReply, err := openCodePermissionReplyForLaunch(cfg)
	if err != nil {
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: permission policy: %w", err)
	}

	launchID := newOpenCodeLaunchID()
	d.deliveryMu.Lock()
	d.mu.Lock()
	if d.promptAwaitingRunner != nil {
		close(d.promptAwaitingRunner.done)
	}
	if d.interruptIdleFence != nil {
		close(d.interruptIdleFence.done)
	}
	if d.summarizeFence != nil && !d.summarizeFence.retired {
		d.summarizeFence.retired = true
		close(d.summarizeFence.retiredCh)
	}
	d.sessionGeneration++
	d.sessionResumeFallback = nil
	d.pendingRuntimeInputs = nil
	d.promptAwaitingRunner = nil
	d.activePromptMessageID = ""
	d.pendingSessionError = nil
	d.pendingSessionErrorFinalIdle = false
	d.contextTerminalDrain = false
	d.autoCompactionUserMessageID = ""
	d.autoCompactionAssistantMessageID = ""
	d.postBoundaryAwaitFreshBusy = false
	d.interruptCleanupDrain = false
	d.interruptIdleFence = nil
	d.sseDeliveryBlocked = true
	d.generationRetiring = false
	d.translatedLines = make(chan string, 512)
	d.textBuf.Reset()
	d.resetOpenCodeRunnerMessageAdmissionLocked()
	d.countedStepFinish = make(map[string]struct{})
	d.contextCompacted = false
	d.sessionStatus = ""
	d.turnActive = false
	d.permissionReply = permissionReply
	d.lastRetryStatusKey = ""
	d.compactQueued = false
	d.summarizeInProgress = false
	d.summarizeGeneration = 0
	d.summarizeFence = nil
	d.proactiveCompactionThreshold = resolveProactiveIdleCompactionThreshold(cfg)
	d.lastInputContextTokens = 0
	d.lastSummarizeCompletedAt = time.Time{}
	d.httpClient = &http.Client{Timeout: 120 * time.Second}
	d.workDir = cfg.WorkDir
	d.launchModel = cfg.Model
	d.runtimeDebugEnabled = cfg.RuntimeDebugEnabled
	d.serveLogPath = ""
	d.runtimeDebugLog = nil
	d.runtimeDebugAgentID = strings.TrimSpace(cfg.AgentID)
	d.runtimeDebugLaunchID = launchID
	d.serverURL = ""
	d.password = ""
	d.serverCancel = nil
	if cfg.Provider != nil {
		d.providerID = cfg.Provider.ID
	}
	d.mu.Unlock()
	d.deliveryMu.Unlock()

	// Generate authentication and prepare configuration before waiting for the
	// process-wide startup gate.
	password := generateRandomPassword()

	binary, err := resolveOpenCodeLaunchBinary(cfg)
	if err != nil {
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: resolve binary: %w", err)
	}
	cfg.Binary = binary

	// Runtime diagnostics are written as AgentTeam-owned, scrubbed JSONL. Keep the
	// raw OpenCode process log level bounded because upstream DEBUG logs may
	// include request or prompt bodies that AgentTeam must not persist.
	logLevel := "WARN"

	env, err := buildOpenCodeServeEnv(cfg, password)
	if err != nil {
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: build serve env: %w", err)
	}
	stderr := newCapturingPrefixWriter(fmt.Sprintf("[opencode %s] ", cfg.AgentID), openCodeCapturedOutputMaxBytes)
	stdout := newCapturingPrefixWriter(fmt.Sprintf("[opencode-stdout %s] ", cfg.AgentID), openCodeCapturedOutputMaxBytes)
	diagnosticLogs, err := d.attachOpenCodeDiagnosticLogs(cfg, stderr, stdout)
	if err != nil {
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: prepare serve log: %w", err)
	}
	serveLogPath := diagnosticLogs.serve.Path()

	diagnostics := openCodeLaunchDiagnostics(cfg, env)
	slog.Info("spawning opencode serve",
		"agent_id", cfg.AgentID,
		"launch_id", launchID,
		"binary", binary,
		"workdir", cfg.WorkDir,
		"model", cfg.Model,
		"runtime_debug", cfg.RuntimeDebugEnabled,
		"opencode_log_level", logLevel,
		"serve_log_path", serveLogPath,
		"runtime_debug_log_path", diagnosticLogs.runtimeDebug.Path(),
	)
	slog.Info("opencode launch diagnostics", append([]any{"agent_id", cfg.AgentID, "launch_id", launchID}, diagnostics.logAttrs()...)...)

	d.emitRuntimeDebug("opencode serve spawn requested", map[string]any{
		"binary":                 binary,
		"workdir":                cfg.WorkDir,
		"model":                  cfg.Model,
		"provider_id":            d.providerID,
		"log_level":              logLevel,
		"serve_log_path":         serveLogPath,
		"runtime_debug_log_path": diagnosticLogs.runtimeDebug.Path(),
		"storage_scope":          openCodeStorageScope(diagnostics.isolationFlags),
		"isolation_flags":        diagnostics.isolationFlags,
	})

	startedServer, startupErr := d.startOpenCodeServe(ctx, openCodeServeStart{
		cfg:         cfg,
		env:         env,
		password:    password,
		logLevel:    logLevel,
		stderr:      stderr,
		stdout:      stdout,
		diagnostics: diagnostics,
		onProcessStarted: func(server openCodeStartedServer) {
			slog.Info("opencode serve process started",
				"agent_id", cfg.AgentID,
				"launch_id", launchID,
				"pid", server.process.cmd.Process.Pid,
				"url", server.url,
			)
			d.emitRuntimeDebug("opencode serve process started", map[string]any{
				"pid": server.process.cmd.Process.Pid,
				"url": server.url,
			})
		},
	})
	if startupErr != nil {
		d.stopOpenCodeServer(startedServer, password)
		logIgnoredCleanupError("failed to close opencode diagnostics logs after startup failure", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: %w", startupErr)
	}
	serverProcess := startedServer.process
	serverURL := startedServer.url
	cmd := serverProcess.cmd
	serverCancel := serverProcess.cancel
	d.mu.Lock()
	if d.serverURL == serverURL && d.password == password {
		d.serverCancel = serverCancel
	}
	d.mu.Unlock()
	slog.Info("opencode serve is healthy", "agent_id", cfg.AgentID, "launch_id", launchID, "url", serverURL)
	d.emitRuntimeDebug("opencode serve health check passed", map[string]any{"url": serverURL})

	// 5. Create a session (or resume an existing one).
	sessionID, resumed, pendingRevert, err := d.createOrResumeSession(ctx, cfg)
	if err != nil {
		d.stopOpenCodeServer(startedServer, password)
		logIgnoredCleanupError("failed to close opencode diagnostics logs after session failure", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: create session: %w", err)
	}

	d.mu.Lock()
	freshSessionFallback := d.sessionResumeFallback
	d.sessionResumeFallback = nil
	d.sessionID = sessionID
	// A freshly started server has no in-memory Runner for this session. Resume
	// bootstrap or the initial prompt below installs the busy reservation before
	// dispatching prompt_async.
	d.sessionStatus = "idle"
	d.mu.Unlock()

	slog.Info("opencode session ready",
		"agent_id", cfg.AgentID,
		"launch_id", launchID,
		"session_id", sessionID,
	)
	d.emitRuntimeDebug("opencode session ready", map[string]any{"session_id": sessionID, "resumed": resumed})

	// 6. Subscribe to SSE events for this workspace.
	sseCtx, sseCancel := context.WithCancel(ctx)
	sseReady := make(chan openCodeSSEReadyEvent, 16)
	sseDone := make(chan struct{})
	go func() {
		defer close(sseDone)
		d.subscribeSSE(sseCtx, cfg, sseReady)
	}()

	d.emitRuntimeDebug("opencode sse subscription waiting", map[string]any{"timeout": openCodeSSEReadyTimeout.String()})
	if err := waitForOpenCodeSSEReady(ctx, sseReady, openCodeSSEReadyTimeout, serverProcess, stderr.Tail); err != nil {
		sseCancel()
		d.stopOpenCodeServer(startedServer, password)
		logIgnoredCleanupError("failed to close opencode diagnostics logs after sse failure", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
		return drivercontract.SpawnResult{}, fmt.Errorf("opencode: subscribe SSE: %w", err)
	}

	d.emitRuntimeDebug("opencode sse subscription ready", map[string]any{"session_id": sessionID})

	// session_init must precede any resume-bootstrap events that OpenCode can
	// publish as soon as prompt_async is accepted.
	d.translatedLines <- fmt.Sprintf(`{"kind":%q,"sessionId":"%s"}`, consts.ParsedEventKindSessionInit, sessionID)

	initialPrompt := carrier.LaunchSurface("prompt")
	hasInitialPrompt := strings.TrimSpace(initialPrompt) != "" || len(cfg.PromptAttachments) > 0
	if resumed {
		bootstrapCtx, bootstrapCancel := context.WithTimeout(ctx, openCodeResumeBootstrapTimeout)
		d.deliveryMu.Lock()
		var bootstrapErr error
		if hasInitialPrompt || pendingRevert {
			bootstrapErr = d.refreshOpenCodeResumedSessionMessageHighWater(bootstrapCtx, sessionID)
		} else {
			bootstrapErr = d.recoverInterruptedResumedSession(bootstrapCtx, sessionID)
		}
		d.deliveryMu.Unlock()
		bootstrapCancel()
		if bootstrapErr != nil {
			d.emitRuntimeDebug("opencode resumed session prompt bootstrap failed", map[string]any{
				"session_id": sessionID,
				"error":      bootstrapErr,
			})
			sseCancel()
			d.stopOpenCodeServer(startedServer, password)
			logIgnoredCleanupError("failed to close opencode diagnostics logs after resume bootstrap failure", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
			return drivercontract.SpawnResult{}, fmt.Errorf("opencode: resume prompt bootstrap: %w", bootstrapErr)
		}
	}

	// Capture the initial message id only after a resumed session's complete
	// history scan advanced the allocator high-water. Otherwise
	// PrepareRecoveryPointForMessage could pin an id older than an existing
	// assistant and the initial prompt would be ignored by MessageV2.latest.
	preparedRecoveryPoint := prepareSpawnRecoveryPoint(d, cfg.InitialDeliveryMessageID, sessionID)

	// 7. Submit the initial prompt through prompt_async. Waiting for the short
	// HTTP acceptance here closes the race where Spawn returned before the
	// initial turn installed its busy reservation.
	if hasInitialPrompt {
		input := drivercontract.RuntimeInput{
			CompiledPrompt: carrier,
			Resume:         resumed,
			Attachments:    cfg.PromptAttachments,
			ID:             cfg.InitialDeliveryMessageID,
		}
		d.deliveryMu.Lock()
		initialPromptErr := d.sendPromptAsync(ctx, input, cfg.Model, "")
		d.deliveryMu.Unlock()
		if initialPromptErr != nil {
			d.emitRuntimeDebug("opencode initial prompt failed; retiring startup generation", map[string]any{
				"session_id": sessionID,
				"error":      initialPromptErr,
			})
			sseCancel()
			d.stopOpenCodeServer(startedServer, password)
			d.clearPromptStateAfterGenerationRetired(sessionID)
			logIgnoredCleanupError("failed to close opencode diagnostics logs after initial prompt failure", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
			return drivercontract.SpawnResult{}, fmt.Errorf("opencode: initial prompt: %w", initialPromptErr)
		}
	}

	// 8. Create a pipe reader that reads from translatedLines for the process
	// stdout parsing agentteam in Process.
	translatedPipeReader, translatedWriter := io.Pipe()
	translatedReader := newOpenCodeTranslatedReader(translatedPipeReader)
	forwardStop := make(chan struct{})
	forwardAbort := make(chan struct{})
	forwardDone := make(chan struct{})
	var forwardStopOnce sync.Once
	var forwardAbortOnce sync.Once
	go func() {
		defer close(forwardDone)
		forwardOpenCodeTranslatedLines(
			d.translatedLines,
			translatedWriter,
			forwardStop,
			forwardAbort,
			d.openCodeOversizedTranslatedEvent,
		)
	}()

	// 9. Provide a dummy stdin pipe so that the process layer's stdinPipe nil
	// check passes. Actual message delivery is handled via HTTP in
	// EncodeStdinMessage, which fires HTTP requests internally; the data
	// written to this pipe is intentionally discarded.
	stdinReader, stdinWriter := io.Pipe()
	go func() {
		// Drain and discard anything written to the dummy stdin pipe.
		buf := make([]byte, 4096)
		for {
			if _, err := stdinReader.Read(buf); err != nil {
				return
			}
		}
	}()

	cleanupOnce := sync.Once{}
	abortForwarding := func() {
		forwardAbortOnce.Do(func() {
			close(forwardAbort)
			// Unblock an io.Pipe write even when a direct Spawn caller did not
			// install Process.parseStdout as the consumer.
			logIgnoredCleanupError("failed to close opencode translated reader while aborting output", translatedReader.Close())
		})
	}
	cleanup := func() {
		cleanupOnce.Do(func() {
			sseCancel()
			abortForwarding()
			logIgnoredCleanupError("failed to close opencode dummy stdin reader during cleanup", stdinReader.Close())
			logIgnoredCleanupError("failed to close opencode diagnostics logs during cleanup", diagnosticLogs.Close(), "agent_id", cfg.AgentID, "path", serveLogPath)
		})
	}

	return drivercontract.SpawnResult{
		Cmd:                   cmd,
		Stdin:                 stdinWriter,
		Stdout:                translatedReader,
		StderrTail:            stderr.Tail,
		SessionID:             sessionID,
		SessionResumed:        strings.TrimSpace(cfg.SessionID) == sessionID,
		FreshSessionFallback:  freshSessionFallback,
		PreparedRecoveryPoint: preparedRecoveryPoint,
		Wait: func() error {
			<-serverProcess.exited
			// Terminal paths cancel the serve process before enqueueing their
			// synthetic Error/TurnEnd pair. Wait for the synchronous SSE
			// producer and any asynchronous summarize producer before stopping
			// the output forwarder; otherwise process exit can overtake buffered
			// lifecycle events and leave durable ownership permanently busy.
			producersCompleted := false
			select {
			case <-sseDone:
				producersCompleted = true
			case <-forwardAbort:
			}
			if producersCompleted {
				if workerDone := d.openCodeSummarizeWorkerDone(); workerDone != nil {
					select {
					case <-workerDone:
					case <-forwardAbort:
					}
				}
			}
			forwardStopOnce.Do(func() { close(forwardStop) })
			select {
			case <-forwardDone:
			case <-forwardAbort:
			}
			select {
			case <-translatedReader.consumed:
			case <-forwardAbort:
			}
			d.clearOpenCodeServerState(startedServer, password)
			cleanup()
			return serverProcess.waitErr
		},
		Terminate: func() {
			d.stopOpenCodeServer(startedServer, password)
			sseCancel()
			abortForwarding()
		},
		Cleanup: cleanup,
	}, nil
}

// forwardOpenCodeTranslatedLines owns the translated stdout writer. A normal
// stop is issued only after all generation-scoped producers have completed, so
// it drains every line already admitted to the queue before closing the pipe.
// Abort is reserved for explicit teardown, where Process already rejects
// ordinary parsed events and blocking on an absent reader would deadlock.
func forwardOpenCodeTranslatedLines(
	lines <-chan string,
	writer *io.PipeWriter,
	stop <-chan struct{},
	abort <-chan struct{},
	oversized func(int) string,
) {
	if writer == nil {
		return
	}
	defer writer.Close()
	writeLine := func(line string) bool {
		if len(line)+1 > maxProcessLineBytes {
			if oversized == nil {
				return false
			}
			line = oversized(len(line) + 1)
			if strings.TrimSpace(line) == "" || len(line)+1 > maxProcessLineBytes {
				return false
			}
			_, _ = io.WriteString(writer, line+"\n")
			return false
		}
		_, err := io.WriteString(writer, line+"\n")
		return err == nil
	}
	for {
		select {
		case <-abort:
			return
		case <-stop:
			for {
				select {
				case <-abort:
					return
				case line, ok := <-lines:
					if !ok || !writeLine(line) {
						return
					}
				default:
					return
				}
			}
		case line, ok := <-lines:
			if !ok || !writeLine(line) {
				return
			}
		}
	}
}

// openCodeTranslatedReader acknowledges EOF only after Process has requested
// the read following the final translated line. Process handles a complete
// line before requesting that read, so Wait can use consumed as an exact
// terminal-event handling fence rather than merely a pipe-write fence.
type openCodeTranslatedReader struct {
	reader   *io.PipeReader
	consumed chan struct{}
	once     sync.Once
}

func newOpenCodeTranslatedReader(reader *io.PipeReader) *openCodeTranslatedReader {
	return &openCodeTranslatedReader{reader: reader, consumed: make(chan struct{})}
}

func (reader *openCodeTranslatedReader) Read(p []byte) (int, error) {
	if reader == nil || reader.reader == nil {
		return 0, io.ErrClosedPipe
	}
	n, err := reader.reader.Read(p)
	if err != nil {
		reader.finish()
	}
	return n, err
}

func (reader *openCodeTranslatedReader) Close() error {
	if reader == nil || reader.reader == nil {
		return nil
	}
	err := reader.reader.Close()
	reader.finish()
	return err
}

func (reader *openCodeTranslatedReader) finish() {
	if reader == nil {
		return
	}
	reader.once.Do(func() { close(reader.consumed) })
}

func (d *OpenCodeDriver) openCodeOversizedTranslatedEvent(bytes int) string {
	d.mu.Lock()
	sessionID := strings.TrimSpace(d.sessionID)
	serverCancel := d.beginOpenCodeGenerationRetirementLocked()
	d.mu.Unlock()
	if serverCancel != nil {
		serverCancel()
	}
	ev := translatedEvent{
		Kind:      consts.ParsedEventKindError,
		SessionID: sessionID,
		Message: fmt.Sprintf(
			"opencode translated event exceeded the %d-byte process line limit (%d bytes)",
			maxProcessLineBytes,
			bytes,
		),
		Metadata: map[string]any{"runtime_transport_fatal": true},
	}
	data, _ := json.Marshal(ev)
	return string(data)
}