kandev/apps/backend/internal/agent/runtime/lifecycle/manager_events.go

package lifecycle

import (
	"context"
	"strings"
	"time"

	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"

	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
	"github.com/kandev/kandev/internal/agentctl/tracing"
	"github.com/kandev/kandev/internal/agentctl/types/streams"
	"github.com/kandev/kandev/internal/events"
	v1 "github.com/kandev/kandev/pkg/api/v1"
)

const (
	toolStatusComplete = "complete"
	toolStatusFailed   = "failed"
)

// handleMessageChunkEvent handles a "message_chunk" agent event, accumulating and flushing on newlines.
func (m *Manager) handleMessageChunkEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	if event.Role == "user" || event.Text == "" {
		return
	}
	// ACP message chunks do not carry the lifecycle prompt generation. The
	// messageMu acquisition is therefore the only reliable turn boundary: a
	// chunk observed before an atomic reset is detached with the old turn,
	// while one observed after reset is retained for the replacement turn.
	m.appendAssistantHistoryChunk(execution, event.Text)
	if event.ProtocolMessageID != "" {
		m.flushPendingLegacyMessage(execution)
		m.publishProtocolMessage(execution, event.ProtocolMessageID, event.Text)
		return
	}
	execution.messageMu.Lock()
	execution.messageBuffer.WriteString(event.Text)
	bufferLenAfterWrite := execution.messageBuffer.Len()
	m.logger.Debug("message_chunk written to buffer",
		zap.String("execution_id", execution.ID),
		zap.String("operation_id", event.OperationID),
		zap.Int("text_length", len(event.Text)),
		zap.Int("buffer_length_after", bufferLenAfterWrite))

	bufContent := execution.messageBuffer.String()
	lastNewline := strings.LastIndex(bufContent, "\n")
	if lastNewline == -1 {
		execution.messageMu.Unlock()
		return
	}
	toFlush := bufContent[:lastNewline+1]
	remainder := bufContent[lastNewline+1:]
	execution.messageBuffer.Reset()
	execution.messageBuffer.WriteString(remainder)
	execution.messageMu.Unlock()

	if strings.TrimSpace(toFlush) != "" {
		m.publishStreamingMessage(execution, toFlush)
	}
}

// handleReasoningEvent handles a "reasoning" agent event, accumulating and flushing on newlines.
func (m *Manager) handleReasoningEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	if event.ReasoningText == "" {
		return
	}
	if event.ProtocolMessageID != "" {
		m.flushPendingLegacyThinking(execution)
		m.publishProtocolThinking(execution, event.ProtocolMessageID, event.ReasoningText)
		return
	}
	execution.messageMu.Lock()
	execution.thinkingBuffer.WriteString(event.ReasoningText)

	bufContent := execution.thinkingBuffer.String()
	lastNewline := strings.LastIndex(bufContent, "\n")
	if lastNewline == -1 {
		execution.messageMu.Unlock()
		return
	}
	toFlush := bufContent[:lastNewline+1]
	remainder := bufContent[lastNewline+1:]
	execution.thinkingBuffer.Reset()
	execution.thinkingBuffer.WriteString(remainder)
	execution.messageMu.Unlock()

	if strings.TrimSpace(toFlush) != "" {
		m.publishStreamingThinking(execution, toFlush)
	}
}

// extractErrorMessage returns the best error message from an agent event.
// Priority: Error field > Text field > default message.
func extractErrorMessage(event *agentctl.AgentEvent) string {
	if event.Error != "" {
		return event.Error
	}
	if event.Text != "" {
		return event.Text
	}
	return "agent error completion"
}

