From d6888c2075e234ff5c7e7f365a25c386c58491e0 Mon Sep 17 00:00:00 2001 From: sam123ben Date: Thu, 20 Aug 2026 06:54:20 +1000 Subject: [PATCH 1/3] fix(substrate): make sandbox agents resume reliably Signed-off-by: sam123ben --- .../substrate/agent_lifecycle.go | 2 ++ .../substrate/agent_lifecycle_test.go | 23 +++++++++++++++++++ .../substrate/lifecycle_shared.go | 10 +++++++- 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go b/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go index 019a3cfbf..50e19ffc0 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle.go +++ b/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 { diff --git a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle_test.go b/go/core/pkg/sandboxbackend/substrate/agent_lifecycle_test.go index 3ff8da8d9..439f78b68 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_lifecycle_test.go +++ b/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) }) } } diff --git a/go/core/pkg/sandboxbackend/substrate/lifecycle_shared.go b/go/core/pkg/sandboxbackend/substrate/lifecycle_shared.go index 29d43f62b..30a5fd8da 100644 --- a/go/core/pkg/sandboxbackend/substrate/lifecycle_shared.go +++ b/go/core/pkg/sandboxbackend/substrate/lifecycle_shared.go @@ -289,16 +289,24 @@ func resolvePodEnv(ctx context.Context, kube client.Reader, namespace string, en } ref := variable.ValueFrom.SecretKeyRef secret := &corev1.Secret{} + var value []byte + var ok bool if localSecret != nil && localSecret.Name == ref.Name { secret = localSecret + if stringValue, found := secret.StringData[ref.Key]; found { + value, ok = []byte(stringValue), true + } else { + value, ok = secret.Data[ref.Key] + } } else if err := kube.Get(ctx, types.NamespacedName{Namespace: namespace, Name: ref.Name}, secret); err != nil { if ref.Optional != nil && *ref.Optional && apierrors.IsNotFound(err) { resolved[i].ValueFrom = nil continue } return nil, err + } else { + value, ok = secret.Data[ref.Key] } - value, ok := secret.Data[ref.Key] if !ok { if ref.Optional != nil && *ref.Optional { resolved[i].ValueFrom = nil From cb2fc2f05e3e6571845e611228ed150742b3bc66 Mon Sep 17 00:00:00 2001 From: sam123ben Date: Thu, 20 Aug 2026 07:35:10 +1000 Subject: [PATCH 2/3] fix(a2a): preserve single-object application errors Signed-off-by: sam123ben --- go/core/internal/a2a/a2a_sdk_compat_test.go | 44 +++++++++++++++++++++ go/go.mod | 2 +- go/go.sum | 2 + 3 files changed, 47 insertions(+), 1 deletion(-) create mode 100644 go/core/internal/a2a/a2a_sdk_compat_test.go diff --git a/go/core/internal/a2a/a2a_sdk_compat_test.go b/go/core/internal/a2a/a2a_sdk_compat_test.go new file mode 100644 index 000000000..68047d81e --- /dev/null +++ b/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) + } +} diff --git a/go/go.mod b/go/go.mod index ab1be5fba..311af1f7e 100644 --- a/go/go.mod +++ b/go/go.mod @@ -5,7 +5,7 @@ go 1.26.5 require ( // core dependencies dario.cat/mergo v1.0.2 - github.com/a2aproject/a2a-go/v2 v2.3.1 + github.com/a2aproject/a2a-go/v2 v2.5.0 github.com/abiosoft/ishell/v2 v2.0.2 github.com/anthropics/anthropic-sdk-go v1.61.0 github.com/aws/aws-sdk-go-v2/config v1.32.33 diff --git a/go/go.sum b/go/go.sum index c7f8db3c6..21e1f246f 100644 --- a/go/go.sum +++ b/go/go.sum @@ -78,6 +78,8 @@ github.com/OpenPeeDeeP/depguard/v2 v2.2.1 h1:vckeWVESWp6Qog7UZSARNqfu/cZqvki8zsu github.com/OpenPeeDeeP/depguard/v2 v2.2.1/go.mod h1:q4DKzC4UcVaAvcfd41CZh0PWpGgzrVxUYBlgKNGquUo= github.com/a2aproject/a2a-go/v2 v2.3.1 h1:QWMdOX2UsJ8BJmjs952eo1FRyGsOVl0gFCKeM76AgGE= github.com/a2aproject/a2a-go/v2 v2.3.1/go.mod h1:mkZr8y2bUgAVQsjs/5fHK7xrRlAHDybMEyxWh2tKRC8= +github.com/a2aproject/a2a-go/v2 v2.5.0 h1:ZdcFoxv+nZTUV0i2ue5hES76YCANFPG9vjqd7vK8yWM= +github.com/a2aproject/a2a-go/v2 v2.5.0/go.mod h1:NcRp/ZHxgMzDj12/BteIC2gOjljuEBKaGRfEdJ2lNSI= github.com/abiosoft/ishell v2.0.0+incompatible h1:zpwIuEHc37EzrsIYah3cpevrIc8Oma7oZPxr03tlmmw= github.com/abiosoft/ishell v2.0.0+incompatible/go.mod h1:HQR9AqF2R3P4XXpMpI0NAzgHf/aS6+zVXRj14cVk9qg= github.com/abiosoft/ishell/v2 v2.0.2 h1:5qVfGiQISaYM8TkbBl7RFO6MddABoXpATrsFbVI+SNo= From f13d370df494826e532596c51aadd281bc343201 Mon Sep 17 00:00:00 2001 From: sam123ben Date: Thu, 20 Aug 2026 08:25:13 +1000 Subject: [PATCH 3/3] fix(substrate): scope session actors to owners Signed-off-by: sam123ben --- .../a2a/substrate_sandbox_transport.go | 24 +++++---- .../a2a/substrate_sandbox_transport_test.go | 20 +++++++ go/core/internal/service/session/service.go | 4 +- .../sandboxbackend/substrate/agent_actor.go | 52 ++++++++++++------- .../substrate/agent_actor_test.go | 21 ++++++-- .../substrate/config_hash_test.go | 17 +++--- 6 files changed, 92 insertions(+), 46 deletions(-) diff --git a/go/core/internal/a2a/substrate_sandbox_transport.go b/go/core/internal/a2a/substrate_sandbox_transport.go index eef62ec50..2c60c0926 100644 --- a/go/core/internal/a2a/substrate_sandbox_transport.go +++ b/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) }() } diff --git a/go/core/internal/a2a/substrate_sandbox_transport_test.go b/go/core/internal/a2a/substrate_sandbox_transport_test.go index 6825831e4..33890734e 100644 --- a/go/core/internal/a2a/substrate_sandbox_transport_test.go +++ b/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() diff --git a/go/core/internal/service/session/service.go b/go/core/internal/service/session/service.go index 6ae275dd1..bc41daef2 100644 --- a/go/core/internal/service/session/service.go +++ b/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) } } diff --git a/go/core/pkg/sandboxbackend/substrate/agent_actor.go b/go/core/pkg/sandboxbackend/substrate/agent_actor.go index 82cfe7810..f053453c6 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_actor.go +++ b/go/core/pkg/sandboxbackend/substrate/agent_actor.go @@ -49,7 +49,7 @@ 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") } @@ -57,11 +57,15 @@ func (b *SandboxAgentActorBackend) EnsureSessionActor(ctx context.Context, sa *v 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,22 +145,24 @@ 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 @@ -161,7 +170,7 @@ func (b *SandboxAgentActorBackend) sessionActorRef(ctx context.Context, sa *v1al 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---) 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]) } diff --git a/go/core/pkg/sandboxbackend/substrate/agent_actor_test.go b/go/core/pkg/sandboxbackend/substrate/agent_actor_test.go index 6f30811a8..5717937cb 100644 --- a/go/core/pkg/sandboxbackend/substrate/agent_actor_test.go +++ b/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"}} diff --git a/go/core/pkg/sandboxbackend/substrate/config_hash_test.go b/go/core/pkg/sandboxbackend/substrate/config_hash_test.go index 1d35f3928..b4cdca07d 100644 --- a/go/core/pkg/sandboxbackend/substrate/config_hash_test.go +++ b/go/core/pkg/sandboxbackend/substrate/config_hash_test.go @@ -47,26 +47,25 @@ func TestSandboxAgentActorTemplateNameWithHash(t *testing.T) { require.LessOrEqual(t, len(sandboxAgentActorTemplateName(long, "deadbeefdeadbeef")), 63) } -func TestSandboxAgentSessionActorIDIsSessionStable(t *testing.T) { +func TestSandboxAgentSessionActorIDIsOwnerAndSessionStable(t *testing.T) { t.Parallel() sa := &v1alpha3.SandboxAgent{ObjectMeta: metav1.ObjectMeta{Name: "my-agent", Namespace: "kagent"}} - // One session ⇔ one actor: the id is derived from the session alone, so it survives config - // AND shape rollouts (the actor's template binding lives on the actor record, not in the id). - id := SandboxAgentSessionActorID(sa, "sess-1") - require.Equal(t, "asr-kagent-my-agent-sess-1", id) - require.Equal(t, id, SandboxAgentSessionActorID(sa, "sess-1")) - require.NotEqual(t, id, SandboxAgentSessionActorID(sa, "sess-2"), "sessions never share an actor") + // One owner/session pair maps to one stable actor across config and shape rollouts. + id := SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "sess-1") + require.Equal(t, id, SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "sess-1")) + require.NotEqual(t, id, SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", "sess-2")) + require.NotEqual(t, id, SandboxAgentPrincipalSessionActorID(sa, "bob@example.com", "sess-1")) // Keeps the per-agent prefix so agent-level cleanup still matches by prefix. prefix := sandboxAgentActorPrefix(sa) require.True(t, strings.HasPrefix(id, prefix+"-")) // Over-budget ids fall back to a deterministic hashed form. - long := SandboxAgentSessionActorID(sa, strings.Repeat("s", 80)) + long := SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", strings.Repeat("s", 80)) require.LessOrEqual(t, len(long), 63) require.True(t, strings.HasPrefix(long, sandboxAgentIDPrefix+"-")) - require.Equal(t, long, SandboxAgentSessionActorID(sa, strings.Repeat("s", 80))) + require.Equal(t, long, SandboxAgentPrincipalSessionActorID(sa, "alice@example.com", strings.Repeat("s", 80))) } func TestBuildActorTemplateShapeHashIdentity(t *testing.T) {