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

package agentruntime

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

	agentsessionstore "github.com/tutti-os/tutti/packages/agent/daemon/activity"
	activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
	replay "github.com/tutti-os/tutti/packages/agent/session-replay"
	"github.com/tutti-os/tutti/packages/agent/store-sqlite/canonical"
)

func (c *Controller) enqueueSessionReport(ctx context.Context, session Session, events []activityshared.Event) {
	if session.IsSideConversation() {
		return
	}
	c.enqueueSessionReportWithInitializationState(ctx, session, events, false)
}

// enqueueInitializedSessionReport publishes the release batch after Host has
// already committed the canonical Session. The Controller intentionally keeps
// its in-memory publication marker until all queued runtime callbacks are
// drained, so this path must not mistake that ordering marker for an
// uninitialized canonical Session and hide the report again.
func (c *Controller) enqueueInitializedSessionReport(ctx context.Context, session Session, events []activityshared.Event) {
	c.enqueueSessionReportWithInitializationState(ctx, session, events, true)
}

func (c *Controller) enqueueSessionReportWithInitializationState(
	ctx context.Context,
	session Session,
	events []activityshared.Event,
	canonicalInitialized bool,
) {
	if session.IsSideConversation() {
		return
	}
	c.observeGoalControlLifecycle(ctx, session, events)
	report := c.prepareSessionReportWithInitializationState(session, events, canonicalInitialized)
	c.observeProviderObservations(ctx, session, report.ProviderObservations)
	if len(report.GoalReconcileRequests) > 0 {
		control := report
		control.TimelineItems = nil
		control.StatePatches = nil
		control.MessageUpdates = nil
		control.SessionAudits = nil
		report.GoalReconcileRequests = nil
		_ = c.reportGoalReconcileControl(ctx, control)
	}
	c.enqueueReport(ctx, report)
}

func (c *Controller) prepareSessionReport(
	session Session,
	events []activityshared.Event,
) agentsessionstore.ReportActivityInput {
	return c.prepareSessionReportWithInitializationState(session, events, false)
}

func (c *Controller) prepareSessionReportWithInitializationState(
	session Session,
	events []activityshared.Event,
	canonicalInitialized bool,
) agentsessionstore.ReportActivityInput {
	c.mu.Lock()
	publicationPending := !canonicalInitialized && c.sessionPublicationPendingLocked(sessionKey(session.RoomID, session.AgentSessionID))
	c.mu.Unlock()
	if publicationPending {
		// A still-provisional runtime has not crossed the durable submitted-intent
		// barrier. Keep incidental provider events hidden until that barrier
		// publishes the canonical prompt. The normal initial-content path removes
		// this marker immediately after the barrier, so an explicit rejection is
		// projected as a visible failed Turn rather than compensated away.
		session.Visible = false
	}
	report := reportActivityInput(session, events)
	c.enrichReportStatePatchesWithSessionMetadata(session, &report)
	if publicationPending {
		hideProvisionalSessionReport(&report)
		report.MessageUpdates = nil
		report.SessionAudits = nil
	}
	return report
}

// reportSessionBeforePublish establishes the commit-before-publish barrier
// for terminal facts. Live projections read the canonical Store while
// handling the published event, so publishing a settled event before its
// active-turn pointer is cleared creates an invalid snapshot and forces every
// caller to reconnect. The bounded wait keeps a broken reporter fail-fast and
// deliberately withholds the uncommitted event rather than advertising state
// that the canonical store does not contain.
func (c *Controller) reportSessionBeforePublish(
	ctx context.Context,
	session Session,
	events []activityshared.Event,
) error {
	if session.IsSideConversation() {
		return nil
	}
	if c == nil || c.reporter == nil {
		return nil
	}
	c.observeGoalControlLifecycle(ctx, session, events)
	report := c.prepareSessionReport(session, events)
	if len(report.StatePatches) == 0 && len(report.MessageUpdates) == 0 &&
		len(report.SessionAudits) == 0 && len(report.GoalReconcileRequests) == 0 {
		return nil
	}
	c.observeProviderObservations(ctx, session, report.ProviderObservations)
	reportCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer cancel()
	request := reportRequest{
		ctx:    reportCtx,
		report: report,
		done:   make(chan error, 1),
	}
	if c.reportQueue == nil {
		return c.report(request.ctx, request)
	}
	queueDepth := c.reportQueue.enqueue(request)
	select {
	case err := <-request.done:
		return err
	case <-reportCtx.Done():
		slog.Warn(
			"agent session terminal activity report barrier timed out",
			"event", "agent_session.activity_report.terminal_barrier_timeout",
			"room_id", session.RoomID,
			"agent_session_id", session.AgentSessionID,
			"queue_depth", queueDepth,
			"error", reportCtx.Err(),
		)
		return reportCtx.Err()
	}
}

