feat(substrate): the compiled ActorTemplate carries the revision's egress policy as its default

go/core/internal/service/agentinstance/workflow.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"net/netip"
 	"slices"
 	"strings"
 	"time"
@@ -13,12 +12,10 @@ import (
 	"github.com/google/uuid"
 	apiv1alpha1 "github.com/kagent-dev/kagent/go/api/gen/kagent/api/v1alpha1"
 	"github.com/kagent-dev/kagent/go/core/internal/database"
-	"github.com/kagent-dev/kagent/go/core/internal/egress"
 	"github.com/kagent-dev/kagent/go/core/internal/substrate"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
-	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 type workflowStore interface {
@@ -164,7 +161,7 @@ func (w *ActorWorkflow) run(ctx context.Context, instanceID string, requestedKin
 	atespace, name := revision.ActorTemplateAtespace, substrate.ActorName(instance.Id)
 	var policy *ateapipb.EgressPolicy
 	if kind == apiv1alpha1.AgentInstanceOperation_AGENT_INSTANCE_OPERATION_CREATE {
-		policy, err = actorEgressPolicy(atespace, revision.EgressDestinations, revision.Credentials)
+		policy, err = substrate.ActorEgressPolicy(atespace, revision.EgressDestinations, revision.Credentials)
 		if err != nil {
 			return w.failPreparation(ctx, operation, fmt.Errorf("build Actor %s/%s egress policy: %w", atespace, name, err))
 		}
@@ -304,48 +301,3 @@ func validActorIdentity(actor *ateapipb.Actor, revision *database.RuntimeRevisio
 	return metadata.GetName() == name && metadata.GetAtespace() == revision.ActorTemplateAtespace && metadata.GetUid() != "" &&
 		ref.GetAtespace() == revision.ActorTemplateAtespace && ref.GetName() == revision.ActorTemplateName
 }
-
-// actorEgressPolicy compiles destinations into an actor's default allowlist.
-// Credential bindings are already canonicalized by the store.
-func actorEgressPolicy(atespace string, destinations []string, credentials []egress.Credential) (*ateapipb.EgressPolicy, error) {
-	var hostnames, cidrs []string
-	for _, destination := range destinations {
-		if ip, err := netip.ParseAddr(destination); err == nil && ip.Zone() == "" {
-			ip = ip.Unmap()
-			cidrs = append(cidrs, netip.PrefixFrom(ip, ip.BitLen()).String())
-			continue
-		}
-		hostname := strings.TrimSuffix(strings.ToLower(destination), ".")
-		if len(validation.IsDNS1123Subdomain(hostname)) != 0 {
-			return nil, fmt.Errorf("invalid egress destination %q", destination)
-		}
-		hostnames = append(hostnames, hostname)
-	}
-	policy := &ateapipb.EgressPolicy{Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: "default"}}
-	for _, binding := range credentials {
-		if !slices.Contains(hostnames, binding.Hostname) {
-			return nil, fmt.Errorf("credential destination %q is not allowed", binding.Hostname)
-		}
-		var rule *ateapipb.HostnameRule
-		if len(policy.Rules) > 0 {
-			rule = policy.Rules[len(policy.Rules)-1].GetHostnames()
-		}
-		if rule == nil || rule.Patterns[0] != binding.Hostname {
-			rule = &ateapipb.HostnameRule{Patterns: []string{binding.Hostname}, Effects: &ateapipb.EgressRuleEffects{}}
-			policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: rule})
-		}
-		rule.Effects.InjectStaticHeaders = append(rule.Effects.InjectStaticHeaders, &ateapipb.CredentialHeaderInjection{Header: binding.Header, Prefix: binding.Prefix, CredentialUri: binding.URI})
-	}
-	if len(hostnames) > 0 {
-		slices.Sort(hostnames)
-		policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: &ateapipb.HostnameRule{Patterns: slices.Compact(hostnames)}})
-	}
-	if len(cidrs) > 0 {
-		slices.Sort(cidrs)
-		policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Cidrs: &ateapipb.CIDRRule{Cidrs: slices.Compact(cidrs)}})
-	}
-	if len(policy.Rules) > 256 {
-		return nil, fmt.Errorf("egress policy exceeds 256 rules")
-	}
-	return policy, nil
-}

go/core/internal/service/agentinstance/workflow_test.go

@@ -318,18 +318,18 @@ func TestActorCreationRetainsEgressPolicyFailure(t *testing.T) {
 }
 
 func TestActorEgressPolicy(t *testing.T) {
-	policy, err := actorEgressPolicy("team-a", []string{"API.Example.com.", "api.example.com", "192.0.2.1", "2001:db8::1", "::ffff:192.0.2.1"}, nil)
+	policy, err := substrate.ActorEgressPolicy("team-a", []string{"API.Example.com.", "api.example.com", "192.0.2.1", "2001:db8::1", "::ffff:192.0.2.1"}, nil)
 	require.NoError(t, err)
 	require.Equal(t, &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "default"}, policy.Metadata)
 	require.Len(t, policy.Rules, 2)
 	require.Equal(t, []string{"api.example.com"}, policy.Rules[0].GetHostnames().GetPatterns())
 	require.Equal(t, []string{"192.0.2.1/32", "2001:db8::1/128"}, policy.Rules[1].GetCidrs().GetCidrs())
