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) != "" }