package agent

import (
	"encoding/json"
	"fmt"
	"strconv"
	"strings"

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

// piAssistantStream is the per-root-run streaming dedup state for the current
// assistant message. It is reset on each assistant message_start and cleared at
// message_end.
type piAssistantStream struct {
	started bool

	// textEmittedLen/thinkEmittedLen track append-only emitted length per
	// contentIndex so text_end/message_end reconcile only a missing suffix.
	textEmittedLen  map[int]int
	thinkEmittedLen map[int]int
	// toolEmitted tracks contentIndexes already projected as a tool call from
	// the assistant message content (avoid double projection with
	// tool_execution_*).
	toolEmitted map[int]bool

	// usage is the latest cumulative provider usage retained across updates.
	usage *piUsage
	// Final message identity from message_end.
	model        string
	responseID   string
	timestamp    int64
	stopReason   string
	usageEmitted bool
}

func newPiAssistantStream() *piAssistantStream {
	return &piAssistantStream{
		textEmittedLen:  make(map[int]int),
		thinkEmittedLen: make(map[int]int),
		toolEmitted:     make(map[int]bool),
	}
}

func (d *PiRPCDriver) handleEvent(frame piRPCFrame) {
	sessionID := d.currentSessionID()
	switch frame.Type {
	case "agent_start":
		d.onAgentStart(sessionID)
	case "turn_start":
		d.onTurnStart(sessionID)
	case "message_start":
		d.onMessageStart(sessionID, frame)
	case "message_update":
		d.onMessageUpdate(sessionID, frame)
	case "message_end":
		d.onMessageEnd(sessionID, frame)
	case "tool_execution_start":
		d.onToolExecutionStart(sessionID, frame)
	case "tool_execution_end":
		d.onToolExecutionEnd(sessionID, frame)
	case "turn_end":
		d.onTurnEnd(sessionID)
	case "agent_end":
		d.onAgentEnd(sessionID, frame)
	case "agent_settled":
		d.onAgentSettled(sessionID)
	case "compaction_start":
		d.onCompactionStart(sessionID, frame)
	case "compaction_end":
		d.onCompactionEnd(sessionID, frame)
	case "queue_update":
		d.onQueueUpdate(sessionID, frame)
	case "auto_retry_start":
		d.onAutoRetryStart(sessionID, frame)
	case "auto_retry_end":
		d.onAutoRetryEnd(sessionID, frame)
	case "bash_execution_update":
		d.onBashExecutionUpdate(sessionID, frame)
	case "extension_error":
		d.onExtensionError(sessionID, frame)
	case "error":
		d.onRuntimeError(sessionID, frame)
	}
}

func (d *PiRPCDriver) currentSessionID() string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return d.sessionID
}

// rootGenerationLocked returns the current root-run generation, or 0 when no
// root run is active.
func (d *PiRPCDriver) rootGenerationLocked() uint64 {
	if !d.rootRunActive {
		return 0
	}
	return d.rootRunGeneration
}

// onAgentStart marks the start of one low-level agent run. The authoritative
// AgentTeam root-start metadata is emitted exactly once per root-run generation.
func (d *PiRPCDriver) onAgentStart(sessionID string) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.agentRunActive = true
	d.agentRunCount++
	emitAuthoritativeStart := !d.rootStartEmitted
	if emitAuthoritativeStart {
		d.rootStartEmitted = true
	}
	d.mu.Unlock()
	if emitAuthoritativeStart {
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindThinking,
			Text:      "",
			SessionID: sessionID,
			Metadata:  map[string]any{codexAuthoritativeTurnStartMetadataKey: true},
		})
	}
}

// onTurnStart marks one assistant/tool cycle beginning. It never emits a AgentTeam
// root event.
func (d *PiRPCDriver) onTurnStart(sessionID string) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.turnActive = true
	d.mu.Unlock()
}

// onTurnEnd marks one assistant/tool cycle complete. It never emits a AgentTeam root
// TurnEnd; only agent_settled does. The completed-turn counter is kept for
// rewind/usage.
func (d *PiRPCDriver) onTurnEnd(sessionID string) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.turnActive = false
	d.turnCount++
	d.mu.Unlock()
}