func (c *Controller) SetGoalControlLifecycleObserver(observer GoalControlLifecycleObserver) {
	if c == nil {
		return
	}
	c.goalControlObserverMu.Lock()
	c.goalControlObserver = observer
	c.goalControlObserverMu.Unlock()
}

func (c *Controller) observeGoalControlLifecycle(
	ctx context.Context,
	session Session,
	events []activityshared.Event,
) {
	if c == nil || len(events) == 0 {
		return
	}
	c.goalControlObserverMu.RLock()
	observer := c.goalControlObserver
	c.goalControlObserverMu.RUnlock()
	if observer == nil {
		return
	}
	for _, event := range events {
		if event.Type != activityshared.EventGoalControlApplied {
			continue
		}
		metadata := event.Payload.Metadata
		observation := GoalControlAppliedObservation{
			WorkspaceID:      session.RoomID,
			AgentSessionID:   firstNonEmptyString(event.AgentSessionID, session.AgentSessionID),
			OperationID:      stringFromPayload(metadata, "operationId"),
			Revision:         payloadInt64(metadata, "revision"),
			RepairEpoch:      payloadInt64(metadata, "repairEpoch"),
			Action:           stringFromPayload(metadata, "action"),
			ProviderTurnID:   stringFromPayload(metadata, "providerTurnId"),
			Observed:         payloadObject(metadata["goal"]),
			OccurredAtUnixMS: event.OccurredAtUnixMS,
			ExecutionPending: payloadBoolValue(metadata, "executionPending"),
		}
		if err := observer.ObserveGoalControlApplied(ctx, observation); err != nil {
			slog.Warn(
				"record runtime goal control application failed",
				"event", "agent_session.goal_control.observe_applied_failed",
				"room_id", observation.WorkspaceID,
				"agent_session_id", observation.AgentSessionID,
				"operation_id", observation.OperationID,
				"revision", observation.Revision,
				"error", err,
			)
		}
	}
}

func (c *Controller) SetProviderObservationObserver(observer ProviderObservationObserver) {
	if c == nil {
		return
	}
	c.providerObservationMu.Lock()
	c.providerObservationObserver = observer
	c.providerObservationMu.Unlock()
}

func (c *Controller) observeProviderObservations(
	ctx context.Context,
	session Session,
	observations []replay.ProviderObservationBatch,
) {
	if c == nil || len(observations) == 0 {
		return
	}
	c.providerObservationMu.RLock()
	observer := c.providerObservationObserver
	c.providerObservationMu.RUnlock()
	if observer == nil {
		return
	}
	if err := observer.ObserveProviderObservations(
		ctx,
		session.RoomID,
		session.AgentSessionID,
		observations,
	); err != nil {
		slog.Warn(
			"record provider observation candidate failed",
			"event", "agent_session.provider_observation.record_failed",
			"room_id", session.RoomID,
			"agent_session_id", session.AgentSessionID,
			"error", err,
		)
	}
}

// reportSubmittedTurnDurable is the acceptance barrier for a user submission.
// The durable reporter commits the submitted Turn and canonical user message
// together before Exec may publish the transition, start provider work, or
// return success.
func (c *Controller) reportSubmittedTurnDurable(
	ctx context.Context,
	session Session,
	events []activityshared.Event,
	keepProvisional bool,
) error {
	if session.IsSideConversation() {
		return nil
	}
	if c == nil || c.reporter == nil {
		// Reporter-less controllers are used as standalone runtimes and have no
		// durable projection. The wired tuttid runtime always provides a reporter.
		return nil
	}
	if keepProvisional {
		session.Visible = false
	}
	report := reportActivityInput(session, events)
	c.enrichReportStatePatchesWithSessionMetadata(session, &report)
	if keepProvisional {
		hideProvisionalSessionReport(&report)
	}
	return c.reporter.ReportSubmitProvenance(ctx, report)
}

