refactor: move substrate runtime into v2 (#2596)

## Summary

- move the Substrate client, gateway, config, and list helpers under
`core/v2`
- remove the legacy `sandboxbackend` implementation and dormant session
actor lifecycle
- update v2 controller and system/session services to use the v2 package

## Testing

- `go test ./core/v2/substrate ./core/v2/agentinstance
./core/internal/service/session ./core/internal/service/system
./core/cmd/controller-v2`
- `go test -skip 'TestE2E.*' ./core/...`\n- `git diff --check`

Signed-off-by: Eitan Yarmush <eitan.yarmush@solo.io>

Author: EItanya

go/core/cmd/controller-v2/main.go

@@ -37,12 +37,12 @@ import (
 	taskservice "github.com/kagent-dev/kagent/go/core/internal/service/task"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
 	"github.com/kagent-dev/kagent/go/core/pkg/migrations"
-	legacysubstrate "github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
 	"github.com/kagent-dev/kagent/go/core/v2/a2agateway"
 	"github.com/kagent-dev/kagent/go/core/v2/agentinstance"
 	"github.com/kagent-dev/kagent/go/core/v2/checkpoint"
 	v2controller "github.com/kagent-dev/kagent/go/core/v2/controller"
 	v2mcp "github.com/kagent-dev/kagent/go/core/v2/mcp"
+	"github.com/kagent-dev/kagent/go/core/v2/substrate"
 	"golang.org/x/sync/errgroup"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -105,7 +105,7 @@ func main() {
 		log.Fatalf("add reconciler to controller manager: %v", err)
 	}
 
-	actors, err := legacysubstrate.Dial(ctx, legacysubstrate.Config{
+	actors, err := substrate.Dial(ctx, substrate.Config{
 		AteAPIEndpoint: env("SUBSTRATE_ATE_API_ENDPOINT", "dns:///api.ate-system.svc:443"),
 		CAFile:         os.Getenv("SUBSTRATE_ATE_API_CA_FILE"),
 		ClientCertFile: os.Getenv("SUBSTRATE_ATE_API_CLIENT_CERT_FILE"),
@@ -122,7 +122,7 @@ func main() {
 	instances := agentinstance.NewService(store, authorizer, instanceWorkflow)
 	checkpoints := checkpoint.NewService(store, authorizer, actors, instanceWorkflow)
 	gatewayDialer, err := a2agateway.NewRuntimeDialer(
-		env("SUBSTRATE_ATENET_ROUTER_URL", legacysubstrate.DefaultAtenetRouterURL),
+		env("SUBSTRATE_ATENET_ROUTER_URL", substrate.DefaultAtenetRouterURL),
 		authenticator,
 	)
 	if err != nil {

go/core/internal/service/session/service.go

@@ -14,9 +14,6 @@ import (
 	"github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors"
 	"github.com/kagent-dev/kagent/go/core/internal/utils"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
-	apierrors "k8s.io/apimachinery/pkg/api/errors"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 type Store interface {
@@ -34,15 +31,9 @@ type Store interface {
 	DeleteSessionShare(context.Context, string, string, string) error
 }
 
-type SandboxActorCleaner interface {
-	DeleteSandboxAgentSessionActor(context.Context, *v1alpha3.SandboxAgent, string) (bool, error)
-}
-
 type Service struct {
-	store        Store
-	kube         client.Client
-	actorCleaner SandboxActorCleaner
-	token        func() (string, error)
+	store Store
+	token func() (string, error)
 }
 
 type Option func(*Service)
@@ -80,13 +71,6 @@ func NewService(store Store, options ...Option) *Service {
 	return service
 }
 
-func WithSandboxLifecycle(kube client.Client, cleaner SandboxActorCleaner) Option {
-	return func(service *Service) {
-		service.kube = kube
-		service.actorCleaner = cleaner
-	}
-}
-
 func WithShareTokenGenerator(generator func() (string, error)) Option {
 	return func(service *Service) {
 		if generator != nil {
@@ -155,18 +139,12 @@ func (s *Service) Create(ctx context.Context, request CreateRequest) (*database.
 		return nil, serviceerrors.NewInvalidArgument(fmt.Sprintf("Agent ref is invalid, please check the agent ref %s", request.AgentRef), err)
 	}
 	if agent.WorkloadType == v1alpha3.WorkloadModeSandbox {
-		_, isSubstrateSandbox, err := s.lookupSubstrateSandboxAgent(ctx, request.AgentRef)
+		existing, err := s.store.ListSessionsForAgentAllUsers(ctx, agentID)
 		if err != nil {
-			return nil, serviceerrors.NewInternal("Failed to inspect sandbox agent", err)
+			return nil, serviceerrors.NewInternal("Failed to list sessions for agent", err)
 		}
-		if !isSubstrateSandbox {
-			existing, err := s.store.ListSessionsForAgentAllUsers(ctx, agentID)
-			if err != nil {
-				return nil, serviceerrors.NewInternal("Failed to list sessions for agent", err)
-			}
-			if len(existing) > 0 {
-				return nil, serviceerrors.NewAlreadyExists("Sandbox agents support only one chat session", fmt.Errorf("a session already exists for this agent"))
-			}
+		if len(existing) > 0 {
+			return nil, serviceerrors.NewAlreadyExists("Sandbox agents support only one chat session", fmt.Errorf("a session already exists for this agent"))
 		}
 	}
 
@@ -261,23 +239,9 @@ func (s *Service) Delete(ctx context.Context, sessionID string) error {
 	if s.store == nil {
 		return serviceerrors.NewInternal("Failed to delete session", fmt.Errorf("database client is not configured"))
 	}
-
-	var cleanup *v1alpha3.SandboxAgent
-	if s.actorCleaner != nil {
-		if session, getErr := s.store.GetSession(ctx, sessionID, userID); getErr == nil && session != nil && session.AgentID != nil {
-			if sandboxAgent, lookupErr := s.substrateSandboxAgentForSession(ctx, session); lookupErr == nil {
-				cleanup = sandboxAgent
-			}
-		}
-	}
 	if err := s.store.DeleteSession(ctx, sessionID, userID); err != nil {
 		return serviceerrors.NewInternal("Failed to delete session", err)
 	}
-	if cleanup != nil {
-		if _, err := s.actorCleaner.DeleteSandboxAgentSessionActor(ctx, cleanup, sessionID); err != nil {
-			ctrllog.FromContext(ctx).Error(err, "failed to delete substrate session actor", "sessionID", sessionID)
-		}
-	}
 	return nil
 }
 
@@ -393,47 +357,6 @@ func (s *Service) DeleteShare(ctx context.Context, sessionID, token string) erro
 	return nil
 }
 
-func (s *Service) lookupSubstrateSandboxAgent(ctx context.Context, agentRef string) (*v1alpha3.SandboxAgent, bool, error) {
-	if s.kube == nil {
-		return nil, false, nil
-	}
-	ref := strings.TrimSpace(agentRef)
-	if ref == "" {
-		return nil, false, nil
-	}
-	kubernetesRef := utils.ConvertToKubernetesIdentifier(ref)
-	namespacedName, err := utils.ParseRefString(kubernetesRef, "")
-	if err != nil {
-		return nil, false, nil
-	}
-	sandboxAgent := &v1alpha3.SandboxAgent{}
-	if err := s.kube.Get(ctx, namespacedName, sandboxAgent); err != nil {
-		if apierrors.IsNotFound(err) {
-			return nil, false, nil
-		}
-		return nil, false, err
-	}
-	return sandboxAgent, true, nil
-}
-
-func (s *Service) substrateSandboxAgentForSession(ctx context.Context, session *database.Session) (*v1alpha3.SandboxAgent, error) {
-	if session == nil || session.AgentID == nil {
-		return nil, nil
-	}
-	agent, err := s.store.GetAgent(ctx, *session.AgentID)
-	if err != nil {
-		return nil, err
-	}
-	if agent.WorkloadType != v1alpha3.WorkloadModeSandbox {
-		return nil, nil
-	}
-	sandboxAgent, isSubstrate, err := s.lookupSubstrateSandboxAgent(ctx, *session.AgentID)
-	if err != nil || !isSubstrate {
-		return nil, err
-	}
-	return sandboxAgent, nil
-}
-
 func authenticatedPrincipal(ctx context.Context) (auth.Principal, error) {
 	session, ok := auth.AuthSessionFrom(ctx)
 	if !ok || session == nil {

go/core/internal/service/system/service.go

@@ -12,7 +12,7 @@ import (
 	"github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors"
 	"github.com/kagent-dev/kagent/go/core/internal/version"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
-	"github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
+	"github.com/kagent-dev/kagent/go/core/v2/substrate"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -330,8 +330,8 @@ func (s *Service) listSubstrateCRs(ctx context.Context, namespace string) ([]Sub
 			WorkerSelector:  labelSelectorString(ctx, actorTemplate.Spec.WorkerSelector),
 			ManagedByKagent: actorTemplate.Labels["app.kubernetes.io/managed-by"] == "kagent",
 		}
-		if agentName := substrate.SandboxAgentNameFromLabels(actorTemplate.Labels); agentName != "" {
-			entry.HarnessName = agentName
+		if harnessName := actorTemplate.Labels[substrate.RevisionHarnessLabel]; harnessName != "" {
+			entry.HarnessName = harnessName
 		}
 		actorTemplates = append(actorTemplates, entry)
 	}

go/core/internal/service/system/service_test.go

@@ -11,7 +11,7 @@ import (
 	"github.com/kagent-dev/kagent/go/core/internal/service/serviceerrors"
 	"github.com/kagent-dev/kagent/go/core/internal/service/system"
 	pkgAuth "github.com/kagent-dev/kagent/go/core/pkg/auth"
-	"github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
+	"github.com/kagent-dev/kagent/go/core/v2/substrate"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
@@ -126,7 +126,7 @@ func TestGetSubstrateStatus(t *testing.T) {
 			&atev1alpha1.ActorTemplate{
 				ObjectMeta: metav1.ObjectMeta{Namespace: "team", Name: "template", Labels: map[string]string{
 					"app.kubernetes.io/managed-by": "kagent",
-					substrate.SandboxAgentLabelKey: "agent",
+					substrate.RevisionHarnessLabel: "agent",
 				}},
 				Spec:   atev1alpha1.ActorTemplateSpec{SandboxClass: atev1alpha1.SandboxClassGvisor},
 				Status: atev1alpha1.ActorTemplateStatus{Phase: atev1alpha1.PhaseReady},