tutti/packages/agent/daemon/runtime/controller_turn_state.go

package agentruntime

import (
	"context"
	"fmt"
	"strings"

	agentsessionstore "github.com/tutti-os/tutti/packages/agent/daemon/activity"
	activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
)

func submittedTurnActivityEvents(
	submitCtx context.Context,
	session Session,
	content []PromptContentBlock,
	displayPrompt string,
	turnID string,
	capabilityRefs []activityshared.CapabilityReference,
	submittedTitle string,
) []activityshared.Event {
	eventContext, ok := activityEventContext(session, "turn-submitted:"+turnID, turnID)
	if !ok {
		return nil
	}
	explicitDisplayPrompt, visibleText := explicitAndVisiblePromptText(content, displayPrompt)
	prompt := newUserPromptActivityEvent(
		submitCtx,
		session,
		content,
		explicitDisplayPrompt,
		visibleText,
		turnID,
		nil,
	)
	event := activityshared.NewTurnUpdated(eventContext, turnID, activityshared.TurnPhaseSubmitted)
	event.Payload.Title = strings.TrimSpace(submittedTitle)
	event.Payload.Metadata = map[string]any{"turnOrigin": "user_prompt"}
	// The controller owns the submit moment; it publishes the submitted
	// lifecycle snapshot so downstream layers copy instead of recomputing
	// (ADR 0008).
	activityshared.StampTurnLifecycleSnapshot(&event, activityshared.TurnLifecycleSnapshot{
		Origin:       activityshared.TurnLifecycleOriginController,
		ActiveTurnID: turnID,
		Phase:        string(activityshared.TurnPhaseSubmitted),
	})
	activityshared.StampTurnCapabilityReferences(&event, capabilityRefs)
	return []activityshared.Event{prompt, event}
}

// guidanceTurnCapabilityReferenceStatePatch records provenance on the
// already-active canonical turn without creating a second turn or claiming a
// lifecycle transition. In particular, it carries no phase or lifecycle
// snapshot: the durable projection merges only the references.
func guidanceTurnCapabilityReferenceStatePatch(
	session Session,
	turnID string,
	capabilityRefs []activityshared.CapabilityReference,
) (agentsessionstore.WorkspaceAgentStatePatch, bool) {
	capabilityRefs = activityshared.NormalizeCapabilityReferences(capabilityRefs)
	turnID = strings.TrimSpace(turnID)
	if turnID == "" || len(capabilityRefs) == 0 {
		return agentsessionstore.WorkspaceAgentStatePatch{}, false
	}
	mapped := make([]agentsessionstore.WorkspaceAgentCapabilityReference, 0, len(capabilityRefs))
	for _, reference := range capabilityRefs {
		mapped = append(mapped, agentsessionstore.WorkspaceAgentCapabilityReference{
			Capability: reference.Capability,
			Source:     reference.Source,
		})
	}
	return agentsessionstore.WorkspaceAgentStatePatch{
		AgentSessionID:    strings.TrimSpace(session.AgentSessionID),
		Provider:          strings.TrimSpace(session.Provider),
		ProviderSessionID: strings.TrimSpace(session.ProviderSessionID),
		OccurredAtUnixMS:  nextEventUnixMS(),
		Turn: &agentsessionstore.WorkspaceAgentTurnPatch{
			TurnID:         turnID,
			CapabilityRefs: mapped,
		},
	}, true
}

// eventsCarryAdapterLifecycleSnapshot reports whether the batch contains an
// adapter-origin lifecycle snapshot — the signal that this session's provider
// publishes authoritative snapshots (ADR 0008).
func eventsCarryAdapterLifecycleSnapshot(events []activityshared.Event) bool {
	for _, event := range events {
		if snapshot, ok := activityshared.TurnLifecycleSnapshotFromEvent(event); ok &&
			snapshot.Origin == activityshared.TurnLifecycleOriginAdapter {
			return true
		}
	}
	return false
}

// applyTurnLifecycleSnapshots copies stamped lifecycle snapshots onto the
// session record. Consumers copy, never merge: the snapshot is the turn
// owner's full statement of the lifecycle at that moment.
func applyTurnLifecycleSnapshots(session Session, events []activityshared.Event) Session {
	for _, event := range events {
		snapshot, ok := activityshared.TurnLifecycleSnapshotFromEvent(event)
		if !ok {
			continue
		}
		session = applyTurnLifecycleSnapshot(session, snapshot, strings.TrimSpace(event.Payload.TurnID))
	}
	return session
}