// onAgentEnd marks one low-level agent run complete. A run may be followed by
// an automatic retry, an overflow compaction, or queued steer/follow-up, so no
// AgentTeam root event is emitted here. When agent_end carries messages and
// willRetry, the last assistant's authoritative usage/stopReason/errorMessage
// are derived even if the normal assistant message_end was absent (e.g.
// abort-before-agent_start or early provider failure). A willRetry run keeps
// the retry context so the next run can replace the final root outcome.
func (d *PiRPCDriver) onAgentEnd(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.agentRunActive = false
	if frame.WillRetry {
		// Retry context retained: a later agent run replaces the final outcome.
		// Do not terminalize and do not emit this run's error as final.
		d.mu.Unlock()
		return
	}
	// Final low-level run (no retry follows): derive authoritative assistant
	// identity/usage/stop/error from agent_end.messages when the driver did not
	// observe a normal message_end.
	if len(frame.Messages) > 0 {
		lastAssistant := piLastAssistantMessage(frame.Messages)
		if lastAssistant != nil {
			d.applyDerivedAssistantLocked(lastAssistant)
		}
	}
	d.mu.Unlock()
}

// piLastAssistantMessage returns the last assistant message in a list, or nil.
func piLastAssistantMessage(messages []piAgentMessage) *piAgentMessage {
	for i := len(messages) - 1; i >= 0; i-- {
		msg := messages[i]
		if strings.TrimSpace(msg.Role) == "assistant" {
			clone := msg
			return &clone
		}
	}
	return nil
}

// applyDerivedAssistantLocked updates the root assistant stream's final
// identity/usage/stop/error from a finalized assistant message snapshot. It is
// idempotent per-generation: message_end/turn_end.message/agent_end.messages
// cannot double-emit. When the assistant's stable invocation identity was
// already finalized (e.g. agent_end.messages repeats the message_end assistant),
// the stream is marked finalized so onAgentSettled never re-emits Usage, while
// stop/error/model refinements still apply.
func (d *PiRPCDriver) applyDerivedAssistantLocked(message *piAgentMessage) {
	assistant := d.assistant
	if assistant == nil {
		assistant = newPiAssistantStream()
		d.assistant = assistant
	}
	if assistant.model == "" {
		assistant.model = firstNonEmptyString(message.ResponseModel, message.Model)
	}
	if assistant.responseID == "" {
		assistant.responseID = message.ResponseID
	}
	if assistant.timestamp == 0 {
		assistant.timestamp = message.Timestamp
	}
	if assistant.stopReason == "" {
		assistant.stopReason = message.StopReason
	}
	if message.Usage != nil {
		assistant.usage = message.Usage
	}
	// If this invocation was already finalized (message_end), do not let the
	// duplicate agent_end snapshot cause a second Usage at settlement.
	invocationID := piAssistantInvocationID(d.sessionID, assistant)
	if invocationID != "" && d.finalizedAssistantIDs[invocationID] {
		assistant.usageEmitted = true
	}
	d.rootStopReason = firstNonEmptyString(message.StopReason, d.rootStopReason)
	// A final provider failure carries an error message; retain it for the
	// settlement error emission.
	if strings.TrimSpace(message.ErrorMessage) != "" {
		d.rootError = firstNonEmptyString(d.rootError, message.ErrorMessage)
	}
	if strings.TrimSpace(message.StopReason) == "error" && d.rootError == "" {
		d.rootError = "Pi assistant run failed"
	}
}

// onAgentSettled is the ONLY authoritative fully-settled boundary. After
// retry/compaction/queue work finishes, it enqueues exactly one AgentTeam root
// TurnEnd (tagged with the completed generation, with the final stop reason)
// followed by RuntimeIdle FIRST, then releases the generation's waiter so a
// replacement can never consume an old idle after a new handoff.
func (d *PiRPCDriver) onAgentSettled(sessionID string) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	gen := d.rootRunGeneration
	turnID := d.rootRunTurnID
	settled := d.rootRunSettled
	stopReason := d.rootStopReason
	rootError := d.rootError
	aborted := d.rootAborted
	assistant := d.assistant
	d.rootRunActive = false
	d.assistant = nil
	d.mu.Unlock()

	if assistant != nil && !assistant.usageEmitted && assistant.usage != nil {
		d.emitUsage(sessionID, assistant)
	}

	// Enqueue the terminal boundary before releasing the waiter.
	d.emitTerminal(gen, turnID, sessionID, stopReason, rootError, stopReason, aborted)
	d.settleRootRunLocked(settled)
}