-	policy, err = actorEgressPolicy("team-a", nil, nil)
+	policy, err = substrate.ActorEgressPolicy("team-a", nil, nil)
 	require.NoError(t, err)
 	require.Empty(t, policy.Rules, "no destinations must deny all egress")
 	for _, destination := range []string{"", "*", "https://api.example.com", "api.example.com:443", "192.0.2.0/24", "fe80::1%eth0"} {
 		t.Run(destination, func(t *testing.T) {
-			_, err := actorEgressPolicy("team-a", []string{destination}, nil)
+			_, err := substrate.ActorEgressPolicy("team-a", []string{destination}, nil)
 			require.Error(t, err)
 		})
 	}
@@ -340,9 +340,9 @@ func TestActorEgressCredentialsRequireAllowedDestination(t *testing.T) {
 		{Hostname: "api.example.com", Header: "authorization", Prefix: "Bearer ", URI: "ate-secret://kubernetes.io/team/auth/token"},
 		{Hostname: "api.example.com", Header: "x-api-key", URI: "ate-secret://kubernetes.io/team/auth/key"},
 	}
-	_, err := actorEgressPolicy("team", []string{"other.example.com"}, bindings)
+	_, err := substrate.ActorEgressPolicy("team", []string{"other.example.com"}, bindings)
 	require.ErrorContains(t, err, "is not allowed")
-	policy, err := actorEgressPolicy("team", []string{"api.example.com", "other.example.com"}, bindings)
+	policy, err := substrate.ActorEgressPolicy("team", []string{"api.example.com", "other.example.com"}, bindings)
 	require.NoError(t, err)
 	require.Len(t, policy.Rules, 2)
 	require.Equal(t, []string{"api.example.com"}, policy.Rules[0].GetHostnames().GetPatterns())

go/core/internal/substrate/actor_template.go

@@ -75,6 +75,13 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R
 	if len(actorEnv) > 32 {
 		return nil, fmt.Errorf("runtime revision has %d environment variables; Substrate supports at most 32", len(actorEnv))
 	}
+	// The template carries the revision's allowlist so the golden actor, which
+	// Substrate creates without any client call, is admitted to the same
+	// destinations as the instances that boot from its snapshot.
+	egressPolicy, err := ActorEgressPolicy(spec.Namespace, spec.EgressDestinations, spec.Credentials)
+	if err != nil {
+		return nil, fmt.Errorf("compile egress policy: %w", err)
+	}
 
 	template := &ateapipb.ActorTemplate{
 		Metadata: &ateapipb.ResourceMetadata{Atespace: spec.Namespace, Name: name},
@@ -95,7 +102,8 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R
 			}, TimeoutSeconds: 30},
 			VolumeMounts: []*ateapipb.VolumeMount{{Name: durableDataVolume, MountPath: durableDataMount}, {Name: egressTrustVolume, MountPath: egressTrustMount}},
 		}},
-		WorkerSelector: workerSelectorForPool(workerKey),
+		WorkerSelector:      workerSelectorForPool(workerKey),
+		DefaultEgressPolicy: &ateapipb.EgressPolicyTemplate{Rules: egressPolicy.GetRules()},
 		SnapshotsConfig: &ateapipb.SnapshotsConfig{
 			StorageLocation: spec.SnapshotLocation,
 			OnPause:         ateapipb.SnapshotContentScope_SNAPSHOT_CONTENT_SCOPE_FULL,
@@ -127,12 +135,13 @@ func actorTemplateSpec(template *ateapipb.ActorTemplate) *ateapipb.ActorTemplate
 			Atespace: template.GetMetadata().GetAtespace(),
 			Name:     template.GetMetadata().GetName(),
 		},
-		WorkerSelector:  template.GetWorkerSelector(),
-		Containers:      template.GetContainers(),
-		Volumes:         template.GetVolumes(),
-		SnapshotsConfig: template.GetSnapshotsConfig(),
-		SandboxConfig:   template.GetSandboxConfig(),
-		Resources:       template.GetResources(),
+		WorkerSelector:      template.GetWorkerSelector(),
+		Containers:          template.GetContainers(),
+		Volumes:             template.GetVolumes(),
+		SnapshotsConfig:     template.GetSnapshotsConfig(),
+		SandboxConfig:       template.GetSandboxConfig(),
+		Resources:           template.GetResources(),
+		DefaultEgressPolicy: template.GetDefaultEgressPolicy(),
 	}
 }
 

