tutti/packages/agent/host/host.go

package agenthost

import (
	"context"
	"strings"
	"sync"
	"time"
)

type Config struct {
	CanonicalStore          CanonicalStore
	RuntimeRailPlacement    RuntimeSessionRailPlacementResolver
	InteractionTrees        CanonicalInteractionTreeStore
	TurnSubmissions         TurnSubmissionStore
	EffectiveHistory        EffectiveHistoryStore
	SessionManagement       SessionManagementStore
	SessionBatchManagement  SessionBatchManagementStore
	SessionDeletionGuard    SessionDeletionGuard
	SessionPurge            SessionPurgeStore
	DeletedSessions         DeletedSessionStore
	HistoricalState         HistoricalSessionStateStore
	SessionForks            SessionForkStore
	SessionForkRecovery     SessionForkRecoveryStore
	SessionForkRuntime      SessionForkRuntime
	SessionForkContext      SessionForkContextPolicy
	SessionForkState        SessionForkProviderStateBinder
	SessionForkAttachments  SessionForkAttachmentStager
	Runtime                 RuntimeController
	HistoryRuntime          RuntimeHistoryController
	RuntimePreparation      RuntimePreparationPort
	SettingsPolicy          SettingsPolicy
	Attachments             AttachmentMaterializer
	Clock                   Clock
	SessionLocker           SessionLocker
	RuntimeStartGate        RuntimeStartGate
	LifecycleObserver       LifecycleObserver
	TerminalFailureObserver TerminalFailureObserver
	CommitObserver          CommitObserver
	RuntimeOperations       RuntimeOperationStore
	OperationEvents         RuntimeOperationEventPublisher
	OperationOwner          string
	Scheduler               Scheduler
	StaleTurnSettler        StaleTurnSettler
	GoalStore               GoalStateStore
	GoalFences              GoalGenerationFenceStore
	GoalRuntime             GoalRuntimeController
	GoalInbox               GoalReconcileInboxStore
	GoalOwner               string
	GoalClock               Clock
	GoalAttemptTimeout      time.Duration
	GoalRecoveryBudget      time.Duration
	GoalMaxAttempts         int
	GoalDispatchDeadline    time.Duration
	GoalActor               *SessionActor
	SessionMutationActor    *SessionActor

	// EditRetryDisabled neutralizes the durable edit-and-retry feature (PR
	// #1681). When set, new edit-retry operations are refused and any operation
	// left over from before it was disabled is quarantined during recovery
	// instead of engaging the saga. The zero value keeps the feature enabled so
	// its unit/conformance tests still exercise it; production wiring sets this
	// true. Remove once the saga's resend/recovery gap is fixed.
	EditRetryDisabled       bool
	SideConversationRuntime SideConversationRuntime
}

type Host struct {
	store                     CanonicalStore
	runtimeRailPlacement      RuntimeSessionRailPlacementResolver
	interactionTrees          CanonicalInteractionTreeStore
	turnSubmissions           TurnSubmissionStore
	effectiveHistory          EffectiveHistoryStore
	sessionManagement         SessionManagementStore
	sessionBatchManagement    SessionBatchManagementStore
	sessionDeletionGuard      SessionDeletionGuard
	sessionPurge              SessionPurgeStore
	deletedSessions           DeletedSessionStore
	historicalState           HistoricalSessionStateStore
	sessionForks              SessionForkStore
	sessionForkRecovery       SessionForkRecoveryStore
	sessionForkRuntime        SessionForkRuntime
	sideRuntime               SideConversationRuntime
	sessionForkContext        SessionForkContextPolicy
	sessionForkState          SessionForkProviderStateBinder
	sessionForkAttachments    SessionForkAttachmentStager
	runtime                   RuntimeController
	historyRuntime            RuntimeHistoryController
	preparation               RuntimePreparationPort
	settingsPolicy            SettingsPolicy
	attachments               AttachmentMaterializer
	clock                     Clock
	locker                    SessionLocker
	startupGate               RuntimeStartGate
	observer                  LifecycleObserver
	terminalFailure           TerminalFailureObserver
	commitObserver            CommitObserver
	operations                RuntimeOperationStore
	events                    RuntimeOperationEventPublisher
	owner                     string
	scheduler                 Scheduler
	staleTurns                StaleTurnSettler
	goals                     GoalStateStore
	goalFences                GoalGenerationFenceStore
	goalRuntime               GoalRuntimeController
	goalInbox                 GoalReconcileInboxStore
	goalOwner                 string
	goalClock                 Clock
	goalAttemptTimeout        time.Duration
	goalRecoveryBudget        time.Duration
	goalMaxAttempts           int
	goalDispatchDeadline      time.Duration
	goalActor                 *SessionActor
	sessionMutationActor      *SessionActor
	workspaceRuntimeAdmission *workspaceRuntimeAdmission
	editRetryDisabled         bool
	goalFencesRestored        sync.Map
	sideMu                    sync.Mutex
	sideConversations         map[string]sideConversationRegistration
}