// handleCompleteEventMarkState marks the execution state after a complete event:
// failed+removed on error, ready on success.
func (m *Manager) handleCompleteEventMarkState(execution *AgentExecution, event *agentctl.AgentEvent, isError bool) {
	if isError {
		errorMsg := extractErrorMessage(event)
		// A turn aborted by backend graceful shutdown is not an agent failure.
		// Redirect it to a benign stop so the session stays resumable and the UI
		// shows no red error banner. MarkCompleted applies the same guard, but
		// routing here keeps the misleading "marking as failed" WARN out of logs.
		if m.IsShuttingDown() {
			_ = m.markStoppedDuringShutdown(execution, 1, errorMsg)
			return
		}
		m.logger.Warn("error completion received, marking execution as failed",
			zap.String("execution_id", execution.ID),
			zap.String("task_id", execution.TaskID),
			zap.String("error", errorMsg),
			zap.String("event_error", event.Error),
			zap.String("event_text", event.Text),
			zap.Any("event_data", event.Data),
			zap.String("agent_command", execution.AgentCommand),
			zap.String("acp_session_id", execution.ACPSessionID))
		if err := m.MarkCompleted(execution.ID, 1, errorMsg); err != nil {
			m.logger.Error("failed to mark execution as failed after error completion",
				zap.String("execution_id", execution.ID),
				zap.Error(err))
		}
		return
	}
	// Empty wakeup turn fallback: if the wakeup-driven turn produced no
	// turn-content events (no message_chunk/tool_call/etc), recordActivity
	// won't have flipped Ready → Running. MarkReady would then early-return
	// on the Ready guard and silently drop AgentReady — same suppression
	// the wakeup fix is designed to prevent.
	//
	// Publishing AgentReady alone is not enough: the orchestrator's
	// handleAgentReady (`event_handlers_agent.go:205`) ignores AgentReady
	// when session.State is not Running/Starting, and after the previous
	// turn ended the session is WaitingForInput. We mirror what
	// recordActivity does for the non-empty case — flip Ready → Running
	// and publish AgentRunning first, so the orchestrator's session state
	// catches up — then let MarkReady do its normal Running → Ready
	// transition and publish AgentReady.
	if execution.Status == v1.AgentStatusReady {
		m.logger.Info("flipping Ready→Running for empty wakeup turn before publishing AgentReady",
			zap.String("execution_id", execution.ID),
			zap.String("session_id", execution.SessionID))
		if err := m.UpdateStatus(execution.ID, v1.AgentStatusRunning); err != nil {
			m.logger.Warn("failed to persist empty wakeup turn running status",
				zap.String("execution_id", execution.ID),
				zap.Error(err))
		}
		m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution)
	}
	if err := m.MarkReady(execution.ID); err != nil {
		m.logger.Error("failed to mark execution as ready after complete",
			zap.String("execution_id", execution.ID),
			zap.Error(err))
	}
}

// handleCompleteEventSignal sends the completion signal on the promptDoneCh channel.
func handleCompleteEventSignal(execution *AgentExecution, event *agentctl.AgentEvent, isError bool) {
	stopReason := "end_turn"
	errorMsg := ""
	if isError {
		stopReason = "error"
		errorMsg = extractErrorMessage(event)
	} else if event.Data != nil {
		// Read StopReason from the complete event (set by ACP adapter from PromptResponse)
		if sr, ok := event.Data["stop_reason"].(string); ok && sr != "" {
			stopReason = sr
		}
	}
	select {
	case execution.promptDoneCh <- PromptCompletionSignal{
		StopReason:       stopReason,
		IsError:          isError,
		Error:            errorMsg,
		PromptGeneration: event.PromptGeneration,
	}:
	default:
		// Channel full or no one waiting — that's fine (e.g., initial prompt in goroutine)
	}
}

type promptCompletionClaim struct {
	execution      *AgentExecution
	readyPayload   AgentEventPayload
	runningPayload AgentEventPayload
	publishRunning bool
	locked         bool
}

