fix(controller): report and stop retrying an ActorTemplate Substrate refuses

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,17 +270,71 @@ 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())
 				}
 			}
 		}
 	}
 }
 
+// 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) {

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