fix(a2a): return not found for expired and missing sessions

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

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):

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},

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)

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:

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

@@ -189,26 +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: "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}}
+				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.
@@ -222,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)
 				}
 			})

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