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

package agentruntime

import (
	"log/slog"
	"strings"

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

func (a *CodexAppServerAdapter) applyTokenUsage(agentSessionID string, params map[string]any) {
	usage, ok := appServerTokenUsageState(params)
	if !ok {
		return
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return
	}
	appSession.usage = mergeACPUsageState(appSession.usage, usage)
}

// appServerTokenUsageState parses a thread/tokenUsage/updated payload into the
// context-window portion of acpUsageState. It is shared between the live
// notification path (applyTokenUsage) and the resume handshake, where codex
// replays token usage before the session is stored.
//
// ThreadTokenUsage schema: "last" = most-recent API call breakdown, "total" =
// cumulative thread totals. Use last.inputTokens (context fill sent to the
// model) as the most accurate indicator of how full the window is. Fall back to
// last.totalTokens (includes response tokens — slightly high but still
// per-request), then total.totalTokens only when "last" is absent entirely.
// Using total.totalTokens as primary causes a false compact alert: after 10
// calls of 27 K tokens each the cumulative reaches 270 K and exceeds the 258 K
// per-request window even though each call individually used only ~10 %.
//
// A non-positive last.inputTokens also triggers the fallback chain: the
// post-compaction frame reports last.inputTokens=0 while last.totalTokens holds
// the real compacted context size. Treating that literal 0 as the context fill
// would display "0" right after a compaction instead of the compacted size.
func appServerTokenUsageState(params map[string]any) (acpUsageState, bool) {
	tokenUsage := payloadObject(params["tokenUsage"])
	if len(tokenUsage) == 0 {
		return acpUsageState{}, false
	}
	last := payloadObject(tokenUsage["last"])
	used, usedOK := firstInt64Value(last, "inputTokens")
	if !usedOK || used <= 0 {
		used, usedOK = firstInt64Value(last, "totalTokens")
	}
	if !usedOK || used <= 0 {
		used, usedOK = firstInt64Value(payloadObject(tokenUsage["total"]), "totalTokens")
	}
	window, windowOK := firstInt64Value(tokenUsage, "modelContextWindow")
	if !usedOK || !windowOK {
		return acpUsageState{}, false
	}
	return acpUsageState{
		contextUsedTokens:   used,
		contextWindowTokens: window,
		contextKnown:        true,
	}, true
}

func (a *CodexAppServerAdapter) applyRateLimits(agentSessionID string, snapshot map[string]any) bool {
	if len(snapshot) == 0 {
		return false
	}
	quotas := appServerRateLimitQuotas(snapshot)
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return false
	}
	appSession.rateLimits = clonePayload(snapshot)
	appSession.startupRateLimitsReady = true
	if len(quotas) > 0 {
		appSession.usage = mergeACPUsageState(appSession.usage, acpUsageState{quotas: quotas})
	}
	return true
}

func (a *CodexAppServerAdapter) applyAccountUpdate(agentSessionID string, params map[string]any) {
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return
	}
	if appSession.account == nil {
		appSession.account = map[string]any{}
	}
	if authMode := asString(params["authMode"]); authMode != "" {
		appSession.account["authMode"] = authMode
	}
	if planType := asString(params["planType"]); planType != "" {
		appSession.account["planType"] = planType
	}
}

// applyGoalUpdate stores the latest goal snapshot and reports the status
// transition so callers can emit user-visible notices when the goal stops
// progressing (paused/blocked/usageLimited/budgetLimited).
func (a *CodexAppServerAdapter) applyGoalUpdate(agentSessionID string, goal map[string]any) (oldStatus, newStatus string, statusChanged bool) {
	goal = normalizedCodexGoal(goal)
	if len(goal) == 0 {
		return "", "", false
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return "", "", false
	}
	return applyCodexGoalUpdateLocked(appSession, agentSessionID, goal)
}

func applyCodexGoalUpdateLocked(appSession *codexAppServerSession, agentSessionID string, goal map[string]any) (oldStatus, newStatus string, statusChanged bool) {
	oldStatus = strings.TrimSpace(asString(appSession.goal["status"]))
	appSession.goal = clonePayload(goal)
	appSession.goalStateVersion++
	newStatus = strings.TrimSpace(asString(appSession.goal["status"]))
	if newStatus != "active" {
		appSession.goalContinuationClaim = nil
	}
	if oldStatus != newStatus {
		slog.Info("agent session app-server goal status changed",
			"event", "agent_session.app_server.goal.status_changed",
			"agent_session_id", agentSessionID,
			"old_status", oldStatus,
			"new_status", newStatus,
		)
	}
	return oldStatus, newStatus, oldStatus != newStatus
}

