tutti/packages/agent/daemon/runtime.go

package agentdaemon

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

	agentruntime "github.com/tutti-os/tutti/packages/agent/daemon/runtime"
)

var ErrHostMetadataRequired = errors.New("agent daemon host metadata is required")
var ErrProcessTransportRequired = errors.New("agent daemon process transport is required")

const (
	defaultLiveSessionReaperIdleAfter     = 30 * time.Minute
	defaultLiveSessionReaperSweepInterval = 5 * time.Minute

	// shutdownCloseAllLiveSessionsTimeout bounds how long Runtime.Close waits
	// for CloseAllLiveSessions to force-terminate every live provider
	// process. Each process close is already internally bounded (SIGTERM,
	// then SIGKILL after a short grace period; see localProcessConnection),
	// so this is a backstop against an unexpectedly large number of live
	// sessions, not the primary timeout mechanism.
	shutdownCloseAllLiveSessionsTimeout = 15 * time.Second
)

type ActivityReporter = agentruntime.ActivityReporter
type DurableActivityReporter = agentruntime.DurableActivityReporter
type Adapter = agentruntime.Adapter
type ClientInfo = agentruntime.ClientInfo
type Controller = agentruntime.Controller
type HostMetadata = agentruntime.HostMetadata
type ProcessTransport = agentruntime.ProcessTransport
type RecordingProcessTransport = agentruntime.RecordingProcessTransport
type ReplayPlaybackState = agentruntime.ReplayPlaybackState
type ReplayProcessTransport = agentruntime.ReplayProcessTransport
type SessionReplayProcessRegistration = agentruntime.SessionReplayProcessRegistration
type SessionReplayProcessTransport = agentruntime.SessionReplayProcessTransport
type SessionRecordingProcessTransport = agentruntime.SessionRecordingProcessTransport
type ProviderCommand = agentruntime.ProviderCommand
type ProviderCommandResolver = agentruntime.ProviderCommandResolver
type ProviderLaunchPrepareInput = agentruntime.ProviderLaunchPrepareInput
type ProviderLaunchPrepareResult = agentruntime.ProviderLaunchPrepareResult
type ProviderLaunchPreparer = agentruntime.ProviderLaunchPreparer
type ProviderLaunchPreparerAdapter = agentruntime.ProviderLaunchPreparerAdapter
type AdapterResolver = agentruntime.AdapterResolver
type CommandNetworkAccessPolicy = agentruntime.CommandNetworkAccessPolicy

type Config struct {
	Reporter                   DurableActivityReporter
	ProcessTransport           ProcessTransport
	HostMetadata               HostMetadata
	ProviderCommandResolver    ProviderCommandResolver
	ProviderLaunchPreparer     ProviderLaunchPreparer
	AdapterResolver            AdapterResolver
	CommandNetworkAccessPolicy CommandNetworkAccessPolicy
	Adapters                   []Adapter
	LiveSessionReaper          LiveSessionReaperConfig
}

type LiveSessionReaperConfig struct {
	Enabled       *bool
	IdleAfter     time.Duration
	SweepInterval time.Duration
}

type Runtime struct {
	controller       *Controller
	processTransport ProcessTransport
	cancel           context.CancelFunc
	done             chan struct{}
	closeOnce        sync.Once
}

func NewRuntime(config Config) (*Runtime, error) {
	var controller *Controller
	if len(config.Adapters) > 0 {
		agentruntime.ApplyProviderLaunchPreparer(config.Adapters, config.ProviderLaunchPreparer)
		controller = agentruntime.NewController(config.Adapters, config.Reporter)
	} else {
		if !hasCompleteHostMetadata(config.HostMetadata) {
			return nil, ErrHostMetadataRequired
		}
		if config.ProcessTransport == nil {
			return nil, ErrProcessTransportRequired
		}
		controller = agentruntime.NewDefaultControllerWithOptions(
			config.Reporter,
			config.ProcessTransport,
			agentruntime.ControllerOptions{
				HostMetadata:               config.HostMetadata,
				ProviderCommandResolver:    config.ProviderCommandResolver,
				ProviderLaunchPreparer:     config.ProviderLaunchPreparer,
				AdapterResolver:            config.AdapterResolver,
				CommandNetworkAccessPolicy: config.CommandNetworkAccessPolicy,
			},
		)
	}
	runtime := &Runtime{
		controller:       controller,
		processTransport: config.ProcessTransport,
	}
	runtime.startLiveSessionReaper(config.LiveSessionReaper)
	return runtime, nil
}

func NewLocalProcessTransport() ProcessTransport {
	return agentruntime.NewLocalProcessTransport()
}

func NewRecordingProcessTransport(
	base ProcessTransport,
	directory string,
) (*RecordingProcessTransport, error) {
	return agentruntime.NewRecordingProcessTransport(base, directory)
}

func NewReplayProcessTransport(directory string) (*ReplayProcessTransport, error) {
	return agentruntime.NewReplayProcessTransport(directory)
}

