fix(substrate): make sandbox agents resume reliably

go/core/internal/a2a/a2a_sdk_compat_test.go

@@ -0,0 +1,44 @@
+package a2a_test
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	a2atype "github.com/a2aproject/a2a-go/v2/a2a"
+	a2aclient "github.com/a2aproject/a2a-go/v2/a2aclient"
+)
+
+func TestJSONRPCClientDecodesSingleObjectErrorData(t *testing.T) {
+	t.Parallel()
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		var request struct {
+			ID json.RawMessage `json:"id"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+			t.Errorf("decode request: %v", err)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":` + string(request.ID) + `,"error":{"code":-32603,"message":"agent execution failed","data":{"@type":"type.googleapis.com/google.rpc.ErrorInfo","domain":"a2a-protocol.org","reason":"INTERNAL_ERROR"}}}`))
+	}))
+	t.Cleanup(server.Close)
+
+	transport := a2aclient.NewJSONRPCTransport(server.URL, server.Client())
+	_, err := transport.SendMessage(t.Context(), a2aclient.ServiceParams{}, &a2atype.SendMessageRequest{
+		Message: a2atype.NewMessage(a2atype.MessageRoleUser, a2atype.NewTextPart("test")),
+	})
+	if err == nil {
+		t.Fatal("SendMessage() succeeded, want application error")
+	}
+	if !strings.Contains(err.Error(), "agent execution failed") {
+		t.Fatalf("SendMessage() error = %q, want application error", err)
+	}
+	if strings.Contains(err.Error(), "failed to decode response") {
+		t.Fatalf("SendMessage() error obscured application error: %v", err)
+	}
+}

go/core/internal/a2a/substrate_sandbox_transport.go