// onMessageStart handles a user or assistant message beginning. A user message
// whose content carries a AgentTeam runtime-input marker is the authoritative
// acknowledgement for the correlated input id.
func (d *PiRPCDriver) onMessageStart(sessionID string, frame piRPCFrame) {
	if frame.Message == nil {
		return
	}
	switch strings.TrimSpace(frame.Message.Role) {
	case "user":
		// User-message surfacing is a durable input acknowledgement, not a
		// turn-lifecycle event. Pi may flush the JSONL/user frame after
		// agent_settled, so it must remain admissible while its exact
		// correlation is still owned.
		d.ackInputIfMarked(sessionID, frame.Message)
	case "assistant":
		d.beginAssistantMessage(sessionID)
	}
}

func (d *PiRPCDriver) beginAssistantMessage(sessionID string) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.assistant = newPiAssistantStream()
	d.mu.Unlock()
}

// onMessageUpdate translates streaming deltas. pi emits these frames as:
// {"type":"message_update","usage":{...},"assistantMessageEvent":{...}}.
func (d *PiRPCDriver) onMessageUpdate(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	assistant := d.assistant
	if assistant == nil {
		assistant = newPiAssistantStream()
		d.assistant = assistant
	}
	if len(frame.Usage) > 0 {
		var usage piUsage
		if json.Unmarshal(frame.Usage, &usage) == nil {
			assistant.usage = &usage
		}
	}
	d.mu.Unlock()

	if len(frame.AssistantMessageEvent) == 0 {
		return
	}
	var ame piAssistantMessageEvent
	if json.Unmarshal(frame.AssistantMessageEvent, &ame) != nil {
		return
	}
	d.translateAssistantDelta(sessionID, ame)
}

func (d *PiRPCDriver) translateAssistantDelta(sessionID string, ame piAssistantMessageEvent) {
	d.mu.Lock()
	assistant := d.assistant
	if assistant == nil || !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	idx := ame.ContentIndex
	switch ame.Type {
	case "text_start":
		// Delta follows; nothing to emit.
	case "text_delta":
		delta := ame.Delta
		if delta != "" {
			emitted := assistant.textEmittedLen[idx]
			// text_delta is a delta; append it exactly once.
			assistant.textEmittedLen[idx] = emitted + len(delta)
			d.mu.Unlock()
			d.emit(translatedEvent{Kind: consts.ParsedEventKindText, Text: delta, SessionID: sessionID})
			return
		}
	case "text_end":
		// content is the full text block for this index; emit only the missing
		// suffix vs what text_delta already emitted.
		content := ame.Content
		emitted := assistant.textEmittedLen[idx]
		if len(content) > emitted {
			assistant.textEmittedLen[idx] = len(content)
			d.mu.Unlock()
			d.emit(translatedEvent{Kind: consts.ParsedEventKindText, Text: content[emitted:], SessionID: sessionID})
			return
		}
	case "thinking_start":
		// Delta follows.
	case "thinking_delta":
		// Pi commonly emits one reasoning delta per token. Projecting each token
		// as a AgentTeam thinking event creates a burst of near-identical activity and
		// trajectory frames (DeepSeek can emit dozens for a one-line reply).
		// Wait for thinking_end's authoritative full block instead. If Pi omits
		// thinking_end, finalizeAssistantMessage reconciles the same full block
		// from message_end, so reasoning is still acked exactly once.
	case "thinking_end":
		content := ame.Content
		emitted := assistant.thinkEmittedLen[idx]
		if len(content) > emitted {
			assistant.thinkEmittedLen[idx] = len(content)
			d.mu.Unlock()
			d.emit(translatedEvent{Kind: consts.ParsedEventKindThinking, Text: content[emitted:], SessionID: sessionID})
			return
		}
	case "toolcall_start", "toolcall_delta":
		// The full tool call arrives at toolcall_end; tool_execution_start is the
		// authoritative tool_call projection. Nothing is emitted here to avoid
		// double projection.
	case "toolcall_end":
		// tool_execution_start is the single authoritative tool_call source
		// (it carries toolCallId/toolName/args). Mark the content index so the
		// assistant message snapshot does not re-project it, but do not emit a
		// duplicate tool_call.
		if ame.ToolCall != nil && ame.ToolCall.Name != "" {
			assistant.toolEmitted[idx] = true
		}
	}
	d.mu.Unlock()
}