func applyTurnLifecycleSnapshot(session Session, snapshot activityshared.TurnLifecycleSnapshot, eventTurnID string) Session {
	switch snapshot.Origin {
	case activityshared.TurnLifecycleOriginAdapter:
		// Snapshots reach the record over two channels (Exec emit closure and
		// the session event sink); drop anything older than what we applied.
		if snapshot.Seq != 0 && snapshot.Seq <= session.LifecycleSeq {
			return session
		}
		session.LifecycleSeq = snapshot.Seq
		session.LifecycleAuthority = true
	case activityshared.TurnLifecycleOriginController:
		// The controller only authors the submit moment and the settle
		// fallback; neither may clobber a different live provider turn.
		if session.TurnLifecycle != nil && session.TurnLifecycle.ActiveTurnID != nil {
			current := strings.TrimSpace(*session.TurnLifecycle.ActiveTurnID)
			if runtimeTurnLifecyclePhaseIsLive(session.TurnLifecycle.Phase) &&
				current != "" &&
				current != strings.TrimSpace(snapshot.ActiveTurnID) &&
				current != eventTurnID {
				return session
			}
		}
	default:
		return session
	}
	lifecycle := TurnLifecycle{Phase: snapshot.Phase, Settling: snapshot.Settling}
	if turnID := strings.TrimSpace(snapshot.ActiveTurnID); turnID != "" {
		lifecycle.ActiveTurnID = &turnID
	}
	if outcome := strings.TrimSpace(snapshot.Outcome); outcome != "" {
		lifecycle.Outcome = &outcome
	}
	session.TurnLifecycle = &lifecycle
	return session
}

// sessionLevelStatusFromEvents extracts the genuinely session-scoped status
// signals from a batch: session failure/completion and explicit effective
// statuses. Unlike the legacy fold it never defaults an unknown or empty
// effective status to ready — for authority sessions readiness is derived
// from the lifecycle, not from metadata refreshes.
func sessionLevelStatusFromEvents(events []activityshared.Event) string {
	status := ""
	for _, event := range events {
		switch event.Type {
		case activityshared.EventSessionFailed:
			status = SessionStatusFailed
		case activityshared.EventSessionCompleted:
			status = SessionStatusCompleted
		case activityshared.EventSessionUpdated:
			switch strings.TrimSpace(event.Payload.EffectiveStatus) {
			case string(activityshared.SessionStatusWorking):
				status = SessionStatusWorking
			case string(activityshared.SessionStatusWaiting):
				status = SessionStatusWaiting
			case string(activityshared.SessionStatusCompleted):
				status = SessionStatusCompleted
			case string(activityshared.SessionStatusFailed):
				status = SessionStatusFailed
			case string(activityshared.SessionStatusPaused):
				status = SessionStatusCanceled
			}
		}
	}
	return status
}

// statusForAuthoritySession is THE status derivation for snapshot-authority
// sessions: a pure function of the copied lifecycle plus session-level
// signals. No other code path may write Status for these sessions.
func statusForAuthoritySession(session Session, batchSessionLevel string) string {
	if batchSessionLevel == SessionStatusFailed || batchSessionLevel == SessionStatusCompleted {
		return batchSessionLevel
	}
	lifecycle := session.TurnLifecycle
	if lifecycle != nil && runtimeTurnLifecyclePhaseIsLive(lifecycle.Phase) {
		if activityshared.TurnLifecyclePhaseIsWaiting(lifecycle.Phase) {
			return SessionStatusWaiting
		}
		return SessionStatusWorking
	}
	if lifecycle != nil && lifecycle.Phase == "settled" {
		if lifecycle.Outcome != nil {
			switch strings.TrimSpace(*lifecycle.Outcome) {
			case string(activityshared.TurnOutcomeFailed):
				return SessionStatusFailed
			case string(activityshared.TurnOutcomeInterrupted), "canceled":
				return SessionStatusCanceled
			}
		}
		return SessionStatusReady
	}
	if batchSessionLevel != "" {
		return batchSessionLevel
	}
	if current := strings.TrimSpace(session.Status); current != "" {
		return current
	}
	return SessionStatusReady
}

