feat(pii): apply PII middleware to the Open Responses API

core/http/endpoints/openresponses/websocket.go

@@ -16,6 +16,7 @@ import (
 	"github.com/mudler/LocalAI/core/config"
 	"github.com/mudler/LocalAI/core/http/middleware"
 	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/core/services/routing/pii"
 	"github.com/mudler/LocalAI/core/templates"
 	"github.com/mudler/LocalAI/pkg/functions"
 	"github.com/mudler/LocalAI/pkg/model"
@@ -72,6 +73,7 @@ func WebSocketEndpoint(application *application.Application) echo.HandlerFunc {
 	ml := application.ModelLoader()
 	evaluator := application.TemplatesEvaluator()
 	appConfig := application.ApplicationConfig()
+	piiPolicy := application.PIIPolicyResolver()
 
 	return func(c echo.Context) error {
 		owner := ownerFromContext(c)
@@ -96,13 +98,13 @@ func WebSocketEndpoint(application *application.Application) echo.HandlerFunc {
 
 		xlog.Debug("WebSocket Responses connection established", "address", ws.RemoteAddr().String())
 
-		handleWebSocketConnection(connCtx, conn, owner, cl, ml, evaluator, appConfig)
+		handleWebSocketConnection(connCtx, conn, owner, cl, ml, evaluator, appConfig, piiPolicy)
 		return nil
 	}
 }
 
 // handleWebSocketConnection runs the read loop for a single WebSocket connection.
-func handleWebSocketConnection(connCtx context.Context, conn *lockedConn, owner string, cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator *templates.Evaluator, appConfig *config.ApplicationConfig) {
+func handleWebSocketConnection(connCtx context.Context, conn *lockedConn, owner string, cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator *templates.Evaluator, appConfig *config.ApplicationConfig, piiPolicy pii.PolicyResolver) {
 	// Responses created with store=false remain available only to this WebSocket
 	// connection, matching the Responses WebSocket continuation contract without
 	// leaking zero-data-retention state into the process-wide response store.
@@ -172,7 +174,7 @@ func handleWebSocketConnection(connCtx context.Context, conn *lockedConn, owner
 				releaseOnce.Do(inflight.Unlock)
 			}
 			defer release()
-			handleWSResponseCreate(connCtx, conn, connectionStore, release, owner, wsMsg.Generate, &wsMsg.OpenResponsesRequest, cl, ml, evaluator, appConfig)
+			handleWSResponseCreate(connCtx, conn, connectionStore, release, owner, wsMsg.Generate, &wsMsg.OpenResponsesRequest, cl, ml, evaluator, appConfig, piiPolicy)
 		}()
 	}
 }
@@ -181,7 +183,7 @@ func handleWebSocketConnection(connCtx context.Context, conn *lockedConn, owner
 // It reuses the existing background stream infrastructure: the request is processed via
 // handleBackgroundStream which buffers events into the store, and a forwarder goroutine
 // reads those events and sends them over the WebSocket.
-func handleWSResponseCreate(connCtx context.Context, conn *lockedConn, connectionStore *ResponseStore, release func(), owner string, generate *bool, input *schema.OpenResponsesRequest, cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator *templates.Evaluator, appConfig *config.ApplicationConfig) {
+func handleWSResponseCreate(connCtx context.Context, conn *lockedConn, connectionStore *ResponseStore, release func(), owner string, generate *bool, input *schema.OpenResponsesRequest, cl *config.ModelConfigLoader, ml *model.ModelLoader, evaluator *templates.Evaluator, appConfig *config.ApplicationConfig, piiPolicy pii.PolicyResolver) {
 	createdAt := time.Now().Unix()
 	responseID := fmt.Sprintf("resp_%s", uuid.New().String())
 	fail := func(errType, message, param string) {
@@ -213,6 +215,19 @@ func handleWSResponseCreate(connCtx context.Context, conn *lockedConn, connectio
 		return
 	}
 
+	// The PII request middleware is HTTP middleware and never sees messages
+	// arriving over this socket. Rather than serve a PII-enabled model
+	// unfiltered (the same content would be blocked or masked on
+	// POST /v1/responses), refuse it here: fail closed, as the middleware
+	// does when its detector is unavailable.
+	if piiPolicy != nil {
+		if enabled, detectors := piiPolicy(cfg); enabled && len(detectors) > 0 {
+			xlog.Warn("WebSocket Responses: refusing PII-enabled model, PII filtering is not applied on the WebSocket transport", "model", input.Model)
+			fail("pii_unsupported_transport", "PII filtering is enabled for this model and is not available over the Responses WebSocket; use POST /v1/responses", "model")
+			return
+		}
+	}
+
 	// Set up context with cancellation tied to connection lifetime
 	reqCtx, reqCancel := context.WithCancel(connCtx)
 	defer reqCancel()

core/http/endpoints/openresponses/websocket_pii_test.go

@@ -0,0 +1,79 @@
+// SPDX-License-Identifier: MIT
+package openresponses
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/core/services/routing/pii"
+	"github.com/mudler/LocalAI/pkg/system"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// The PII request middleware is HTTP middleware, so messages arriving over
+// the Responses WebSocket never pass through it. These specs drive the real
+// socket loop and prove a PII-enabled model is refused there instead of
+// being served unfiltered.
+var _ = Describe("WebSocket Responses PII guard", func() {
+	// dial runs the real connection loop behind a test server with the given
+	// policy and returns the first event answering one prewarm request
+	// (generate=false, so no backend is needed).
+	dial := func(policy pii.PolicyResolver) schema.ORStreamEvent {
+		GinkgoHelper()
+		dir := GinkgoT().TempDir()
+		appCfg := config.NewApplicationConfig()
+		appCfg.SystemState = &system.SystemState{Model: system.Model{ModelsPath: dir}}
+		cl := config.NewModelConfigLoader(dir)
+		m := config.ModelConfig{Name: "m", Backend: "test"}
+		m.Model = "m.gguf"
+		cl.ReplaceModelConfigs([]config.ModelConfig{m})
+
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			ws, err := wsUpgrader.Upgrade(w, r, nil)
+			if err != nil {
+				return
+			}
+			defer ws.Close()
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			handleWebSocketConnection(ctx, &lockedConn{Conn: ws}, "owner", cl, nil, nil, appCfg, policy)
+		}))
+		DeferCleanup(srv.Close)
+
+		conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http"), nil)
+		Expect(err).ToNot(HaveOccurred())
+		DeferCleanup(conn.Close)
+		Expect(conn.SetReadDeadline(time.Now().Add(10 * time.Second))).To(Succeed())
+		Expect(conn.WriteMessage(websocket.TextMessage,
+			[]byte(`{"type":"response.create","generate":false,"store":false,"model":"m","input":"my key is sk-secret"}`))).To(Succeed())
+
+		var ev schema.ORStreamEvent
+		Expect(conn.ReadJSON(&ev)).To(Succeed())
+		return ev
+	}
+
+	It("refuses a model whose PII policy is enabled", func() {
+		ev := dial(func(any) (bool, []string) { return true, []string{"det"} })
+		Expect(ev.Type).To(Equal("error"))
+		Expect(ev.Error).ToNot(BeNil())
+		Expect(ev.Error.Type).To(Equal("pii_unsupported_transport"))
+		Expect(ev.Error.Message).ToNot(ContainSubstring("sk-secret"), "the refusal must not echo the request content")
+	})
+
+	It("serves a model without a PII policy", func() {
+		ev := dial(func(any) (bool, []string) { return false, nil })
+		Expect(ev.Type).To(Equal("response.created"))
+	})
+
+	It("serves a PII-enabled model with no detectors, as the HTTP middleware does", func() {
+		ev := dial(func(any) (bool, []string) { return true, nil })
+		Expect(ev.Type).To(Equal("response.created"))
+	})
+})

core/http/routes/openresponses.go

@@ -9,6 +9,8 @@ import (
 	"github.com/mudler/LocalAI/core/http/endpoints/openresponses"
 	"github.com/mudler/LocalAI/core/http/middleware"
 	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/core/services/routing/pii"
+	"github.com/mudler/LocalAI/core/services/routing/piiadapter"
 	"github.com/mudler/xlog"
 )
 
@@ -41,16 +43,7 @@ func RegisterOpenResponsesRoutes(app *echo.Echo,
 		agentControl,
 	)
 
-	responsesMiddleware := []echo.MiddlewareFunc{
-		// Intercept requests where the model name matches an agent — route directly
-		// to the agent pool without going through the model config resolution pipeline.
-		localai.AgentResponsesInterceptor(application),
-		middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser()),
-		middleware.TraceMiddleware(application),
-		re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_CHAT)),
-		re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenResponsesRequest) }),
-		setOpenResponsesRequestContext(re),
-	}
+	responsesMiddleware := openResponsesMiddleware(re, application)
 
 	// Main Open Responses endpoint
 	app.POST("/v1/responses", responsesHandler, responsesMiddleware...)