func (m *Manager) claimPromptCompletion(
	execution *AgentExecution,
	event *agentctl.AgentEvent,
	isError bool,
) (promptCompletionClaim, bool) {
	claim := promptCompletionClaim{}
	if event.PromptGeneration == 0 {
		return claim, true
	}

	execution.promptLifecycleMu.Lock()
	claim.locked = true
	claimed := false
	err := m.executionStore.WithLock(execution.ID, func(current *AgentExecution) {
		if current != execution || current.promptGeneration != event.PromptGeneration {
			return
		}
		if current.promptCompletionGeneration == event.PromptGeneration {
			return
		}
		current.promptCompletionGeneration = event.PromptGeneration
		claimed = true
		claim.execution = current
		if isError {
			return
		}
		if current.Status != v1.AgentStatusReady {
			current.firstActivityOnce.Do(func() {
				claim.publishRunning = true
				claim.runningPayload = newAgentEventPayload(current)
			})
		}
		current.Status = v1.AgentStatusReady
		claim.readyPayload = newAgentEventPayload(current)
	})
	if err == nil && claimed {
		return claim, true
	}

	execution.promptLifecycleMu.Unlock()
	m.logger.Debug("ignoring completion for superseded prompt generation",
		zap.String("execution_id", execution.ID),
		zap.Uint64("event_prompt_generation", event.PromptGeneration))
	return promptCompletionClaim{}, false
}

func completeEventResult(event *agentctl.AgentEvent) (bool, string) {
	isError := false
	if event.Data != nil {
		isError, _ = event.Data["is_error"].(bool)
	}
	if isError {
		return true, "error"
	}
	if event.Data != nil {
		if stopReason, ok := event.Data["stop_reason"].(string); ok && stopReason != "" {
			return false, stopReason
		}
	}
	return false, "end_turn"
}

func (m *Manager) finishPromptCompletion(
	execution *AgentExecution,
	event *agentctl.AgentEvent,
	isError bool,
	claim promptCompletionClaim,
) {
	handleCompleteEventSignal(execution, event, isError)
	if event.PromptGeneration == 0 || isError {
		if isError {
			setProviderError(execution, event.ProviderError)
		}
		m.handleCompleteEventMarkState(execution, event, isError)
		if claim.locked {
			execution.promptLifecycleMu.Unlock()
		}
		return
	}

	m.persistExecutorRunning(context.Background(), claim.execution)
	execution.promptLifecycleMu.Unlock()
	if claim.publishRunning {
		m.eventPublisher.publishAgentEventPayload(context.Background(), events.AgentRunning, claim.runningPayload)
	}
	m.eventPublisher.publishAgentEventPayload(context.Background(), events.AgentReady, claim.readyPayload)
}

func setProviderError(execution *AgentExecution, providerError *streams.ProviderError) {
	if execution == nil || providerError == nil || !providerError.Valid() {
		return
	}
	copy := *providerError
	if providerError.ResetAt != nil {
		resetAt := *providerError.ResetAt
		copy.ResetAt = &resetAt
	}
	execution.ProviderError = &copy
}

// handleCompleteEvent handles a "complete" agent event: flushes buffers, marks state, and signals SendPrompt.
func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl.AgentEvent) bool {
	isError, stopReason := completeEventResult(event)
	claim, claimed := m.claimPromptCompletion(execution, event, isError)
	if !claimed {
		return false
	}
	m.releaseActivity(executionActivityKey(execution.ID))

	execution.lastActivityAtMu.Lock()
	execution.lastActivityAt = time.Now()
	execution.lastActivityAtMu.Unlock()

	// Check buffer content BEFORE any processing
	execution.messageMu.Lock()
	bufferContentBeforeFlush := execution.messageBuffer.String()
	currentMsgID := execution.currentMessageID
	execution.messageMu.Unlock()

	bufferPreview := bufferContentBeforeFlush
	if len(bufferPreview) > 100 {
		bufferPreview = bufferPreview[:100] + "..."
	}

	// Create a turn_end span on the session trace
	_, turnSpan := tracing.TraceTurnEnd(execution.SessionTraceContext(), execution.ID, execution.SessionID)
	turnSpan.SetAttributes(
		attribute.String("stop_reason", stopReason),
		attribute.Bool("is_error", isError),
	)
	turnSpan.End()

	m.logger.Info("agent turn complete",
		zap.String("execution_id", execution.ID),
		zap.String("operation_id", event.OperationID),
		zap.String("session_id", event.SessionID),
		zap.String("current_msg_id", currentMsgID),
		zap.Int("buffer_length", len(bufferContentBeforeFlush)),
		zap.String("buffer_preview", bufferPreview),
		zap.Bool("is_error", isError))

	// Flush the message buffer to publish any remaining content as a streaming message.
	flushedText := m.flushMessageBuffer(execution)
	execution.messageMu.Lock()
	execution.clearProtocolMessageCorrelationLocked()
	execution.messageMu.Unlock()
	if flushedText != "" {
		event.Text = flushedText
	}
	m.flushAssistantHistory(execution)

	m.logger.Info("complete event processed",
		zap.String("execution_id", execution.ID),
		zap.String("operation_id", event.OperationID))

	// Signal promptDoneCh BEFORE publishing agent.ready via MarkReady.
	// MarkReady publishes agent.ready synchronously, which triggers handleAgentReady
	// in the orchestrator. If handleAgentReady launches a follow-up prompt (queued
	// message), the new SendPrompt drains promptDoneCh. If we signal AFTER MarkReady,
	// the drain races with the first SendPrompt's receive and can steal the signal,
	// leaving the first SendPrompt hung and the second prompt's completion event
	// never reaching the event bus.
	m.finishPromptCompletion(execution, event, isError, claim)
	return true
}