go/core/internal/substrate/actor_template_test.go

@@ -9,6 +9,7 @@ import (
 	a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1"
 
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"github.com/kagent-dev/kagent/go/core/internal/egress"
 	"github.com/kagent-dev/kagent/go/core/internal/translator"
 	"google.golang.org/protobuf/proto"
 	corev1 "k8s.io/api/core/v1"
@@ -23,7 +24,9 @@ func TestActorTemplateForRevision(t *testing.T) {
 		WorkerPoolName: "default", SnapshotLocation: "snapshots",
 		ConfigJSON: []byte(`{"instruction":"help"}`), AgentCard: &a2apb.AgentCard{Name: "helper", Version: "v1", Capabilities: &a2apb.AgentCapabilities{Streaming: new(true)},
 			SupportedInterfaces: []*a2apb.AgentInterface{{Url: "http://127.0.0.1:80", ProtocolBinding: "GRPC", ProtocolVersion: "1.0"}}, DefaultInputModes: []string{"text"}, DefaultOutputModes: []string{"text"}},
-		Environment: []corev1.EnvVar{{Name: "API_KEY", Value: translator.CredentialPlaceholder}},
+		Environment:        []corev1.EnvVar{{Name: "API_KEY", Value: translator.CredentialPlaceholder}},
+		EgressDestinations: []string{"github.com", "api.anthropic.com"},
+		Credentials:        []egress.Credential{{Hostname: "api.anthropic.com", Header: "x-api-key", URI: "ate-secret://kubernetes.io/agents/anthropic/key"}},
 	}
 	revisionID, err := spec.Digest()
 	if err != nil {
@@ -72,6 +75,42 @@ func TestActorTemplateForRevision(t *testing.T) {
 	if environment["KAGENT_CONFIG_JSON"].Value != string(spec.ConfigJSON) {
 		t.Fatal("config was not embedded as a non-secret literal")
 	}
+	wantPolicy, err := ActorEgressPolicy("agents", spec.EgressDestinations, spec.Credentials)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !proto.Equal(&ateapipb.EgressPolicy{Rules: template.GetDefaultEgressPolicy().GetRules()}, &ateapipb.EgressPolicy{Rules: wantPolicy.GetRules()}) {
+		t.Fatalf("template default egress policy = %v, want %v", template.GetDefaultEgressPolicy(), wantPolicy)
+	}
+	if injected := template.GetDefaultEgressPolicy().GetRules()[0].GetHostnames().GetEffects().GetInjectStaticHeaders(); len(injected) != 1 || injected[0].GetHeader() != "x-api-key" {
+		t.Fatalf("credential injection missing from the template policy: %v", template.GetDefaultEgressPolicy())
+	}
+}
+
+func TestActorTemplateForRevisionDeniesAllEgressWithoutDestinations(t *testing.T) {
+	spec := &translator.Revision{
+		Namespace: "agents", AgentTemplateName: "helper", HarnessName: "kagent",
+		Image:          "agent.example/image@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+		WorkerPoolName: "default", SnapshotLocation: "snapshots",
+		ConfigJSON: []byte(`{}`), 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"}},
+	}
+	revisionID, err := spec.Digest()
+	if err != nil {
+		t.Fatal(err)
+	}
+	template, err := ActorTemplateForRevision(spec, revisionID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if template.GetDefaultEgressPolicy() == nil || len(template.GetDefaultEgressPolicy().GetRules()) != 0 {
+		t.Fatalf("a revision without destinations must carry an empty (deny-all) policy, got %v", template.GetDefaultEgressPolicy())
+	}
+	invalid := *spec
+	invalid.EgressDestinations = []string{"https://github.com"}
+	if _, err := ActorTemplateForRevision(&invalid, revisionID); err == nil {
+		t.Fatal("an invalid destination must fail template compilation")
+	}
 }
 
 func TestActorTemplateStampsTheRevisionOnTheResource(t *testing.T) {
@@ -120,4 +159,9 @@ func TestActorTemplateSpecEqualIgnoresServerFields(t *testing.T) {
 	if ActorTemplateSpecEqual(left, right) {
 		t.Fatal("different container image was accepted")
 	}
+	right = proto.CloneOf(left)
+	right.DefaultEgressPolicy = &ateapipb.EgressPolicyTemplate{Rules: []*ateapipb.EgressRule{{Hostnames: &ateapipb.HostnameRule{Patterns: []string{"github.com"}}}}}
+	if ActorTemplateSpecEqual(left, right) {
+		t.Fatal("different default egress policy was accepted")
+	}
 }

go/core/internal/substrate/egress.go

@@ -3,11 +3,16 @@ package substrate
 import (
 	"context"
 	"fmt"
+	"net/netip"
+	"slices"
+	"strings"
 
 	"github.com/agent-substrate/substrate/pkg/proto/ateapipb"
+	"github.com/kagent-dev/kagent/go/core/internal/egress"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
+	"k8s.io/apimachinery/pkg/util/validation"
 )
 
 // EnsureActorEgressPolicy can be retried after a lost create response. A prepared
@@ -32,3 +37,48 @@ func (c *Client) EnsureActorEgressPolicy(ctx context.Context, atespace, name str
 	}
 	return nil
 }
+
+// ActorEgressPolicy compiles destinations into an actor's default allowlist.
+// Credential bindings are already canonicalized by the store.
+func ActorEgressPolicy(atespace string, destinations []string, credentials []egress.Credential) (*ateapipb.EgressPolicy, error) {
+	var hostnames, cidrs []string
+	for _, destination := range destinations {
+		if ip, err := netip.ParseAddr(destination); err == nil && ip.Zone() == "" {
+			ip = ip.Unmap()
+			cidrs = append(cidrs, netip.PrefixFrom(ip, ip.BitLen()).String())
+			continue
+		}
+		hostname := strings.TrimSuffix(strings.ToLower(destination), ".")
+		if len(validation.IsDNS1123Subdomain(hostname)) != 0 {
+			return nil, fmt.Errorf("invalid egress destination %q", destination)
+		}
+		hostnames = append(hostnames, hostname)
+	}
+	policy := &ateapipb.EgressPolicy{Metadata: &ateapipb.ResourceMetadata{Atespace: atespace, Name: "default"}}
+	for _, binding := range credentials {
+		if !slices.Contains(hostnames, binding.Hostname) {
+			return nil, fmt.Errorf("credential destination %q is not allowed", binding.Hostname)
+		}
+		var rule *ateapipb.HostnameRule
+		if len(policy.Rules) > 0 {
+			rule = policy.Rules[len(policy.Rules)-1].GetHostnames()
+		}
+		if rule == nil || rule.Patterns[0] != binding.Hostname {
+			rule = &ateapipb.HostnameRule{Patterns: []string{binding.Hostname}, Effects: &ateapipb.EgressRuleEffects{}}
+			policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: rule})
+		}
+		rule.Effects.InjectStaticHeaders = append(rule.Effects.InjectStaticHeaders, &ateapipb.CredentialHeaderInjection{Header: binding.Header, Prefix: binding.Prefix, CredentialUri: binding.URI})
+	}
+	if len(hostnames) > 0 {
+		slices.Sort(hostnames)
+		policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Hostnames: &ateapipb.HostnameRule{Patterns: slices.Compact(hostnames)}})
+	}
+	if len(cidrs) > 0 {
+		slices.Sort(cidrs)
+		policy.Rules = append(policy.Rules, &ateapipb.EgressRule{Cidrs: &ateapipb.CIDRRule{Cidrs: slices.Compact(cidrs)}})
+	}
+	if len(policy.Rules) > 256 {
+		return nil, fmt.Errorf("egress policy exceeds 256 rules")
+	}
+	return policy, nil
+}

