fix(substrate): let the golden actor fetch git and S3 skill sources

docs/architecture/configuration-and-compilation.md

@@ -216,9 +216,14 @@ millicores and memory bytes for every harness, rounding positive fractions up
 to the next unit. Standalone sandbox limits remain under `sandboxes` configuration.
 
 Harness and SandboxTemplate snapshot locations require Substrate's
-`PerTemplateSnapshotLocation` preview gate. The CLI installer enables it; direct
-Substrate installations must include it in their Helm `preview` list. The
+`PerTemplateSnapshotLocation` preview gate, and the golden actor's egress policy
+requires `GoldenEgressPolicy`. The CLI installer enables both; direct Substrate
+installations must include them in their Helm `preview` list. The
 Substrate client opts ActorTemplate creation requests into preview features.
+Without `GoldenEgressPolicy`, Substrate rejects ActorTemplates for Agents with
+git or S3 skill sources. The Agent reports `ActorTemplateRejected` with a message
+that names the gate. The controller retries with backoff, so the Agent becomes
+Ready after the gate is turned on.
 
 A missing WorkerPool reports `WorkerPoolNotFound`; an unsupported class reports
 `RevisionInvalid`. Neither produces a desired ActorTemplate. WorkerPool updates
@@ -308,9 +313,12 @@ download cache or fall back to an in-runtime OCI pull.
 
 Package limits and path/symlink validation still apply. Build skill images from
 scratch with just package contents; arbitrary ORAS artifact layouts are not
-supported. Existing Git and S3 fetches remain runtime operations and retain
-their golden-runtime networking limitations. The Python ADK does not yet consume
-the `agent_plugins` runtime contract. BYO images must consume that contract to
+supported. The runtime fetches git and S3 sources when the golden actor starts.
+The ActorTemplate's golden egress policy allows only the origins (scheme, host,
+and port) of those sources, for the root agent and its shared agents. Sessions
+restore the fetched sources from the golden snapshot and reach those origins
+through their own egress policy. The Python ADK does not yet consume the
+`agent_plugins` runtime contract. BYO images must consume that contract to
 load these skills; the Go ADK BYO fixture does so.
 
 ### Runtime configuration

go/core/internal/controller/preparation_test.go

@@ -22,6 +22,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 
 	kagentfake "github.com/kagent-dev/kagent/go/api/clientset/versioned/fake"
 	apiv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1"
