feat: mark finished runs as inactive

Inactive runs will not be processed by the controller and will be removed from
its cache. These runs will still be accessible from the API.

This logic is implemented such that other objects can be marked as inactive and
the controller will similarly ignore them.

Signed-off-by: Donnie Adams <[email protected]>

Author: thedadams

go.mod

@@ -23,7 +23,7 @@ require (
 	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de
 	github.com/mhale/smtpd v0.8.3
 	github.com/obot-platform/kinm v0.0.0-20250116162656-270198b40c6d
-	github.com/obot-platform/nah v0.0.0-20250227230402-c8ac47e1add7
+	github.com/obot-platform/nah v0.0.0-20250305151239-35a77e89a2eb
 	github.com/obot-platform/namegenerator v0.0.0-20241217121223-fc58bdb7dca2
 	github.com/obot-platform/obot/apiclient v0.0.0-00010101000000-000000000000
 	github.com/obot-platform/obot/logger v0.0.0-20241217130503-4004a5c69f32

go.sum

@@ -460,8 +460,8 @@ github.com/oasdiff/yaml3 v0.0.0-20241210130736-a94c01f36349 h1:t05Ww3DxZutOqbMN+
 github.com/oasdiff/yaml3 v0.0.0-20241210130736-a94c01f36349/go.mod h1:y5+oSEHCPT/DGrS++Wc/479ERge0zTFxaF8PbGKcg2o=
 github.com/obot-platform/kinm v0.0.0-20250116162656-270198b40c6d h1:GzMvRkssr4jAa2YvQiv9eXhjuNpaZVab3GajE7+cQ3s=
 github.com/obot-platform/kinm v0.0.0-20250116162656-270198b40c6d/go.mod h1:RzrH0geIlbiTHDGZ8bpCk5k1hwdU9uu3l4zJn9n0pZU=
-github.com/obot-platform/nah v0.0.0-20250227230402-c8ac47e1add7 h1:IuG8TZqqxz5P1ZHj2aCxEEn6LGgHXeUVhkG/ELh2a8U=
-github.com/obot-platform/nah v0.0.0-20250227230402-c8ac47e1add7/go.mod h1:KG1jLO9FeYvCPGI0iDqe5oqDqOFLd3/dt/iwuMianmI=
+github.com/obot-platform/nah v0.0.0-20250305151239-35a77e89a2eb h1:Cgf8prbkB2gwFAI2vsXRN1U7enKkFMYJjaUT+uNR7dc=
+github.com/obot-platform/nah v0.0.0-20250305151239-35a77e89a2eb/go.mod h1:be9DBO6pFx4yiKs9bm4GcuZgGX4OVpeRJclA9ojxzmo=
 github.com/obot-platform/namegenerator v0.0.0-20241217121223-fc58bdb7dca2 h1:jiyBM/TYxU6UNVS9ff8Y8n55DOKDYohKkIZjfHpjfTY=
 github.com/obot-platform/namegenerator v0.0.0-20241217121223-fc58bdb7dca2/go.mod h1:isbKX6EfvvG/ojjFB2ZLyz27+2xoG3yRmpTSE+ytWEs=
 github.com/olekukonko/tablewriter v0.0.6-0.20230925090304-df64c4bbad77 h1:3bMMZ1f+GPXFQ1uNaYbO/uECWvSfqEA+ZEXn1rFAT88=

pkg/api/handlers/threads.go

@@ -158,7 +158,7 @@ func (a *ThreadHandler) Update(req api.Context) error {
 		return err
 	}
 
-	// Don't allow update of tools here, do it with the /tools endpoing
+	// Don't allow update of tools here, do it with the /tools endpoint
 	newThread.Tools = existing.Spec.Manifest.Tools
 
 	existing.Spec.Manifest = newThread

pkg/controller/handlers/inactive/inactive.go

@@ -0,0 +1,27 @@
+package inactive
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/obot-platform/nah/pkg/backend"
+	kclient "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+func RemoveFromCache(ctx context.Context, backend backend.Backend, obj kclient.Object) error {
+	gvk := obj.GetObjectKind().GroupVersionKind()
+	if gvk.Kind == "" {
+		var err error
+		gvk, err = backend.GVKForObject(obj, backend.Scheme())
+		if err != nil {
+			return fmt.Errorf("failed to get GVK for object: %w", err)
+		}
+	}
+
+	informer, err := backend.GetInformerForKind(ctx, gvk)
+	if err != nil {
+		return fmt.Errorf("failed to get informer for kind %s: %w", gvk.String(), err)
+	}
+
+	return informer.GetStore().Delete(obj)
+}