func New(config Config) *Host {
	goalActor := config.GoalActor
	if goalActor == nil {
		goalActor = NewSessionActor()
	}
	sessionMutationActor := config.SessionMutationActor
	if sessionMutationActor == nil {
		sessionMutationActor = NewSessionActor()
	}
	host := &Host{
		store: config.CanonicalStore, runtimeRailPlacement: config.RuntimeRailPlacement,
		interactionTrees: config.InteractionTrees,
		turnSubmissions:  config.TurnSubmissions, effectiveHistory: config.EffectiveHistory,
		sessionManagement: config.SessionManagement, sessionBatchManagement: config.SessionBatchManagement, sessionDeletionGuard: config.SessionDeletionGuard, sessionPurge: config.SessionPurge,
		deletedSessions: config.DeletedSessions,
		sessionForks:    config.SessionForks, sessionForkRuntime: config.SessionForkRuntime,
		historicalState:    config.HistoricalState,
		sideRuntime:        config.SideConversationRuntime,
		sessionForkContext: config.SessionForkContext, sessionForkState: config.SessionForkState,
		sessionForkAttachments: config.SessionForkAttachments,
		runtime:                config.Runtime,
		historyRuntime:         config.HistoryRuntime,
		sessionForkRecovery:    config.SessionForkRecovery,
		preparation:            config.RuntimePreparation, settingsPolicy: config.SettingsPolicy, attachments: config.Attachments,
		clock: config.Clock, locker: config.SessionLocker, startupGate: config.RuntimeStartGate,
		observer: config.LifecycleObserver, terminalFailure: config.TerminalFailureObserver,
		commitObserver: config.CommitObserver,
		operations:     config.RuntimeOperations, events: config.OperationEvents,
		owner: config.OperationOwner, scheduler: config.Scheduler, staleTurns: config.StaleTurnSettler,
		goals: config.GoalStore, goalFences: config.GoalFences, goalRuntime: config.GoalRuntime, goalInbox: config.GoalInbox,
		goalOwner: config.GoalOwner, goalClock: config.GoalClock,
		goalAttemptTimeout: config.GoalAttemptTimeout, goalRecoveryBudget: config.GoalRecoveryBudget,
		goalMaxAttempts: config.GoalMaxAttempts, goalDispatchDeadline: config.GoalDispatchDeadline,
		goalActor: goalActor, sessionMutationActor: sessionMutationActor,
		workspaceRuntimeAdmission: newWorkspaceRuntimeAdmission(),
		editRetryDisabled:         config.EditRetryDisabled,
		sideConversations:         make(map[string]sideConversationRegistration),
	}
	if host.runtimeRailPlacement == nil {
		host.runtimeRailPlacement, _ = host.store.(RuntimeSessionRailPlacementResolver)
	}
	if host.interactionTrees == nil {
		host.interactionTrees, _ = host.store.(CanonicalInteractionTreeStore)
	}
	if host.deletedSessions == nil {
		host.deletedSessions, _ = config.SessionBatchManagement.(DeletedSessionStore)
	}
	if host.sessionForkRecovery == nil {
		host.sessionForkRecovery, _ = host.sessionForks.(SessionForkRecoveryStore)
	}
	// Durable runtime and goal failures reach TerminalFailureObserver through
	// the same wrappers, so an adapter that wires only failure analytics still
	// observes them.
	observesCommits := host.commitObserver != nil || host.terminalFailure != nil
	if host.operations != nil && observesCommits {
		host.operations = &observedRuntimeOperationStore{RuntimeOperationStore: host.operations, host: host}
	}
	if host.effectiveHistory != nil && observesCommits {
		host.effectiveHistory = &observedEffectiveHistoryStore{
			EffectiveHistoryStore: host.effectiveHistory,
			host:                  host,
		}
	}
	if host.goals != nil && observesCommits {
		host.goals = &observedGoalStateStore{GoalStateStore: host.goals, host: host}
	}
	if registrar, ok := config.GoalRuntime.(GoalRuntimeControlLifecycleRegistrar); ok {
		registrar.SetGoalControlAppliedSink(host.ObserveRuntimeGoalControlApplied)
	}
	return host
}