// onMessageEnd handles a completed message. The authoritative assistant message
// snapshot finalizes text/thinking suffix reconciliation, emits exactly one
// Usage, and records the final stop reason.
func (d *PiRPCDriver) onMessageEnd(sessionID string, frame piRPCFrame) {
	if frame.Message == nil {
		return
	}
	switch strings.TrimSpace(frame.Message.Role) {
	case "user":
		d.ackInputIfMarked(sessionID, frame.Message)
	case "assistant":
		d.mu.Lock()
		active := d.rootRunActive
		d.mu.Unlock()
		if !active {
			return
		}
		d.finalizeAssistantMessage(sessionID, frame.Message)
	}
}

func (d *PiRPCDriver) finalizeAssistantMessage(sessionID string, message *piAgentMessage) {
	blocks := decodePiContentBlocks(message.Content)

	d.mu.Lock()
	assistant := d.assistant
	if assistant == nil {
		assistant = newPiAssistantStream()
		d.assistant = assistant
	}
	// Compute the stable invocation identity first; if this assistant was already
	// finalized in this root generation (e.g. agent_end.messages repeats the same
	// message_end assistant), do not re-emit Usage/text/tool projections.
	assistant.model = firstNonEmptyString(message.ResponseModel, message.Model)
	assistant.responseID = message.ResponseID
	if message.Timestamp != 0 {
		assistant.timestamp = message.Timestamp
	}
	invocationID := piAssistantInvocationID(sessionID, assistant)
	alreadyFinalized := invocationID != "" && d.finalizedAssistantIDs[invocationID]

	// Duplicate snapshots may still refine/confirm final stop/error/model state,
	// but must never re-emit Usage/text/tool projections for a finalized identity.
	if !alreadyFinalized {
		// Reconcile each content block: emit a missing text/thinking suffix, project
		// tool calls that were not already emitted by toolcall_end.
		for idx, block := range blocks {
			switch block.Type {
			case "text":
				emitted := assistant.textEmittedLen[idx]
				if len(block.Text) > emitted {
					assistant.textEmittedLen[idx] = len(block.Text)
					d.mu.Unlock()
					d.emit(translatedEvent{Kind: consts.ParsedEventKindText, Text: block.Text[emitted:], SessionID: sessionID})
					d.mu.Lock()
				}
			case "thinking":
				emitted := assistant.thinkEmittedLen[idx]
				if len(block.Thinking) > emitted {
					assistant.thinkEmittedLen[idx] = len(block.Thinking)
					d.mu.Unlock()
					d.emit(translatedEvent{Kind: consts.ParsedEventKindThinking, Text: block.Thinking[emitted:], SessionID: sessionID})
					d.mu.Lock()
				}
			case "toolCall":
				// tool_execution_start is the single authoritative tool_call source;
				// the assistant message snapshot must not double-project it.
				if block.Name != "" {
					assistant.toolEmitted[idx] = true
				}
			}
		}
	}
	// Capture final message identity + stop reason. The finalized assistant
	// message is authoritative: its usage is the provider-final cumulative value
	// (including zero), even when message_update.usage stayed zero while
	// streaming.
	assistant.stopReason = firstNonEmptyString(message.StopReason, assistant.stopReason)
	if message.Usage != nil {
		assistant.usage = message.Usage
	}
	// Authoritative usage snapshot: emit exactly once per stable invocation
	// identity in this root generation.
	emitUsageNow := false
	if invocationID != "" && !alreadyFinalized {
		if d.finalizedAssistantIDs == nil {
			d.finalizedAssistantIDs = make(map[string]bool)
		}
		d.finalizedAssistantIDs[invocationID] = true
		assistant.usageEmitted = true
		if assistant.usage != nil {
			emitUsageNow = true
		}
	}
	d.rootStopReason = message.StopReason
	assistantFinal := *assistant
	d.assistant = nil
	d.mu.Unlock()
	if emitUsageNow {
		d.emitUsage(sessionID, &assistantFinal)
	}
}

