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

package agentruntime

import (
	"fmt"
	"log/slog"
	"strings"
)

type codexAppServerTurnKind string

const (
	codexAppServerTurnKindNormal  codexAppServerTurnKind = "normal"
	codexAppServerTurnKindCompact codexAppServerTurnKind = "compact"
	// codexAppServerTurnKindGoalAdopted marks a turn that codex started on its
	// own to continue an active goal, adopted by the reducer so its output is
	// tracked like any Exec-driven turn.
	codexAppServerTurnKindGoalAdopted codexAppServerTurnKind = "goal-adopted"
)

type codexAppServerTurnPhase string

const (
	codexAppServerTurnPhaseIdle         codexAppServerTurnPhase = "idle"
	codexAppServerTurnPhaseRunning      codexAppServerTurnPhase = "running"
	codexAppServerTurnPhaseCompacting   codexAppServerTurnPhase = "compacting"
	codexAppServerTurnPhaseInterrupting codexAppServerTurnPhase = "interrupting"
	codexAppServerTurnPhaseCompleted    codexAppServerTurnPhase = "completed"
	codexAppServerTurnPhaseFailed       codexAppServerTurnPhase = "failed"
	codexAppServerTurnPhaseCanceled     codexAppServerTurnPhase = "canceled"
)

type codexAppServerTurnTerminal struct {
	turn  map[string]any
	err   error
	phase codexAppServerTurnPhase
}

func (phase codexAppServerTurnPhase) terminal() bool {
	switch phase {
	case codexAppServerTurnPhaseCompleted,
		codexAppServerTurnPhaseFailed,
		codexAppServerTurnPhaseCanceled:
		return true
	default:
		return false
	}
}