// handleToolCallEvent processes the "tool_call" agent event: flushes the message buffer
// and stores the tool call in session history.
// Returns the (possibly updated) event.
//
// Subagent-internal tool calls (ParentToolCallID set) stream on the same session
// concurrently with the parent agent's own text, so they must NOT flush the
// buffer — flushing would split the parent's in-flight streaming message into
// separate DB rows mid-sentence, breaking markdown that spans the boundary.
func (m *Manager) handleToolCallEvent(execution *AgentExecution, event agentctl.AgentEvent) agentctl.AgentEvent {
	if event.ParentToolCallID == "" {
		execution.setActiveTool(activeTopLevelTool{
			ToolCallID: event.ToolCallID,
			Name:       event.ToolName,
			Title:      event.ToolTitle,
			Status:     event.ToolStatus,
		})
		// flushMessageBuffer publishes any remaining buffered content through
		// the streaming path itself and always returns "".
		m.flushMessageBuffer(execution)
	}
	m.flushAssistantHistory(execution)
	if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" {
		if err := m.historyManager.AppendToolCall(execution.SessionID, event); err != nil {
			m.logger.Warn("failed to store tool call to history", zap.Error(err))
		}
	}
	m.logger.Debug("tool call started",
		zap.String("execution_id", execution.ID),
		zap.String("tool_call_id", event.ToolCallID),
		zap.String("tool_name", event.ToolName))
	return event
}

// handleToolUpdateEvent stores completed tool results in session history.
func (m *Manager) handleToolUpdateEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	if event.ParentToolCallID == "" && isTerminalToolUpdate(event) {
		m.flushStreamCoalescer(execution)
		execution.clearActiveTool(event.ToolCallID)
	}
	if m.historyManager != nil && execution.historyEnabled && execution.SessionID != "" && event.ToolStatus == toolStatusComplete {
		m.flushAssistantHistory(execution)
		if err := m.historyManager.AppendToolResult(execution.SessionID, event); err != nil {
			m.logger.Warn("failed to store tool result to history", zap.Error(err))
		}
	}
}

// handleErrorEvent processes a raw "error" as an error completion so generation
// ownership remains held across validation, buffer flushing, and state mutation.
// The raw error event is not published to the frontend stream; the agent failure
// path (handleAgentFailed) sets session FAILED with the error message.
func (m *Manager) handleErrorEvent(execution *AgentExecution, event agentctl.AgentEvent) bool {
	data := make(map[string]any, len(event.Data)+1)
	for key, value := range event.Data {
		data[key] = value
	}
	data["is_error"] = true
	event.Data = data
	return m.handleCompleteEvent(execution, &event)
}

// handleContextWindowEvent processes the "context_window" agent event: logs and publishes it.
// Returns true because no further stream publishing is needed.
func (m *Manager) handleContextWindowEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	m.logger.Debug("context window update received",
		zap.String("execution_id", execution.ID),
		zap.Int64("size", event.ContextWindowSize),
		zap.Int64("used", event.ContextWindowUsed),
		zap.Float64("efficiency", event.ContextEfficiency))
	m.eventPublisher.PublishContextWindow(
		execution,
		event.ContextWindowSize,
		event.ContextWindowUsed,
		event.ContextWindowRemaining,
		event.ContextEfficiency,
	)
}