// piAssistantInvocationID derives the stable ModelInvocationID for a finalized
// assistant: the provider responseId when present, otherwise a deterministic
// session:assistantTimestamp identity (matching emitUsage's fallback).
func piAssistantInvocationID(sessionID string, assistant *piAssistantStream) string {
	invocationID := strings.TrimSpace(assistant.responseID)
	if invocationID == "" && assistant.timestamp > 0 {
		invocationID = sessionID + ":" + strconv.FormatInt(assistant.timestamp, 10)
	}
	return invocationID
}

// emitUsage emits exactly one Usage ParsedEvent for a finalized assistant
// message with a stable invocation identity.
func (d *PiRPCDriver) emitUsage(sessionID string, assistant *piAssistantStream) {
	if assistant == nil || assistant.usage == nil {
		return
	}
	usage := assistant.usage
	invocationID := piAssistantInvocationID(sessionID, assistant)
	d.emit(translatedEvent{
		Kind:                consts.ParsedEventKindUsage,
		SessionID:           sessionID,
		InputTokens:         uint64(usage.Input),
		CachedInputTokens:   uint64(usage.CacheRead),
		OutputTokens:        uint64(usage.Output),
		ActiveContextTokens: uint64(usage.TotalTokens),
		Model:               assistant.model,
		ModelInvocationID:   invocationID,
	})
}

// ackInputIfMarked emits runtime_input_acked for a user message whose
// content carries the AgentTeam runtime-input marker, and releases the correlated
// steer-in-flight entry exactly once.
func (d *PiRPCDriver) ackInputIfMarked(sessionID string, message *piAgentMessage) {
	if message == nil {
		return
	}
	inputID := piRuntimeInputIDFromContent(message.Content)
	if inputID == "" {
		return
	}
	d.mu.Lock()
	// Only an input that this driver has accepted may advance durable AgentTeam
	// ownership. This rejects quoted/forged markers and stale transcript frames
	// while still allowing the marker to arrive after root settlement.
	correlated := d.steerInFlight[inputID] || d.promptInputID == inputID || d.initialAckWaiters[inputID] != nil
	if !correlated {
		d.mu.Unlock()
		return
	}
	// Emit exactly once per generation even when message_start and message_end
	// both carry the marker.
	if d.ackedInputIDs == nil {
		d.ackedInputIDs = make(map[string]bool)
	}
	if d.ackedInputIDs[inputID] {
		d.mu.Unlock()
		return
	}
	d.ackedInputIDs[inputID] = true
	delete(d.steerInFlight, inputID)
	delete(d.startupSteerIdleObservedAt, inputID)
	if d.promptInputID == inputID {
		d.promptInputID = ""
	}
	waiter := d.initialAckWaiters[inputID]
	if waiter != nil {
		delete(d.initialAckWaiters, inputID)
	}
	d.mu.Unlock()
	if waiter != nil {
		waiter <- nil
	}
	d.emit(translatedEvent{
		Kind:           consts.ParsedEventKindRuntimeInputAcked,
		RuntimeInputID: inputID,
		SessionID:      sessionID,
	})
}

// onToolExecutionStart projects a tool call from the top-level tool fields.
func (d *PiRPCDriver) onToolExecutionStart(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.mu.Unlock()
	if strings.TrimSpace(frame.ToolName) == "" {
		return
	}
	d.emit(translatedEvent{
		Kind:       consts.ParsedEventKindToolCall,
		ToolName:   frame.ToolName,
		ToolCallID: frame.ToolCallID,
		ToolInput:  string(frame.Args),
		SessionID:  sessionID,
	})
}

