From 9be7d57fcb8ea1a8409c7d058aac73a86cb8361a Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Thu, 16 Jul 2026 21:34:17 +0200 Subject: [PATCH 1/3] refactor: hide pgx errors behind database.ErrNotFound Single-row reads in the postgres client map pgx.ErrNoRows to an exported database.ErrNotFound sentinel, wrapped with call-site context. Consumers outside the database package (share-token middleware, error-response logging, sandbox session materialization) now match on the sentinel, so pgx no longer leaks through the database.Client interface. Signed-off-by: QuentinBisson --- go/api/database/errors.go | 8 +++++ .../a2a/substrate_sandbox_transport.go | 3 +- .../a2a/substrate_sandbox_transport_test.go | 3 +- go/core/internal/database/client_postgres.go | 23 +++++++++----- go/core/internal/database/client_test.go | 30 +++++++++++++++++++ go/core/internal/httpserver/middleware.go | 4 +-- .../internal/httpserver/middleware_error.go | 4 +-- .../server_share_middleware_test.go | 7 ++--- 8 files changed, 63 insertions(+), 19 deletions(-) create mode 100644 go/api/database/errors.go 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/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..ee2519c132 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) 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) From ff43b38020b140806bfad1ed2a6614bfb99507bc Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Thu, 16 Jul 2026 21:45:20 +0200 Subject: [PATCH 2/3] fix: gate REST not-found responses on database.ErrNotFound Session, session-share, and task handlers mapped every database read error to 404, so a backend outage read as "not found" to API clients that treat 404 as a definitive signal. RespondNotFoundOrError writes 404 only for database.ErrNotFound and 500 for anything else. Signed-off-by: QuentinBisson --- .../internal/httpserver/handlers/helpers.go | 13 ++++++++ .../handlers/helpers_notfound_test.go | 31 +++++++++++++++++++ .../httpserver/handlers/session_shares.go | 6 ++-- .../internal/httpserver/handlers/sessions.go | 12 +++---- go/core/internal/httpserver/handlers/tasks.go | 2 +- 5 files changed, 54 insertions(+), 10 deletions(-) create mode 100644 go/core/internal/httpserver/handlers/helpers_notfound_test.go 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) From 8dab542e3df2a93bb73ceb662830b127d7e6317f Mon Sep 17 00:00:00 2001 From: QuentinBisson Date: Thu, 16 Jul 2026 22:04:50 +0200 Subject: [PATCH 3/3] fix: keep backend error detail out of 5xx response bodies RespondWithError echoed the underlying error into the JSON body for every status code, so 500s exposed backend internals such as database connection errors. 5xx bodies now carry only the generic message; the underlying error stays in the server log. Signed-off-by: QuentinBisson --- go/core/internal/httpserver/middleware_error.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/go/core/internal/httpserver/middleware_error.go b/go/core/internal/httpserver/middleware_error.go index ee2519c132..8c2f79e79b 100644 --- a/go/core/internal/httpserver/middleware_error.go +++ b/go/core/internal/httpserver/middleware_error.go @@ -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