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") + }) + }) +} 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"}} +}