From 5833378794bf15c267fd1231b3b09da621d4e17a Mon Sep 17 00:00:00 2001
From: Haitao Pan <manbuzhe2009@qq.com>
Date: Tue, 2 Jun 2026 16:21:17 +0800
Subject: [PATCH] Filter raw gateway SSE events for direct requests

---
 internal/acp/http_handler.go      | 39 ++++++++++++++-----------
 internal/acp/web_contract_test.go | 48 +++++++++++++++++++++++++++++++
 2 files changed, 71 insertions(+), 16 deletions(-)

diff --git a/internal/acp/http_handler.go b/internal/acp/http_handler.go
index 80d39e1..f4d9400 100644
--- a/internal/acp/http_handler.go
+++ b/internal/acp/http_handler.go
@@ -213,20 +213,22 @@ func (s *Server) handleRPCWithTransform(
 		if !stream {
 			return
 		}
-		if openClawGatewayTask {
-			if reason := openClawGatewayNotificationDropReason(message); reason != "" {
-				log.Printf(
-					"level=warn component=acp_sse event=notification_dropped path=%q rpcMethod=%q requestId=%q sessionId=%q threadId=%q reason=%q notificationMethod=%q",
-					r.URL.Path,
-					request.Method,
-					fmt.Sprint(request.ID),
-					shared.StringArg(request.Params, "sessionId", ""),
-					shared.StringArg(request.Params, "threadId", ""),
-					reason,
-					shared.StringArg(message, "method", ""),
-				)
-				return
-			}
+		if reason := gatewaySSEBridgeNotificationDropReason(
+			request.Method,
+			openClawGatewayTask,
+			message,
+		); reason != "" {
+			log.Printf(
+				"level=warn component=acp_sse event=notification_dropped path=%q rpcMethod=%q requestId=%q sessionId=%q threadId=%q reason=%q notificationMethod=%q",
+				r.URL.Path,
+				request.Method,
+				fmt.Sprint(request.ID),
+				shared.StringArg(request.Params, "sessionId", ""),
+				shared.StringArg(request.Params, "threadId", ""),
+				reason,
+				shared.StringArg(message, "method", ""),
+			)
+			return
 		}
 		streamWriter.write(message)
 	}
@@ -281,9 +283,14 @@ func (s *Server) handleRPCWithTransform(
 	_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, response))
 }
 
-func openClawGatewayNotificationDropReason(message map[string]any) string {
+func gatewaySSEBridgeNotificationDropReason(
+	rpcMethod string,
+	openClawGatewayTask bool,
+	message map[string]any,
+) string {
 	method := strings.TrimSpace(shared.StringArg(message, "method", ""))
-	if strings.HasPrefix(method, "xworkmate.gateway.") {
+	if strings.HasPrefix(method, "xworkmate.gateway.") &&
+		(openClawGatewayTask || strings.TrimSpace(rpcMethod) == "xworkmate.gateway.request") {
 		return "raw_gateway_event"
 	}
 	if !openClawGatewayNotificationWithinLimit(message) {
diff --git a/internal/acp/web_contract_test.go b/internal/acp/web_contract_test.go
index f12949e..206382d 100644
--- a/internal/acp/web_contract_test.go
+++ b/internal/acp/web_contract_test.go
@@ -834,6 +834,54 @@ func TestHTTPHandlerTasksGetReturnsCompletedOpenClawResult(t *testing.T) {
 	}
 }
 
+func TestGatewayRequestSSEFiltersRawGatewayEvents(t *testing.T) {
+	tests := []struct {
+		name                string
+		rpcMethod           string
+		openClawGatewayTask bool
+		notificationMethod  string
+		wantDropReason      string
+	}{
+		{
+			name:               "direct gateway request raw push",
+			rpcMethod:          "xworkmate.gateway.request",
+			notificationMethod: "xworkmate.gateway.push",
+			wantDropReason:     "raw_gateway_event",
+		},
+		{
+			name:                "openclaw task raw push",
+			rpcMethod:           "session.message",
+			openClawGatewayTask: true,
+			notificationMethod:  "xworkmate.gateway.snapshot",
+			wantDropReason:      "raw_gateway_event",
+		},
+		{
+			name:               "ordinary session notification",
+			rpcMethod:          "session.message",
+			notificationMethod: "session.update",
+			wantDropReason:     "",
+		},
+		{
+			name:               "non openclaw gateway raw push remains scoped out",
+			rpcMethod:          "session.message",
+			notificationMethod: "xworkmate.gateway.push",
+			wantDropReason:     "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := gatewaySSEBridgeNotificationDropReason(
+				tt.rpcMethod,
+				tt.openClawGatewayTask,
+				map[string]any{"method": tt.notificationMethod},
+			)
+			if got != tt.wantDropReason {
+				t.Fatalf("expected drop reason %q, got %q", tt.wantDropReason, got)
+			}
+		})
+	}
+}
+
 func TestSafeSSEStreamDropsLateNotificationsAfterClose(t *testing.T) {
 	writer := &panicSSEWriter{header: http.Header{}}
 	stream := newSafeSSEStream(context.Background(), writer, safeSSEStreamMeta{})