func (a *CodexAppServerAdapter) goalStateFence(agentSessionID string) (*codexAppServerSession, uint64) {
	if a == nil {
		return nil, 0
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return nil, 0
	}
	return appSession, appSession.goalStateVersion
}

func (a *CodexAppServerAdapter) applyGoalUpdateAtFence(
	agentSessionID string,
	expectedSession *codexAppServerSession,
	expectedVersion uint64,
	goal map[string]any,
) bool {
	goal = normalizedCodexGoal(goal)
	if len(goal) == 0 || expectedSession == nil || a == nil {
		return false
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession != expectedSession || appSession.goalStateVersion != expectedVersion {
		return false
	}
	applyCodexGoalUpdateLocked(appSession, agentSessionID, goal)
	return true
}

func (a *CodexAppServerAdapter) applyGoalClear(agentSessionID string) {
	a.mu.Lock()
	defer a.mu.Unlock()
	appSession := a.sessions[strings.TrimSpace(agentSessionID)]
	if appSession == nil {
		return
	}
	appSession.goalStateVersion++
	if appSession.goal != nil {
		slog.Info("agent session app-server goal cleared",
			"event", "agent_session.app_server.goal.cleared",
			"agent_session_id", agentSessionID,
			"old_status", strings.TrimSpace(asString(appSession.goal["status"])),
		)
	}
	appSession.goal = nil
	appSession.goalContinuationClaim = nil
}

func appServerRateLimitQuotas(snapshot map[string]any) []map[string]any {
	quotas := make([]map[string]any, 0, 2)
	for _, window := range []struct {
		key       string
		quotaType string
	}{
		{key: "primary", quotaType: "session"},
		{key: "secondary", quotaType: "weekly"},
	} {
		entry := payloadObject(snapshot[window.key])
		if len(entry) == 0 {
			continue
		}
		usedPercent, ok := acpFloatValue(entry["usedPercent"])
		if !ok {
			continue
		}
		if usedPercent < 0 {
			usedPercent = 0
		}
		if usedPercent > 100 {
			usedPercent = 100
		}
		quota := map[string]any{
			"quotaType":        appServerRateLimitQuotaType(entry, window.quotaType),
			"percentRemaining": 100 - usedPercent,
		}
		if resetsAt, ok := int64Value(entry["resetsAt"]); ok && resetsAt > 0 {
			if resetsAt < 1_000_000_000_000 {
				resetsAt *= 1000
			}
			quota["resetsAtUnixMs"] = resetsAt
		}
		quotas = append(quotas, quota)
	}
	if len(quotas) == 0 {
		return nil
	}
	return quotas
}

// Keep duration semantics aligned with codexUsageQuotaType in
// apps/desktop/src/main/agentProviderUsageProbe.ts. Active sessions use this
// daemon mapper; empty-session /status uses the desktop probe.
func appServerRateLimitQuotaType(entry map[string]any, fallback string) string {
	durationMins, ok := int64Value(entry["windowDurationMins"])
	if !ok {
		return fallback
	}
	switch durationMins {
	case 5 * 60:
		return "session"
	case 7 * 24 * 60:
		return "weekly"
	default:
		return fallback
	}
}

func appServerSystemNoticeEvent(session Session, turnID string, noticeKind string, title string, detail string, metadata ...map[string]any) activityshared.Event {
	update := map[string]any{
		"sessionUpdate": "system_notice",
		"kind":          "agent_system_notice",
		"noticeKind":    noticeKind,
	}
	if title != "" {
		update["title"] = title
	}
	if title == appServerContextCompactedTitle {
		update["noticeCommand"] = "compact"
		update["noticeCommandStatus"] = "completed"
	}
	if title == appServerCompactingContextTitle {
		update["noticeCommand"] = "compact"
		update["noticeCommandStatus"] = "running"
	}
	if detail != "" {
		update["detail"] = detail
	}
	for _, extra := range metadata {
		for key, value := range extra {
			if value != nil {
				update[key] = value
			}
		}
	}
	event, _ := acpSystemNoticeEvent(session, turnID, update, "system_notice", true)
	return event
}

// --- server -> client requests (approvals, user input) ---