ai-workspace-lab/xworkmate-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Filter raw gateway SSE events for direct requests

Browse Source
This commit is contained in:
Haitao Pan 2026-06-02 16:21:17 +08:00
parent 1874b28bc1
commit 5833378794
2 changed files with 71 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
internal/acp/http_handler.go
View File

@ -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) {

48
internal/acp/web_contract_test.go
View File

@ -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{})