@@ -74,6 +67,28 @@ func RegisterOpenResponsesRoutes(app *echo.Echo,
 	app.POST("/responses/:id/cancel", cancelResponseHandler, middleware.TraceMiddleware(application))
 }
 
+// openResponsesMiddleware is the middleware chain of POST /v1/responses and
+// POST /responses. It is a separate function so tests can drive the real
+// chain in front of a stub handler.
+func openResponsesMiddleware(re *middleware.RequestExtractor, application *application.Application) []echo.MiddlewareFunc {
+	return []echo.MiddlewareFunc{
+		// Intercept requests where the model name matches an agent — route directly
+		// to the agent pool without going through the model config resolution pipeline.
+		localai.AgentResponsesInterceptor(application),
+		middleware.UsageMiddleware(application.StatsRecorder(), application.FallbackUser()),
+		middleware.TraceMiddleware(application),
+		re.BuildFilteredFirstAvailableDefaultModel(config.BuildUsecaseFilterFn(config.FLAG_CHAT)),
+		re.SetModelAndConfig(func() schema.LocalAIRequest { return new(schema.OpenResponsesRequest) }),
+		setOpenResponsesRequestContext(re),
+		// PII filtering runs once the model config is resolved and merged
+		// (it gates on the served model's pii block), and before the handler
+		// turns the input items into prompt messages. Without it, content
+		// that a model's PII policy blocks or masks on /v1/chat/completions
+		// would reach the same model unfiltered through this endpoint.
+		pii.RequestMiddleware(application.PIIRedactor(), application.PIIEvents(), piiadapter.OpenResponses(), application.FallbackUser(), pii.WithNERResolver(application.PIINERResolver()), pii.WithPolicyResolver(application.PIIPolicyResolver())),
+	}
+}
+
 // setOpenResponsesRequestContext sets up the context and cancel function for Open Responses requests
 func setOpenResponsesRequestContext(re *middleware.RequestExtractor) echo.MiddlewareFunc {
 	return func(next echo.HandlerFunc) echo.HandlerFunc {

core/http/routes/openresponses_pii_test.go

@@ -0,0 +1,136 @@
+// SPDX-License-Identifier: MIT
+package routes
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/labstack/echo/v4"
+	"github.com/mudler/LocalAI/core/application"
+	"github.com/mudler/LocalAI/core/config"
+	"github.com/mudler/LocalAI/core/http/middleware"
+	"github.com/mudler/LocalAI/core/schema"
+	"github.com/mudler/LocalAI/pkg/system"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// Drives the real POST /v1/responses middleware chain (openResponsesMiddleware)
+// in front of a stub handler, with an in-process pattern detector, to prove
+// the PII policy that guards /v1/chat/completions also guards this endpoint.
+var _ = Describe("Open Responses PII filtering", func() {
+	var (
+		e        *echo.Echo
+		cancel   context.CancelFunc
+		reached  bool
+		received *schema.OpenResponsesRequest
+	)
+
+	BeforeEach(func() {
+		tmp := GinkgoT().TempDir()
+		modelsDir := filepath.Join(tmp, "models")
+		Expect(os.MkdirAll(modelsDir, 0o755)).To(Succeed())
+		st, err := system.GetSystemState(
+			system.WithModelPath(modelsDir),
+			system.WithBackendPath(filepath.Join(tmp, "backends")),
+		)
+		Expect(err).ToNot(HaveOccurred())
+
+		var ctx context.Context
+		ctx, cancel = context.WithCancel(context.Background())
+		app, err := application.New(config.WithContext(ctx), config.WithSystemState(st))
+		Expect(err).ToNot(HaveOccurred())
+
+		// The pattern tier runs in-process; no backend is loaded.
+		files := map[string]string{
+			"secret-filter.yaml": `name: secret-filter
+backend: pattern
+pii_detection:
+  default_action: mask
+  patterns:
+    - name: SECRET
+      match: "sk-test-[A-Za-z0-9]+"
+      action: block
+    - name: TOKEN
+      match: "tok-[A-Za-z0-9]+"
+      action: mask
+`,
+			"guarded.yaml": `name: guarded
+backend: test
+pii:
+  enabled: true
+  detectors: [secret-filter]
+`,
+		}
+		for name, body := range files {
+			p := filepath.Join(modelsDir, name)
+			Expect(os.WriteFile(p, []byte(body), 0o644)).To(Succeed())
+			Expect(app.ModelConfigLoader().ReadModelConfig(p)).To(Succeed())
+		}
+
+		re := middleware.NewRequestExtractor(app.ModelConfigLoader(), app.ModelLoader(), app.ApplicationConfig())
+		reached, received = false, nil
+		e = echo.New()
+		e.POST("/v1/responses", func(c echo.Context) error {
+			reached = true
+			received = c.Get(middleware.CONTEXT_LOCALS_KEY_LOCALAI_REQUEST).(*schema.OpenResponsesRequest)
+			return c.NoContent(http.StatusNoContent)
+		}, openResponsesMiddleware(re, app)...)
+	})
+
+	AfterEach(func() { cancel() })
+
+	post := func(body string) *httptest.ResponseRecorder {
+		req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(body))
+		req.Header.Set("Content-Type", "application/json")
+		rec := httptest.NewRecorder()
+		e.ServeHTTP(rec, req)
+		return rec
+	}
+
+	It("blocks a secret in a string input with pii_blocked", func() {
+		rec := post(`{"model":"guarded","input":"my key is sk-test-ABC123"}`)
+		Expect(rec.Code).To(Equal(http.StatusBadRequest))
+		Expect(rec.Body.String()).To(ContainSubstring("pii_blocked"))
+		Expect(rec.Body.String()).ToNot(ContainSubstring("sk-test-ABC123"), "the block response must not echo the secret")
+		Expect(reached).To(BeFalse(), "a blocked request must not reach the handler")
+	})
+
+	It("blocks a secret hidden in a content part, instructions or tool output", func() {
+		for _, body := range []string{
+			`{"model":"guarded","input":[{"type":"message","role":"user","content":[{"type":"input_text","text":"key sk-test-ABC123"}]}]}`,
+			`{"model":"guarded","instructions":"use sk-test-ABC123","input":"hi"}`,
+			`{"model":"guarded","input":[{"type":"function_call_output","call_id":"c1","output":"sk-test-ABC123"}]}`,
+			`{"model":"guarded","input":[{"type":"function_call_output","call_id":"c1","output":["sk-test-ABC123"]}]}`,
+			`{"model":"guarded","input":[{"type":"function_call_output","call_id":"c1","output":[{"token":"sk-test-ABC123"}]}]}`,
+		} {
+			reached = false
+			rec := post(body)
+			Expect(rec.Code).To(Equal(http.StatusBadRequest), body)
+			Expect(rec.Body.String()).To(ContainSubstring("pii_blocked"), body)
+			Expect(reached).To(BeFalse(), body)
+		}
+	})
+
+	It("masks a maskable value before the handler sees it", func() {
+		rec := post(`{"model":"guarded","input":[{"type":"message","role":"user","content":"token tok-XYZ789 please"}]}`)
+		Expect(rec.Code).To(Equal(http.StatusNoContent))
+		Expect(reached).To(BeTrue())
+		content := received.Input.([]any)[0].(map[string]any)["content"].(string)
+		Expect(content).ToNot(ContainSubstring("tok-XYZ789"))
+		Expect(content).To(HavePrefix("token "))
+		Expect(content).To(HaveSuffix(" please"))
+	})
+
+	It("passes clean input through unchanged", func() {
+		rec := post(`{"model":"guarded","instructions":"be brief","input":"what is the capital of France?"}`)
+		Expect(rec.Code).To(Equal(http.StatusNoContent))
+		Expect(reached).To(BeTrue())
+		Expect(received.Input).To(Equal("what is the capital of France?"))
+		Expect(received.Instructions).To(Equal("be brief"))
+	})
+})