// handleAvailableCommandsEvent processes the "available_commands" agent event.
func (m *Manager) handleAvailableCommandsEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	if len(event.AvailableCommands) == 0 {
		return
	}
	execution.SetAvailableCommands(event.AvailableCommands)
	m.logger.Debug("stored available commands",
		zap.String("execution_id", execution.ID),
		zap.String("session_id", execution.SessionID),
		zap.Int("command_count", len(event.AvailableCommands)))
	m.eventPublisher.PublishAvailableCommands(execution, event.AvailableCommands)
}

// turnContentEventTypes is the set of agent event types that unambiguously
// signal a real turn is in progress — assistant text, reasoning, tool work,
// plan/permission updates. These are the only events that drive the
// Ready → Running flip for wakeup-driven turns; boot/metadata events
// (agent_capabilities, available_commands, session_mode/models/status,
// context_window, etc.) can arrive *after* MarkBootReady has put the
// execution into Ready (e.g. claude-agent-acp emits available_commands
// asynchronously ~50ms after session/new, well after dispatchInitialPrompt
// has fired MarkBootReady for a no-prompt task) and must NOT be treated
// as turn starts. Terminal events (complete/error) are excluded too —
// they own their own status transitions via handleCompleteEvent and a
// dedicated empty-turn fallback in handleCompleteEventMarkState.
var turnContentEventTypes = map[string]struct{}{
	"message_chunk":      {},
	"reasoning":          {},
	"tool_call":          {},
	"tool_update":        {},
	"plan":               {},
	"agent_plan":         {},
	"permission_request": {},
}

func isTerminalToolUpdate(event agentctl.AgentEvent) bool {
	if event.Type != "tool_update" {
		return false
	}
	switch event.ToolStatus {
	case toolStatusComplete, "completed", "success", "error", toolStatusFailed, "cancelled":
		return true
	default:
		return false
	}
}

// recordActivity updates the last-activity timestamp and, on the very first
// event from an execution, publishes AgentRunning to transition STARTING → RUNNING.
//
// Wakeup-driven turns also flip Ready → Running here. The adapter's wakeup
// scheduler fires synthetic prompts directly via Adapter.Prompt, bypassing
// SessionManager.SendPrompt — so nothing on the lifecycle side flips the
// execution back to Running for those turns. Without that flip, MarkReady's
// duplicate-suppression guard (manager_interaction.go:896 — early-returns
// when execution.Status is already Ready) silently drops the wakeup turn's
// AgentReady event, the orchestrator never calls completeTurnForSession,
// and workflow on_turn_complete + queued-message dispatch silently break.
//
// The flip is gated on `turnContentEventTypes` so post-boot metadata events
// (available_commands_update arriving 50ms after MarkBootReady, etc.) don't
// accidentally re-arm a freshly-booted no-prompt session as Running.
func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.AgentEvent) {
	execution.lastActivityAtMu.Lock()
	execution.lastActivityAt = time.Now()
	execution.lastActivityAtMu.Unlock()

	// Gate firstActivityOnce on `Status != Ready` so a delayed metadata
	// event arriving after MarkBootReady can't accidentally fire
	// AgentRunning. In practice the adapter always emits agent_capabilities
	// during Initialize (before MarkBootReady), so firstActivityOnce fires
	// while Status is still Running — this is defensive hardening.
	if execution.Status != v1.AgentStatusReady {
		execution.firstActivityOnce.Do(func() {
			m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution)
		})
		return
	}
	if m.executionStore == nil {
		return
	}
	if isTerminalToolUpdate(event) {
		return
	}
	if _, ok := turnContentEventTypes[event.Type]; !ok {
		return
	}
	if err := m.UpdateStatus(execution.ID, v1.AgentStatusRunning); err != nil {
		m.logger.Warn("failed to persist wakeup-driven running status",
			zap.String("execution_id", execution.ID),
			zap.Error(err))
		return
	}
	m.logger.Info("wakeup-driven turn detected; flipping execution back to Running",
		zap.String("execution_id", execution.ID),
		zap.String("session_id", execution.SessionID),
		zap.String("trigger_event_type", event.Type))
	m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution)
}