// onToolExecutionEnd projects a tool output from the final top-level result.
// partialResult on tool_execution_update is cumulative and intentionally not
// projected per-update to avoid duplicate tool outputs.
func (d *PiRPCDriver) onToolExecutionEnd(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	d.mu.Unlock()
	if strings.TrimSpace(frame.ToolName) == "" {
		return
	}
	text := piToolExecutionResultText(frame.Result)
	if text != "" {
		d.emit(translatedEvent{
			Kind:       consts.ParsedEventKindToolOutput,
			ToolName:   frame.ToolName,
			ToolCallID: frame.ToolCallID,
			Text:       text,
			SessionID:  sessionID,
		})
	}
}

func piToolExecutionResultText(raw json.RawMessage) string {
	if len(raw) == 0 {
		return ""
	}
	var result struct {
		Content []piContentBlock `json:"content"`
	}
	if json.Unmarshal(raw, &result) != nil {
		return ""
	}
	parts := make([]string, 0, len(result.Content))
	for _, block := range result.Content {
		if block.Type == "text" && strings.TrimSpace(block.Text) != "" {
			parts = append(parts, block.Text)
		}
	}
	return strings.Join(parts, "\n")
}

func (d *PiRPCDriver) onCompactionStart(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	d.nativeCompactionInProgress = true
	d.nativeCompactionRootGeneration = 0
	if d.rootRunActive {
		d.nativeCompactionRootGeneration = d.rootRunGeneration
	}
	d.mu.Unlock()
	d.emit(translatedEvent{
		Kind:      consts.ParsedEventKindRuntimeCompacting,
		SessionID: sessionID,
		Metadata:  map[string]any{"pi_event": "compaction_start", "reason": frame.Reason},
	})
}

// nativeCompactionCanIdleLocked proves that compaction_end is not racing a
// still-live Pi lifecycle. A root run, prompt preflight, or any input
// correlation can still produce work after compaction_end; publishing
// RuntimeIdle in those cases would make Process flush an active turn or queued
// delivery. When compaction_start was observed inside a root, its generation
// remains a fence after rootRunActive drops so a late end cannot duplicate the
// root's agent_settled boundary.
func (d *PiRPCDriver) nativeCompactionCanIdleLocked() bool {
	if d.nativeCompactionRootGeneration != 0 {
		return false
	}
	return !d.rootRunActive &&
		d.promptAccepted == nil &&
		strings.TrimSpace(d.promptInputID) == "" &&
		len(d.steerInFlight) == 0 &&
		len(d.initialAckWaiters) == 0
}

