diff --git a/go/api/database/errors.go b/go/api/database/errors.go new file mode 100644 index 0000000000..aee9734c8c --- /dev/null +++ b/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") diff --git a/go/core/internal/a2a/substrate_sandbox_transport.go b/go/core/internal/a2a/substrate_sandbox_transport.go index b4d8108de9..8d9ed64d6d 100644 --- a/go/core/internal/a2a/substrate_sandbox_transport.go +++ b/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) diff --git a/go/core/internal/a2a/substrate_sandbox_transport_test.go b/go/core/internal/a2a/substrate_sandbox_transport_test.go index a4e31dabb2..6edd7ad6aa 100644 --- a/go/core/internal/a2a/substrate_sandbox_transport_test.go +++ b/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) } }) diff --git a/go/core/internal/database/client_postgres.go b/go/core/internal/database/client_postgres.go index 4e0bdc66d9..eea5454ce5 100644 --- a/go/core/internal/database/client_postgres.go +++ b/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 @@ -292,7 +301,7 @@ func (c *postgresClient) StoreTask(ctx context.Context, task *a2a.Task) error { func (c *postgresClient) GetTask(ctx context.Context, taskID string) (*a2a.Task, error) { row, err := c.q.GetTask(ctx, taskID) 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) } @@ -338,7 +347,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) } @@ -393,7 +402,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 } @@ -450,7 +459,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 } diff --git a/go/core/internal/database/client_test.go b/go/core/internal/database/client_test.go index a8c263d7de..fd7db259bd 100644 --- a/go/core/internal/database/client_test.go +++ b/go/core/internal/database/client_test.go @@ -781,3 +781,33 @@ 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() + + 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"); 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 }}, + } + + 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) + }) + } +} diff --git a/go/core/internal/httpserver/handlers/helpers.go b/go/core/internal/httpserver/handlers/helpers.go index 2afedc7119..f69a71fe49 100644 --- a/go/core/internal/httpserver/handlers/helpers.go +++ b/go/core/internal/httpserver/handlers/helpers.go @@ -3,6 +3,7 @@ package handlers import ( "context" "encoding/json" + stderrors "errors" "fmt" "net/http" "reflect" @@ -10,6 +11,7 @@ import ( "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") diff --git a/go/core/internal/httpserver/handlers/helpers_notfound_test.go b/go/core/internal/httpserver/handlers/helpers_notfound_test.go new file mode 100644 index 0000000000..f38e7c84f1 --- /dev/null +++ b/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) + }) + } +} diff --git a/go/core/internal/httpserver/handlers/session_shares.go b/go/core/internal/httpserver/handlers/session_shares.go index 5a7a7b97c7..e69133fa55 100644 --- a/go/core/internal/httpserver/handlers/session_shares.go +++ b/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 } diff --git a/go/core/internal/httpserver/handlers/sessions.go b/go/core/internal/httpserver/handlers/sessions.go index 63935605ed..2db363225f 100644 --- a/go/core/internal/httpserver/handlers/sessions.go +++ b/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 } diff --git a/go/core/internal/httpserver/handlers/tasks.go b/go/core/internal/httpserver/handlers/tasks.go index 7993794d14..c626d58a88 100644 --- a/go/core/internal/httpserver/handlers/tasks.go +++ b/go/core/internal/httpserver/handlers/tasks.go @@ -35,7 +35,7 @@ func (h *TasksHandler) HandleGetTask(w ErrorResponseWriter, r *http.Request) { task, err := h.DatabaseService.GetTask(r.Context(), taskID) if err != nil { - w.RespondWithError(errors.NewNotFoundError("Task not found", err)) + RespondNotFoundOrError(w, "Task not found", err) return } wireVersion, err := utils.NegotiateA2AWireVersion(r) diff --git a/go/core/internal/httpserver/middleware.go b/go/core/internal/httpserver/middleware.go index fb190da386..cf7a86dd87 100644 --- a/go/core/internal/httpserver/middleware.go +++ b/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) diff --git a/go/core/internal/httpserver/middleware_error.go b/go/core/internal/httpserver/middleware_error.go index d8abcc2e14..8c2f79e79b 100644 --- a/go/core/internal/httpserver/middleware_error.go +++ b/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 diff --git a/go/core/internal/httpserver/server_share_middleware_test.go b/go/core/internal/httpserver/server_share_middleware_test.go index ba4d0b27a2..07fe58d3d0 100644 --- a/go/core/internal/httpserver/server_share_middleware_test.go +++ b/go/core/internal/httpserver/server_share_middleware_test.go @@ -6,7 +6,6 @@ import ( "net/http/httptest" "testing" - "github.com/jackc/pgx/v5" dbpkg "github.com/kagent-dev/kagent/go/api/database" authimpl "github.com/kagent-dev/kagent/go/core/internal/httpserver/auth" "github.com/kagent-dev/kagent/go/core/pkg/auth" @@ -87,7 +86,7 @@ func TestShareTokenMiddleware(t *testing.T) { { name: "invalid token returns 403", getShare: func(_ context.Context, _ string) (*dbpkg.SessionShare, error) { - return nil, pgx.ErrNoRows + return nil, dbpkg.ErrNotFound }, buildReq: func() *http.Request { r := httptest.NewRequest(http.MethodGet, "/api/sessions/sess-1", nil) @@ -98,11 +97,11 @@ func TestShareTokenMiddleware(t *testing.T) { wantShareCtx: false, }, { - // Revocation deletes the session_share row; subsequent lookups return pgx.ErrNoRows, + // Revocation deletes the session_share row; subsequent lookups return database.ErrNotFound, // so revoked tokens are rejected immediately — no grace period. name: "revoked token returns 403", getShare: func(_ context.Context, _ string) (*dbpkg.SessionShare, error) { - return nil, pgx.ErrNoRows + return nil, dbpkg.ErrNotFound }, buildReq: func() *http.Request { r := httptest.NewRequest(http.MethodGet, "/api/sessions/sess-1", nil)