// handleStreamDisconnect handles unexpected updates stream disconnections.
// It proactively updates execution status and publishes an error event so the
// orchestrator can transition the session state without waiting for a future prompt.
func (m *Manager) handleStreamDisconnect(
	execution *AgentExecution,
	err error,
	promptGeneration uint64,
) {
	disconnectFields := []zap.Field{
		zap.String("execution_id", execution.ID),
		zap.String("session_id", execution.SessionID),
		zap.Uint64("prompt_generation", promptGeneration),
		zap.Error(err),
	}
	// A disconnect during graceful shutdown is expected: StopAllAgents stops the
	// agent process, which drops the agentctl WebSocket. Logging it at WARN turns
	// routine teardown into noise, so downgrade to DEBUG while shutting down. The
	// failed-status/error-event handling below is unchanged either way.
	if m.IsShuttingDown() {
		m.logger.Debug("agent updates stream disconnected during shutdown", disconnectFields...)
	} else {
		m.logger.Warn("agent updates stream disconnected", disconnectFields...)
	}

	if promptGeneration != 0 {
		execution.promptLifecycleMu.Lock()
		defer execution.promptLifecycleMu.Unlock()

		var claimed bool
		var updated *AgentExecution
		statusErr := m.executionStore.WithLock(execution.ID, func(current *AgentExecution) {
			if current != execution || current.promptGeneration != promptGeneration {
				return
			}
			current.Status = v1.AgentStatusFailed
			updated = current
			claimed = true
		})
		if statusErr != nil {
			m.logger.Warn("failed to persist stream disconnect failed status",
				zap.String("execution_id", execution.ID),
				zap.Error(statusErr))
			return
		}
		if !claimed {
			m.logger.Debug("ignoring stream disconnect for superseded prompt generation",
				zap.String("execution_id", execution.ID),
				zap.Uint64("disconnect_prompt_generation", promptGeneration))
			return
		}

		m.flushMessageBuffer(execution)
		m.flushAssistantHistory(execution)
		m.persistExecutorRunning(context.Background(), updated)
		m.publishStreamDisconnectError(execution, err)
		return
	}

	// The stream callback may race with the caller starting another prompt
	// after promptDoneCh is signaled. Drain the partial assistant transcript
	// here as well as at prompt setup; the shared buffer lock makes either
	// path the single owner and prevents a later reset from dropping it.
	m.flushMessageBuffer(execution)
	m.flushAssistantHistory(execution)

	if err := m.UpdateStatus(execution.ID, v1.AgentStatusFailed); err != nil {
		m.logger.Warn("failed to persist stream disconnect failed status",
			zap.String("execution_id", execution.ID),
			zap.Error(err))
	}

	m.publishStreamDisconnectError(execution, err)
}

func (m *Manager) publishStreamDisconnectError(execution *AgentExecution, err error) {
	m.eventPublisher.PublishAgentctlEvent(
		context.Background(), events.AgentctlError, execution,
		"agent stream disconnected: "+err.Error(),
	)
}

