From c08e3b6129254aa9ccc17371dccabbb01b7840ca Mon Sep 17 00:00:00 2001 From: Jet Chiang Date: Thu, 15 Oct 2026 17:56:51 -0400 Subject: [PATCH 1/2] session error Signed-off-by: Jet Chiang --- docs/architecture/runtime-and-lifecycle.md | 10 +++ go/core/internal/a2agateway/http.go | 2 + go/core/internal/a2agateway/http_test.go | 61 ++++++++++++++++++- .../service/session/expiration_test.go | 24 ++++++++ .../internal/service/session/interactions.go | 5 +- .../service/session/interactions_test.go | 5 ++ go/core/test/e2e/session_expiration_test.go | 16 +++++ 7 files changed, 121 insertions(+), 2 deletions(-) diff --git a/docs/architecture/runtime-and-lifecycle.md b/docs/architecture/runtime-and-lifecycle.md index 01208061b0..53fb8a1090 100644 --- a/docs/architecture/runtime-and-lifecycle.md +++ b/docs/architecture/runtime-and-lifecycle.md @@ -115,6 +115,16 @@ the session, shares, runtime row, and creation receipt. GetSession returns NotFo CreateSession or ForkSession with the same caller/request ID can create a fresh session. Explicit client deletion still retains its tombstone and request ID. +A2A continuation using a missing or expired context ID returns `TaskNotFound` +(JSON-RPC `-32001`, gRPC `NOT_FOUND`), matching continuation using an old task ID. +Ownership-filtered lookup gives the same response for another user's Session; +the response does not distinguish expiry from absence or hidden ownership. +Authorization-check failures still return permission denied. Clients can start a +new conversation after `TaskNotFound` by sending without a context or task ID. +Service-level not-found errors use this same A2A mapping, including missing Agents. +Agent-card HTTP discovery returns 404 for not-found errors and 403 for permission +denials. Starting a new conversation requires an existing, authorized Agent. + The retention rule preserves the existing audit policy: A2A context, tasks, event history, and explicit checkpoints remain in PostgreSQL. Retained checkpoints can still be forked after the source session expires and retain their own snapshot diff --git a/go/core/internal/a2agateway/http.go b/go/core/internal/a2agateway/http.go index 03542b71f0..384c0ed4b4 100644 --- a/go/core/internal/a2agateway/http.go +++ b/go/core/internal/a2agateway/http.go @@ -45,6 +45,8 @@ func NewHTTPHandler(gateway a2asrv.RequestHandler, authenticator auth.AuthProvid status = http.StatusUnauthorized case errors.Is(err, a2atype.ErrUnauthorized): status = http.StatusForbidden + case errors.Is(err, a2atype.ErrTaskNotFound): + status = http.StatusNotFound case errors.Is(err, a2atype.ErrInvalidRequest): status = http.StatusBadRequest case errors.Is(err, a2atype.ErrUnsupportedOperation): diff --git a/go/core/internal/a2agateway/http_test.go b/go/core/internal/a2agateway/http_test.go index 817619229c..9235c9bb12 100644 --- a/go/core/internal/a2agateway/http_test.go +++ b/go/core/internal/a2agateway/http_test.go @@ -136,6 +136,65 @@ func TestGatewayCreatesConversationAndReusesInitialMessage(t *testing.T) { } } +func TestGatewayMissingConversationAcrossTransports(t *testing.T) { + for _, protocol := range []a2atype.TransportProtocol{a2atype.TransportProtocolJSONRPC, a2atype.TransportProtocolGRPC} { + for _, test := range []struct { + name string + storeErr error + taskErr error + authErr error + want error + }{ + {name: "missing context", storeErr: database.ErrNotFound, want: a2atype.ErrTaskNotFound}, + {name: "expired task", taskErr: database.ErrNotFound, storeErr: database.ErrNotFound, want: a2atype.ErrTaskNotFound}, + {name: "permission denied", authErr: errors.New("denied"), want: a2atype.ErrUnauthorized}, + {name: "store failure", storeErr: errors.New("private database details"), want: a2atype.ErrInternalError}, + } { + t.Run(string(protocol)+"/"+test.name, func(t *testing.T) { + store := &gatewayTestStore{session: gatewayTestSession(), err: test.storeErr, taskErr: test.taskErr} + dialer := &gatewayTestDialer{} + gateway := newTestGateway(store, &gatewayTestAuthorizer{err: test.authErr}, dialer, "") + transport := newGatewayTestTransport(t, startCoreTestServer(t, gateway, nil), protocol) + params := a2aclient.ServiceParams{"authorization": {"Bearer valid"}} + for _, ids := range []struct { + contextID string + taskID a2atype.TaskID + }{ + {contextID: gatewayTestID}, + {taskID: "old-task"}, + {}, // A new conversation must first resolve the Agent. + } { + request := gatewayTestRequest() + request.Message.ContextID, request.Message.TaskID = ids.contextID, ids.taskID + _, err := transport.SendMessage(t.Context(), params, request) + require.ErrorIs(t, err, test.want) + events := 0 + for event, err := range transport.SendStreamingMessage(t.Context(), params, request) { + require.Nil(t, event) + require.ErrorIs(t, err, test.want) + events++ + } + require.Equal(t, 1, events) + } + _, err := transport.GetTask(t.Context(), params, &a2atype.GetTaskRequest{Tenant: gatewayTestAgent, ID: "old-task"}) + require.ErrorIs(t, err, test.want) + _, err = transport.GetExtendedAgentCard(t.Context(), params, &a2atype.GetExtendedAgentCardRequest{Tenant: gatewayTestAgent}) + require.ErrorIs(t, err, test.want) + events := 0 + for event, err := range transport.SubscribeToTask(t.Context(), params, &a2atype.SubscribeToTaskRequest{Tenant: gatewayTestAgent, ID: "old-task"}) { + require.Nil(t, event) + require.ErrorIs(t, err, test.want) + events++ + } + require.Equal(t, 1, events) + require.Nil(t, dialer.session, "unavailable conversations must not reach the runtime") + require.Zero(t, store.reserveCalls) + require.False(t, store.unscoped, "session lookup must retain ownership filtering") + }) + } + } +} + func TestHTTPAgentURLIsAuthoritative(t *testing.T) { for _, tc := range []struct { name, path, tenant string @@ -298,7 +357,7 @@ func TestHTTPGatewayRejectsInvalidAccess(t *testing.T) { {name: "card needs authentication", method: "GET", path: gatewayTestAgent + a2asrv.WellKnownAgentCardPath, want: 401}, {name: "RPC needs authentication", method: "POST", path: gatewayTestAgent, want: 401}, {name: "card needs authorization", method: "GET", path: gatewayTestAgent + a2asrv.WellKnownAgentCardPath, authorized: true, authzErr: errors.New("denied"), want: 403}, - {name: "missing session is hidden", method: "GET", path: gatewayTestAgent + a2asrv.WellKnownAgentCardPath, authorized: true, storeErr: database.ErrNotFound, want: 403}, + {name: "missing Agent", method: "GET", path: gatewayTestAgent + a2asrv.WellKnownAgentCardPath, authorized: true, storeErr: database.ErrNotFound, want: 404}, {name: "internal errors are hidden", method: "GET", path: gatewayTestAgent + a2asrv.WellKnownAgentCardPath, authorized: true, storeErr: errors.New("private connection details"), want: 500}, {name: "invalid session ID", method: "GET", path: "invalid!/assistant" + a2asrv.WellKnownAgentCardPath, authorized: true, want: 400}, {name: "unknown route", method: "GET", path: gatewayTestAgent + "/unknown", authorized: true, want: 404}, diff --git a/go/core/internal/service/session/expiration_test.go b/go/core/internal/service/session/expiration_test.go index aa5b72c8d2..5624305dc9 100644 --- a/go/core/internal/service/session/expiration_test.go +++ b/go/core/internal/service/session/expiration_test.go @@ -15,6 +15,7 @@ import ( "go.opentelemetry.io/otel/sdk/metric/metricdata" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/types" "github.com/kagent-dev/kagent/go/core/internal/database" ) @@ -80,9 +81,32 @@ func TestSessionExpirationRetainsCheckpointHistory(t *testing.T) { store, source := lifecycleFixture(t) actors := &lifecycleTestActors{actors: map[string]*ateapipb.Actor{}} _, checkpointID := lifecycleForkFixture(t, store, actors, source) + interactions := NewInteractionService(store, nil, NewService(store, serviceTestAuthorizer{}, nil), false) + agent := types.NamespacedName{Namespace: source.Agent.Namespace, Name: source.Agent.Name} + // Real ownership filtering must hide an existing conversation in exactly + // the same way as a missing or expired conversation, without reading history. + _, err := interactions.ListTasks(serviceTestContext("visitor"), agent, &a2a.ListTasksRequest{ContextID: source.ContextId}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) + _, err = interactions.ListTasks(serviceTestContext(source.Creator), agent, &a2a.ListTasksRequest{ContextID: uuid.NewString()}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) + live, err := interactions.ListTasks(serviceTestContext(source.Creator), agent, &a2a.ListTasksRequest{ContextID: source.ContextId}) + require.NoError(t, err) + require.Len(t, live.Tasks, 1) + _, err = interactions.GetTask(serviceTestContext("visitor"), agent, &a2a.GetTaskRequest{ID: live.Tasks[0].ID}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) worker, err := NewExpirationWorker(store, NewActorWorkflow(store, actors, make(chan struct{}, 1), time.Second), time.Hour, time.Minute) require.NoError(t, err) require.NoError(t, worker.expire(t.Context(), source.Id, time.Now())) + _, err = interactions.ListTasks(serviceTestContext(source.Creator), agent, &a2a.ListTasksRequest{ContextID: source.ContextId}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) + _, err = interactions.GetTask(serviceTestContext(source.Creator), agent, &a2a.GetTaskRequest{ID: live.Tasks[0].ID}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) + for _, taskID := range []a2a.TaskID{"", live.Tasks[0].ID} { + message := a2a.NewMessage(a2a.MessageRoleUser, a2a.NewTextPart("Continue")) + message.ContextID, message.TaskID = source.ContextId, taskID + _, err = interactions.PrepareSend(serviceTestContext(source.Creator), agent, &a2a.SendMessageRequest{Message: message}) + require.ErrorIs(t, err, a2a.ErrTaskNotFound) + } // A new fork after expiration reconstructs the retained conversation. fork, created, err := store.ForkSession(t.Context(), checkpointID, source.Creator, uuid.NewString(), uuid.NewString()) require.NoError(t, err) diff --git a/go/core/internal/service/session/interactions.go b/go/core/internal/service/session/interactions.go index 3fac42e253..f9a90a7635 100644 --- a/go/core/internal/service/session/interactions.go +++ b/go/core/internal/service/session/interactions.go @@ -541,8 +541,11 @@ func interactionError(ctx context.Context, err error) error { switch serviceerrors.CodeOf(err) { case serviceerrors.CodeUnauthenticated: return a2atype.ErrUnauthenticated - case serviceerrors.CodePermissionDenied, serviceerrors.CodeNotFound: + case serviceerrors.CodePermissionDenied: return a2atype.ErrUnauthorized + case serviceerrors.CodeNotFound: + // A2A's standard not-found error covers missing Agents and conversations. + return a2atype.ErrTaskNotFound case serviceerrors.CodeInvalidArgument, serviceerrors.CodeAlreadyExists: return a2atype.NewError(a2atype.ErrInvalidRequest, serviceerrors.MessageOf(err)) case serviceerrors.CodeFailedPrecondition, serviceerrors.CodeAborted: diff --git a/go/core/internal/service/session/interactions_test.go b/go/core/internal/service/session/interactions_test.go index 8531bb45f0..4774ca5346 100644 --- a/go/core/internal/service/session/interactions_test.go +++ b/go/core/internal/service/session/interactions_test.go @@ -197,6 +197,7 @@ func TestInteractionsEnforcePermissionsWithoutGateway(t *testing.T) { {name: "unauthenticated", ctx: context.Background(), agent: agent, want: a2atype.ErrUnauthenticated}, {name: "missing Agent", ctx: serviceTestContext("visitor"), want: a2atype.ErrInvalidRequest}, {name: "denied caller", ctx: serviceTestContext("visitor"), agent: agent, want: a2atype.ErrUnauthorized}, + {name: "missing session", ctx: serviceTestContext("visitor"), agent: agent, want: a2atype.ErrTaskNotFound}, {name: "read-only share", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: agent}, {name: "other Agent", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: types.NamespacedName{Namespace: "team-a", Name: "other"}, want: a2atype.ErrUnauthorized}, } { @@ -205,6 +206,10 @@ func TestInteractionsEnforcePermissionsWithoutGateway(t *testing.T) { Agent: &apiv1.ResourceReference{Namespace: agent.Namespace, Name: agent.Name}} sessionStore := &serviceTestStore{getResult: stored} authorizer := &recordingAuthorizer{denied: map[string]bool{id: true}} + if test.name == "missing session" { + sessionStore.getResult, sessionStore.getErr = nil, database.ErrNotFound + authorizer.denied[id] = false + } tasks := &interactionTestStore{task: &a2atype.Task{ID: "task", ContextID: id, Status: a2atype.TaskStatus{State: a2atype.TaskStateCompleted}}} service := NewInteractionService(tasks, nil, NewService(sessionStore, authorizer, nil), false) want := test.want diff --git a/go/core/test/e2e/session_expiration_test.go b/go/core/test/e2e/session_expiration_test.go index 52d2f1a86f..a045e2b468 100644 --- a/go/core/test/e2e/session_expiration_test.go +++ b/go/core/test/e2e/session_expiration_test.go @@ -8,6 +8,8 @@ import ( "time" "github.com/a2aproject/a2a-go/v2/a2a" + a2apb "github.com/a2aproject/a2a-go/v2/a2apb/v1" + "github.com/a2aproject/a2a-go/v2/a2apb/v1/pbconv" "github.com/google/uuid" "github.com/stretchr/testify/require" "google.golang.org/grpc" @@ -56,6 +58,20 @@ func TestSessionIdleExpiration(t *testing.T) { _, err := fixture.sessions.GetSession(fixture.ctx, &apiv1.GetSessionRequest{SessionId: id}) return status.Code(err) == codes.NotFound }, 2*time.Minute, time.Second, "idle session did not expire") + for _, message := range []*a2a.Message{ + {ID: uuid.NewString(), ContextID: conversation.contextID}, + {ID: uuid.NewString(), ContextID: uuid.NewString()}, + {ID: uuid.NewString(), TaskID: task.ID}, + } { + message.Role = a2a.MessageRoleUser + message.Parts = a2a.ContentParts{a2a.NewTextPart("Continue")} + continuation, err := pbconv.ToProtoSendMessageRequest(&a2a.SendMessageRequest{Tenant: conversation.tenant, Message: message}) + require.NoError(t, err) + _, err = fixture.client.SendMessage(fixture.ctx, continuation) + require.Equal(t, codes.NotFound, status.Code(err), "expired and missing conversations must be distinguishable from authorization failures") + } + _, err = fixture.client.GetTask(fixture.ctx, &a2apb.GetTaskRequest{Tenant: conversation.tenant, Id: string(task.ID)}) + require.Equal(t, codes.NotFound, status.Code(err)) actor, err := findSubstrateActor(fixture.ctx, fixture.system, request.Agent.Namespace, substrate.ActorName(id)) require.NoError(t, err) require.Nil(t, actor, "expiration must remove the Substrate Actor") From 7a9a40509ecf4765688e041a0fa975dabec97c44 Mon Sep 17 00:00:00 2001 From: Jet Chiang Date: Thu, 15 Oct 2026 23:37:21 -0400 Subject: [PATCH 2/2] cleanup test Signed-off-by: Jet Chiang --- .../service/session/interactions_test.go | 34 ++++++++++--------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/go/core/internal/service/session/interactions_test.go b/go/core/internal/service/session/interactions_test.go index 4774ca5346..5d870eaaf3 100644 --- a/go/core/internal/service/session/interactions_test.go +++ b/go/core/internal/service/session/interactions_test.go @@ -189,31 +189,33 @@ func TestInteractionsEnforcePermissionsWithoutGateway(t *testing.T) { }}, } { for _, test := range []struct { - name string - ctx context.Context - agent types.NamespacedName - want error + name string + ctx context.Context + agent types.NamespacedName + denied bool + getErr error + want error }{ {name: "unauthenticated", ctx: context.Background(), agent: agent, want: a2atype.ErrUnauthenticated}, {name: "missing Agent", ctx: serviceTestContext("visitor"), want: a2atype.ErrInvalidRequest}, - {name: "denied caller", ctx: serviceTestContext("visitor"), agent: agent, want: a2atype.ErrUnauthorized}, - {name: "missing session", ctx: serviceTestContext("visitor"), agent: agent, want: a2atype.ErrTaskNotFound}, - {name: "read-only share", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: agent}, - {name: "other Agent", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: types.NamespacedName{Namespace: "team-a", Name: "other"}, want: a2atype.ErrUnauthorized}, + {name: "denied caller", ctx: serviceTestContext("visitor"), agent: agent, denied: true, want: a2atype.ErrUnauthorized}, + {name: "missing session", ctx: serviceTestContext("visitor"), agent: agent, getErr: database.ErrNotFound, want: a2atype.ErrTaskNotFound}, + {name: "read-only share", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: agent, denied: true}, + {name: "other Agent", ctx: auth.ShareContextTo(serviceTestContext("visitor"), &auth.ShareContext{SessionID: id, UserID: "owner"}), agent: types.NamespacedName{Namespace: "team-a", Name: "other"}, denied: true, want: a2atype.ErrUnauthorized}, } { t.Run(operation.name+"/"+test.name, func(t *testing.T) { stored := &apiv1.Session{Id: id, ContextId: id, State: apiv1.RuntimeState_RUNTIME_STATE_SUSPENDED, Agent: &apiv1.ResourceReference{Namespace: agent.Namespace, Name: agent.Name}} - sessionStore := &serviceTestStore{getResult: stored} - authorizer := &recordingAuthorizer{denied: map[string]bool{id: true}} - if test.name == "missing session" { - sessionStore.getResult, sessionStore.getErr = nil, database.ErrNotFound - authorizer.denied[id] = false + sessionStore := &serviceTestStore{getResult: stored, getErr: test.getErr} + authorizer := &recordingAuthorizer{denied: map[string]bool{id: test.denied}} + if test.getErr != nil { + sessionStore.getResult = nil } tasks := &interactionTestStore{task: &a2atype.Task{ID: "task", ContextID: id, Status: a2atype.TaskStatus{State: a2atype.TaskStateCompleted}}} service := NewInteractionService(tasks, nil, NewService(sessionStore, authorizer, nil), false) want := test.want - if test.name == "read-only share" && operation.verb != auth.VerbGet { + _, shared := auth.ShareContextFrom(test.ctx) + if shared && operation.verb != auth.VerbGet { want = a2atype.ErrUnauthorized } // No A2A SDK tenant or interceptor is installed for this call. @@ -227,10 +229,10 @@ func TestInteractionsEnforcePermissionsWithoutGateway(t *testing.T) { if want == nil && (tasks.taskReads != 1 || sessionStore.getCreator != "owner") { t.Fatal("shared history was not read through its owner") } - if test.name == "unauthenticated" && tasks.taskLookups != 0 { + if _, authenticated := auth.AuthSessionFrom(test.ctx); !authenticated && tasks.taskLookups != 0 { t.Fatal("unauthenticated operation looked up a task") } - if test.name == "denied caller" && (authorizer.verb != operation.verb || authorizer.resource.Type != "Session" || authorizer.resource.Name != id) { + if test.denied && !shared && (authorizer.verb != operation.verb || authorizer.resource.Type != "Session" || authorizer.resource.Name != id) { t.Fatalf("operation used the wrong permission: %+v", authorizer) } })