@@ -16,7 +16,6 @@ import (
 	"github.com/kagent-dev/kagent/go/api/v1alpha3"
 	"github.com/kagent-dev/kagent/go/core/internal/utils"
 	"github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
-	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 // substrateSandboxSessionRoundTripper routes each A2A request to the session actor identified by contextId.
@@ -71,15 +70,18 @@ func (t *substrateSandboxSessionRoundTripper) RoundTrip(req *http.Request) (*htt
 	if sessionID == "" {
 		return nil, fmt.Errorf("message contextId (session id) is required for substrate sandbox agents")
 	}
+	userID := strings.TrimSpace(req.Header.Get("X-User-Id"))
+	if userID == "" {
+		return nil, fmt.Errorf("request carries no user identity")
+	}
 
-	// non blocking attempt to ensure that sandbox agent session metadata is persisted to postgres
-	// to support session list and delete cleanup.
-	if err := t.ensureSessionRow(req.Context(), sessionID, req.Header.Get("X-User-Id")); err != nil {
-		ctrllog.FromContext(req.Context()).WithName("substrate-sandbox-transport").Error(err,
-			"failed to ensure session row; continuing without it", "sessionID", sessionID)
+	// Persist owner-scoped session metadata before creating or resuming an actor.
+	// Failing closed keeps untracked sessions out of the shared sandbox runtime.
+	if err := t.ensureSessionRow(req.Context(), sessionID, userID); err != nil {
+		return nil, fmt.Errorf("ensure controller session row: %w", err)
 	}
 
-	res, err := t.actorBackend.EnsureSessionActor(req.Context(), t.sandboxAgent, sessionID)
+	res, err := t.actorBackend.EnsureSessionActor(req.Context(), t.sandboxAgent, userID, sessionID)
 	if err != nil {
 		return nil, err
 	}
@@ -93,7 +95,7 @@ func (t *substrateSandboxSessionRoundTripper) RoundTrip(req *http.Request) (*htt
 
 	resp, err := actorRT.RoundTrip(req)
 	if err != nil {
-		t.scheduleSuspendSession(sessionID)
+		t.scheduleSuspendSession(userID, sessionID)
 		return nil, err
 	}
 
@@ -102,7 +104,7 @@ func (t *substrateSandboxSessionRoundTripper) RoundTrip(req *http.Request) (*htt
 	resp.Body = &suspendSessionActorOnClose{
 		ReadCloser: resp.Body,
 		suspend: func() {
-			t.scheduleSuspendSession(sessionID)
+			t.scheduleSuspendSession(userID, sessionID)
 		},
 	}
 	return resp, nil
@@ -132,14 +134,14 @@ func (t *substrateSandboxSessionRoundTripper) ensureSessionRow(ctx context.Conte
 	return nil
 }
 
-func (t *substrateSandboxSessionRoundTripper) scheduleSuspendSession(sessionID string) {
+func (t *substrateSandboxSessionRoundTripper) scheduleSuspendSession(userID, sessionID string) {
 	if t == nil || t.actorBackend == nil || t.sandboxAgent == nil {
 		return
 	}
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 		defer cancel()
-		_ = t.actorBackend.SuspendSessionActor(ctx, t.sandboxAgent, sessionID)
+		_ = t.actorBackend.SuspendSessionActor(ctx, t.sandboxAgent, userID, sessionID)
 	}()
 }
 

go/core/internal/a2a/substrate_sandbox_transport_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"net/http"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -12,6 +13,7 @@ import (
 	"github.com/kagent-dev/kagent/go/api/v1alpha3"
 	coredatabase "github.com/kagent-dev/kagent/go/core/internal/database"
 	"github.com/kagent-dev/kagent/go/core/internal/dbtest"
+	"github.com/kagent-dev/kagent/go/core/pkg/sandboxbackend/substrate"
 )
 
 func TestSuspendSessionActorOnClose(t *testing.T) {
@@ -146,6 +148,24 @@ func TestExtractA2AContextID(t *testing.T) {
 	}
 }
 
+func TestSubstrateSandboxTransportRequiresOwnerIdentity(t *testing.T) {
+	t.Parallel()
+
+	rt := &substrateSandboxSessionRoundTripper{
+		sandboxAgent: &v1alpha3.SandboxAgent{},
+		actorBackend: &substrate.SandboxAgentActorBackend{},
+	}
+	req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://example.test", strings.NewReader(`{"params":{"message":{"contextId":"shared-session"}}}`))
+	if err != nil {
+		t.Fatalf("NewRequestWithContext: %v", err)
+	}
+
+	_, err = rt.RoundTrip(req)
+	if err == nil || !strings.Contains(err.Error(), "user identity") {
+		t.Fatalf("RoundTrip() error = %v, want missing user identity", err)
+	}
+}
+
 func TestSuspendSessionActorOnClosePreservesBody(t *testing.T) {
 	t.Parallel()
 

go/core/internal/service/session/service.go

@@ -35,7 +35,7 @@ type Store interface {
 }
 
 type SandboxActorCleaner interface {
-	DeleteSandboxAgentSessionActor(context.Context, *v1alpha3.SandboxAgent, string) (bool, error)
+	DeleteSandboxAgentSessionActor(context.Context, *v1alpha3.SandboxAgent, string, string) (bool, error)
 }
 
 type Service struct {
@@ -274,7 +274,7 @@ func (s *Service) Delete(ctx context.Context, sessionID string) error {
 		return serviceerrors.NewInternal("Failed to delete session", err)
 	}
 	if cleanup != nil {
-		if _, err := s.actorCleaner.DeleteSandboxAgentSessionActor(ctx, cleanup, sessionID); err != nil {
+		if _, err := s.actorCleaner.DeleteSandboxAgentSessionActor(ctx, cleanup, userID, sessionID); err != nil {
 			ctrllog.FromContext(ctx).Error(err, "failed to delete substrate session actor", "sessionID", sessionID)
 		}
 	}

go/core/pkg/sandboxbackend/substrate/agent_actor.go

@@ -49,19 +49,23 @@ func NewSandboxAgentActorBackend(client *Client, kube client.Client, atenetRoute
 // building a new golden, so a shape change can briefly make chat return "no free workers"; on a
 // multi-replica pool the spare workers keep serving existing actors, so a rollout does not hit
 // that error. Scaling the WorkerPool is the remedy for capacity pressure, not in-process retries.
-func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, sessionID string) (sandboxbackend.EnsureResult, error) {
+func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, userID, sessionID string) (sandboxbackend.EnsureResult, error) {
 	if sa == nil {
 		return sandboxbackend.EnsureResult{}, fmt.Errorf("SandboxAgent is required")
 	}
 	sessionID = strings.TrimSpace(sessionID)
 	if sessionID == "" {
 		return sandboxbackend.EnsureResult{}, fmt.Errorf("session id is required")
 	}
+	userID = strings.TrimSpace(userID)
+	if userID == "" {
+		return sandboxbackend.EnsureResult{}, fmt.Errorf("user identity is required")
+	}
 	if b == nil || b.client == nil {
 		return sandboxbackend.EnsureResult{}, fmt.Errorf("substrate ate-api client is required")
 	}
 
-	actorID, tmplName, err := b.sessionActorRef(ctx, sa, sessionID)
+	actorID, tmplName, err := b.sessionActorRef(ctx, sa, userID, sessionID)
 	if err != nil {
 		return sandboxbackend.EnsureResult{}, err
 	}
@@ -105,11 +109,14 @@ func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v
 }
 
 // SuspendSessionActor checkpoints and frees the worker for a chat session actor.
-func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, sessionID string) error {
+func (b *SandboxAgentActorBackend) SuspendSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, userID, sessionID string) error {
 	if sa == nil {
 		return nil
 	}
-	actorID, _, err := b.sessionActorRef(ctx, sa, sessionID)
+	if strings.TrimSpace(userID) == "" {
+		return fmt.Errorf("user identity is required")
+	}
+	actorID, _, err := b.sessionActorRef(ctx, sa, userID, sessionID)
 	if err != nil {
 		return err
 	}
@@ -138,30 +145,32 @@ func (b *SandboxAgentActorBackend) DeleteSandboxAgentActor(ctx context.Context,
 	return deleteActor(ctx, b.client, atespace, actorID)
 }
 
-// DeleteSandboxAgentSessionActor deletes the actor for a single chat session. One session ⇔ one
-// actor with a session-derived id, so a single deterministic delete covers the session's whole
-// life regardless of how many shape rollouts it survived.
-func (b *SandboxAgentActorBackend) DeleteSandboxAgentSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, sessionID string) (bool, error) {
+// DeleteSandboxAgentSessionActor deletes the actor for one owner/session pair. A single
+// deterministic delete covers that pair's whole life across shape rollouts.
+func (b *SandboxAgentActorBackend) DeleteSandboxAgentSessionActor(ctx context.Context, sa *v1alpha3.SandboxAgent, userID, sessionID string) (bool, error) {
 	if sa == nil {
 		return true, nil
 	}
-	return b.DeleteSandboxAgentActor(ctx, sa.Namespace, SandboxAgentSessionActorID(sa, sessionID))
+	if strings.TrimSpace(userID) == "" {
+		return false, fmt.Errorf("user identity is required")
+	}
+	return b.DeleteSandboxAgentActor(ctx, sa.Namespace, SandboxAgentPrincipalSessionActorID(sa, userID, sessionID))
 }
 
-// sessionActorRef returns the session's stable actor id plus the agent's CURRENT template name.
+// sessionActorRef returns the owner/session pair's stable actor id plus the agent's CURRENT template name.
 // The template name is only used when the actor does not exist yet (first message): an existing
 // actor is resumed under its original template — substrate stores the template name on the actor
 // record and rebuilds the workload spec from it — which is what pins a session to the shape it
 // was created under for its entire life.
-func (b *SandboxAgentActorBackend) sessionActorRef(ctx context.Context, sa *v1alpha3.SandboxAgent, sessionID string) (actorID, templateName string, err error) {
+func (b *SandboxAgentActorBackend) sessionActorRef(ctx context.Context, sa *v1alpha3.SandboxAgent, userID, sessionID string) (actorID, templateName string, err error) {
 	tmpl, err := ResolveCurrentActorTemplate(ctx, b.kube, sa.Namespace, sa.Name)
 	if err != nil {
 		return "", "", err
 	}
 	if tmpl == nil {
 		return "", "", fmt.Errorf("no ActorTemplate generated yet for SandboxAgent %s/%s", sa.Namespace, sa.Name)
 	}
-	return SandboxAgentSessionActorID(sa, sessionID), tmpl.Name, nil
+	return SandboxAgentPrincipalSessionActorID(sa, userID, sessionID), tmpl.Name, nil
 }
 
 // DeleteAllSandboxAgentActors deletes legacy per-agent actors and all session actors for a SandboxAgent.
@@ -228,17 +237,22 @@ func sandboxAgentActorPrefix(sa *v1alpha3.SandboxAgent) string {
 	return SandboxAgentActorID(sa)
 }
 
-// SandboxAgentSessionActorID returns the ate-api actor id for a SandboxAgent chat session,
-// derived from the session alone: one session ⇔ one actor for the session's entire life, across
-// config AND shape rollouts (the actor's template binding lives on the actor record, not in the
-// id). The id keeps the agent prefix (asr-<ns>-<name>-) so per-agent cleanup still matches.
-func SandboxAgentSessionActorID(sa *v1alpha3.SandboxAgent, sessionID string) string {
-	raw := fmt.Sprintf("%s-%s", sandboxAgentActorPrefix(sa), sanitizeSessionID(sessionID))
+// SandboxAgentPrincipalSessionActorID returns an opaque actor ID scoped to both
+// the authenticated owner and the caller-visible session ID.
+func SandboxAgentPrincipalSessionActorID(sa *v1alpha3.SandboxAgent, userID, sessionID string) string {
+	userID = strings.TrimSpace(userID)
+	owner := sha256.Sum256([]byte(userID))
+	raw := fmt.Sprintf(
+		"%s-u%x-%s",
+		sandboxAgentActorPrefix(sa),
+		owner[:6],
+		sanitizeSessionID(sessionID),
+	)
 	raw = strings.ToLower(strings.ReplaceAll(raw, "_", "-"))
 	if len(raw) <= 63 && dns1123Label.MatchString(raw) {
 		return raw
 	}
-	sum := sha256.Sum256([]byte(sa.Namespace + "/" + sa.Name + "/" + sessionID))
+	sum := sha256.Sum256([]byte(sa.Namespace + "/" + sa.Name + "/" + userID + "/" + sessionID))
 	return fmt.Sprintf("%s-%x", sandboxAgentIDPrefix, sum[:12])
 }
 

go/core/pkg/sandboxbackend/substrate/agent_actor_test.go

@@ -64,12 +64,11 @@ func (c *statusActorClient) ResumeActor(_ context.Context, in *ateapipb.ResumeAc
 	return &ateapipb.ResumeActorResponse{Actor: a}, nil
 }
 
-// TestDeleteSandboxAgentSessionActor covers the one-session-one-actor delete: a single
-// deterministic id covers the session's whole life, regardless of rollouts it survived.
+// TestDeleteSandboxAgentSessionActor covers the owner/session actor delete across rollouts.
 func TestDeleteSandboxAgentSessionActor(t *testing.T) {
 	t.Parallel()
 	sa := reapAgent()
-	actorID := SandboxAgentSessionActorID(sa, "sess-1")
+	actorID := SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "sess-1")
 
 	rec := &statusActorClient{actors: map[string]*ateapipb.Actor{}}
 	rec.add(&ateapipb.Actor{Metadata: &ateapipb.ResourceMetadata{Name: actorID}, Status: &ateapipb.ActorStatus{State: ateapipb.ActorState_ACTOR_STATE_SUSPENDED}})
@@ -79,7 +78,7 @@ func TestDeleteSandboxAgentSessionActor(t *testing.T) {
 	var done bool
 	var err error
 	for range 3 {
-		done, err = b.DeleteSandboxAgentSessionActor(context.Background(), sa, "sess-1")
+		done, err = b.DeleteSandboxAgentSessionActor(context.Background(), sa, "alice@example.com", "sess-1")
 		require.NoError(t, err)
 		if done {
 			break
@@ -89,11 +88,23 @@ func TestDeleteSandboxAgentSessionActor(t *testing.T) {
 	require.Equal(t, []string{actorID}, rec.deleted)
 
 	// Missing actor is already done.
-	done, err = b.DeleteSandboxAgentSessionActor(context.Background(), sa, "sess-never")
+	done, err = b.DeleteSandboxAgentSessionActor(context.Background(), sa, "alice@example.com", "sess-never")
 	require.NoError(t, err)
 	require.True(t, done)
 }
 
+func TestSandboxAgentPrincipalSessionActorIDIsOwnerScoped(t *testing.T) {
+	t.Parallel()
+	sa := reapAgent()
+
+	alice := SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "shared-session")
+	bob := SandboxAgentPrincipalSessionActorID(sa, "bob@example.com", "shared-session")
+
+	require.NotEqual(t, alice, bob)
+	require.Equal(t, alice, SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "shared-session"))
+	require.NotContains(t, alice, "alice")
+}
+
 // reapAgent is the SandboxAgent used by the reap tests.
 func reapAgent() *v1alpha3.SandboxAgent {
 	return &v1alpha3.SandboxAgent{ObjectMeta: metav1.ObjectMeta{Name: "agent", Namespace: "kagent"}}

go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go

@@ -98,6 +98,7 @@ func (p *Lifecycle) buildSandboxAgentActorTemplate(
 					Path: "/.well-known/agent-card.json",
 					Port: substrateKagentListenPort,
 				},
+				TimeoutSeconds: 30,
 			},
 		}},
 		WorkerSelector: workerSelectorForPool(wpKey),
@@ -175,6 +176,7 @@ func applyDurableDirSessionStore(spec *atev1alpha1.ActorTemplateSpec) {
 		MountPath: durableDataMount,
 	})
 	spec.SnapshotsConfig.OnCommit = atev1alpha1.SnapshotScopeData
+	spec.SnapshotsConfig.OnResume = atev1alpha1.OnResumeConfig{FromData: atev1alpha1.ResumeSourceColdBoot}
 }
 
 func findKagentContainer(containers []corev1.Container) *corev1.Container {

go/core/pkg/sandboxbackend/substrate/agent_lifecycle_test.go

@@ -47,6 +47,27 @@ func TestActorTemplateEnvFromPodEnv(t *testing.T) {
 	require.Equal(t, []atev1alpha1.EnvVar{{Name: "LITERAL", Value: "ok"}}, got)
 }
 
+func TestResolvePodEnvUsesStringDataFromDesiredSecret(t *testing.T) {
+	t.Parallel()
+
+	localSecret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "agent-config", Namespace: "kagent"},
+		StringData: map[string]string{"config.json": `{"app":"isolated"}`},
+	}
+	environment := []corev1.EnvVar{{
+		Name: "KAGENT_CONFIG_JSON",
+		ValueFrom: &corev1.EnvVarSource{SecretKeyRef: &corev1.SecretKeySelector{
+			LocalObjectReference: corev1.LocalObjectReference{Name: localSecret.Name},
+			Key:                  "config.json",
+		}},
+	}}
+
+	resolved, err := resolvePodEnv(context.Background(), nil, "kagent", environment, localSecret)
+	require.NoError(t, err)
+	require.Equal(t, `{"app":"isolated"}`, resolved[0].Value)
+	require.Nil(t, resolved[0].ValueFrom)
+}
+
 func TestBuildSubstrateDeclarativeCommand(t *testing.T) {
 	t.Parallel()
 
@@ -286,10 +307,12 @@ func TestBuildSandboxAgentActorTemplateDurableDirSessions(t *testing.T) {
 			require.NotNil(t, c.Readyz)
 			require.Equal(t, "/.well-known/agent-card.json", c.Readyz.HTTPGet.Path)
 			require.Equal(t, substrateKagentListenPort, c.Readyz.HTTPGet.Port)
+			require.Equal(t, int32(30), c.Readyz.TimeoutSeconds)
 			// Durable-dir sessions suspend with Data scope (cheap per-turn snapshots + config
 			// refresh on resume); pause keeps Full for the golden build.
 			require.Equal(t, atev1alpha1.SnapshotScopeData, tmpl.Spec.SnapshotsConfig.OnCommit)
 			require.Equal(t, atev1alpha1.SnapshotScopeFull, tmpl.Spec.SnapshotsConfig.OnPause)
+			require.Equal(t, atev1alpha1.ResumeSourceColdBoot, tmpl.Spec.SnapshotsConfig.OnResume.FromData)
 		})
 	}
 }