package agent

import (
	"encoding/json"
	"errors"
	"math"
	"strings"

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

var errOpenCodeSSEReplayGap = errors.New("opencode SSE stream lost after readiness")

// handleSSEEvent processes a single SSE event payload.
// Wire format: {"type":"<event-type>","properties":{...}}
func (d *OpenCodeDriver) handleSSEEvent(data string) {
	var event struct {
		Type       string          `json:"type"`
		Properties json.RawMessage `json:"properties"`
	}
	if err := json.Unmarshal([]byte(data), &event); err != nil {
		d.emitRuntimeDebug("opencode sse event parse failed", map[string]any{
			"error":      err,
			"data_bytes": len(data),
		})
		return
	}

	d.mu.Lock()
	currentSessionID := d.sessionID
	generationRetiring := d.generationRetiring
	contextTerminalDrain := d.contextTerminalDrain
	interruptDraining := d.interruptIdleFence != nil || d.interruptCleanupDrain
	d.mu.Unlock()
	if generationRetiring {
		d.emitRuntimeDebug("opencode sse event ignored for retiring generation", map[string]any{
			"event_type": event.Type,
			"session_id": currentSessionID,
		})
		return
	}
	if interruptDraining {
		switch event.Type {
		case "message.updated", "message.part.updated", "message.part.delta", "session.error", "permission.asked":
			d.emitRuntimeDebug("opencode interrupt cleanup event ignored", map[string]any{
				"event_type": event.Type,
				"session_id": currentSessionID,
			})
			return
		}
	}
	if contextTerminalDrain {
		switch event.Type {
		case "message.part.updated", "message.part.delta":
			// SessionProcessor.halt publishes its error idle before its
			// ensuring(cleanup) message/tool updates, and Runner.finishRun can
			// then publish another idle. Those frames belong to the already
			// terminal context-overflow turn. Only a fresh busy/retry status
			// proves native summarize (or another run) has begun.
			d.emitRuntimeDebug("opencode post-context terminal cleanup ignored", map[string]any{
				"event_type": event.Type,
				"session_id": currentSessionID,
			})
			return
		}
	}
	d.emitRuntimeDebug("opencode sse event received", map[string]any{
		"event_type": event.Type,
		"session_id": currentSessionID,
	})

	switch event.Type {
	case "message.updated":
		d.handleMessageUpdated(event.Properties, currentSessionID)

	case "message.part.updated":
		d.handlePartUpdated(event.Properties, currentSessionID)

	case "message.part.delta":
		d.handlePartDelta(event.Properties, currentSessionID)

	case "session.status":
		d.handleSessionStatus(event.Properties, currentSessionID)

	case "session.error":
		d.handleSessionError(event.Properties, currentSessionID)

	case "session.compacted":
		d.handleSessionCompacted(event.Properties, currentSessionID)

	case "permission.asked":
		d.handleOpenCodePermissionAsked(event.Properties, currentSessionID)
	default:
		d.emitRuntimeDebug("opencode sse event ignored", map[string]any{
			"event_type": event.Type,
			"session_id": currentSessionID,
		})
	}
}

type openCodeRuntimeInputAckedEvent struct {
	RuntimeInputID string
	SessionID      string
}

// openCodeTokens mirrors the token object OpenCode carries on assistant
// messages and step-finish parts. Per OpenCode upstream (session/session.ts),
// input is the NON-cached input (the AI SDK's cumulative input minus cache read
// and cache write), output excludes reasoning, and cache has separate
// read/write buckets.
type openCodeTokens struct {
	Input     float64 `json:"input"`
	Output    float64 `json:"output"`
	Reasoning float64 `json:"reasoning"`
	Cache     struct {
		Read  float64 `json:"read"`
		Write float64 `json:"write"`
	} `json:"cache"`
}

type openCodeMessageUpdatedProperties struct {
	SessionID string                     `json:"sessionID"`
	Info      openCodeMessageUpdatedInfo `json:"info"`
}

type openCodeMessageUpdatedInfo struct {
	ID        string          `json:"id"`
	ParentID  string          `json:"parentID"`
	SessionID string          `json:"sessionID"`
	Role      string          `json:"role"`
	ModelID   string          `json:"modelID"`
	Tokens    openCodeTokens  `json:"tokens"`
	Finish    string          `json:"finish"`
	Summary   bool            `json:"summary"`
	Error     json.RawMessage `json:"error"`
	Time      struct {
		Completed *int64 `json:"completed"`
	} `json:"time"`
}

func (t openCodeTokens) snapshot() openCodeUsageSnapshot {
	return openCodeUsageSnapshot{
		input:      nonNegativeTokenCount(t.Input),
		cachedRead: nonNegativeTokenCount(t.Cache.Read),
		cacheWrite: nonNegativeTokenCount(t.Cache.Write),
		output:     nonNegativeTokenCount(t.Output),
		reasoning:  nonNegativeTokenCount(t.Reasoning),
	}
}

// openCodeUsageSnapshot is the per-step token breakdown decoded from an
// OpenCode tokens object.
type openCodeUsageSnapshot struct {
	input      uint64
	cachedRead uint64
	cacheWrite uint64
	output     uint64
	reasoning  uint64
}

func (u openCodeUsageSnapshot) empty() bool {
	return u.input == 0 && u.cachedRead == 0 && u.cacheWrite == 0 && u.output == 0 && u.reasoning == 0
}

// inputTotal returns total input tokens (non-cached input plus cache read and
// cache write) so cached is a subset of input, matching the shared contract.
func (u openCodeUsageSnapshot) inputTotal() uint64 {
	return u.input + u.cachedRead + u.cacheWrite
}

// outputTotal returns total output tokens (model output plus reasoning).
func (u openCodeUsageSnapshot) outputTotal() uint64 {
	return u.output + u.reasoning
}

func nonNegativeTokenCount(value float64) uint64 {
	if value <= 0 || math.IsNaN(value) || math.IsInf(value, 0) {
		return 0
	}
	if value >= float64(math.MaxUint64) {
		return math.MaxUint64
	}
	return uint64(value)
}

// handleMessageUpdated tracks per-message metadata from OpenCode message.updated
// events: it acknowledges runtime inputs and refreshes the proactive-compaction
// context signal plus the model used to attribute per-step usage. Token usage
// itself is accumulated from step-finish parts in handlePartUpdated, because
// message.updated only carries the last step's replacement value.
func (d *OpenCodeDriver) handleMessageUpdated(props json.RawMessage, currentSessionID string) {
	var p openCodeMessageUpdatedProperties
	if err := json.Unmarshal(props, &p); err != nil {
		d.emitRuntimeDebug("opencode message.updated parse failed", map[string]any{"error": err})
		return
	}

	messageSessionID := strings.TrimSpace(p.SessionID)
	if messageSessionID == "" {
		messageSessionID = strings.TrimSpace(p.Info.SessionID)
	}
	if currentSessionID != "" && messageSessionID != currentSessionID {
		d.emitRuntimeDebug("opencode message.updated ignored for another session", map[string]any{
			"message_session_id": messageSessionID,
			"current_session_id": currentSessionID,
			"role":               p.Info.Role,
		})
		return
	}
	d.mu.Lock()
	generationRetiring := d.generationRetiring
	d.mu.Unlock()
	if generationRetiring {
		return
	}
	messageID := strings.TrimSpace(p.Info.ID)
	d.observeOpenCodeMessageID(messageID)
	d.observeOpenCodeMessageID(p.Info.ParentID)
	usage := p.Info.Tokens.snapshot()
	d.emitRuntimeDebug("opencode message.updated", map[string]any{
		"session_id": messageSessionID,
		"message_id": messageID,
		"parent_id":  strings.TrimSpace(p.Info.ParentID),
		"role":       p.Info.Role,
		"model":      strings.TrimSpace(p.Info.ModelID),
		"finish":     strings.TrimSpace(p.Info.Finish),
		"summary":    p.Info.Summary,
		"has_error":  openCodeRawErrorPresent(p.Info.Error),
		"has_usage":  !usage.empty(),
	})
	switch p.Info.Role {
	case "user":
		d.mu.Lock()
		contextTerminalDrain := d.contextTerminalDrain
		d.mu.Unlock()
		if contextTerminalDrain {
			return
		}
		d.admitOpenCodeAutoContinuationUser(messageID)
		d.markPendingRuntimeInputPersisted(messageID, messageSessionID)
		return
	case "assistant":
		if !d.admitOpenCodeAssistantEvent(messageID, p.Info.ParentID, p.Info.Summary) {
			d.emitRuntimeDebug("opencode assistant update ignored outside the current Runner", map[string]any{
				"session_id": messageSessionID,
				"message_id": messageID,
				"parent_id":  strings.TrimSpace(p.Info.ParentID),
			})
			return
		}
		acks := d.popPendingRuntimeInputsForAssistant(p.Info.ParentID, messageSessionID)
		d.noteOpenCodeAssistantObserved(
			messageSessionID,
			messageID,
			p.Info.ParentID,
			p.Info.Summary,
		)
		terminalUpdate := d.noteOpenCodeAssistantTerminalUpdate(
			messageSessionID,
			messageID,
			p.Info.ParentID,
			p.Info.Finish,
			p.Info.Summary,
			p.Info.Error,
			p.Info.Time.Completed != nil,
		)
		d.mu.Lock()
		contextTerminalDrain := d.contextTerminalDrain
		d.mu.Unlock()
		if !terminalUpdate && !contextTerminalDrain {
			d.markSessionTurnActiveForAssistant(p.Info.ParentID)
		}
		for _, ack := range acks {
			d.emitRuntimeInputAcked(ack.RuntimeInputID, ack.SessionID)
		}
		if contextTerminalDrain && !terminalUpdate {
			return
		}
		// Token usage is accumulated from step-finish parts (each part is one
		// model invocation); message.updated only carries the last step's
		// replacement value, so it is not emitted as a usage event here. It is
		// used only to refresh the proactive-compaction context signal.
		if !usage.empty() {
			d.mu.Lock()
			if d.generationRetiring {
				d.mu.Unlock()
				return
			}
			// The most recent message.updated reflects the runtime's current
			// total input context; overwrite (not max) so a post-compaction
			// snapshot reports the smaller context.
			d.lastInputContextTokens = usage.inputTotal()
			d.mu.Unlock()
		}
	default:
		return
	}
}

func openCodeRawErrorPresent(raw json.RawMessage) bool {
	trimmed := strings.TrimSpace(string(raw))
	return trimmed != "" && trimmed != "null"
}

func (d *OpenCodeDriver) noteOpenCodeCompactionPart(sessionID string, messageID string, auto *bool) bool {
	trimmedSessionID := strings.TrimSpace(sessionID)
	trimmedMessageID := strings.TrimSpace(messageID)
	if trimmedMessageID == "" || auto == nil {
		return false
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.generationRetiring ||
		d.promptAwaitingRunner != nil ||
		strings.TrimSpace(d.sessionID) != trimmedSessionID {
		return false
	}
	if *auto {
		if !d.turnActive || d.summarizeInProgress {
			return false
		}
		activePromptMessageID := strings.TrimSpace(d.activePromptMessageID)
		if activePromptMessageID == "" {
			return false
		}
		if _, admitted := d.runnerUserMessageIDs[activePromptMessageID]; !admitted {
			return false
		}
		if !d.rekeyPendingRuntimeInputLocked(
			activePromptMessageID,
			trimmedMessageID,
			trimmedSessionID,
		) {
			return false
		}
		d.admitOpenCodeRunnerUserMessageLocked(trimmedMessageID)
		if d.runnerCompactionUserMessageIDs == nil {
			d.runnerCompactionUserMessageIDs = make(map[string]bool)
		}
		d.runnerCompactionUserMessageIDs[trimmedMessageID] = true
		if trimmedMessageID != strings.TrimSpace(d.autoCompactionUserMessageID) {
			d.autoCompactionAssistantMessageID = ""
		}
		d.autoCompactionUserMessageID = trimmedMessageID
		return true
	}
	if fence := d.summarizeFence; d.summarizeInProgress &&
		fence != nil &&
		!fence.retired &&
		fence.sessionGeneration == d.sessionGeneration &&
		fence.summarizeGeneration == d.summarizeGeneration &&
		strings.TrimSpace(fence.sessionID) == trimmedSessionID {
		fence.compactionObserved = true
		if trimmedMessageID != strings.TrimSpace(fence.compactionUserID) {
			fence.summaryAssistantID = ""
		}
		fence.compactionUserID = trimmedMessageID
		d.admitOpenCodeRunnerUserMessageLocked(trimmedMessageID)
		if d.runnerCompactionUserMessageIDs == nil {
			d.runnerCompactionUserMessageIDs = make(map[string]bool)
		}
		d.runnerCompactionUserMessageIDs[trimmedMessageID] = false
		return true
	}
	return false
}

func (d *OpenCodeDriver) handleSessionCompacted(props json.RawMessage, currentSessionID string) {
	var p struct {
		SessionID string `json:"sessionID"`
	}
	if err := json.Unmarshal(props, &p); err != nil {
		d.emitRuntimeDebug("opencode session.compacted parse failed", map[string]any{"error": err})
		return
	}
	trimmedSessionID := strings.TrimSpace(p.SessionID)
	if trimmedSessionID == "" || trimmedSessionID != strings.TrimSpace(currentSessionID) {
		return
	}
	d.mu.Lock()
	if d.generationRetiring {
		d.mu.Unlock()
		return
	}
	if fence := d.summarizeFence; d.summarizeInProgress &&
		fence != nil &&
		!fence.retired &&
		fence.sessionGeneration == d.sessionGeneration &&
		fence.summarizeGeneration == d.summarizeGeneration &&
		strings.TrimSpace(fence.sessionID) == trimmedSessionID &&
		fence.compactionObserved &&
		fence.busyObserved {
		fence.compactedObserved = true
		d.maybeCompleteOpenCodeSummarizeSSEFenceLocked(fence)
	}
	// An upstream automatic compaction has now succeeded. Any following
	// continuation is an ordinary assistant run, so later provider errors must
	// not be mislabeled as compaction failures.
	d.autoCompactionUserMessageID = ""
	d.autoCompactionAssistantMessageID = ""
	d.contextCompacted = true
	d.mu.Unlock()
}

// noteOpenCodeAssistantTerminalUpdate recognizes processor cleanup and failed
// summary updates. For compaction.auto=false, SessionProcessor publishes its
// first idle before this update and Runner publishes another idle afterward.
// For a failed automatic compaction, the summary error update precedes the only
// Runner idle. In both cases this flag identifies the idle at which a terminal
// boundary is safe.
func (d *OpenCodeDriver) noteOpenCodeAssistantTerminalUpdate(
	sessionID string,
	messageID string,
	parentID string,
	finish string,
	summary bool,
	rawError json.RawMessage,
	completed bool,
) bool {
	hasError := openCodeRawErrorPresent(rawError)
	if !completed || (!hasError && !strings.EqualFold(strings.TrimSpace(finish), "error")) {
		return false
	}
	message := runtimeErrorMessageFromRaw(rawError)

	d.mu.Lock()
	defer d.mu.Unlock()
	if d.generationRetiring {
		return true
	}
	trimmedMessageID := strings.TrimSpace(messageID)
	if summary {
		if _, admitted := d.runnerSummaryAssistantMessageIDs[trimmedMessageID]; !admitted {
			return false
		}
	} else {
		if _, admitted := d.runnerAssistantMessageIDs[trimmedMessageID]; !admitted {
			return false
		}
	}
	if strings.TrimSpace(message) == "" && d.pendingSessionError != nil {
		message = d.pendingSessionError.Message
	}
	if strings.TrimSpace(message) == "" {
		message = "opencode assistant finished with an error"
	}
	metadata := map[string]any{}
	if errorName := openCodeRuntimeErrorName(rawError); strings.EqualFold(errorName, "ContextOverflowError") {
		metadata[openCodeErrorNameMetadataKey] = errorName
	}
	transientError := classifyTransientRuntimeErrorRaw(rawError)
	if !transientError.Retryable {
		transientError = classifyTransientRuntimeErrorText(message)
	}
	metadata = mergeTransientRuntimeErrorMetadata(metadata, transientError)
	if summary {
		metadata[contextCompactionFailedMetadataKey] = true
		if d.proactiveSummarizeInProgressLocked(sessionID) {
			fence := d.summarizeFence
			if strings.TrimSpace(fence.summaryAssistantID) == trimmedMessageID &&
				strings.TrimSpace(fence.compactionUserID) == strings.TrimSpace(parentID) {
				metadata[contextCompactionMaintenanceMetadataKey] = true
			}
		}
	}
	d.pendingSessionError = &translatedEvent{
		Kind:      consts.ParsedEventKindError,
		SessionID: strings.TrimSpace(sessionID),
		Message:   message,
		Metadata:  metadata,
	}
	d.pendingSessionErrorFinalIdle = true
	return true
}

// noteOpenCodeAssistantObserved binds summary assistants to the compaction
// message observed in this generation. OpenCode can update arbitrary historic
// messages in a session, so summary/error fields are not lifecycle evidence
// until this exact parent relationship is established.
func (d *OpenCodeDriver) noteOpenCodeAssistantObserved(
	sessionID string,
	messageID string,
	parentID string,
	summary bool,
) {
	if !summary {
		return
	}
	trimmedSessionID := strings.TrimSpace(sessionID)
	trimmedMessageID := strings.TrimSpace(messageID)
	trimmedParentID := strings.TrimSpace(parentID)
	if trimmedMessageID == "" || trimmedParentID == "" {
		return
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.generationRetiring || strings.TrimSpace(d.sessionID) != trimmedSessionID {
		return
	}
	if auto, bound := d.runnerCompactionUserMessageIDs[trimmedParentID]; bound && auto {
		d.autoCompactionAssistantMessageID = trimmedMessageID
	}
	if fence := d.summarizeFence; d.summarizeInProgress &&
		fence != nil &&
		!fence.retired &&
		fence.sessionGeneration == d.sessionGeneration &&
		fence.summarizeGeneration == d.summarizeGeneration &&
		strings.TrimSpace(fence.sessionID) == trimmedSessionID &&
		trimmedParentID == strings.TrimSpace(fence.compactionUserID) {
		fence.summaryAssistantID = trimmedMessageID
	}
}

type openCodePartUpdatedProperties struct {
	SessionID string          `json:"sessionID"`
	Part      json.RawMessage `json:"part"`
}

type openCodeUpdatedPart struct {
	ID        string                    `json:"id"`
	MessageID string                    `json:"messageID"`
	Type      string                    `json:"type"`
	Text      string                    `json:"text"`
	Tool      string                    `json:"tool"`
	CallID    string                    `json:"callID"`
	Tokens    openCodeTokens            `json:"tokens"`
	Auto      *bool                     `json:"auto"`
	Overflow  bool                      `json:"overflow"`
	Synthetic bool                      `json:"synthetic"`
	Metadata  map[string]any            `json:"metadata"`
	State     *openCodeUpdatedPartState `json:"state"`
}

type openCodeUpdatedPartState struct {
	Status string          `json:"status"`
	Input  json.RawMessage `json:"input"`
	Output json.RawMessage `json:"output"`
	Title  string          `json:"title"`
	Error  string          `json:"error"`
}

func openCodeUpdatedPartStateStatus(state *openCodeUpdatedPartState) string {
	if state == nil {
		return ""
	}
	return strings.TrimSpace(state.Status)
}

// handlePartUpdated translates a message.part.updated SSE event.
//
// Wire format in properties:
//
//	{ "sessionID": "...", "part": { "id", "type", ... }, "time": 123 }
//
// Part types handled here: "tool" (lifecycle state) and "step-finish" (one
// model invocation, carrying that step's token usage, accumulated per step).
// We intentionally do NOT emit text/reasoning from part.updated because those
// are delivered incrementally via message.part.delta. part.updated carries the
// full accumulated text, not a delta.
func (d *OpenCodeDriver) handlePartUpdated(props json.RawMessage, currentSessionID string) {
	var p openCodePartUpdatedProperties
	if err := json.Unmarshal(props, &p); err != nil {
		d.emitRuntimeDebug("opencode message.part.updated parse failed", map[string]any{"error": err})
		return
	}

	if p.SessionID != currentSessionID {
		d.emitRuntimeDebug("opencode message.part.updated ignored for another session", map[string]any{
			"part_session_id":    p.SessionID,
			"current_session_id": currentSessionID,
		})
		return
	}
	d.mu.Lock()
	generationRetiring := d.generationRetiring
	d.mu.Unlock()
	if generationRetiring {
		return
	}

	var part openCodeUpdatedPart
	if err := json.Unmarshal(p.Part, &part); err != nil {
		d.emitRuntimeDebug("opencode message.part.updated part parse failed", map[string]any{"error": err})
		return
	}
	d.emitRuntimeDebug("opencode message.part.updated", map[string]any{
		"session_id": p.SessionID,
		"part_id":    strings.TrimSpace(part.ID),
		"message_id": strings.TrimSpace(part.MessageID),
		"part_type":  strings.TrimSpace(part.Type),
		"tool":       strings.TrimSpace(part.Tool),
		"call_id":    strings.TrimSpace(part.CallID),
		"state":      openCodeUpdatedPartStateStatus(part.State),
	})
	if part.Type == "compaction" {
		if !d.noteOpenCodeCompactionPart(p.SessionID, part.MessageID, part.Auto) {
			d.emitRuntimeDebug("opencode compaction part ignored outside the current Runner", map[string]any{
				"session_id": p.SessionID,
				"message_id": strings.TrimSpace(part.MessageID),
				"part_id":    strings.TrimSpace(part.ID),
			})
		}
		return
	}
	if part.Type == "text" && part.Synthetic &&
		d.admitOpenCodeSyntheticRunnerContinuationPart(
			p.SessionID,
			part.MessageID,
			part.Text,
			part.Metadata,
		) {
		d.emitRuntimeDebug("opencode synthetic Runner continuation user admitted", map[string]any{
			"session_id": p.SessionID,
			"message_id": strings.TrimSpace(part.MessageID),
			"part_id":    strings.TrimSpace(part.ID),
		})
		return
	}
	d.mu.Lock()
	_, acceptPart := d.runnerAssistantMessageIDs[strings.TrimSpace(part.MessageID)]
	acceptPart = acceptPart && d.turnActive && d.promptAwaitingRunner == nil
	d.mu.Unlock()
	if !acceptPart {
		d.emitRuntimeDebug("opencode part update ignored outside the admitted current Runner", map[string]any{
			"session_id": p.SessionID,
			"message_id": strings.TrimSpace(part.MessageID),
			"part_id":    strings.TrimSpace(part.ID),
			"part_type":  strings.TrimSpace(part.Type),
		})
		return
	}

	switch part.Type {
	case "tool":
		d.flushTextBuffer()

		if part.State != nil {
			d.markSessionTurnActive()
			toolName := part.Tool
			// OpenCode republishes the same part for each tool lifecycle state.
			// Keep its stable call id so downstream projections update one row.
			toolCallID := strings.TrimSpace(part.CallID)
			if toolCallID == "" {
				toolCallID = strings.TrimSpace(part.ID)
			}
			inputStr := "{}"
			if part.State.Input != nil {
				inputStr = string(part.State.Input)
			}
			kind := consts.ParsedEventKindToolCall
			if status := strings.ToLower(strings.TrimSpace(part.State.Status)); status == "completed" || status == "error" {
				kind = consts.ParsedEventKindToolOutput
			}
			ev := translatedEvent{
				Kind:       kind,
				ToolName:   toolName,
				ToolCallID: toolCallID,
			}
			if kind == consts.ParsedEventKindToolCall {
				ev.ToolInput = inputStr
			} else {
				ev.Text = openCodeToolOutputText(part.State.Output)
				if strings.EqualFold(strings.TrimSpace(part.State.Status), "error") {
					ev.Metadata = map[string]any{"toolError": true}
					if strings.TrimSpace(part.State.Error) != "" {
						ev.Text = strings.TrimSpace(part.State.Error)
					}
				}
			}
			data, _ := json.Marshal(ev)
			d.translatedLines <- string(data)
		}
	case "step-finish":
		d.handleOpenCodeStepFinishUsage(part.ID, part.MessageID, part.Tokens, p.SessionID)
	}
}

// handleOpenCodeStepFinishUsage records token usage for one step-finish part.
// Each step-finish part is one model invocation; accumulating per-step (deduped
// by stable part id) yields the message total, since message.updated only
// carries the last step's replacement value. Model is left empty so recordUsage
// falls back to the process launch model (message.updated, which carries
// modelID, arrives after the step-finish part).
func (d *OpenCodeDriver) handleOpenCodeStepFinishUsage(
	partID string,
	messageID string,
	tokens openCodeTokens,
	sessionID string,
) {
	d.flushTextBuffer()
	step := tokens.snapshot()
	if step.empty() {
		return
	}
	partID = strings.TrimSpace(partID)
	if partID == "" {
		return
	}
	d.mu.Lock()
	if d.generationRetiring {
		d.mu.Unlock()
		return
	}
	if d.countedStepFinish == nil {
		d.countedStepFinish = make(map[string]struct{})
	}
	if _, seen := d.countedStepFinish[partID]; seen {
		d.mu.Unlock()
		return
	}
	d.countedStepFinish[partID] = struct{}{}
	_, summaryInvocation := d.runnerSummaryAssistantMessageIDs[strings.TrimSpace(messageID)]
	contextCompacted := d.contextCompacted && !summaryInvocation
	if contextCompacted {
		d.contextCompacted = false
	}
	d.mu.Unlock()
	usageEv := translatedEvent{
		Kind:                    consts.ParsedEventKindUsage,
		SessionID:               sessionID,
		ModelInvocationID:       partID,
		ContextExposureKnown:    !summaryInvocation,
		ContextExposureExcluded: summaryInvocation,
		ContextCompacted:        contextCompacted,
		InputTokens:             step.inputTotal(),
		CachedInputTokens:       step.cachedRead,
		OutputTokens:            step.outputTotal(),
	}
	data, _ := json.Marshal(usageEv)
	d.translatedLines <- string(data)
}

func openCodeToolOutputText(raw json.RawMessage) string {
	raw = json.RawMessage(strings.TrimSpace(string(raw)))
	if len(raw) == 0 || string(raw) == "null" {
		return ""
	}
	var text string
	if json.Unmarshal(raw, &text) == nil {
		return strings.TrimSpace(text)
	}
	return string(raw)
}

// handlePartDelta accumulates incremental text from message.part.delta events
// into the driver's textBuf. The buffered text is flushed as a single
// consolidated "text" event when the turn ends or when a tool_call part is
// observed, rather than emitting one event per delta.
func (d *OpenCodeDriver) handlePartDelta(props json.RawMessage, currentSessionID string) {
	var p struct {
		SessionID string `json:"sessionID"`
		MessageID string `json:"messageID"`
		PartID    string `json:"partID"`
		Field     string `json:"field"`
		Delta     string `json:"delta"`
	}
	if err := json.Unmarshal(props, &p); err != nil {
		d.emitRuntimeDebug("opencode message.part.delta parse failed", map[string]any{"error": err})
		return
	}

	if p.SessionID != currentSessionID {
		d.emitRuntimeDebug("opencode message.part.delta ignored for another session", map[string]any{
			"part_session_id":    p.SessionID,
			"current_session_id": currentSessionID,
		})
		return
	}

	if p.Field == "text" && p.Delta != "" {
		d.emitRuntimeDebug("opencode message.part.delta text received", map[string]any{
			"session_id":  p.SessionID,
			"message_id":  strings.TrimSpace(p.MessageID),
			"part_id":     strings.TrimSpace(p.PartID),
			"delta_bytes": len(p.Delta),
		})
		d.mu.Lock()
		if d.generationRetiring {
			d.mu.Unlock()
			return
		}
		messageID := strings.TrimSpace(p.MessageID)
		_, admittedAssistant := d.runnerAssistantMessageIDs[messageID]
		if d.promptAwaitingRunner != nil || !d.turnActive || !admittedAssistant {
			d.mu.Unlock()
			d.emitRuntimeDebug("opencode text delta ignored outside the admitted current Runner", map[string]any{
				"session_id": p.SessionID,
				"message_id": messageID,
				"part_id":    strings.TrimSpace(p.PartID),
			})
			return
		}
		_, summaryAssistant := d.runnerSummaryAssistantMessageIDs[messageID]
		if summaryAssistant {
			d.mu.Unlock()
			d.emitRuntimeDebug("opencode internal compaction summary text suppressed", map[string]any{
				"session_id": p.SessionID,
				"message_id": messageID,
				"part_id":    strings.TrimSpace(p.PartID),
			})
			return
		}
		d.noteSessionTurnActiveLocked("", false)
		d.textBuf.WriteString(p.Delta)
		d.mu.Unlock()
	}
}