// onCompactionEnd only confirms a genuinely successful, non-aborted completion.
// Aborted/failed results preserve the reason/detail; a willRetry completion
// keeps the root active (reporting retrying observability) without a false
// terminal error, while a final failure updates the root error outcome for
// settlement. Manual/overflow compaction paths never manufacture a retry
// continuation on false success. A successful standalone native completion
// (including the benign short-session no-op) emits RuntimeIdle only after the
// driver proves that no root, preflight, or correlated delivery remains.
func (d *PiRPCDriver) onCompactionEnd(sessionID string, frame piRPCFrame) {
	// A short session is a benign no-op; an active root's agent_end/settled still
	// closes the run. A standalone no-op needs its own RuntimeIdle boundary.
	if frame.CompactionTooSmallNoop() {
		text := strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(frame.ErrorMessage), "Compaction failed:"))
		d.mu.Lock()
		canIdle := !frame.WillRetry && d.nativeCompactionCanIdleLocked()
		d.nativeCompactionInProgress = false
		d.mu.Unlock()
		metadata := map[string]any{"pi_event": "compaction_end", "reason": frame.Reason, "noop": true}
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindThinking,
			Text:      text,
			SessionID: sessionID,
			Metadata:  metadata,
		})
		if canIdle {
			d.emit(translatedEvent{Kind: consts.ParsedEventKindRuntimeIdle, SessionID: sessionID, Metadata: metadata})
		}
		return
	}
	successful := !frame.Aborted && strings.TrimSpace(frame.ErrorMessage) == "" && frame.CompactionResultPresent()
	d.mu.Lock()
	confirmed := false
	if successful {
		d.contextCompacted = true
		confirmed = true
	}
	// A willRetry completion means the agent will re-run; do not emit a
	// terminal error now, but do not confirm context either (the retry is the
	// continuation).
	if frame.WillRetry {
		confirmed = false
	}
	canIdle := successful && !frame.WillRetry && d.nativeCompactionCanIdleLocked()
	// Consume the standalone compaction lifecycle before releasing the lock.
	// A root generation fence is retained so a late duplicate compaction_end
	// cannot publish a second idle boundary after agent_settled.
	d.nativeCompactionInProgress = false
	if frame.Aborted {
		// Aborted compaction: preserve the reason, keep the root active.
		d.mu.Unlock()
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindThinking,
			Text:      "Pi context compaction aborted",
			SessionID: sessionID,
			Metadata:  map[string]any{"pi_event": "compaction_end", "reason": frame.Reason, "aborted": true},
		})
		return
	}
	if strings.TrimSpace(frame.ErrorMessage) != "" && !frame.WillRetry {
		// Final compaction failure: update the root error outcome. The Error
		// event below is the single projection of this final failure.
		d.rootError = firstNonEmptyString(d.rootError, frame.ErrorMessage)
		d.rootErrorEmitted = true
	}
	d.mu.Unlock()

	metadata := map[string]any{"pi_event": "compaction_end", "reason": frame.Reason}
	if confirmed {
		metadata[contextCompactionConfirmedMetadataKey] = true
		metadata["confirmed"] = true
	} else if frame.WillRetry {
		metadata["willRetry"] = true
	}
	text := "Pi context compaction completed"
	if frame.WillRetry {
		text = "Pi context compaction will retry"
	} else if !successful && strings.TrimSpace(frame.ErrorMessage) != "" {
		text = "Pi context compaction failed: " + frame.ErrorMessage
	}
	d.emit(translatedEvent{
		Kind:      consts.ParsedEventKindThinking,
		Text:      text,
		SessionID: sessionID,
		Metadata:  metadata,
	})
	if !successful && !frame.WillRetry && strings.TrimSpace(frame.ErrorMessage) != "" {
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindError,
			Message:   firstNonEmptyString(frame.ErrorMessage, "Pi context compaction failed"),
			SessionID: sessionID,
			Metadata:  map[string]any{"will_retry": false},
		})
	}
	if canIdle {
		d.emit(translatedEvent{Kind: consts.ParsedEventKindRuntimeIdle, SessionID: sessionID, Metadata: metadata})
	}
}

func (d *PiRPCDriver) onQueueUpdate(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	active := d.rootRunActive
	d.mu.Unlock()
	if !active {
		return
	}
	if len(frame.Steering) == 0 && len(frame.FollowUp) == 0 {
		return
	}
	parts := make([]string, 0, len(frame.Steering)+len(frame.FollowUp))
	for _, s := range frame.Steering {
		parts = append(parts, "steering: "+s)
	}
	for _, f := range frame.FollowUp {
		parts = append(parts, "follow-up: "+f)
	}
	d.emit(translatedEvent{
		Kind:      consts.ParsedEventKindThinking,
		Text:      strings.Join(parts, "\n"),
		SessionID: sessionID,
		Metadata:  map[string]any{"pi_event": "queue_update"},
	})
}

func (d *PiRPCDriver) onAutoRetryStart(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	active := d.rootRunActive
	d.mu.Unlock()
	if !active {
		return
	}
	d.emit(translatedEvent{
		Kind:      consts.ParsedEventKindThinking,
		Text:      fmt.Sprintf("Pi retry %d/%d", frame.Attempt, frame.MaxAttempts),
		SessionID: sessionID,
		Metadata: map[string]any{
			"pi_event": "auto_retry_start",
			"attempt":  frame.Attempt,
			"max":      frame.MaxAttempts,
		},
	})
}