func NewSessionReplayProcessTransport(
	registrations []SessionReplayProcessRegistration,
) (*SessionReplayProcessTransport, error) {
	return agentruntime.NewSessionReplayProcessTransport(registrations)
}

func NewSessionRecordingProcessTransport(
	base ProcessTransport,
) (*SessionRecordingProcessTransport, error) {
	return agentruntime.NewSessionRecordingProcessTransport(base)
}

func MustRuntime(config Config) *Runtime {
	runtime, err := NewRuntime(config)
	if err != nil {
		panic(err)
	}
	return runtime
}

func (r *Runtime) Controller() *Controller {
	if r == nil {
		return nil
	}
	return r.controller
}

func (r *Runtime) Close() {
	if r == nil {
		return
	}
	r.closeOnce.Do(func() {
		r.closeAllLiveSessions()
		if finalizer, ok := r.processTransport.(interface{ Finalize() error }); ok {
			if err := finalizer.Finalize(); err != nil {
				slog.Error(
					"agent process transport finalization failed",
					"event", "agent_session.process_transport.finalize_failed",
					"error", err,
				)
			}
		}
		if r.cancel != nil {
			r.cancel()
		}
		if r.done != nil {
			<-r.done
		}
	})
}

// closeAllLiveSessions force-terminates every live provider process, including
// Codex app-server, the Claude Code SDK sidecar, and subprocess adapters for
// providers that use ACP,
// before the daemon process exits. A spawned subprocess is not killed automatically
// just because tuttid exits — it is reparented to init and keeps running —
// so without this step, every daemon shutdown (or a desktop-parent-monitor
// triggered shutdown after the host app disappears) would orphan any
// in-flight provider processes, leaving them running unmanaged against the
// session's working directory indefinitely.
func (r *Runtime) closeAllLiveSessions() {
	if r == nil || r.controller == nil {
		return
	}
	ctx, cancel := context.WithTimeout(context.Background(), shutdownCloseAllLiveSessionsTimeout)
	defer cancel()
	result := r.controller.CloseAllLiveSessions(ctx)
	if result.Scanned == 0 && result.ResourceCleanupAttempted == 0 {
		return
	}
	slog.Info("agent live session shutdown close completed",
		"event", "agent_session.shutdown_close.completed",
		"scanned", result.Scanned,
		"closed", result.Closed,
		"skipped_cleanup_budget", result.SkippedCleanupBudget,
		"failed", result.Failed,
		"resource_cleanup_attempted", result.ResourceCleanupAttempted,
		"resource_cleanup_cleaned", result.ResourceCleanupCleaned,
		"resource_cleanup_failed", result.ResourceCleanupFailed,
	)
}

func (r *Runtime) startLiveSessionReaper(config LiveSessionReaperConfig) {
	if r == nil || r.controller == nil || !liveSessionReaperEnabled(config) {
		return
	}
	idleAfter := config.IdleAfter
	if idleAfter <= 0 {
		idleAfter = defaultLiveSessionReaperIdleAfter
	}
	sweepInterval := config.SweepInterval
	if sweepInterval <= 0 {
		sweepInterval = defaultLiveSessionReaperSweepInterval
	}
	ctx, cancel := context.WithCancel(context.Background())
	r.cancel = cancel
	r.done = make(chan struct{})
	go func() {
		defer close(r.done)
		ticker := time.NewTicker(sweepInterval)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
				result := r.controller.ReleaseIdleLiveSessions(ctx, agentruntime.ReleaseIdleLiveSessionsInput{
					IdleAfter: idleAfter,
					Now:       time.Now(),
				})
				if result.Scanned == 0 && result.ResourceCleanupAttempted == 0 {
					continue
				}
				slog.Info("agent live session reaper sweep completed",
					"event", "agent_session.live_reaper.sweep_completed",
					"scanned", result.Scanned,
					"released", result.Released,
					"skipped_fresh", result.SkippedFresh,
					"skipped_active_turn", result.SkippedActiveTurn,
					"skipped_unsupported", result.SkippedUnsupported,
					"skipped_not_live", result.SkippedNotLive,
					"skipped_busy", result.SkippedBusy,
					"skipped_cleanup_budget", result.SkippedCleanupBudget,
					"failed", result.Failed,
					"resource_cleanup_attempted", result.ResourceCleanupAttempted,
					"resource_cleanup_cleaned", result.ResourceCleanupCleaned,
					"resource_cleanup_failed", result.ResourceCleanupFailed,
				)
			}
		}
	}()
}

func liveSessionReaperEnabled(config LiveSessionReaperConfig) bool {
	if config.Enabled == nil {
		return true
	}
	return *config.Enabled
}

func hasCompleteHostMetadata(host HostMetadata) bool {
	return strings.TrimSpace(host.ClientInfo.Name) != "" &&
		strings.TrimSpace(host.ClientInfo.Title) != "" &&
		strings.TrimSpace(host.ClientInfo.Version) != "" &&
		strings.TrimSpace(host.WorkspaceEnvName) != "" &&
		strings.TrimSpace(host.OpenClawSessionKeyPrefix) != ""
}