// submitAvailabilityForAuthoritySession derives SubmitAvailability from the
// same copied lifecycle, replacing the hand-written variants that used to
// live in the lifecycle fold, the reporter's codex patch path, and the
// reconcile.
func submitAvailabilityForAuthoritySession(session Session) *SubmitAvailability {
	lifecycle := session.TurnLifecycle
	if lifecycle != nil && runtimeTurnLifecyclePhaseIsLive(lifecycle.Phase) {
		if activityshared.TurnLifecyclePhaseIsWaiting(lifecycle.Phase) {
			return blockedSubmitAvailability("waiting")
		}
		return blockedSubmitAvailability("active_turn")
	}
	return availableSubmitAvailability()
}

// foldTurnSessionEvents applies an emitted turn event batch to the session
// record. Snapshot-authority sessions copy lifecycle snapshots and derive
// Status/SubmitAvailability purely (ADR 0008); legacy sessions keep the
// historic folding path until their provider publishes snapshots (Phase B).
func (c *Controller) foldTurnSessionEvents(session Session, events []activityshared.Event, execTurnID string) Session {
	events = eventsOwnedBySession(events, session.AgentSessionID)
	if len(events) == 0 {
		return session
	}
	for _, event := range events {
		if event.Type == activityshared.EventRootProviderTurnStarted &&
			!session.IsSideConversation() {
			session.Resumable = true
			break
		}
	}
	previousStatus := session.Status
	if session.LifecycleAuthority || eventsCarryAdapterLifecycleSnapshot(events) {
		session = applySessionEventsBase(session, events)
		session = applyTurnLifecycleSnapshots(session, events)
		session.Status = statusForAuthoritySession(session, sessionLevelStatusFromEvents(events))
		session.SubmitAvailability = submitAvailabilityForAuthoritySession(session)
		return session
	}
	session = applySessionEvents(session, events)
	session = applyTurnLifecycleFromEvents(session, events)
	if turnEventsAreTerminal(events) {
		session = reconcileFinishedTurnStatus(session)
	} else if execTurnID != "" {
		session = c.preserveActiveTurnStatus(session, execTurnID, previousStatus)
	}
	return session
}

func eventsOwnedBySession(events []activityshared.Event, agentSessionID string) []activityshared.Event {
	agentSessionID = strings.TrimSpace(agentSessionID)
	owned := make([]activityshared.Event, 0, len(events))
	for _, event := range events {
		eventSessionID := strings.TrimSpace(event.AgentSessionID)
		if eventSessionID == "" || eventSessionID == agentSessionID {
			owned = append(owned, event)
		}
	}
	return owned
}

func applyTurnLifecycleFromEvents(session Session, events []activityshared.Event) Session {
	for _, event := range events {
		phase := turnLifecyclePhaseFromEvent(event)
		if phase == "" {
			continue
		}
		turnID := strings.TrimSpace(event.Payload.TurnID)
		if turnID == "" {
			continue
		}
		lifecycle := TurnLifecycle{Phase: phase}
		if phase == "settled" {
			outcome := turnLifecycleOutcomeFromEvent(event)
			if outcome != "" {
				lifecycle.Outcome = &outcome
			}
			session.SubmitAvailability = availableSubmitAvailability()
		} else {
			activeTurnID := turnID
			lifecycle.ActiveTurnID = &activeTurnID
			if phase == "waiting" {
				session.SubmitAvailability = blockedSubmitAvailability("waiting")
			} else {
				session.SubmitAvailability = blockedSubmitAvailability("active_turn")
			}
		}
		session.TurnLifecycle = &lifecycle
	}
	return session
}

func turnLifecyclePhaseFromEvent(event activityshared.Event) string {
	switch event.Type {
	case activityshared.EventTurnStarted, activityshared.EventRootProviderTurnStarted:
		return "running"
	case activityshared.EventTurnUpdated:
		switch strings.TrimSpace(event.Payload.TurnPhase) {
		case "submitted":
			return "submitted"
		case string(activityshared.TurnPhaseWaiting), string(activityshared.TurnPhaseWaitingApproval), string(activityshared.TurnPhaseWaitingInput):
			return "waiting"
		case string(activityshared.TurnPhaseRunning), string(activityshared.TurnPhaseWorking):
			return "running"
		}
	case activityshared.EventTurnCompleted, activityshared.EventTurnFailed:
		return "settled"
	default:
		if string(event.Type) == EventTurnCanceled {
			return "settled"
		}
	}
	return ""
}