func hideProvisionalSessionReport(report *agentsessionstore.ReportActivityInput) {
	if report == nil {
		return
	}
	for index := range report.StatePatches {
		report.StatePatches[index].RuntimeContext = clonePayload(
			report.StatePatches[index].RuntimeContext,
		)
		if report.StatePatches[index].RuntimeContext == nil {
			report.StatePatches[index].RuntimeContext = make(map[string]any)
		}
		report.StatePatches[index].RuntimeContext["visible"] = false
		report.StatePatches[index].RuntimeContext["provisional"] = true
	}
}

// reportProviderAcceptanceDurable is the acceptance barrier for a provider
// root turn. Edit-retry dispatch completion may only follow this report.
func (c *Controller) reportProviderAcceptanceDurable(
	ctx context.Context,
	session Session,
	events []activityshared.Event,
) (bool, error) {
	if c == nil || c.reporter == nil || !containsDurableProviderAcceptance(events) {
		return false, nil
	}
	report := c.prepareSessionReport(session, events)
	// Mirror enqueueSessionReport: observe batches before the durable commit so
	// checkpoint candidates exist when ObserveReplayCommitted runs.
	c.observeProviderObservations(ctx, session, report.ProviderObservations)
	reportCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
	defer cancel()
	request := reportRequest{
		ctx:    reportCtx,
		report: report,
		done:   make(chan error, 1),
	}
	if c.reportQueue == nil {
		return true, c.report(request.ctx, request)
	}
	c.reportQueue.enqueue(request)
	select {
	case err := <-request.done:
		return true, err
	case <-reportCtx.Done():
		return true, reportCtx.Err()
	}
}

