refactor: hide pgx errors behind database.ErrNotFound

go/api/database/errors.go

@@ -0,0 +1,8 @@
+package database
+
+import "errors"
+
+// ErrNotFound reports that the requested record does not exist (or is not
+// visible to the given user). Match with errors.Is; implementations wrap it
+// with call-site context.
+var ErrNotFound = errors.New("record not found")

go/core/internal/a2a/substrate_sandbox_transport.go

@@ -12,7 +12,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/jackc/pgx/v5"
 	"github.com/kagent-dev/kagent/go/api/database"
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
 	"github.com/kagent-dev/kagent/go/core/internal/utils"
@@ -123,7 +122,7 @@ func (t *substrateSandboxSessionRoundTripper) ensureSessionRow(ctx context.Conte
 	if err == nil {
 		return nil
 	}
-	if !errors.Is(err, pgx.ErrNoRows) {
+	if !errors.Is(err, database.ErrNotFound) {
 		return fmt.Errorf("get session %q: %w", sessionID, err)
 	}
 	agentID := utils.ConvertToPythonIdentifier(t.sandboxAgent.Namespace + "/" + t.sandboxAgent.Name)

go/core/internal/a2a/substrate_sandbox_transport_test.go

@@ -8,7 +8,6 @@ import (
 	"sync/atomic"
 	"testing"
 
-	"github.com/jackc/pgx/v5"
 	"github.com/kagent-dev/kagent/go/api/database"
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
 	coredatabase "github.com/kagent-dev/kagent/go/core/internal/database"
@@ -131,7 +130,7 @@ func TestEnsureSessionRow(t *testing.T) {
 		if err := rt.ensureSessionRow(ctx, "sess-3", ""); err == nil {
 			t.Fatal("expected an error when the request carries no user identity")
 		}
-		if _, err := db.GetSession(ctx, "sess-3", ""); !errors.Is(err, pgx.ErrNoRows) {
+		if _, err := db.GetSession(ctx, "sess-3", ""); !errors.Is(err, database.ErrNotFound) {
 			t.Fatalf("expected no row, got %v", err)
 		}
 	})

go/core/internal/database/client_postgres.go

@@ -56,10 +56,19 @@ func (c *postgresClient) StoreAgent(ctx context.Context, agent *dbpkg.Agent) err
 	})
 }
 
+// notFoundOr maps the driver's no-rows error to dbpkg.ErrNotFound so callers
+// outside this package match on the exported sentinel, never on pgx.
+func notFoundOr(err error) error {
+	if errors.Is(err, pgx.ErrNoRows) {
+		return dbpkg.ErrNotFound
+	}
+	return err
+}
+
 func (c *postgresClient) GetAgent(ctx context.Context, id string) (*dbpkg.Agent, error) {
 	row, err := c.q.GetAgent(ctx, id)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get agent %s: %w", id, err)
+		return nil, fmt.Errorf("failed to get agent %s: %w", id, notFoundOr(err))
 	}
 	return toAgent(row), nil
 }
