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

package agentruntime

import (
	"context"
	"log/slog"
	"strings"
	"time"

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

const (
	canonicalTerminalCommitRetryInitialBackoff = 100 * time.Millisecond
	canonicalTerminalCommitRetryMaxBackoff     = 5 * time.Second
)

type rootTurnCompletionAuthority uint8

const (
	rootTurnCompletionAwaitProviderAggregation rootTurnCompletionAuthority = iota
	rootTurnCompletionDirectCanonical
)

// classifyRootTurnCompletion distinguishes a terminal fact for the exact
// canonical Turn from provider-root lifecycle facts. Dispatch disposition is
// deliberately absent: admission and completion are independent contracts.
func classifyRootTurnCompletion(
	turnID string,
	events []activityshared.Event,
) rootTurnCompletionAuthority {
	if turnHasTerminalEvent(events, turnID) {
		return rootTurnCompletionDirectCanonical
	}
	return rootTurnCompletionAwaitProviderAggregation
}

// pendingCanonicalTerminalCommit retains a terminal fact that the Controller
// accepted in memory but could not yet confirm through the durable reporter.
// The owning blocking-exec goroutine keeps the exact active-turn fence until
// this record commits or daemon reconciliation proves the Turn already settled.
type pendingCanonicalTerminalCommit struct {
	session Session
	events  []activityshared.Event
}

func (pending *pendingCanonicalTerminalCommit) merge(
	session Session,
	events []activityshared.Event,
) {
	if pending == nil || len(events) == 0 {
		return
	}
	pending.session = session
	pending.events = append(
		pending.events,
		unemittedActivityEvents(events, pending.events)...,
	)
}

func (pending *pendingCanonicalTerminalCommit) directCanonical(turnID string) bool {
	return pending != nil && classifyRootTurnCompletion(
		turnID,
		pending.events,
	) == rootTurnCompletionDirectCanonical
}

// convergeCanonicalTerminalCommit retries the same idempotent terminal facts
// after the bounded synchronous barrier fails. It intentionally outlives the
// submit caller's context: the submitted Turn is already durable, so only a
// successful terminal commit or an exact daemon reconciliation may release the
// active-turn fence.
func (c *Controller) convergeCanonicalTerminalCommit(
	ctx context.Context,
	turnID string,
	pending *pendingCanonicalTerminalCommit,
) bool {
	if c == nil || pending == nil || len(pending.events) == 0 {
		return false
	}
	retryCtx := context.WithoutCancel(ctx)
	backoff := canonicalTerminalCommitRetryInitialBackoff
	attempt := 1
	for {
		timer := time.NewTimer(backoff)
		<-timer.C

		active, ok := c.activeTurn(
			pending.session.RoomID,
			pending.session.AgentSessionID,
		)
		if !ok || strings.TrimSpace(active.turnID) != strings.TrimSpace(turnID) {
			// ReconcileRootTurnSettlement may win this race after proving the
			// canonical Store is already settled. Never republish after that.
			return false
		}

		session := c.preserveCurrentSessionSettings(pending.session)
		accepted, ok := c.storeTurnSession(session, turnID)
		if !ok {
			return false
		}
		pending.session = accepted
		err := c.reportSessionBeforePublish(retryCtx, accepted, pending.events)
		if err == nil {
			if !c.isProvisionalSession(accepted) {
				c.publish(accepted, pending.events)
			}
			return pending.directCanonical(turnID)
		}
		slog.Warn(
			"agent session terminal activity report remains pending",
			"event", "agent_session.activity_report.terminal_commit_retry",
			"room_id", accepted.RoomID,
			"agent_session_id", accepted.AgentSessionID,
			"turn_id", strings.TrimSpace(turnID),
			"attempt", attempt,
			"retry_backoff", backoff,
			"error", err,
		)

		if backoff < canonicalTerminalCommitRetryMaxBackoff {
			backoff *= 2
			if backoff > canonicalTerminalCommitRetryMaxBackoff {
				backoff = canonicalTerminalCommitRetryMaxBackoff
			}
		}
		attempt++
	}
}

// finalizeBlockingExecTurn owns the active-turn fence transition after a
// blocking adapter invocation exits. Root-provider adapters normally wait for
// canonical provider aggregation. The only providerless exception is an exact
// canonical terminal that already crossed the synchronous durable-report
// barrier; it needs no provider identity to become authoritative.
func (c *Controller) finalizeBlockingExecTurn(
	session Session,
	turnID string,
	rootProviderLifecycle bool,
	directCanonicalTerminalCommitted bool,
) {
	if !rootProviderLifecycle || directCanonicalTerminalCommitted {
		_, _ = c.finishTurn(session, turnID)
		return
	}
	_, _ = c.storeTurnSession(session, turnID)
}