// flushSessionReports waits until every report enqueued earlier for this
// Session has crossed the durable reporter. Goal-generation fencing uses this
// after the adapter's publication handoff, so Host cannot observe an idle
// Session immediately before an already-published Goal start commits.
func (c *Controller) flushSessionReports(ctx context.Context, session Session) error {
	if c == nil || c.reportQueue == nil {
		return nil
	}
	request := reportRequest{
		ctx: context.WithoutCancel(ctx),
		report: agentsessionstore.ReportActivityInput{
			WorkspaceID: session.RoomID,
			Source:      eventSourceFromSession(session),
		},
		barrier: true,
		done:    make(chan error, 1),
	}
	c.reportQueue.enqueue(request)
	select {
	case err := <-request.done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

func containsDurableProviderAcceptance(events []activityshared.Event) bool {
	for _, event := range events {
		if event.Type != activityshared.EventRootProviderTurnStarted {
			continue
		}
		source, _ := event.Payload.Metadata["acceptanceSource"].(string)
		if source == string(AcceptanceSourceTurnStartResponse) ||
			source == string(AcceptanceSourceHistoryRead) {
			return true
		}
	}
	return false
}

func (c *Controller) reportGoalReconcileControl(ctx context.Context, report agentsessionstore.ReportActivityInput) error {
	if c == nil || c.reporter == nil {
		return errors.New("durable goal reconcile reporter is unavailable")
	}
	return c.reporter.Report(ctx, report)
}

func (c *Controller) reportGoalReconcileDurable(ctx context.Context, session Session, request GoalReconcileDurableRequest) error {
	if session.IsSideConversation() {
		return nil
	}
	report := agentsessionstore.ReportActivityInput{
		WorkspaceID: session.RoomID,
		Connector:   &canonical.ConnectorInfo{ID: session.Provider, Version: "agent-gui-runtime"},
		Source:      eventSourceFromSession(session),
		GoalReconcileRequests: []agentsessionstore.WorkspaceAgentGoalReconcileRequest{{
			RequestID: request.RequestID, Phase: request.Phase, AgentSessionID: session.AgentSessionID,
			ProviderTurnID: request.ProviderTurnID, Reason: request.Reason, FenceMode: request.FenceMode,
			ExpectedOperationID: request.ExpectedOperationID, ExpectedRevision: request.ExpectedRevision,
			ExpectedRepairEpoch: request.ExpectedRepairEpoch, QuiesceSucceeded: request.QuiesceSucceeded,
			QuiesceError: request.QuiesceError,
		}},
	}
	return c.reportGoalReconcileControl(ctx, report)
}

func (c *Controller) enqueueSessionSnapshotReport(ctx context.Context, session Session) {
	if session.IsSideConversation() {
		return
	}
	report := agentsessionstore.ReportActivityInput{
		WorkspaceID: session.RoomID,
		Connector: &canonical.ConnectorInfo{
			ID:      session.Provider,
			Version: "agent-gui-runtime",
		},
		Source: eventSourceFromSession(session),
	}
	c.enrichReportWithSessionSnapshot(session, &report)
	c.enqueueReport(ctx, report)
}

func (c *Controller) enqueueSessionStatePatchReport(
	ctx context.Context,
	session Session,
	patch agentsessionstore.WorkspaceAgentStatePatch,
) {
	if session.IsSideConversation() {
		return
	}
	report := agentsessionstore.ReportActivityInput{
		WorkspaceID: session.RoomID,
		Connector: &canonical.ConnectorInfo{
			ID:      session.Provider,
			Version: "agent-gui-runtime",
		},
		Source:       eventSourceFromSession(session),
		StatePatches: []agentsessionstore.WorkspaceAgentStatePatch{patch},
	}
	c.enqueueReport(ctx, report)
}

func (c *Controller) enrichReportWithSessionSnapshot(session Session, report *agentsessionstore.ReportActivityInput) {
	if report == nil {
		return
	}
	snapshot := c.sessionStateSnapshot(session)
	if snapshot.AgentSessionID == "" {
		return
	}
	patch := statePatchFromSessionStateSnapshot(snapshot)
	if len(report.StatePatches) == 0 {
		report.StatePatches = append(report.StatePatches, patch)
		return
	}
	enrichReportStatePatchesWithSessionMetadata(report, patch)
}

func (c *Controller) enrichReportStatePatchesWithSessionMetadata(
	session Session,
	report *agentsessionstore.ReportActivityInput,
) {
	if report == nil || len(report.StatePatches) == 0 {
		return
	}
	snapshot := c.sessionStateSnapshot(session)
	if snapshot.AgentSessionID == "" {
		return
	}
	snapshotPatch := statePatchFromSessionStateSnapshot(snapshot)
	enrichReportStatePatchesWithSessionMetadata(report, snapshotPatch)
	if session.UserTitleSet {
		// A user-established title is the authoritative title source. Never let
		// a stale provider title carried by an event payload override it here;
		// the session's accepted title always wins in the persisted report.
		title := strings.TrimSpace(snapshotPatch.Title)
		for index := range report.StatePatches {
			if !statePatchMatchesSession(report.StatePatches[index], snapshotPatch.AgentSessionID) {
				continue
			}
			report.StatePatches[index].Title = title
		}
	}
}

func (c *Controller) enrichStreamStateEventsWithSessionSnapshot(
	session Session,
	events []StreamEvent,
) {
	if c == nil || len(events) == 0 {
		return
	}
	snapshot := c.sessionStateSnapshot(session)
	if snapshot.AgentSessionID == "" {
		return
	}
	snapshotPatch := statePatchFromSessionStateSnapshot(snapshot)
	for index := range events {
		if events[index].EventType != StreamEventStatePatch {
			continue
		}
		patch, ok := events[index].Data.(agentsessionstore.WorkspaceAgentStatePatch)
		if !ok {
			continue
		}
		tmp := agentsessionstore.ReportActivityInput{
			StatePatches: []agentsessionstore.WorkspaceAgentStatePatch{patch},
		}
		enrichReportStatePatchesWithSessionMetadata(&tmp, snapshotPatch)
		tmp.StatePatches[0].TurnLifecycle = cloneTurnLifecycle(snapshotPatch.TurnLifecycle)
		tmp.StatePatches[0].SubmitAvailability = cloneSubmitAvailability(snapshotPatch.SubmitAvailability)
		if session.UserTitleSet && statePatchMatchesSession(patch, snapshotPatch.AgentSessionID) {
			// Same ownership rule as the report enrichment: a user-established
			// title is the authoritative title source for the stream projection.
			tmp.StatePatches[0].Title = strings.TrimSpace(snapshotPatch.Title)
		}
		events[index].Data = tmp.StatePatches[0]
	}
}

func statePatchMatchesSession(
	patch agentsessionstore.WorkspaceAgentStatePatch,
	sessionID string,
) bool {
	sessionID = strings.TrimSpace(sessionID)
	return sessionID != "" && strings.TrimSpace(patch.AgentSessionID) == sessionID
}

// enrichReportStatePatchesWithSessionMetadata fills stable session metadata on
// persisted event reports. Canonical turn lifecycle is intentionally excluded:
// only an event's explicit Turn patch may advance a WorkspaceAgentTurn.
func enrichReportStatePatchesWithSessionMetadata(
	report *agentsessionstore.ReportActivityInput,
	patch agentsessionstore.WorkspaceAgentStatePatch,
) {
	if report == nil {
		return
	}
	for index := range report.StatePatches {
		if patch.AgentSessionID != "" &&
			report.StatePatches[index].AgentSessionID != "" &&
			strings.TrimSpace(report.StatePatches[index].AgentSessionID) != strings.TrimSpace(patch.AgentSessionID) {
			continue
		}
		report.StatePatches[index].Settings = clonePayload(patch.Settings)
		if patch.Capabilities != nil {
			report.StatePatches[index].Capabilities = canonical.CloneCapabilitySnapshot(patch.Capabilities)
		}
		report.StatePatches[index].RuntimeContext = clonePayload(patch.RuntimeContext)
		report.StatePatches[index].RuntimeContextPatch = nil
		if report.StatePatches[index].Provider == "" {
			report.StatePatches[index].Provider = patch.Provider
		}
		if report.StatePatches[index].ProviderSessionID == "" {
			report.StatePatches[index].ProviderSessionID = patch.ProviderSessionID
		}
		if report.StatePatches[index].Model == "" {
			report.StatePatches[index].Model = patch.Model
		}
		if report.StatePatches[index].PermissionModeID == "" {
			report.StatePatches[index].PermissionModeID = patch.PermissionModeID
		}
		if report.StatePatches[index].CWD == "" {
			report.StatePatches[index].CWD = patch.CWD
		}
		if report.StatePatches[index].Title == "" {
			report.StatePatches[index].Title = patch.Title
		}
	}
}

func (c *Controller) enqueueReport(ctx context.Context, report agentsessionstore.ReportActivityInput) {
	if len(report.TimelineItems) == 0 && len(report.StatePatches) == 0 && len(report.MessageUpdates) == 0 && len(report.SessionAudits) == 0 && len(report.GoalReconcileRequests) == 0 {
		return
	}
	if c.reporter == nil {
		return
	}
	request := reportRequest{
		ctx:    context.WithoutCancel(ctx),
		report: report,
	}
	timelineItemsForLog, statePatchesForLog := SummarizeReportActivityInputForLog(report)
	slog.Debug(
		"agent session activity report enqueued",
		"event", "agent_session.activity_report.enqueued",
		"room_id", report.WorkspaceID,
		"agent_session_id", report.Source.AgentID,
		"provider", report.Source.Provider,
		"provider_session_id", report.Source.ProviderSessionID,
		"timeline_item_count", len(report.TimelineItems),
		"state_patch_count", len(report.StatePatches),
		"message_update_count", len(report.MessageUpdates),
		"session_audit_count", len(report.SessionAudits),
		"timeline_items", timelineItemsForLog,
		"state_patches", statePatchesForLog,
	)
	if c.reportQueue == nil {
		_ = c.report(request.ctx, request)
		return
	}
	depth := c.reportQueue.enqueue(request)
	if depth >= 1024 && depth%1024 == 0 {
		slog.Warn(
			"agent session activity report queue backlog is growing",
			"event", "agent_session.activity_report.queue_backlog",
			"room_id", report.WorkspaceID,
			"agent_session_id", report.Source.AgentID,
			"provider", report.Source.Provider,
			"provider_session_id", report.Source.ProviderSessionID,
			"queue_depth", depth,
			"timeline_item_count", len(report.TimelineItems),
			"state_patch_count", len(report.StatePatches),
			"message_update_count", len(report.MessageUpdates),
			"session_audit_count", len(report.SessionAudits),
			"timeline_items", timelineItemsForLog,
			"state_patches", statePatchesForLog,
		)
	}
}

func (c *Controller) runReportWorker() {
	if c.reportQueue == nil {
		return
	}
	coalescer := newStreamingReportCoalescer(defaultStreamingReportCoalesceWindow)
	defer coalescer.stop()
	for {
		// Do not let a continuously populated report queue starve the streaming
		// coalescer's timer.
		select {
		case <-coalescer.ready():
			for _, pending := range coalescer.flushAll() {
				_ = c.report(pending.ctx, pending)
			}
		default:
		}
		if request, ok := c.reportQueue.dequeue(); ok {
			for _, next := range coalescer.add(request) {
				_ = c.report(next.ctx, next)
			}
			continue
		}
		select {
		case <-c.reportQueue.ready():
		case <-coalescer.ready():
			for _, pending := range coalescer.flushAll() {
				_ = c.report(pending.ctx, pending)
			}
		}
	}
}

func (c *Controller) report(ctx context.Context, request reportRequest) (reportErr error) {
	if request.done != nil {
		defer func() {
			request.done <- reportErr
			close(request.done)
		}()
	}
	if request.barrier {
		return nil
	}
	if c.reporter == nil {
		return errors.New("agent session activity reporter is unavailable")
	}
	if request.submitProvenance {
		reportErr = c.reporter.ReportSubmitProvenance(ctx, request.report)
	} else {
		reportErr = c.reporter.Report(ctx, request.report)
	}
	if reportErr != nil {
		timelineItemsForLog, statePatchesForLog := SummarizeReportActivityInputForLog(request.report)
		slog.Error(
			"agent session activity report failed",
			"event", "agent_session.activity_report.controller_failed",
			"room_id", request.report.WorkspaceID,
			"agent_session_id", request.report.Source.AgentID,
			"provider", request.report.Source.Provider,
			"provider_session_id", request.report.Source.ProviderSessionID,
			"timeline_item_count", len(request.report.TimelineItems),
			"state_patch_count", len(request.report.StatePatches),
			"message_update_count", len(request.report.MessageUpdates),
			"session_audit_count", len(request.report.SessionAudits),
			"timeline_items", timelineItemsForLog,
			"state_patches", statePatchesForLog,
			"submit_provenance", request.submitProvenance,
			"error", reportErr,
		)
	}
	return reportErr
}

func sessionKey(roomID, agentSessionID string) string {
	return roomID + "/" + agentSessionID
}

func firstNonEmpty(values ...string) string {
	for _, value := range values {
		if strings.TrimSpace(value) != "" {
			return strings.TrimSpace(value)
		}
	}
	return ""
}

func deriveSessionStatusFromEvents(events []activityshared.Event, fallback string) string {
	status := strings.TrimSpace(fallback)
	for _, event := range events {
		switch event.Type {
		case activityshared.EventSessionFailed, activityshared.EventTurnFailed:
			status = SessionStatusFailed
		case activityshared.EventSessionCompleted:
			status = SessionStatusCompleted
		case activityshared.EventTurnCompleted:
			if strings.TrimSpace(event.Payload.TurnOutcome) == string(activityshared.TurnOutcomeInterrupted) {
				status = SessionStatusCanceled
			} else {
				status = SessionStatusReady
			}
		case activityshared.EventTurnUpdated:
			if event.Payload.TurnPhase == string(activityshared.TurnPhaseWaitingApproval) ||
				event.Payload.TurnPhase == string(activityshared.TurnPhaseWaitingInput) {
				status = SessionStatusWaiting
			} else if event.Payload.TurnPhase == string(activityshared.TurnPhaseWorking) ||
				event.Payload.TurnPhase == string(activityshared.TurnPhaseRunning) ||
				event.Payload.TurnPhase == string(activityshared.TurnPhaseSubmitted) {
				status = SessionStatusWorking
			}
		case activityshared.EventSessionUpdated:
			if next := sessionStatusFromActivity(event.Payload.EffectiveStatus); next != "" {
				status = next
			}
		case activityshared.EventTurnStarted:
			status = SessionStatusWorking
		}
	}
	return firstNonEmpty(status, SessionStatusReady)
}

func normalizeSessionStatus(status string) string {
	switch strings.TrimSpace(status) {
	case SessionStatusReady:
		return SessionStatusReady
	case SessionStatusWorking:
		return SessionStatusWorking
	case SessionStatusWaiting:
		return SessionStatusWaiting
	case SessionStatusCanceled:
		return SessionStatusCanceled
	case SessionStatusFailed:
		return SessionStatusFailed
	case SessionStatusCompleted:
		return SessionStatusCompleted
	default:
		return ""
	}
}