pkg/controller/handlers/runs/runs.go

@@ -5,8 +5,10 @@ import (
 	"time"
 
 	"github.com/gptscript-ai/go-gptscript"
+	"github.com/obot-platform/nah/pkg/backend"
 	"github.com/obot-platform/nah/pkg/router"
 	"github.com/obot-platform/nah/pkg/untriggered"
+	"github.com/obot-platform/obot/pkg/controller/handlers/inactive"
 	"github.com/obot-platform/obot/pkg/invoke"
 	v1 "github.com/obot-platform/obot/pkg/storage/apis/obot.obot.ai/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -16,10 +18,14 @@ import (
 
 type Handler struct {
 	invoker *invoke.Invoker
+	backend backend.Backend
 }
 
-func New(invoker *invoke.Invoker) *Handler {
-	return &Handler{invoker: invoker}
+func New(invoker *invoke.Invoker, backend backend.Backend) *Handler {
+	return &Handler{
+		invoker: invoker,
+		backend: backend,
+	}
 }
 
 func (*Handler) DeleteRunState(req router.Request, _ router.Response) error {
@@ -79,3 +85,17 @@ func (h *Handler) DeleteFinished(req router.Request, _ router.Response) error {
 	}
 	return nil
 }
+
+func (h *Handler) MarkInactive(req router.Request, _ router.Response) error {
+	run := req.Object.(*v1.Run)
+	if run.DeletionTimestamp.IsZero() && run.Status.State == gptscript.Continue && run.Labels[v1.LabelInactive] != "true" {
+		v1.SetInactive(run)
+		if err := req.Client.Update(req.Ctx, run); err != nil {
+			return err
+		}
+
+		return inactive.RemoveFromCache(req.Ctx, h.backend, run)
+	}
+
+	return nil
+}

pkg/controller/handlers/threads/threads.go

@@ -8,9 +8,11 @@ import (
 
 	"github.com/gptscript-ai/go-gptscript"
 	"github.com/obot-platform/nah/pkg/router"
+	"github.com/obot-platform/nah/pkg/untriggered"
 	v1 "github.com/obot-platform/obot/pkg/storage/apis/obot.obot.ai/v1"
 	"github.com/obot-platform/obot/pkg/system"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	kclient "sigs.k8s.io/controller-runtime/pkg/client"
 )
 
@@ -288,3 +290,25 @@ func (t *Handler) CleanupEphemeralThreads(req router.Request, _ router.Response)
 
 	return kclient.IgnoreNotFound(req.Delete(thread))
 }
+
+func (t *Handler) ActivateRuns(req router.Request, _ router.Response) error {
+	var runs v1.RunList
+	// This must be uncached since inactive things aren't in the cache.
+	if err := req.List(untriggered.UncachedList(&runs), &kclient.ListOptions{
+		Namespace:     req.Namespace,
+		FieldSelector: fields.SelectorFromSet(map[string]string{"spec.threadName": req.Object.GetName()}),
+	}); err != nil {
+		return fmt.Errorf("failed to list runs for thread %s: %w", req.Object.GetName(), err)
+	}
+
+	for _, run := range runs.Items {
+		if !v1.IsActive(&run) {
+			v1.SetActive(&run)
+			if err := req.Client.Update(req.Ctx, &run); err != nil {
+				return fmt.Errorf("failed to update run %q to active: %w", run.Name, err)
+			}
+		}
+	}
+
+	return nil
+}

pkg/controller/routes.go

@@ -40,7 +40,7 @@ func (c *Controller) setupRoutes() error {
 	knowledgeset := knowledgeset.New(c.services.Invoker)
 	knowledgesource := knowledgesource.NewHandler(c.services.Invoker, c.services.GPTClient)
 	knowledgefile := knowledgefile.New(c.services.Invoker, c.services.GPTClient, c.services.KnowledgeSetIngestionLimit)
-	runs := runs.New(c.services.Invoker)
+	runs := runs.New(c.services.Invoker, c.services.Router.Backend())
 	webHooks := webhook.New()
 	cronJobs := cronjob.New()
 	oauthLogins := oauthapp.NewLogin(c.services.Invoker, c.services.ServerURL)
@@ -55,6 +55,8 @@ func (c *Controller) setupRoutes() error {
 	root.Type(&v1.Run{}).HandlerFunc(cleanup.Cleanup)
 	root.Type(&v1.Run{}).HandlerFunc(runs.Resume)
 	root.Type(&v1.Run{}).HandlerFunc(workflow.GetTaskResult)
+	// This handler should be the last one so that deleting Runs will not be marked as inactive.
+	root.Type(&v1.Run{}).HandlerFunc(runs.MarkInactive)
 
 	// Threads
 	root.Type(&v1.Thread{}).HandlerFunc(cleanup.Cleanup)
@@ -66,6 +68,7 @@ func (c *Controller) setupRoutes() error {
 	root.Type(&v1.Thread{}).HandlerFunc(threads.CleanupEphemeralThreads)
 	root.Type(&v1.Thread{}).HandlerFunc(threads.SetCreated)
 	root.Type(&v1.Thread{}).FinalizeFunc(v1.ThreadFinalizer, credentialCleanup.Remove)
+	root.Type(&v1.Thread{}).FinalizeFunc(v1.ThreadFinalizer+"-child-cleanup", threads.ActivateRuns)
 
 	// KnowledgeSummary
 	root.Type(&v1.KnowledgeSummary{}).HandlerFunc(cleanup.Cleanup)

pkg/services/config.go

@@ -7,6 +7,8 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
+	"reflect"
+	"slices"
 	"sort"
 	"strings"
 
@@ -18,6 +20,7 @@ import (
 	"github.com/gptscript-ai/gptscript/pkg/sdkserver"
 	"github.com/obot-platform/nah"
 	"github.com/obot-platform/nah/pkg/apply"
+	nfields "github.com/obot-platform/nah/pkg/fields"
 	"github.com/obot-platform/nah/pkg/leader"
 	"github.com/obot-platform/nah/pkg/router"
 	"github.com/obot-platform/nah/pkg/runtime"
@@ -43,9 +46,12 @@ import (
 	"github.com/obot-platform/obot/pkg/system"
 	coordinationv1 "k8s.io/api/coordination/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/authentication/authenticator"
 	"k8s.io/apiserver/pkg/authentication/request/union"
+	kcache "sigs.k8s.io/controller-runtime/pkg/cache"
+	kclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	// Setup nah logging
 	_ "github.com/obot-platform/nah/pkg/logrus"
@@ -76,6 +82,7 @@ type Config struct {
 	AuthAdminEmails            []string `usage:"Emails of admin users"`
 	AgentsDir                  string   `usage:"The directory to auto load agents on start (default $XDG_CONFIG_HOME/.obot/agents)"`
 	StaticDir                  string   `usage:"The directory to serve static files from"`
+	IgnoreInactive             bool     `usage:"Ignore inactive objects" default:"false"`
 
 	// Sendgrid webhook
 	SendgridWebhookUsername string `usage:"The username for the sendgrid webhook to authenticate with"`
@@ -279,17 +286,27 @@ func New(ctx context.Context, config Config) (*Services, error) {
 			return nil, err
 		}
 	} else {
-		var notFound gptscript.ErrNotFound
-		if err := c.DeleteCredential(ctx, system.DefaultNamespace, system.KnowledgeCredID); err != nil && !errors.As(err, &notFound) {
+		if err := c.DeleteCredential(ctx, system.DefaultNamespace, system.KnowledgeCredID); err != nil && !errors.As(err, &gptscript.ErrNotFound{}) {
 			return nil, err
 		}
 	}
 
+	byObjectFieldSelectors := make(map[kclient.Object]kcache.ByObject)
+	if config.IgnoreInactive {
+		for _, t := range scheme.Scheme.AllKnownTypes() {
+			if v, ok := reflect.New(t).Interface().(nfields.Fields); ok && slices.Contains(v.FieldNames(), v1.LabelInactive) {
+				// Only add the field selector requirement to objects that support the field selector.
+				byObjectFieldSelectors[v.(kclient.Object)] = kcache.ByObject{Field: fields.OneTermEqualSelector(v1.LabelInactive, "")}
+			}
+		}
+	}
+
 	r, err := nah.NewRouter("obot-controller", &nah.Options{
-		DefaultRESTConfig: restConfig,
-		Scheme:            scheme.Scheme,
-		ElectionConfig:    leader.NewDefaultElectionConfig("", "obot-controller", restConfig),
-		HealthzPort:       -1,
+		RESTConfig:     restConfig,
+		Scheme:         scheme.Scheme,
+		ByObject:       byObjectFieldSelectors,
+		ElectionConfig: leader.NewDefaultElectionConfig("", "obot-controller", restConfig),
+		HealthzPort:    -1,
 		GVKThreadiness: map[schema.GroupVersionKind]int{
 			v1.SchemeGroupVersion.WithKind("KnowledgeFile"): config.KnowledgeFileWorkers,
 		},

pkg/storage/apis/obot.obot.ai/v1/inactive.go

@@ -0,0 +1,20 @@
+package v1
+
+import kclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+const LabelInactive = "obot_inactive_object"
+
+func SetInactive(o kclient.Object) {
+	if o.GetLabels() == nil {
+		o.SetLabels(make(map[string]string, 1))
+	}
+	o.GetLabels()[LabelInactive] = "true"
+}
+
+func SetActive(o kclient.Object) {
+	delete(o.GetLabels(), LabelInactive)
+}
+
+func IsActive(o kclient.Object) bool {
+	return o.GetLabels()[LabelInactive] != "true"
+}

pkg/storage/apis/obot.obot.ai/v1/run.go

@@ -2,6 +2,7 @@ package v1
 
 import (
 	gptscriptclient "github.com/gptscript-ai/go-gptscript"
+	"github.com/obot-platform/nah/pkg/fields"
 	"github.com/obot-platform/obot/apiclient/types"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -23,6 +24,11 @@ const (
 	AuthProviderSyncAnnotation  = "obot.ai/auth-provider-sync"
 )
 
+var (
+	_ fields.Fields = (*Run)(nil)
+	_ DeleteRefs    = (*Run)(nil)
+)
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 type Run struct {
@@ -40,6 +46,8 @@ func (in *Run) Has(field string) bool {
 func (in *Run) Get(field string) string {
 	if in != nil {
 		switch field {
+		case LabelInactive:
+			return in.Labels[LabelInactive]
 		case "spec.threadName":
 			return in.Spec.ThreadName
 		case "spec.previousRunName":
@@ -51,7 +59,7 @@ func (in *Run) Get(field string) string {
 }
 
 func (in *Run) FieldNames() []string {
-	return []string{"spec.threadName", "spec.previousRunName"}
+	return []string{"spec.threadName", "spec.previousRunName", LabelInactive}
 }
 
 func (in *Run) GetColumns() [][]string {