func turnLifecycleOutcomeFromEvent(event activityshared.Event) string {
	switch event.Type {
	case activityshared.EventTurnFailed:
		return "failed"
	case activityshared.EventTurnCompleted:
		if strings.TrimSpace(event.Payload.TurnOutcome) == string(activityshared.TurnOutcomeInterrupted) {
			return "canceled"
		}
		return "completed"
	default:
		if string(event.Type) == EventTurnCanceled {
			return "canceled"
		}
		return strings.TrimSpace(event.Payload.TurnOutcome)
	}
}

func cloneRuntimeSubmitAvailability(value *SubmitAvailability) *SubmitAvailability {
	if value == nil {
		return nil
	}
	return &SubmitAvailability{
		State:  strings.TrimSpace(value.State),
		Reason: strings.TrimSpace(value.Reason),
	}
}

func cloneRuntimeCompletedCommand(value *CompletedCommand) *CompletedCommand {
	if value == nil {
		return nil
	}
	return &CompletedCommand{
		Kind:   strings.TrimSpace(value.Kind),
		Status: strings.TrimSpace(value.Status),
	}
}

func cloneRuntimeTurnLifecycle(value *TurnLifecycle) *TurnLifecycle {
	if value == nil {
		return nil
	}
	var activeTurnID *string
	if value.ActiveTurnID != nil {
		active := strings.TrimSpace(*value.ActiveTurnID)
		activeTurnID = &active
	}
	var outcome *string
	if value.Outcome != nil {
		next := strings.TrimSpace(*value.Outcome)
		outcome = &next
	}
	return &TurnLifecycle{
		ActiveTurnID:     activeTurnID,
		Phase:            strings.TrimSpace(value.Phase),
		Settling:         value.Settling,
		Outcome:          outcome,
		CompletedCommand: cloneRuntimeCompletedCommand(value.CompletedCommand),
	}
}

func (c *Controller) preserveActiveTurnStatus(session Session, turnID string, previousStatus string) Session {
	if c == nil || session.Status != SessionStatusReady {
		return session
	}
	key := sessionKey(session.RoomID, session.AgentSessionID)
	c.mu.Lock()
	active, ok := c.turns[key]
	c.mu.Unlock()
	if ok && active.turnID == turnID {
		session.Status = firstNonEmpty(previousStatus, SessionStatusWorking)
	}
	return session
}

func (c *Controller) preserveCurrentSessionSettings(session Session) Session {
	if c == nil {
		return session
	}
	key := sessionKey(session.RoomID, session.AgentSessionID)
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.preserveCurrentSessionSettingsLocked(key, session)
}

func (c *Controller) preserveCurrentSessionSettingsLocked(key string, session Session) Session {
	if c == nil {
		return session
	}
	current, ok := c.sessions[key]
	if !ok ||
		strings.TrimSpace(current.RoomID) != strings.TrimSpace(session.RoomID) ||
		strings.TrimSpace(current.AgentSessionID) != strings.TrimSpace(session.AgentSessionID) ||
		strings.TrimSpace(current.Provider) != strings.TrimSpace(session.Provider) {
		return session
	}
	session.PermissionModeID = strings.TrimSpace(current.PermissionModeID)
	if current.Settings != nil {
		settings := normalizeSessionSettings(current.Settings, current.Provider, session.PermissionModeID)
		session.Settings = cloneSessionSettings(settings)
	} else {
		session.Settings = nil
	}
	session.RuntimeContext = runtimeContextWithSessionSettings(session.RuntimeContext, session.SettingsValue())
	return session
}

func runtimeContextWithSessionSettings(runtimeContext map[string]any, settings SessionSettings) map[string]any {
	next := clonePayload(runtimeContext)
	if next == nil {
		next = map[string]any{}
	}
	next["permissionModeId"] = strings.TrimSpace(settings.PermissionModeID)
	next["planMode"] = settings.PlanMode
	next["model"] = strings.TrimSpace(settings.Model)
	next["reasoningEffort"] = strings.TrimSpace(settings.ReasoningEffort)
	next["speed"] = strings.TrimSpace(settings.Speed)
	return next
}

func sessionHasDifferentLiveTurn(session Session, turnID string) bool {
	if !sessionHasLiveTurnLifecycle(session) {
		return false
	}
	return runtimeTurnLifecycleActiveTurnID(session.TurnLifecycle) != strings.TrimSpace(turnID)
}