func (a *CodexAppServerAdapter) transitionActiveTurnPhase(
	agentSessionID string,
	turn *codexAppServerActiveTurn,
	phase codexAppServerTurnPhase,
) {
	if a == nil || turn == nil || phase == "" {
		return
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil || appSession.activeTurn != turn || turn.phase.terminal() {
		return
	}
	if phase == codexAppServerTurnPhaseCompacting {
		turn.kind = codexAppServerTurnKindCompact
	}
	turn.phase = phase
}

// completeActiveTurn transitions the reducer-owned turn projection to a
// terminal phase after the `turn/completed` notification or an already-terminal
// initial turn snapshot. The blocking Exec wrapper observes the terminal
// channel; it no longer owns terminal classification.
func (a *CodexAppServerAdapter) completeActiveTurn(agentSessionID string, turn map[string]any) {
	a.settleActiveTurn(agentSessionID, asString(turn["id"]), func(activeTurn *codexAppServerActiveTurn) codexAppServerTurnTerminal {
		phase := appServerProjectedTurnTerminalPhase(turn, activeTurn.forceCanceled)
		appServerLogTurnTerminalShadowMismatch(agentSessionID, turn, phase)
		return codexAppServerTurnTerminal{turn: turn, phase: phase}
	})
}

func (a *CodexAppServerAdapter) failActiveTurnFromAppServerError(agentSessionID string, params map[string]any) {
	err := appServerNotificationError(params)
	a.settleActiveTurn(agentSessionID, asString(params["turnId"]), func(*codexAppServerActiveTurn) codexAppServerTurnTerminal {
		return codexAppServerTurnTerminal{err: err, phase: codexAppServerTurnPhaseFailed}
	})
}

// settleActiveTurn owns the shared settle sequence — lock, session lookup,
// provider-turn-id match, phase transition, activeTurnID clear, non-blocking
// terminal send — so the completion and failure paths cannot diverge on the
// guards. terminalFor runs under the adapter lock with the matched active turn.
//
// A mismatched non-empty provider turn id settles anyway when the recorded id
// was never confirmed by turn/started: a turn/start issued while another turn
// is already running responds with a stub id that codex never starts — the
// input is steered into the running turn, so that turn's same-thread terminal
// is this turn's terminal (live-verified against codex 0.142.5,
// TestLiveProtocolTurnStartDuringActiveTurn). Requiring an exact match there
// wedged awaitTurnCompletion forever. Goal-adopted turns are the only active
// turns allowed to settle from an empty terminal id because their ownership was
// established by the preceding turn/started notification.
func (a *CodexAppServerAdapter) settleActiveTurn(
	agentSessionID string,
	providerTurnID string,
	terminalFor func(*codexAppServerActiveTurn) codexAppServerTurnTerminal,
) {
	if a == nil {
		return
	}
	var activeTurn *codexAppServerActiveTurn
	var terminal codexAppServerTurnTerminal
	steerAdoptedTurnID := ""
	droppedTurnID := ""
	a.mu.Lock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession != nil {
		activeTurn = appSession.activeTurn
		if activeTurn != nil {
			expected := strings.TrimSpace(appSession.activeTurnID)
			actual := strings.TrimSpace(providerTurnID)
			switch {
			case expected == "" || actual == expected:
			case actual == "" && activeTurn.kind == codexAppServerTurnKindGoalAdopted:
			case actual != "" && !appSession.activeTurnStartConfirmed:
				steerAdoptedTurnID = expected
			default:
				droppedTurnID = expected
				activeTurn = nil
			}
		}
		if activeTurn != nil {
			terminal = terminalFor(activeTurn)
			activeTurn.phase = terminal.phase
			appSession.activeTurnID = ""
			appSession.activeTurnStartConfirmed = false
		}
	}
	emits := activeTurn != nil && activeTurn.settleEmits
	a.mu.Unlock()
	if droppedTurnID != "" {
		slog.Warn("agent session app-server turn terminal dropped on provider turn id mismatch",
			"event", "agent_session.app_server.turn.terminal_id_mismatch",
			"agent_session_id", agentSessionID,
			"recorded_turn_id", droppedTurnID,
			"terminal_turn_id", providerTurnID,
		)
	}
	if activeTurn == nil {
		return
	}
	if steerAdoptedTurnID != "" {
		slog.Warn("agent session app-server steered turn adopted running turn terminal",
			"event", "agent_session.app_server.turn.steer_terminal_adopted",
			"agent_session_id", agentSessionID,
			"stub_turn_id", steerAdoptedTurnID,
			"terminal_turn_id", providerTurnID,
		)
	}
	select {
	case activeTurn.terminal <- terminal:
	default:
	}
	if emits {
		a.finalizeSettledTurn(agentSessionID, activeTurn, terminal)
	}
}

func appServerProjectedTurnTerminalPhase(turn map[string]any, forceCanceled bool) codexAppServerTurnPhase {
	if forceCanceled {
		return codexAppServerTurnPhaseCanceled
	}
	switch strings.TrimSpace(asString(turn["status"])) {
	case "failed":
		return codexAppServerTurnPhaseFailed
	case "interrupted", "canceled":
		return codexAppServerTurnPhaseCanceled
	default:
		return codexAppServerTurnPhaseCompleted
	}
}

func appServerLegacyTurnTerminalPhase(turn map[string]any) codexAppServerTurnPhase {
	switch strings.TrimSpace(asString(turn["status"])) {
	case "failed":
		return codexAppServerTurnPhaseFailed
	case "interrupted", "canceled":
		return codexAppServerTurnPhaseCanceled
	default:
		return codexAppServerTurnPhaseCompleted
	}
}

func appServerLogTurnTerminalShadowMismatch(
	agentSessionID string,
	turn map[string]any,
	projected codexAppServerTurnPhase,
) {
	legacy := appServerLegacyTurnTerminalPhase(turn)
	if projected == legacy {
		return
	}
	slog.Warn("agent session app-server turn projection terminal mismatch",
		"event", "agent_session.app_server.turn_projection.shadow_mismatch",
		"agent_session_id", agentSessionID,
		"provider_turn_id", asString(turn["id"]),
		"status", asString(turn["status"]),
		"projected_phase", string(projected),
		"legacy_phase", string(legacy),
	)
}

func appServerNotificationError(params map[string]any) error {
	turnError := payloadObject(params["error"])
	message := strings.TrimSpace(asStringRaw(turnError["message"]))
	if message == "" {
		message = strings.TrimSpace(asStringRaw(params["message"]))
	}
	if message == "" {
		return fmt.Errorf("codex app-server turn failed")
	}
	return fmt.Errorf("%s", message)
}