@@ -139,7 +140,7 @@ func TestUnresolvedPoolReleasesAbandonedRevision(t *testing.T) {
 }
 
 func TestPreparationErrorsPublishStatusAndRecover(t *testing.T) {
-	for _, test := range []struct {
+	tests := []struct {
 		name      string
 		templates fakeActorTemplates
 		store     fakeRuntimeRevisionStore
@@ -151,7 +152,9 @@ func TestPreparationErrorsPublishStatusAndRecover(t *testing.T) {
 		{name: "atespace failed", templates: fakeActorTemplates{ensureErr: status.Error(codes.PermissionDenied, "private-credential")}, code: codes.PermissionDenied},
 		{name: "pair store failed", store: fakeRuntimeRevisionStore{pairErr: errors.New("private-database-connection")}, code: codes.Unknown},
 		{name: "revision store failed", store: fakeRuntimeRevisionStore{revisionErr: errors.New("private-database-connection")}, code: codes.Unknown},
-	} {
+	}
+	for i := range tests {
+		test := &tests[i]
 		t.Run(test.name, func(t *testing.T) {
 			collections, _ := newPreparationTestCollections(t, "microvm")
 			initial := collections.Reconciliations.List()[0]
@@ -238,14 +241,16 @@ func TestPreparationRevisionIsolation(t *testing.T) {
 }
 
 func TestPreparationQueueAndPollerBackoff(t *testing.T) {
-	for _, test := range []struct {
+	tests := []struct {
 		name      string
 		templates fakeActorTemplates
 		store     fakeRuntimeRevisionStore
 	}{
 		{name: "revision being collected", store: fakeRuntimeRevisionStore{pairErr: database.ErrObjectDeleting}},
 		{name: "persistence unavailable", store: fakeRuntimeRevisionStore{revisionErr: errors.New("database unavailable")}},
-	} {
+	}
+	for i := range tests {
+		test := &tests[i]
 		t.Run(test.name, func(t *testing.T) {
 			synctest.Test(t, func(t *testing.T) {
 				collections, _ := newPreparationTestCollections(t, "microvm")
@@ -486,6 +491,84 @@ func TestPreparationTerminalFailures(t *testing.T) {
 	}
 }
 
+// Substrate without the GoldenEgressPolicy preview gate rejects a template
+// with a golden egress policy. Enabling the gate must heal the Agent.
+func TestPreparationRetriesGoldenEgressRejection(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		collections, _ := newPreparationTestCollectionsFor(t, "microvm", gitSkillTemplate())
+		initial := collections.Reconciliations.List()[0]
+		gateOff := field.ErrorList{field.Forbidden(field.NewPath("actor_template", "golden_egress_policy"), "")}.ToAggregate()
+		templates := &fakeActorTemplates{createErr: status.Error(codes.InvalidArgument, gateOff.Error())}
+		store := &fakeRuntimeRevisionStore{}
+		statusClient := kagentfake.NewSimpleClientset(initial.Agent.DeepCopy()).ApiV1alpha3()
+		r := newReconciler(collections, templates, store, statusClient)
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		go func() { assert.NoError(t, r.Start(ctx)) }()
+		synctest.Wait()
+
+		failure := collections.Reconciliations.GetKey(initial.ResourceName()).PreparationFailure
+		require.NotNil(t, failure)
+		require.True(t, failure.Retryable, "enabling the gate must not need a controller restart")
+		published, err := statusClient.Agents(initial.Agent.Namespace).Get(t.Context(), initial.Agent.Name, metav1.GetOptions{})
+		require.NoError(t, err)
+		ready := apimeta.FindStatusCondition(published.Status.Conditions, kagentv1alpha3.AgentConditionReady)
+		require.Equal(t, "ActorTemplateRejected", ready.Reason)
+		require.Contains(t, ready.Message, "GoldenEgressPolicy preview gate")
+		require.NotContains(t, ready.Message, "golden_egress_policy")
+
+		time.Sleep(5 * time.Minute)
+		synctest.Wait()
+		require.Greater(t, store.upserts(), 2, "the queue retries the rejected template")
+		require.Less(t, store.upserts(), 30, "retries back off")
+
+		templates.setCreateErr(nil)
+		time.Sleep(time.Minute)
+		synctest.Wait()
+		require.NotNil(t, templates.created(), "the template is created once Substrate accepts it")
+		require.Nil(t, collections.Reconciliations.GetKey(initial.ResourceName()).PreparationFailure)
+		cancel()
+		synctest.Wait()
+	})
+}
+
+// Other validation errors on a template with a golden egress policy stay
+// terminal: enabling the gate cannot repair them.
+func TestPreparationGoldenEgressOtherRejectionIsTerminal(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		collections, _ := newPreparationTestCollectionsFor(t, "microvm", gitSkillTemplate())
+		initial := collections.Reconciliations.List()[0]
+		require.NotNil(t, initial.Target.ActorTemplate.GetGoldenEgressPolicy())
+		templates := &fakeActorTemplates{createErr: status.Error(codes.InvalidArgument, "actor_template.containers[0].env: Too many")}
+		store := &fakeRuntimeRevisionStore{}
+		statusClient := kagentfake.NewSimpleClientset(initial.Agent.DeepCopy()).ApiV1alpha3()
+		r := newReconciler(collections, templates, store, statusClient)
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		go func() { assert.NoError(t, r.Start(ctx)) }()
+		synctest.Wait()
+
+		failure := collections.Reconciliations.GetKey(initial.ResourceName()).PreparationFailure
+		require.NotNil(t, failure)
+		require.False(t, failure.Retryable)
+		require.Equal(t, "ActorTemplateRejected", failure.Reason)
+		require.NotContains(t, failure.Message, "GoldenEgressPolicy")
+		time.Sleep(5 * time.Minute)
+		synctest.Wait()
+		require.Equal(t, 1, store.upserts(), "a terminal rejection is not retried")
+		cancel()
+		synctest.Wait()
+	})
+}
+
+func gitSkillTemplate() *kagentv1alpha3.AgentTemplate {
+	return &kagentv1alpha3.AgentTemplate{ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "assistant", UID: "template-uid"}, Spec: kagentv1alpha3.AgentTemplateSpec{
+		Skills: []kagentv1alpha3.AgentTemplateSkill{{Name: "review", Source: kagentv1alpha3.ArtifactSource{Git: &kagentv1alpha3.GitArtifact{
+			URL: "https://git.example.com/skills.git", Commit: strings.Repeat("a", 40),
+		}}}},
+	}}
+}
+
 func TestInvalidActorTemplateHasNoCompiledTarget(t *testing.T) {
 	collections, harnesses := newPreparationTestCollections(t, "microvm")
 	initial := collections.Reconciliations.List()[0]

go/core/internal/controller/reconciler.go

@@ -19,6 +19,7 @@ import (
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/util/workqueue"
 
@@ -404,6 +405,16 @@ func (r *Reconciler) reconcileAgent(ctx context.Context, key string) error {
 			return r.observePreparationError(*state, fmt.Errorf("ensure Atespace %s: %w", desiredRef.GetAtespace(), err))
 		}
 		observed, err = r.templates.CreateActorTemplate(ctx, target.ActorTemplate)
+		if status.Code(err) == codes.InvalidArgument && strings.Contains(status.Convert(err).Message(), goldenEgressGateOff) {
+			// The operator can enable the preview gate, so keep retrying with backoff.
+			r.observePreparation(*state, goldenProgress{}, &ReconciliationFailure{
+				Condition: kagentv1alpha3.AgentConditionReady,
+				Reason:    "ActorTemplateRejected",
+				Message:   "Substrate rejected the ActorTemplate. Git and S3 skill sources need Substrate's GoldenEgressPolicy preview gate. Add GoldenEgressPolicy to the preview list in Substrate's Helm values.",
+				Retryable: true,
+			})
+			return fmt.Errorf("create ActorTemplate %s/%s: %w", desiredRef.GetAtespace(), desiredRef.GetName(), err)
+		}
 		if status.Code(err) == codes.InvalidArgument {
 			// Validation rejects the immutable request, so retries cannot repair
 			// it. Backend messages may quote configuration; keep status safe.
@@ -461,6 +472,10 @@ func (r *Reconciler) reconcileAgent(ctx context.Context, key string) error {
 	return nil
 }
 
+// goldenEgressGateOff is how Substrate reports a golden egress policy while its
+// GoldenEgressPolicy preview gate is off. Other validation errors are terminal.
+var goldenEgressGateOff = field.Forbidden(field.NewPath("actor_template", "golden_egress_policy"), "").Error()
+
 func (r *Reconciler) observePreparationError(state AgentReconciliation, err error) error {
 	if state.Target == nil {
 		return err

go/core/internal/controller/reconciler_test.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"context"
 	"errors"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"testing/synctest"
@@ -282,6 +283,9 @@ func (s *failingFinalizationStore) DeleteRuntimeRevision(ctx context.Context, re
 
 type fakeActorTemplates struct {
 	actorEgressStub
+	// mu guards template and createErr for tests that change them while the
+	// reconciler's queue runs.
+	mu                 sync.Mutex
 	template           *ateapipb.ActorTemplate
 	ensureErr          error
 	getErr             error
@@ -295,6 +299,8 @@ type fakeActorTemplates struct {
 func (f *fakeActorTemplates) EnsureAtespace(context.Context, string) error { return f.ensureErr }
 
 func (f *fakeActorTemplates) GetActorTemplate(context.Context, string, string) (*ateapipb.ActorTemplate, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	if f.getErr != nil {
 		return nil, f.getErr
 	}
@@ -305,6 +311,8 @@ func (f *fakeActorTemplates) GetActorTemplate(context.Context, string, string) (
 }
 
 func (f *fakeActorTemplates) CreateActorTemplate(_ context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	if f.createErr != nil {
 		return nil, f.createErr
 	}
@@ -316,6 +324,18 @@ func (f *fakeActorTemplates) CreateActorTemplate(_ context.Context, template *at
 	return f.template, nil
 }
 
+func (f *fakeActorTemplates) setCreateErr(err error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.createErr = err
+}
+
+func (f *fakeActorTemplates) created() *ateapipb.ActorTemplate {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.template
+}
+
 func (f *fakeActorTemplates) DeleteActorTemplate(_ context.Context, _, _, uid string) error {
 	f.deletedUIDs = append(f.deletedUIDs, uid)
 	if f.deleteErr != nil {
@@ -332,6 +352,9 @@ func (f *fakeActorTemplates) DeleteActorTemplate(_ context.Context, _, _, uid st
 
 type fakeRuntimeRevisionStore struct {
 	egressTargetStoreStub
+	// mu guards pair and pairCalls for tests that read them while the
+	// reconciler's queue runs.
+	mu                sync.Mutex
 	pair              *database.AgentDefinition
 	revision          *database.RuntimeRevision
 	markedSuccessful  bool
@@ -374,11 +397,19 @@ func (s *fakeRuntimeRevisionStore) ListAgentEgressSources(context.Context) ([]da
 }
 
 func (s *fakeRuntimeRevisionStore) UpsertAgentDefinition(_ context.Context, pair database.AgentDefinition) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	s.pairCalls++
 	s.pair = &pair
 	return s.pairErr
 }
 
+func (s *fakeRuntimeRevisionStore) upserts() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.pairCalls
+}
+
 func (s *fakeRuntimeRevisionStore) RecordRuntimeRevision(_ context.Context, revision database.RuntimeRevision, options database.RecordRevisionOptions) (database.GoldenRetry, error) {
 	s.recordCalls++
 	if s.revisionErr != nil {

go/core/internal/substrate/actor_template.go

@@ -91,6 +91,10 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R
 	if err != nil {
 		return nil, err
 	}
+	goldenEgress, err := GoldenEgressPolicy(spec.GoldenEgressDestinations)
+	if err != nil {
+		return nil, fmt.Errorf("compile golden egress policy: %w", err)
+	}
 
 	template := &ateapipb.ActorTemplate{
 		Metadata:  &ateapipb.ResourceMetadata{Atespace: spec.Namespace, Name: name},
@@ -118,6 +122,7 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R
 			// Pin the server default so immutable comparison survives creation.
 			GoldenSnapshotConfig: &ateapipb.GoldenSnapshotConfig{Mode: ateapipb.GoldenSnapshotMode_GOLDEN_SNAPSHOT_MODE_ENABLED},
 		},
+		GoldenEgressPolicy: goldenEgress,
 		Volumes: []*ateapipb.Volume{
 			{Name: durableDataVolume, DurableDir: &ateapipb.DurableDirVolumeSource{}},
 			// Substrate regenerates this projection on Run and Restore. A fork
@@ -173,11 +178,12 @@ func actorTemplateSpec(template *ateapipb.ActorTemplate) *ateapipb.ActorTemplate
 			Atespace: template.GetMetadata().GetAtespace(),
 			Name:     template.GetMetadata().GetName(),
 		},
-		WorkerSelector: template.GetWorkerSelector(),
-		Containers:     template.GetContainers(),
-		Volumes:        template.GetVolumes(),
-		SnapshotConfig: template.GetSnapshotConfig(),
-		Resources:      template.GetResources(),
+		WorkerSelector:     template.GetWorkerSelector(),
+		Containers:         template.GetContainers(),
+		Volumes:            template.GetVolumes(),
+		SnapshotConfig:     template.GetSnapshotConfig(),
+		Resources:          template.GetResources(),
+		GoldenEgressPolicy: template.GetGoldenEgressPolicy(),
 	}
 }
 

go/core/internal/substrate/actor_template_test.go

@@ -181,6 +181,58 @@ func mustSpecDigest(t *testing.T, template *ateapipb.ActorTemplate) string {
 	return digest
 }
 
+// The golden actor fetches git and S3 skill sources before its snapshot.
+// Its policy must allow exactly those origins and nothing else.
+func TestActorTemplateGoldenEgressPolicy(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		destinations []string
+		want         *ateapipb.EgressPolicyTemplate
+	}{
+		{name: "OCI only", want: nil},
+		{
+			name:         "git and S3 origins",
+			destinations: []string{"http://git.internal.example:8080", "https://git.example.com:443", "https://objects.example.com:9000"},
+			want: &ateapipb.EgressPolicyTemplate{Rules: []*ateapipb.EgressRule{
+				{Http: &ateapipb.HttpRule{Hostnames: []string{"git.internal.example"}, Ports: &ateapipb.Ports{Numbers: []int32{8080}}}},
+				{Https: &ateapipb.HttpsRule{Hostnames: []string{"git.example.com"}, Ports: &ateapipb.Ports{Numbers: []int32{443}}}},
+				{Https: &ateapipb.HttpsRule{Hostnames: []string{"objects.example.com"}, Ports: &ateapipb.Ports{Numbers: []int32{9000}}}},
+			}},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			spec := &translator.Revision{CPU: "1", Memory: "1Gi", Namespace: "agents", AgentName: "helper", WorkerPoolName: "default",
+				AgentCard: &a2apb.AgentCard{Name: "helper", Version: "v1", Capabilities: &a2apb.AgentCapabilities{},
+					SupportedInterfaces: []*a2apb.AgentInterface{{Url: "http://127.0.0.1:80", ProtocolBinding: "GRPC", ProtocolVersion: "1.0"}},
+					DefaultInputModes:   []string{"text"}, DefaultOutputModes: []string{"text"}},
+				EgressDestinations:       []string{"http://kagent-controller.kagent:8083", "https://api.openai.com:443"},
+				GoldenEgressDestinations: test.destinations,
+			}
+			revisionID, err := spec.Digest()
+			require.NoError(t, err)
+			template, err := ActorTemplateForRevision(spec, revisionID)
+			require.NoError(t, err)
+			require.True(t, proto.Equal(test.want, template.GetGoldenEgressPolicy()), "golden egress policy = %v", template.GetGoldenEgressPolicy())
+		})
+	}
+}
+
+func TestActorTemplateSpecEqualComparesGoldenEgressPolicy(t *testing.T) {
+	left := &ateapipb.ActorTemplate{
+		Metadata:   &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "template"},
+		Containers: []*ateapipb.Container{{Name: "agent", Image: "agent:v1"}},
+		GoldenEgressPolicy: &ateapipb.EgressPolicyTemplate{Rules: []*ateapipb.EgressRule{
+			{Https: &ateapipb.HttpsRule{Hostnames: []string{"git.example.com"}, Ports: &ateapipb.Ports{Numbers: []int32{443}}}},
+		}},
+	}
+	right := proto.CloneOf(left)
+	require.True(t, ActorTemplateSpecEqual(left, right))
+	right.GoldenEgressPolicy.Rules[0].Https.Hostnames = []string{"objects.example.com"}
+	require.False(t, ActorTemplateSpecEqual(left, right), "a changed golden egress policy was accepted")
+	right.GoldenEgressPolicy = nil
+	require.False(t, ActorTemplateSpecEqual(left, right), "a removed golden egress policy was accepted")
+}
+
 func TestActorTemplateEnvironmentValueSize(t *testing.T) {
 	const configOverhead = len(`{"instruction":""}`)
 	for _, test := range []struct {

go/core/internal/substrate/client.go

@@ -149,8 +149,9 @@ func (c *Client) GetActorTemplate(ctx context.Context, atespace, name string) (*
 func (c *Client) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error) {
 	ctx, cancel := c.callCtx(ctx)
 	defer cancel()
-	// Substrate requires request opt-in as well as the server's
-	// PerTemplateSnapshotLocation gate to honor the configured snapshot location.
+	// Substrate requires request opt-in as well as the server's preview gates:
+	// PerTemplateSnapshotLocation for the snapshot location and
+	// GoldenEgressPolicy for the golden actor's egress policy.
 	ctx = metadata.AppendToOutgoingContext(ctx, "ate-preview-allowed", "true")
 	return c.ControlClient.CreateActorTemplate(ctx, &ateapipb.CreateActorTemplateRequest{ActorTemplate: template})
 }