// handlePromptHandoffEvent finalizes lifecycle's shared streaming-buffer and
// completion-wait ownership at the same generation-bearing boundary where the
// ACP adapter made one human successor eligible to inherit prompt ownership.
// It does not mark the execution Ready: foreground/background activity and
// coarse turn completion remain separate orchestrator/provider concerns.
func (m *Manager) handlePromptHandoffEvent(
	execution *AgentExecution,
	event agentctl.AgentEvent,
) {
	handoff, _ := event.Data[streams.AgentEventDataPromptHandoff].(bool)
	if !handoff || event.PromptGeneration == 0 {
		return
	}

	execution.promptLifecycleMu.Lock()
	defer execution.promptLifecycleMu.Unlock()

	ownsGeneration := false
	err := m.executionStore.WithLock(execution.ID, func(current *AgentExecution) {
		ownsGeneration = current == execution &&
			current.promptGeneration == event.PromptGeneration
	})
	if err != nil || !ownsGeneration {
		m.logger.Debug("ignoring prompt handoff for superseded generation",
			zap.String("execution_id", execution.ID),
			zap.Uint64("event_prompt_generation", event.PromptGeneration))
		return
	}

	m.flushMessageBuffer(execution)
	execution.messageMu.Lock()
	execution.clearProtocolMessageCorrelationLocked()
	execution.messageMu.Unlock()
	m.flushAssistantHistory(execution)

	select {
	case execution.promptDoneCh <- PromptCompletionSignal{
		StopReason:       streams.EventTypeForegroundIdle,
		PromptGeneration: event.PromptGeneration,
	}:
	default:
		m.logger.Warn("prompt handoff could not signal lifecycle waiter",
			zap.String("execution_id", execution.ID),
			zap.Uint64("prompt_generation", event.PromptGeneration))
	}
}

// handleAgentEvent processes incoming agent events from the agent
func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) {
	if event.Type == streams.EventTypeMCPAttachment {
		if event.MCPAttachmentAttempt != nil {
			if event.MCPAttachmentAttempt.ExecutionID != "" && event.MCPAttachmentAttempt.ExecutionID != execution.ID {
				m.logger.Warn("dropping stale MCP attachment attempt",
					zap.String("event_execution_id", event.MCPAttachmentAttempt.ExecutionID),
					zap.String("live_execution_id", execution.ID))
				return
			}
			attempt := *event.MCPAttachmentAttempt
			attempt.TaskID = execution.TaskID
			attempt.SessionID = execution.SessionID
			attempt.ExecutionID = execution.ID
			attempt.AgentID = execution.AgentID
			event.MCPAttachmentAttempt = &attempt
		}
		// Attachment diagnostics must not count as model activity or alter turn
		// ownership. They flow directly to the orchestrator for persistence.
		m.eventPublisher.PublishAgentStreamEvent(execution, event)
		return
	}
	if event.PromptGeneration == 0 || (event.Type != toolStatusComplete && event.Type != "error") {
		m.recordActivity(execution, event)
	}

	m.logger.Debug("handleAgentEvent entry",
		zap.String("execution_id", execution.ID),
		zap.String("event_type", event.Type),
		zap.String("operation_id", event.OperationID),
		zap.Int("text_length", len(event.Text)))

	switch event.Type {
	case "message_chunk":
		m.handleMessageChunkEvent(execution, event)
		return

	case "reasoning":
		m.handleReasoningEvent(execution, event)
		return

	case "tool_call":
		event = m.handleToolCallEvent(execution, event)

	case "tool_update":
		m.handleToolUpdateEvent(execution, event)

	case "plan":
		m.logger.Debug("agent plan update",
			zap.String("execution_id", execution.ID))

	case "error":
		m.handleErrorEvent(execution, event)
		return

	case toolStatusComplete:
		if !m.handleCompleteEvent(execution, &event) {
			return
		}

	case "permission_request":
		m.logger.Debug("permission request received",
			zap.String("execution_id", execution.ID),
			zap.String("pending_id", event.PendingID),
			zap.String("title", event.PermissionTitle))
		m.eventPublisher.PublishPermissionRequest(execution, event)
		return

	case "context_window":
		m.handleContextWindowEvent(execution, event)
		return

	case "available_commands":
		m.handleAvailableCommandsEvent(execution, event)
		return

	case "agent_capabilities":
		if len(event.AuthMethods) > 0 {
			execution.SetAuthMethods(event.AuthMethods)
		}

	case "session_mode":
		execution.SetModeState(&CachedModeState{
			CurrentModeID:  event.CurrentModeID,
			AvailableModes: event.AvailableModes,
		})
		// No return — must flow through to PublishAgentStreamEvent so the orchestrator
		// can filter and re-publish to the dedicated session mode subject.

	case "session_models":
		baselineCandidate, settled := execution.SetModelStateApplyingSettlement(&CachedModelState{
			CurrentModelID: event.CurrentModelID,
			Models:         event.SessionModels,
			ConfigOptions:  event.ConfigOptions,
			ConfigSource:   agentEventDataString(event.Data, "config_options_source"),
			ConfigID:       agentEventDataString(event.Data, "config_options_config_id"),
		})
		if settled {
			event.ConfigBaselineCandidate = baselineCandidate.ConfigOptions
			if event.Data == nil {
				event.Data = make(map[string]any)
			}
			event.Data["config_options_settled"] = true
		}
		// No return — must flow through to PublishAgentStreamEvent so the orchestrator
		// can persist the model and re-publish to the dedicated session models subject.

	case streams.EventTypeForegroundIdle:
		m.handlePromptHandoffEvent(execution, event)
	}

	m.eventPublisher.PublishAgentStreamEvent(execution, event)
}