go/core/internal/translator/revision.go

@@ -65,6 +65,11 @@ type Revision struct {
 	EgressDestinations []string
 }
 
+// actorTemplateFormat is bumped when ActorTemplateForRevision changes what it
+// renders from unchanged revision inputs. 2 = default_egress_policy on the
+// template.
+const actorTemplateFormat = 2
+
 // Digest returns the immutable identity of every input that affects runtime
 // behavior. The full digest is the database key; Kubernetes names use a short
 // prefix only for readability.
@@ -83,11 +88,16 @@ func (r *Revision) Digest() (RevisionID, error) {
 		Provenance         json.RawMessage     `json:"provenance"`
 		Credentials        []egress.Credential `json:"credentials,omitempty"`
 		EgressDestinations []string            `json:"egressDestinations"`
+		// ActorTemplateFormat changes when the ActorTemplate compiled from the
+		// same inputs changes shape, so the immutable template is re-created
+		// under a new name instead of conflicting with the stored one.
+		ActorTemplateFormat int `json:"actorTemplateFormat"`
 	}{
 		Namespace: r.Namespace, AgentTemplateName: r.AgentTemplateName, HarnessName: r.HarnessName,
 		Image: r.Image, Command: r.Command, Args: r.Args, Environment: r.Environment, ConfigJSON: r.ConfigJSON,
 		WorkerPoolName: r.WorkerPoolName, SnapshotLocation: r.SnapshotLocation, Provenance: r.Provenance,
 		Credentials: r.Credentials, EgressDestinations: r.EgressDestinations,
+		ActorTemplateFormat: actorTemplateFormat,
 	})
 	if err != nil {
 		return RevisionID{}, fmt.Errorf("marshal runtime revision inputs: %w", err)