@@ -101,7 +110,7 @@ func (c *postgresClient) StoreSession(ctx context.Context, session *dbpkg.Sessio
 func (c *postgresClient) GetSession(ctx context.Context, sessionID, userID string) (*dbpkg.Session, error) {
 	row, err := c.q.GetSession(ctx, dbgen.GetSessionParams{ID: sessionID, UserID: userID})
 	if err != nil {
-		return nil, fmt.Errorf("failed to get session %s: %w", sessionID, err)
+		return nil, fmt.Errorf("failed to get session %s: %w", sessionID, notFoundOr(err))
 	}
 	return toSession(row), nil
 }
@@ -179,7 +188,7 @@ func (c *postgresClient) CreateSessionShare(ctx context.Context, share *dbpkg.Se
 func (c *postgresClient) GetSessionShareByToken(ctx context.Context, token string) (*dbpkg.SessionShare, error) {
 	row, err := c.q.GetSessionShareByToken(ctx, token)
 	if err != nil {
-		return nil, fmt.Errorf("get session share by token: %w", err)
+		return nil, fmt.Errorf("get session share by token: %w", notFoundOr(err))
 	}
 	result := toSessionShare(row)
 	return &result, nil
@@ -321,7 +330,7 @@ func (c *postgresClient) checkTaskOwner(ctx context.Context, taskID, userID stri
 func (c *postgresClient) GetTask(ctx context.Context, taskID, userID string) (*a2a.Task, error) {
 	row, err := c.q.GetTask(ctx, dbgen.GetTaskParams{ID: taskID, UserID: &userID})
 	if err != nil {
-		return nil, fmt.Errorf("failed to get task %s: %w", taskID, err)
+		return nil, fmt.Errorf("failed to get task %s: %w", taskID, notFoundOr(err))
 	}
 	return parseVersionedTask(row.Data, row.ProtocolVersion)
 }
@@ -372,7 +381,7 @@ func (c *postgresClient) StorePushNotification(ctx context.Context, config *a2a.
 func (c *postgresClient) GetPushNotification(ctx context.Context, taskID, configID string) (*a2a.PushConfig, error) {
 	row, err := c.q.GetPushNotification(ctx, dbgen.GetPushNotificationParams{TaskID: taskID, ID: configID})
 	if err != nil {
-		return nil, fmt.Errorf("failed to get push notification: %w", err)
+		return nil, fmt.Errorf("failed to get push notification: %w", notFoundOr(err))
 	}
 	return parseVersionedPushConfig(row.Data, row.ProtocolVersion)
 }
@@ -427,7 +436,7 @@ func (c *postgresClient) ListFeedback(ctx context.Context, userID string) ([]dbp
 func (c *postgresClient) GetTool(ctx context.Context, name string) (*dbpkg.Tool, error) {
 	row, err := c.q.GetTool(ctx, name)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get tool %s: %w", name, err)
+		return nil, fmt.Errorf("failed to get tool %s: %w", name, notFoundOr(err))
 	}
 	return toTool(row), nil
 }
@@ -484,7 +493,7 @@ func (c *postgresClient) RefreshToolsForServer(ctx context.Context, serverName,
 func (c *postgresClient) GetToolServer(ctx context.Context, name string) (*dbpkg.ToolServer, error) {
 	row, err := c.q.GetToolServer(ctx, name)
 	if err != nil {
-		return nil, fmt.Errorf("failed to get tool server %s: %w", name, err)
+		return nil, fmt.Errorf("failed to get tool server %s: %w", name, notFoundOr(err))
 	}
 	return toToolServer(row), nil
 }
@@ -565,7 +574,7 @@ func (c *postgresClient) ListCheckpoints(ctx context.Context, userID, threadID,
 				UserID: userID, ThreadID: threadID, CheckpointNs: checkpointNS, CheckpointID: *checkpointID,
 			})
 			if err != nil {
-				return fmt.Errorf("failed to get checkpoint: %w", err)
+				return fmt.Errorf("failed to get checkpoint: %w", notFoundOr(err))
 			}
 			checkpoints = []dbgen.LgCheckpoint{cp}
 		} else if limit > 0 {

go/core/internal/database/client_test.go

@@ -974,3 +974,38 @@ func TestSearchAgentMemoryConcurrentAccessCount(t *testing.T) {
 		require.NoError(t, err, "concurrent memory search must not fail")
 	}
 }
+
+// TestSingleRowReadsMapMissingToErrNotFound verifies that every single-row
+// read maps the driver's no-rows error to dbpkg.ErrNotFound, so callers can
+// match with errors.Is without importing pgx.
+func TestSingleRowReadsMapMissingToErrNotFound(t *testing.T) {
+	db := setupTestDB(t)
+	client := NewClient(db)
+	ctx := context.Background()
+	missing := "missing"
+
+	tests := []struct {
+		name string
+		read func() error
+	}{
+		{name: "GetAgent", read: func() error { _, err := client.GetAgent(ctx, "missing"); return err }},
+		{name: "GetSession", read: func() error { _, err := client.GetSession(ctx, "missing", "user"); return err }},
+		{name: "GetSessionShareByToken", read: func() error { _, err := client.GetSessionShareByToken(ctx, "missing"); return err }},
+		{name: "GetTask", read: func() error { _, err := client.GetTask(ctx, "missing", "user"); return err }},
+		{name: "GetPushNotification", read: func() error { _, err := client.GetPushNotification(ctx, "missing", "missing"); return err }},
+		{name: "GetTool", read: func() error { _, err := client.GetTool(ctx, "missing"); return err }},
+		{name: "GetToolServer", read: func() error { _, err := client.GetToolServer(ctx, "missing"); return err }},
+		{name: "ListCheckpoints by ID", read: func() error {
+			_, err := client.ListCheckpoints(ctx, "user", "thread", "namespace", &missing, 0)
+			return err
+		}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.read()
+			require.Error(t, err)
+			require.ErrorIs(t, err, dbpkg.ErrNotFound)
+		})
+	}
+}

go/core/internal/httpserver/handlers/helpers.go

@@ -3,13 +3,15 @@ package handlers
 import (
 	"context"
 	"encoding/json"
+	stderrors "errors"
 	"fmt"
 	"net/http"
 	"reflect"
 	"strconv"
 	"strings"
 
 	"github.com/gorilla/mux"
+	"github.com/kagent-dev/kagent/go/api/database"
 	"github.com/kagent-dev/kagent/go/core/internal/httpserver/errors"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
 	corev1 "k8s.io/api/core/v1"
@@ -24,6 +26,17 @@ type ErrorResponseWriter interface {
 	Flush()
 }
 
+// RespondNotFoundOrError writes a 404 only when err is a missing-record error
+// from the database client; anything else is a backend failure and must
+// surface as a 500 rather than masquerade as not-found.
+func RespondNotFoundOrError(w ErrorResponseWriter, notFoundMessage string, err error) {
+	if stderrors.Is(err, database.ErrNotFound) {
+		w.RespondWithError(errors.NewNotFoundError(notFoundMessage, err))
+		return
+	}
+	w.RespondWithError(errors.NewInternalServerError("Internal server error", err))
+}
+
 func RespondWithJSON(w http.ResponseWriter, code int, payload any) {
 	log := ctrllog.Log.WithName("http-helpers")
 

go/core/internal/httpserver/handlers/helpers_notfound_test.go

@@ -0,0 +1,31 @@
+package handlers_test
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/kagent-dev/kagent/go/api/database"
+	"github.com/kagent-dev/kagent/go/core/internal/httpserver/handlers"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRespondNotFoundOrError(t *testing.T) {
+	tests := []struct {
+		name       string
+		err        error
+		wantStatus int
+	}{
+		{name: "missing record maps to 404", err: fmt.Errorf("session x: %w", database.ErrNotFound), wantStatus: http.StatusNotFound},
+		{name: "backend failure maps to 500", err: errors.New("connection refused"), wantStatus: http.StatusInternalServerError},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			w := newMockErrorResponseWriter()
+			handlers.RespondNotFoundOrError(w, "not found", tt.err)
+			require.Equal(t, tt.wantStatus, w.Code)
+		})
+	}
+}

go/core/internal/httpserver/handlers/session_shares.go

@@ -68,7 +68,7 @@ func (h *SessionSharesHandler) HandleCreateSessionShare(w ErrorResponseWriter, r
 
 	// Verify the session belongs to the caller.
 	if _, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID); err != nil {
-		w.RespondWithError(errors.NewNotFoundError("session not found", err))
+		RespondNotFoundOrError(w, "session not found", err)
 		return
 	}
 
@@ -113,7 +113,7 @@ func (h *SessionSharesHandler) HandleListSessionShares(w ErrorResponseWriter, r
 
 	// Verify the session belongs to the caller.
 	if _, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID); err != nil {
-		w.RespondWithError(errors.NewNotFoundError("session not found", err))
+		RespondNotFoundOrError(w, "session not found", err)
 		return
 	}
 
@@ -152,7 +152,7 @@ func (h *SessionSharesHandler) HandleDeleteSessionShare(w ErrorResponseWriter, r
 
 	// Verify the session belongs to the caller before attempting deletion.
 	if _, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID); err != nil {
-		w.RespondWithError(errors.NewNotFoundError("session not found", err))
+		RespondNotFoundOrError(w, "session not found", err)
 		return
 	}
 

go/core/internal/httpserver/handlers/sessions.go

@@ -67,7 +67,7 @@ func (h *SessionsHandler) HandleGetSessionsForAgent(w ErrorResponseWriter, r *ht
 	// agent table as regular agents, so the lookup is uniform.
 	agentID := utils.ConvertToPythonIdentifier(namespace + "/" + agentName)
 	if _, err := h.DatabaseService.GetAgent(r.Context(), agentID); err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Agent not found", err))
+		RespondNotFoundOrError(w, "Agent not found", err)
 		return
 	}
 
@@ -230,7 +230,7 @@ func (h *SessionsHandler) HandleGetSession(w ErrorResponseWriter, r *http.Reques
 	log.V(1).Info("Getting session from database")
 	session, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID)
 	if err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Session not found", err))
+		RespondNotFoundOrError(w, "Session not found", err)
 		return
 	}
 
@@ -333,7 +333,7 @@ func (h *SessionsHandler) HandleUpdateSession(w ErrorResponseWriter, r *http.Req
 
 	session, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID)
 	if err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Session not found", err))
+		RespondNotFoundOrError(w, "Session not found", err)
 		return
 	}
 
@@ -344,7 +344,7 @@ func (h *SessionsHandler) HandleUpdateSession(w ErrorResponseWriter, r *http.Req
 		log = log.WithValues("agentRef", *sessionRequest.AgentRef)
 		agent, err := h.DatabaseService.GetAgent(r.Context(), utils.ConvertToPythonIdentifier(*sessionRequest.AgentRef))
 		if err != nil {
-			w.RespondWithError(errors.NewNotFoundError("Agent not found", err))
+			RespondNotFoundOrError(w, "Agent not found", err)
 			return
 		}
 		session.AgentID = &agent.ID
@@ -426,7 +426,7 @@ func (h *SessionsHandler) HandleListTasksForSession(w ErrorResponseWriter, r *ht
 	// Verify session exists
 	_, err = h.DatabaseService.GetSession(r.Context(), sessionID, userID)
 	if err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Session not found for given ID", err))
+		RespondNotFoundOrError(w, "Session not found for given ID", err)
 		return
 	}
 
@@ -499,7 +499,7 @@ func (h *SessionsHandler) HandleAddEventToSession(w ErrorResponseWriter, r *http
 	// Get session to verify it exists
 	session, err := h.DatabaseService.GetSession(r.Context(), sessionID, userID)
 	if err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Session not found", err))
+		RespondNotFoundOrError(w, "Session not found", err)
 		return
 	}
 

go/core/internal/httpserver/handlers/tasks.go

@@ -43,7 +43,7 @@ func (h *TasksHandler) HandleGetTask(w ErrorResponseWriter, r *http.Request) {
 
 	task, err := h.DatabaseService.GetTask(r.Context(), taskID, userID)
 	if err != nil {
-		w.RespondWithError(errors.NewNotFoundError("Task not found", err))
+		RespondNotFoundOrError(w, "Task not found", err)
 		return
 	}
 	wireVersion, err := utils.NegotiateA2AWireVersion(r)

go/core/internal/httpserver/middleware.go

@@ -9,7 +9,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jackc/pgx/v5"
+	"github.com/kagent-dev/kagent/go/api/database"
 	"github.com/kagent-dev/kagent/go/core/internal/httpserver/handlers"
 	"github.com/kagent-dev/kagent/go/core/pkg/auth"
 	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
@@ -111,7 +111,7 @@ func (s *HTTPServer) shareTokenMiddleware(next http.Handler) http.Handler {
 
 		share, err := s.config.DbClient.GetSessionShareByToken(r.Context(), token)
 		if err != nil {
-			if errors.Is(err, pgx.ErrNoRows) {
+			if errors.Is(err, database.ErrNotFound) {
 				http.Error(w, "Invalid or expired share token", http.StatusForbidden)
 			} else {
 				http.Error(w, "Internal server error", http.StatusInternalServerError)

go/core/internal/httpserver/middleware_error.go

@@ -8,7 +8,7 @@ import (
 	"net"
 	"net/http"
 
-	"github.com/jackc/pgx/v5"
+	"github.com/kagent-dev/kagent/go/api/database"
 	apierrors "github.com/kagent-dev/kagent/go/core/internal/httpserver/errors"
 	"github.com/kagent-dev/kagent/go/core/internal/httpserver/handlers"
 	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
@@ -67,7 +67,7 @@ func (w *errorResponseWriter) RespondWithError(err error) {
 		underlying = err
 	}
 
-	if underlying != nil && !errors.Is(underlying, pgx.ErrNoRows) {
+	if underlying != nil && !errors.Is(underlying, database.ErrNotFound) {
 		log.Error(underlying, message)
 	} else {
 		log.Info(message)
@@ -76,7 +76,9 @@ func (w *errorResponseWriter) RespondWithError(err error) {
 	w.Header().Set("Content-Type", "application/json")
 	w.WriteHeader(statusCode)
 	errMsg := message
-	if underlying != nil {
+	// 5xx bodies carry only the generic message: the underlying error is
+	// backend-internal detail (already logged above) and must not reach clients.
+	if underlying != nil && statusCode < http.StatusInternalServerError {
 		errMsg = message + ": " + underlying.Error()
 	}
 	json.NewEncoder(w).Encode(map[string]string{"error": errMsg}) //nolint:errcheck