func agentEventDataString(data map[string]any, key string) string {
	value, _ := data[key].(string)
	return value
}

// handleGitStatusUpdate processes git status updates from the workspace tracker
func (m *Manager) handleGitStatusUpdate(execution *AgentExecution, update *agentctl.GitStatusUpdate) {
	// Publish git status update to event bus for WebSocket streaming and persistence
	m.eventPublisher.PublishGitStatus(execution, update)
}

// handleGitCommitCreated processes git commit events from the workspace tracker
func (m *Manager) handleGitCommitCreated(execution *AgentExecution, commit *agentctl.GitCommitNotification) {
	// Publish commit event to event bus for WebSocket streaming and orchestrator handling
	m.eventPublisher.PublishGitCommit(execution, commit)
}

// handleGitResetDetected processes git reset events from the workspace tracker
func (m *Manager) handleGitResetDetected(execution *AgentExecution, reset *agentctl.GitResetNotification) {
	// Publish reset event to event bus for orchestrator handling (commit sync)
	m.eventPublisher.PublishGitReset(execution, reset)
}

// handleBranchSwitch processes branch switch events from the workspace tracker
func (m *Manager) handleBranchSwitch(execution *AgentExecution, branchSwitch *agentctl.GitBranchSwitchNotification) {
	// Publish branch switch event to event bus for orchestrator handling (base commit update)
	m.eventPublisher.PublishBranchSwitch(execution, branchSwitch)
}

// handleFileChangeNotification processes file change notifications from the workspace tracker
func (m *Manager) handleFileChangeNotification(execution *AgentExecution, notification *agentctl.FileChangeNotification) {
	m.eventPublisher.PublishFileChange(execution, notification)
}

// handleShellOutput processes shell output from the workspace stream
func (m *Manager) handleShellOutput(execution *AgentExecution, data string) {
	m.eventPublisher.PublishShellOutput(execution, data)
}

// handleProcessOutput processes script process output from the workspace stream
func (m *Manager) handleProcessOutput(execution *AgentExecution, output *agentctl.ProcessOutput) {
	if output == nil {
		return
	}
	m.logger.Debug("lifecycle received process output",
		zap.String("session_id", output.SessionID),
		zap.String("process_id", output.ProcessID),
		zap.String("kind", string(output.Kind)),
		zap.String("stream", output.Stream),
		zap.Int("bytes", len(output.Data)),
	)
	m.eventPublisher.PublishProcessOutput(execution, output)
}

// handleProcessStatus processes script process status updates from the workspace stream
func (m *Manager) handleProcessStatus(execution *AgentExecution, status *agentctl.ProcessStatusUpdate) {
	if status == nil {
		return
	}
	m.logger.Debug("lifecycle received process status",
		zap.String("session_id", status.SessionID),
		zap.String("process_id", status.ProcessID),
		zap.String("status", string(status.Status)),
	)
	m.eventPublisher.PublishProcessStatus(execution, status)
	m.releaseTerminalProcessActivity(status)
}

// handleShellExit processes shell exit events from the workspace stream
func (m *Manager) handleShellExit(execution *AgentExecution, code int) {
	m.eventPublisher.PublishShellExit(execution, code)
}