From c80d2483b1b102860b673c50065a2e677e2e1572 Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Fri, 25 Sep 2026 09:56:22 +0200 Subject: [PATCH 1/2] fix(substrate): reject a runtime environment value Substrate would refuse A compiled revision whose KAGENT_CONFIG_JSON or KAGENT_AGENT_CARD_JSON is longer than Substrate's 32768-character env value limit was only rejected by CreateActorTemplate. ActorTemplateForRevision now counts each value in code points, as Substrate does, and fails with the variable, the limit and the actual size, so the pair reports Compatible=False (ActorTemplateInvalid). Signed-off-by: QuentinBisson --- go/core/internal/substrate/actor_template.go | 37 +++++++++++- .../internal/substrate/actor_template_test.go | 60 +++++++++++++++++++ 2 files changed, 95 insertions(+), 2 deletions(-) diff --git a/go/core/internal/substrate/actor_template.go b/go/core/internal/substrate/actor_template.go index f404069a1c..0514cd1f79 100644 --- a/go/core/internal/substrate/actor_template.go +++ b/go/core/internal/substrate/actor_template.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "strings" + "unicode/utf8" apia2a "github.com/kagent-dev/kagent/go/api/a2a" @@ -25,6 +26,13 @@ const ( durableDataMount = "/data" ) +const ( + // Substrate's ActorTemplate validation caps a container at 32 environment + // variables and each value at 32768 characters, counted in code points. + maxEnvironmentVariables = 32 + maxEnvironmentValueRunes = 32768 +) + const egressTrustVolume = "egress-trust" const egressTrustMount = "/run/kagent/egress" @@ -76,8 +84,11 @@ func ActorTemplateForRevision(spec *translator.Revision, revisionID translator.R if err != nil { return nil, err } - if len(actorEnv) > 32 { - return nil, fmt.Errorf("runtime revision has %d environment variables; Substrate supports at most 32", len(actorEnv)) + if len(actorEnv) > maxEnvironmentVariables { + return nil, fmt.Errorf("runtime revision has %d environment variables; Substrate supports at most %d", len(actorEnv), maxEnvironmentVariables) + } + if err := checkEnvironmentValueSizes(actorEnv); err != nil { + return nil, err } sandboxConfig, err := sandboxConfigForClass(spec.SandboxClass) if err != nil { @@ -208,6 +219,28 @@ func actorTemplateEnvFromPodEnv(environment []corev1.EnvVar) ([]*ateapipb.EnvVar return result, nil } +// checkEnvironmentValueSizes rejects a value Substrate would refuse, so the +// failure reaches the pair's status instead of every CreateActorTemplate call. +func checkEnvironmentValueSizes(environment []*ateapipb.EnvVar) error { + for _, variable := range environment { + size := utf8.RuneCountInString(variable.GetValue()) + if size <= maxEnvironmentValueRunes { + continue + } + switch variable.GetName() { + case "KAGENT_CONFIG_JSON": + return fmt.Errorf("compiled agent config is %d characters after JSON encoding; Substrate accepts at most %d in environment variable %s. Move long reference material, such as a long system prompt, into a skill", + size, maxEnvironmentValueRunes, variable.GetName()) + case "KAGENT_AGENT_CARD_JSON": + return fmt.Errorf("agent card is %d characters after JSON encoding; Substrate accepts at most %d in environment variable %s. Shorten the agent description", + size, maxEnvironmentValueRunes, variable.GetName()) + default: + return fmt.Errorf("environment variable %s is %d characters; Substrate accepts at most %d", variable.GetName(), size, maxEnvironmentValueRunes) + } + } + return nil +} + func sandboxConfigForClass(class atev1alpha1.SandboxClass) (*ateapipb.SandboxConfig, error) { switch class { case "", atev1alpha1.SandboxClassGvisor: diff --git a/go/core/internal/substrate/actor_template_test.go b/go/core/internal/substrate/actor_template_test.go index 856024639a..48be4940e8 100644 --- a/go/core/internal/substrate/actor_template_test.go +++ b/go/core/internal/substrate/actor_template_test.go @@ -2,8 +2,11 @@ package substrate import ( "encoding/json" + "fmt" "slices" + "strings" "testing" + "unicode/utf8" a2atype "github.com/a2aproject/a2a-go/v2/a2a" a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" @@ -167,3 +170,60 @@ func TestActorTemplateSpecEqualIgnoresServerFields(t *testing.T) { t.Fatal("different container image was accepted") } } + +func TestActorTemplateEnvironmentValueSize(t *testing.T) { + // configWithInstruction renders the instruction the way the compilers do, + // with json.Marshal's HTML escaping, and returns the rendered config. + configWithInstruction := func(t *testing.T, instruction string) []byte { + t.Helper() + config, err := json.Marshal(map[string]string{"instruction": instruction}) + require.NoError(t, err) + return config + } + // overhead is what the config costs around an ASCII instruction. + overhead := utf8.RuneCount(configWithInstruction(t, "")) + + for _, tc := range []struct { + name string + instruction string + wantSize int + }{ + {name: "ASCII at the limit", instruction: strings.Repeat("a", maxEnvironmentValueRunes-overhead)}, + {name: "ASCII one past the limit", instruction: strings.Repeat("a", maxEnvironmentValueRunes-overhead+1), wantSize: maxEnvironmentValueRunes + 1}, + {name: "HTML-escaped characters cost six each", instruction: strings.Repeat("<>&", 2000), wantSize: overhead + 6*6000}, + {name: "multibyte characters count once", instruction: strings.Repeat("é", 16000) + strings.Repeat("🙂", 16000)}, + {name: "multibyte characters past the limit", instruction: strings.Repeat("日", maxEnvironmentValueRunes), wantSize: overhead + maxEnvironmentValueRunes}, + } { + t.Run(tc.name, func(t *testing.T) { + spec := &translator.Revision{ + Namespace: "agents", AgentName: "helper", WorkerPoolName: "default", + ConfigJSON: configWithInstruction(t, tc.instruction), AgentCard: sizeTestCard(), + } + revisionID, err := spec.Digest() + require.NoError(t, err) + _, err = ActorTemplateForRevision(spec, revisionID) + if tc.wantSize == 0 { + require.NoError(t, err) + return + } + require.EqualError(t, err, fmt.Sprintf("compiled agent config is %d characters after JSON encoding; Substrate accepts at most 32768 in environment variable KAGENT_CONFIG_JSON. Move long reference material, such as a long system prompt, into a skill", tc.wantSize)) + }) + } + + t.Run("other variables are named", func(t *testing.T) { + spec := &translator.Revision{ + Namespace: "agents", AgentName: "helper", WorkerPoolName: "default", + ConfigJSON: []byte(`{}`), AgentCard: sizeTestCard(), + Environment: []corev1.EnvVar{{Name: "EXTRA", Value: strings.Repeat("x", maxEnvironmentValueRunes+1)}}, + } + revisionID, err := spec.Digest() + require.NoError(t, err) + _, err = ActorTemplateForRevision(spec, revisionID) + require.EqualError(t, err, "environment variable EXTRA is 32769 characters; Substrate accepts at most 32768") + }) +} + +func sizeTestCard() *a2apb.AgentCard { + return &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"}} +} From 9ab598e082f01484d8b7bb98ec75ea477033928b Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Fri, 25 Sep 2026 09:56:22 +0200 Subject: [PATCH 2/2] fix(controller): stop retrying an ActorTemplate Substrate rejects CreateActorTemplate returning InvalidArgument now records a final Compatible=False (ActorTemplateRejected) failure carrying Substrate's validation message, and the Agent is not requeued. The one-second pending-template poll re-added failing Agents with a plain Add, which bypasses the queue's rate limiter and restarts its budget, so a persistent failure was retried about once a second. The poll now leaves a failing Agent to the queue's backoff and, once that budget is spent, retries it every 30 seconds. Signed-off-by: QuentinBisson --- go/core/internal/controller/reconciler.go | 102 ++++++++++++- .../internal/controller/reconciler_test.go | 134 ++++++++++++++++++ 2 files changed, 229 insertions(+), 7 deletions(-) diff --git a/go/core/internal/controller/reconciler.go b/go/core/internal/controller/reconciler.go index 3738c63311..2fa5ed6d21 100644 --- a/go/core/internal/controller/reconciler.go +++ b/go/core/internal/controller/reconciler.go @@ -6,6 +6,7 @@ import ( "fmt" "reflect" "strings" + "sync" "time" atev1alpha1 "github.com/agent-substrate/substrate/pkg/api/v1alpha1" @@ -19,6 +20,7 @@ import ( claudetranslator "github.com/kagent-dev/kagent/go/core/internal/translator/claude" codextranslator "github.com/kagent-dev/kagent/go/core/internal/translator/codex" kagenttranslator "github.com/kagent-dev/kagent/go/core/internal/translator/kagent" + "github.com/kagent-dev/kagent/go/pkg/logging" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" @@ -169,6 +171,10 @@ type Reconciler struct { store runtimeRevisionStore status kagentclient.ApiV1alpha3Interface + agentFailuresMu sync.Mutex + // agentFailures holds when each failing Agent last failed. + agentFailures map[string]agentFailure + agents controllers.Queue agentStatuses controllers.Queue modelConfigStatuses controllers.Queue @@ -199,9 +205,7 @@ func newReconciler( store: store, status: status, } - r.agents = newReconciliationQueue("v2-agents", func(item any) error { - return r.reconcileAgent(context.Background(), item.(string)) - }) + r.agents = newReconciliationQueue("v2-agents", r.reconcileAgentItem) r.agentStatuses = newReconciliationQueue("v2-agent-status", func(item any) error { return r.reconcileAgentStatus(context.Background(), item.(string)) }) @@ -229,12 +233,18 @@ func newReconciler( return r } +const ( + reconciliationMaxAttempts = 10 + // reconciliationRetryCap is the longest a failing key waits between attempts. + reconciliationRetryCap = 30 * time.Second +) + // Ten attempts span about 2.5 minutes. Each queue owns its backoff state; // fresh graph events can enqueue work again after an error budget is exhausted. func newReconciliationQueue(name string, reconcile func(any) error) controllers.Queue { return controllers.NewQueue(name, controllers.WithGenericReconciler(reconcile), - controllers.WithMaxAttempts(10), - controllers.WithRateLimiter(workqueue.NewTypedItemExponentialFailureRateLimiter[any](time.Second, 30*time.Second)), + controllers.WithMaxAttempts(reconciliationMaxAttempts), + controllers.WithRateLimiter(workqueue.NewTypedItemExponentialFailureRateLimiter[any](time.Second, reconciliationRetryCap)), ) } @@ -260,10 +270,10 @@ func (r *Reconciler) pollPendingTemplates(stop <-chan struct{}) { select { case <-stop: return - case <-ticker.C: + case now := <-ticker.C: for _, state := range r.collections.Reconciliations.List() { golden := state.ObservedActorTemplate.GetStatus().GetGoldenSnapshotStatus() - if state.canPrepare() && golden.GetGoldenTag() == nil { + if state.canPrepare() && golden.GetGoldenTag() == nil && r.agentRetryDue(state.ResourceName(), now) { r.agents.Add(state.ResourceName()) } } @@ -271,6 +281,60 @@ func (r *Reconciler) pollPendingTemplates(stop <-chan struct{}) { } } +// agentFailure is how often an Agent failed in a row and when it last did. +type agentFailure struct { + attempts int + at time.Time +} + +// reconcileAgentItem hands a failure back to the queue until its retry budget +// is spent. The queue would then forget the key and the next poll would start +// its backoff over from one second, so the last error of the budget and every +// later one are logged here and the poll retries the Agent at the backoff cap. +func (r *Reconciler) reconcileAgentItem(item any) error { + key := item.(string) + ctx := context.Background() + err := r.reconcileAgent(ctx, key) + if attempts := r.recordAgentAttempt(key, err, time.Now()); attempts >= reconciliationMaxAttempts { + logging.FromContext(ctx).Error("reconcile Agent; retrying at the backoff cap", + "agent", key, "attempts", attempts, "error", err) + return nil + } + return err +} + +// recordAgentAttempt returns how often the Agent failed in a row. The poll +// must not re-add a failing Agent every tick: an Add bypasses the queue's rate +// limiter, so the poll alone would retry a persistent failure every second. +func (r *Reconciler) recordAgentAttempt(key string, err error, now time.Time) int { + r.agentFailuresMu.Lock() + defer r.agentFailuresMu.Unlock() + if err == nil { + delete(r.agentFailures, key) + return 0 + } + if r.agentFailures == nil { + r.agentFailures = map[string]agentFailure{} + } + failure := r.agentFailures[key] + failure.attempts++ + failure.at = now + r.agentFailures[key] = failure + return failure.attempts +} + +// agentRetryDue leaves a failing Agent to the queue's backoff and, once the +// queue's retry budget is spent, retries it at the backoff cap. +func (r *Reconciler) agentRetryDue(key string, now time.Time) bool { + r.agentFailuresMu.Lock() + defer r.agentFailuresMu.Unlock() + failure, failing := r.agentFailures[key] + if !failing { + return true + } + return failure.attempts >= reconciliationMaxAttempts && !now.Before(failure.at.Add(reconciliationRetryCap)) +} + func (r *Reconciler) Start(ctx context.Context) error { r.Run(ctx.Done()) return nil @@ -323,6 +387,12 @@ 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, state.DesiredActorTemplate) + if status.Code(err) == codes.InvalidArgument { + // Substrate validates the template before storing it, so it + // rejects the same revision on every attempt. + r.observePreparation(*state, nil, actorTemplateRejected(err)) + return nil + } if status.Code(err) == codes.AlreadyExists { observed, err = r.templates.GetActorTemplate(ctx, desiredRef.GetAtespace(), desiredRef.GetName()) } @@ -379,6 +449,24 @@ func runtimePreparationFailure(err error, configName string, class atev1alpha1.S } } +// maxRejectionMessageRunes bounds the Substrate message copied into status; +// a validation message can quote the rejected value. +const maxRejectionMessageRunes = 1024 + +// actorTemplateRejected publishes Substrate's validation message, which names +// the rejected field and its limit, as a final failure of the revision. +func actorTemplateRejected(err error) *ReconciliationFailure { + message := status.Convert(err).Message() + if runes := []rune(message); len(runes) > maxRejectionMessageRunes { + message = string(runes[:maxRejectionMessageRunes]) + "…" + } + return &ReconciliationFailure{ + Condition: kagentv1alpha3.AgentConditionCompatible, + Reason: "ActorTemplateRejected", + Message: "Substrate rejected the ActorTemplate: " + message, + } +} + // Observations belong to the Agent's current preparation, independently of how // long sessions or checkpoints keep its old runtime alive in the database. func (r *Reconciler) observePreparation(state AgentReconciliation, template *ateapipb.ActorTemplate, failure *ReconciliationFailure) { diff --git a/go/core/internal/controller/reconciler_test.go b/go/core/internal/controller/reconciler_test.go index 62e5d35bd4..d0e847cb5f 100644 --- a/go/core/internal/controller/reconciler_test.go +++ b/go/core/internal/controller/reconciler_test.go @@ -3,10 +3,13 @@ package controller import ( "context" "errors" + "strings" + "sync" "sync/atomic" "testing" "testing/synctest" "time" + "unicode/utf8" a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" "google.golang.org/protobuf/encoding/protowire" @@ -26,6 +29,7 @@ import ( "istio.io/istio/pkg/kube/krt" "istio.io/istio/pkg/kube/krt/krttest" corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -257,6 +261,7 @@ type fakeActorTemplates struct { ensureErr error getErr error createErr error + creates int deleteErr error deletedBeforeError bool } @@ -274,6 +279,7 @@ func (f *fakeActorTemplates) GetActorTemplate(context.Context, string, string) ( } func (f *fakeActorTemplates) CreateActorTemplate(_ context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error) { + f.creates++ if f.createErr != nil { return nil, f.createErr } @@ -428,3 +434,131 @@ func TestReconciliationQueueRetriesWithBackoff(t *testing.T) { require.NoError(t, queue.WaitForClose(time.Second)) }) } + +func pendingAgentFixture(t *testing.T, stop chan struct{}, templates actorTemplateClient) (*Reconciler, krt.StaticCollection[AgentReconciliation], AgentReconciliation) { + t.Helper() + opts := krt.NewOptionsBuilder(stop, "test", nil) + agent := &kagentv1alpha3.Agent{ObjectMeta: metav1.ObjectMeta{Namespace: "team-a", Name: "assistant", UID: "agent-uid"}} + revision := &v2translator.Revision{AgentCard: &a2apb.AgentCard{Name: "assistant"}} + revisionID, err := revision.Digest() + require.NoError(t, err) + state := AgentReconciliation{ + Agent: agent, + Revision: revision, RevisionID: revisionID, + DesiredActorTemplate: &ateapipb.ActorTemplate{Metadata: &ateapipb.ResourceMetadata{Atespace: "team-a", Name: "assistant-kagent-revision"}}, + } + reconciliations := krt.NewStaticCollection(nil, []AgentReconciliation{state}, opts.WithName("Reconciliations")...) + reconciler := &Reconciler{ + collections: Collections{ + AgentRuntimeObservations: krt.NewStaticCollection[AgentRuntimeObservation](nil, nil, opts.WithName("AgentRuntimeObservations")...), + Reconciliations: reconciliations, + }, + templates: templates, store: &fakeRuntimeRevisionStore{}, + } + return reconciler, reconciliations, state +} + +func TestReconcilerStopsOnActorTemplateRejection(t *testing.T) { + stop := make(chan struct{}) + t.Cleanup(func() { close(stop) }) + rejection := "actor_template.containers[0].env[26].value: Too long: may not be more than 32768 characters" + templates := &fakeActorTemplates{createErr: status.Error(codes.InvalidArgument, rejection)} + reconciler, reconciliations, state := pendingAgentFixture(t, stop, templates) + + require.NoError(t, reconciler.reconcileAgent(t.Context(), state.ResourceName()), "a rejected template must not be requeued") + require.Equal(t, 1, templates.creates) + observed := reconciler.collections.AgentRuntimeObservations.GetKey(state.ResourceName()) + require.NotNil(t, observed) + require.Equal(t, &ReconciliationFailure{ + Condition: kagentv1alpha3.AgentConditionCompatible, + Reason: "ActorTemplateRejected", + Message: "Substrate rejected the ActorTemplate: " + rejection, + }, observed.Failure) + + // The graph copies the observed failure into the Agent. + state.Failure = observed.Failure + reconciliations.UpdateObject(state) + require.NoError(t, reconciler.reconcileAgent(t.Context(), state.ResourceName())) + require.Equal(t, 1, templates.creates, "a rejected revision must not be sent to Substrate again") + + agentStatus := statusForAgent(state, 1, "") + compatible := apimeta.FindStatusCondition(agentStatus.Conditions, kagentv1alpha3.AgentConditionCompatible) + require.NotNil(t, compatible) + require.Equal(t, metav1.ConditionFalse, compatible.Status) + require.Equal(t, "ActorTemplateRejected", compatible.Reason) + require.Contains(t, compatible.Message, rejection) +} + +func TestActorTemplateRejectionMessageIsBounded(t *testing.T) { + failure := actorTemplateRejected(status.Error(codes.InvalidArgument, strings.Repeat("é", 2*maxRejectionMessageRunes))) + require.Equal(t, len("Substrate rejected the ActorTemplate: ")+maxRejectionMessageRunes+1, utf8.RuneCountInString(failure.Message)) +} + +// lockedActorTemplates serialises a fake a running queue calls into. +type lockedActorTemplates struct { + mu sync.Mutex + fake *fakeActorTemplates +} + +func (l *lockedActorTemplates) EnsureAtespace(ctx context.Context, atespace string) error { + l.mu.Lock() + defer l.mu.Unlock() + return l.fake.EnsureAtespace(ctx, atespace) +} + +func (l *lockedActorTemplates) GetActorTemplate(ctx context.Context, atespace, name string) (*ateapipb.ActorTemplate, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fake.GetActorTemplate(ctx, atespace, name) +} + +func (l *lockedActorTemplates) CreateActorTemplate(ctx context.Context, template *ateapipb.ActorTemplate) (*ateapipb.ActorTemplate, error) { + l.mu.Lock() + defer l.mu.Unlock() + return l.fake.CreateActorTemplate(ctx, template) +} + +func (l *lockedActorTemplates) DeleteActorTemplate(ctx context.Context, atespace, name string) error { + l.mu.Lock() + defer l.mu.Unlock() + return l.fake.DeleteActorTemplate(ctx, atespace, name) +} + +func (l *lockedActorTemplates) inspect(f func(*fakeActorTemplates)) { + l.mu.Lock() + defer l.mu.Unlock() + f(l.fake) +} + +func TestPendingPollBacksOffFailingAgents(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + stop := make(chan struct{}) + defer close(stop) + templates := &lockedActorTemplates{fake: &fakeActorTemplates{createErr: status.Error(codes.Unavailable, "substrate unavailable")}} + reconciler, _, _ := pendingAgentFixture(t, stop, templates) + reconciler.agents = newReconciliationQueue("test-pending-poll", reconciler.reconcileAgentItem) + go reconciler.agents.Run(stop) + go reconciler.pollPendingTemplates(stop) + creates := func() (count int) { + templates.inspect(func(f *fakeActorTemplates) { count = f.creates }) + return count + } + + // The queue's attempts take about two minutes. From then on the + // poll retries once per backoff cap. + time.Sleep(3 * time.Minute) + synctest.Wait() + attempts := creates() + require.LessOrEqual(t, attempts, reconciliationMaxAttempts+2, "a failing Agent must back off instead of being retried every poll") + time.Sleep(10 * reconciliationRetryCap) + synctest.Wait() + require.InDelta(t, 10, creates()-attempts, 1, "an Agent whose retry budget is spent must be retried at the backoff cap") + + templates.inspect(func(f *fakeActorTemplates) { f.createErr = nil }) + time.Sleep(reconciliationRetryCap + time.Second) + synctest.Wait() + templates.inspect(func(f *fakeActorTemplates) { + require.NotNil(t, f.template, "a recovered backend must create the template") + }) + }) +}