func settleFinishedTurnLifecycle(session Session, turnID string) Session {
	if session.TurnLifecycle == nil {
		return session
	}
	if runtimeTurnLifecycleActiveTurnID(session.TurnLifecycle) != strings.TrimSpace(turnID) {
		return session
	}
	if !runtimeTurnLifecyclePhaseIsLive(session.TurnLifecycle.Phase) {
		return session
	}
	outcome := "completed"
	if session.TurnLifecycle.Outcome != nil && strings.TrimSpace(*session.TurnLifecycle.Outcome) != "" {
		outcome = strings.TrimSpace(*session.TurnLifecycle.Outcome)
	}
	session.TurnLifecycle = &TurnLifecycle{
		Phase:            "settled",
		Outcome:          &outcome,
		CompletedCommand: cloneRuntimeCompletedCommand(session.TurnLifecycle.CompletedCommand),
	}
	session.SubmitAvailability = availableSubmitAvailability()
	return session
}

func sessionHasLiveTurnLifecycle(session Session) bool {
	if session.TurnLifecycle == nil {
		return false
	}
	return runtimeTurnLifecycleActiveTurnID(session.TurnLifecycle) != "" &&
		runtimeTurnLifecyclePhaseIsLive(session.TurnLifecycle.Phase)
}

func runtimeTurnLifecycleActiveTurnID(value *TurnLifecycle) string {
	if value == nil || value.ActiveTurnID == nil {
		return ""
	}
	return strings.TrimSpace(*value.ActiveTurnID)
}

func runtimeTurnLifecyclePhaseIsLive(phase string) bool {
	// Delegates to the canonical predicate; the phase vocabulary lives in
	// exactly one place (activityshared, mirrored in activity-core for TS).
	return activityshared.TurnLifecyclePhaseIsLive(phase)
}

func unemittedActivityEvents(events []activityshared.Event, emitted []activityshared.Event) []activityshared.Event {
	if len(events) == 0 {
		return nil
	}
	if len(emitted) == 0 {
		return events
	}
	seen := make(map[string]struct{}, len(emitted))
	for _, event := range emitted {
		seen[activityEventIdentity(event)] = struct{}{}
	}
	out := make([]activityshared.Event, 0, len(events))
	for _, event := range events {
		if _, ok := seen[activityEventIdentity(event)]; ok {
			continue
		}
		out = append(out, event)
	}
	return out
}

// retainTurnCallLifecycleEvents keeps adapter-produced close events for turnID
// when synthesizing a context-canceled terminal. Controllers must not discard:
//   - CallFailed closes for open tools
//   - failed/completed assistant or thinking snapshots that settle in-flight
//     stream rows (Claude FinishInterrupted on ctx cancel)
func retainTurnCallLifecycleEvents(events []activityshared.Event, turnID string) []activityshared.Event {
	turnID = strings.TrimSpace(turnID)
	if len(events) == 0 || turnID == "" {
		return nil
	}
	out := make([]activityshared.Event, 0, len(events))
	for _, event := range events {
		if strings.TrimSpace(event.Payload.TurnID) != turnID {
			continue
		}
		switch event.Type {
		case activityshared.EventCallStarted,
			activityshared.EventCallCompleted,
			activityshared.EventCallFailed:
			out = append(out, event)
		case activityshared.EventMessageAppended, activityshared.EventMessageCreated:
			if isRetainedCancelMessageSettlement(event) {
				out = append(out, event)
			}
		}
	}
	return out
}

func isRetainedCancelMessageSettlement(event activityshared.Event) bool {
	role := string(event.Payload.Role)
	if role != string(activityshared.MessageRoleAssistant) &&
		role != string(activityshared.MessageRoleAssistantThinking) {
		return false
	}
	streamState := asString(event.Payload.Metadata["streamState"])
	if streamState == "" {
		streamState = strings.TrimSpace(event.Payload.Status)
	}
	return streamState == messageStreamStateCompleted || streamState == messageStreamStateFailed
}

func activityEventIdentity(event activityshared.Event) string {
	if event.EventID != "" {
		return event.EventID
	}
	return fmt.Sprintf(
		"%s\x00%s\x00%s\x00%s\x00%d",
		event.Type,
		event.AgentSessionID,
		event.ProviderSessionID,
		event.Payload.TurnID,
		event.OccurredAtUnixMS,
	)
}