// observeStep reports a diagnostic lifecycle step. A failed step never emits a
// TerminalFailure on its own; it only names the stage that the enclosing
// command reports once at its boundary.
func (h *Host) observeStep(ctx context.Context, flow, name, workspaceID, sessionID, provider string, startedAt time.Time, err error) {
	if h != nil && h.observer != nil {
		h.observer.ObserveLifecycleStep(ctx, LifecycleStep{
			Flow: flow, Name: name, AgentSessionID: sessionID, Provider: provider, StartedAt: startedAt, Err: err,
		})
	}
	recordCommandFailureStage(ctx, flow, workspaceID, sessionID, provider, name, err)
}

func (h *Host) observeGuidanceTargetFailure(
	ctx context.Context,
	ref SessionRef,
	provider, turnID, clientSubmitID string,
	startedAt time.Time,
	err error,
) {
	if err == nil {
		return
	}
	if h != nil && h.observer != nil {
		h.observer.ObserveLifecycleStep(ctx, LifecycleStep{
			Flow: "guidance", Name: "guidance_target", AgentSessionID: ref.AgentSessionID,
			Provider: provider, StartedAt: startedAt, Err: err,
		})
	}
	h.observeTerminalFailure(ctx, TerminalFailure{
		Flow:           "guidance",
		FailureStage:   "guidance_target",
		WorkspaceID:    ref.WorkspaceID,
		AgentSessionID: ref.AgentSessionID,
		TurnID:         strings.TrimSpace(turnID),
		ClientSubmitID: strings.TrimSpace(clientSubmitID),
		Provider:       provider,
		ErrorCode:      guidanceTargetFailureCode(err),
		ErrorMessage:   err.Error(),
		Retryable:      false,
	})
}

func (h *Host) observeTerminalFailure(ctx context.Context, failure TerminalFailure) {
	if h == nil || h.terminalFailure == nil {
		return
	}
	if h.store != nil && strings.TrimSpace(failure.WorkspaceID) != "" && strings.TrimSpace(failure.AgentSessionID) != "" {
		if session, found, err := h.store.GetSession(ctx, failure.WorkspaceID, failure.AgentSessionID); err == nil && found {
			if strings.TrimSpace(failure.Provider) == "" {
				failure.Provider = strings.TrimSpace(session.Provider)
			}
			failure.IsChildSession = failure.IsChildSession || canonicalSessionIsChild(session)
		}
	}
	if failure.ErrorMessage == "" && failure.ErrorCode == "" {
		return
	}
	// A specific emission owns the incident; the enclosing command boundary
	// must not report the same failure a second time.
	markCommandTerminalFailureEmitted(ctx)
	h.terminalFailure.ObserveTerminalFailure(ctx, failure)
}

type systemClock struct{}

func (systemClock) Now() time.Time { return time.Now() }

func (h *Host) now() time.Time {
	if h != nil && h.clock != nil {
		return h.clock.Now()
	}
	return systemClock{}.Now()
}