func (d *PiRPCDriver) onAutoRetryEnd(sessionID string, frame piRPCFrame) {
	// `success` on auto_retry_end is the retry outcome. A successful retry clears
	// any prior error so the final root result is not poisoned by the earlier
	// transient failure. A final failure updates the root error outcome for
	// settlement.
	failed := frame.Success == nil || !*frame.Success
	d.mu.Lock()
	if !d.rootRunActive {
		d.mu.Unlock()
		return
	}
	if failed {
		detail := firstNonEmptyString(frame.FinalError, "Pi automatic retry failed")
		if d.rootError == "" {
			d.rootError = detail
		}
		// The Error event below is the single projection of this final failure.
		d.rootErrorEmitted = true
	} else {
		// Successful retry: the retried run replaces the final outcome.
		d.rootError = ""
		d.rootErrorEmitted = false
	}
	d.mu.Unlock()
	if failed {
		d.emit(translatedEvent{
			Kind:      consts.ParsedEventKindError,
			Message:   firstNonEmptyString(frame.FinalError, "Pi automatic retry failed"),
			SessionID: sessionID,
			Metadata:  map[string]any{"will_retry": false},
		})
	}
}

func (d *PiRPCDriver) onBashExecutionUpdate(sessionID string, frame piRPCFrame) {
	d.mu.Lock()
	active := d.rootRunActive
	d.mu.Unlock()
	if !active {
		return
	}
	if strings.TrimSpace(frame.Delta) == "" {
		return
	}
	d.emit(translatedEvent{
		Kind:       consts.ParsedEventKindToolOutput,
		ToolName:   "shell",
		ToolCallID: frame.IDString(),
		Text:       frame.Delta,
		SessionID:  sessionID,
	})
}

func (d *PiRPCDriver) onExtensionError(sessionID string, frame piRPCFrame) {
	d.onFatalRuntimeError(sessionID, firstNonEmptyString(frame.Error, frame.ErrorMessage, "Pi extension error"))
}

// onRuntimeError handles provider/model startup failures emitted as a generic
// event by Pi versions that do not wrap the diagnostic in agent_end. The error
// is tied to the active root generation so a late frame cannot poison a later
// prompt or fabricate a second terminal error.
func (d *PiRPCDriver) onRuntimeError(sessionID string, frame piRPCFrame) {
	d.onFatalRuntimeError(sessionID, firstNonEmptyString(frame.Error, frame.ErrorMessage, "Pi runtime error"))
}

func (d *PiRPCDriver) onFatalRuntimeError(sessionID, message string) {
	d.mu.Lock()
	initialStartup := !d.rootRunActive && d.rootRunGeneration == 0
	if !d.rootRunActive && !initialStartup {
		d.mu.Unlock()
		return
	}
	message = firstNonEmptyString(strings.TrimSpace(message), "Pi runtime error")
	if initialStartup {
		// Preserve a provider/model/catalog diagnostic that arrives before the
		// first prompt. There is no root terminal boundary yet, but Process
		// still needs the event for StartupFailure and retry classification.
		d.rootError = firstNonEmptyString(d.rootError, message)
	} else {
		d.rootError = firstNonEmptyString(d.rootError, message)
		d.rootStopReason = "error"
	}
	alreadyEmitted := d.rootErrorEmitted
	d.rootErrorEmitted = true
	d.mu.Unlock()
	if alreadyEmitted {
		return
	}
	d.emit(translatedEvent{
		Kind:      consts.ParsedEventKindError,
		Message:   message,
		SessionID: sessionID,
		Metadata:  map[string]any{"will_retry": false},
	})
}

// piStopReasonForLoop maps a pi stop reason to a AgentTeam ParsedEvent stop reason.
func piStopReasonForLoop(piReason string) string {
	switch strings.TrimSpace(piReason) {
	case "stop":
		return consts.ParsedEventStopReasonEndTurn
	case "length":
		return consts.ParsedEventStopReasonMaxTokens
	case "toolUse":
		return consts.ParsedEventStopReasonToolUse
	case "aborted":
		return "aborted"
	case "error":
		return "error"
	default:
		return consts.ParsedEventStopReasonEndTurn
	}
}

func decodePiContentBlocks(raw json.RawMessage) []piContentBlock {
	if len(raw) == 0 {
		return nil
	}
	// content may be a plain string (user message) or an array of blocks.
	var blocks []piContentBlock
	if json.Unmarshal(raw, &blocks) == nil {
		return blocks
	}
	return nil
}