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

fix: filter OpenClaw gateway SSE notifications

Browse Source
This commit is contained in:
Haitao Pan 2026-05-10 13:55:16 +08:00
parent ed04d91dad
commit 6ad48d4c26
4 changed files with 221 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
internal/acp/http_handler.go
View File

@ -19,6 +19,8 @@ import (
var httpSSEKeepaliveInterval = 20 * time.Second
const openClawGatewayMaxNotificationBytes = 64 * 1024
func (s *Server) Handler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
@ -209,6 +211,21 @@ func (s *Server) handleRPCWithTransform(
if !stream {
return
}
if r.URL.Path == "/gateway/openclaw" {
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
}
}
streamWriter.write(message)
}
if stream {
@ -262,6 +279,28 @@ func (s *Server) handleRPCWithTransform(
_ = json.NewEncoder(w).Encode(shared.ResultEnvelope(request.ID, response))
}
func openClawGatewayNotificationDropReason(message map[string]any) string {
method := strings.TrimSpace(shared.StringArg(message, "method", ""))
if strings.HasPrefix(method, "xworkmate.gateway.") {
return "raw_gateway_event"
}
if !openClawGatewayNotificationWithinLimit(message) {
return "oversized"
}
return ""
}
func openClawGatewayNotificationWithinLimit(message map[string]any) bool {
if message == nil {
return true
}
body, err := json.Marshal(message)
if err != nil {
return false
}
return len(body) <= openClawGatewayMaxNotificationBytes
}
type safeSSEStream struct {
ctx context.Context
w http.ResponseWriter

46
internal/acp/orchestrator.go
View File

@ -182,10 +182,18 @@ func (o *SessionOrchestrator) runOpenClawGatewayChat(
notify func(map[string]any),
) (map[string]any, *shared.RPCError) {
collector := newOpenClawChatCollector()
sessionID := strings.TrimSpace(shared.StringArg(params, "sessionId", ""))
threadID := strings.TrimSpace(shared.StringArg(params, "threadId", sessionID))
if sessionID == "" {
sessionID = threadID
}
notifyWithCollection := func(message map[string]any) {
collector.observe(message)
if notify != nil {
notify(message)
if notify == nil {
return
}
if update := openClawGatewaySessionUpdate(message, sessionID, threadID, turnID); update != nil {
notify(update)
}
}
artifactDeliveryRequired := openClawArtifactDeliveryRequired(params)
@ -882,6 +890,40 @@ func (c *openClawChatCollector) artifactPayload() map[string]any {
return result
}
func openClawGatewaySessionUpdate(notification map[string]any, sessionID string, threadID string, turnID string) map[string]any {
params := shared.AsMap(notification["params"])
event := shared.AsMap(params["event"])
if strings.TrimSpace(shared.StringArg(event, "event", "")) != "chat.run" {
return nil
}
payload := shared.AsMap(event["payload"])
text := firstNonEmptyString(payload, "assistantText", "text", "message", "output", "summary")
if text == "" {
return nil
}
update := map[string]any{
"sessionId": sessionID,
"threadId": threadID,
"turnId": turnID,
"type": "delta",
"event": "delta",
"delta": text,
"text": text,
"pending": true,
"error": false,
}
if isTerminalGatewayPayload(payload) {
update["type"] = "status"
update["event"] = "completed"
update["message"] = text
update["pending"] = false
if strings.EqualFold(strings.TrimSpace(shared.StringArg(payload, "state", "")), "error") {
update["error"] = true
}
}
return shared.NotificationEnvelope("session.update", update)
}
func hasArtifactPayload(payload map[string]any) bool {
if len(payload) == 0 {
return false

29
internal/acp/routing_test.go
View File

@ -1643,6 +1643,8 @@ type acpFakeOpenClawGateway struct {
closeNextChatSend atomic.Bool
alwaysCloseChatSend atomic.Bool
agentWaitDelayMs atomic.Int64
largeGatewayPayloadBytes atomic.Int64
emitAgentDelta atomic.Bool
lastConnectClient atomic.Value
lastArtifactExportParams atomic.Value
lastAgentWaitParams atomic.Value
@ -1831,10 +1833,35 @@ func newAcpFakeOpenClawGateway(t *testing.T) *acpFakeOpenClawGateway {
if strings.Contains(fake.runMessage(runID), "hallucinate-files") {
message = "文件已就绪，点击直接下载👇 三个格式一键收取："
}
if payloadBytes := fake.largeGatewayPayloadBytes.Load(); payloadBytes > 0 {
_ = conn.WriteJSON(map[string]any{
"type": "event",
"event": "health",
"seq": 1,
"payload": map[string]any{
"status": "ok",
"blob": strings.Repeat("x", int(payloadBytes)),
},
})
}
if fake.emitAgentDelta.Load() {
_ = conn.WriteJSON(map[string]any{
"type": "event",
"event": "agent",
"seq": 2,
"payload": map[string]any{
"runId": runID,
"sessionKey": "main",
"stream": "assistant",
"data": map[string]any{"text": "streamed delta"},
"largeIgnored": strings.Repeat("y", 1024),
},
})
}
_ = conn.WriteJSON(map[string]any{
"type": "event",
"event": "chat",
"seq": 1,
"seq": 3,
"payload": map[string]any{
"runId": runID,
"state": "final",

110
internal/acp/web_contract_test.go
View File

@ -275,6 +275,116 @@ func TestHTTPHandlerGatewayOpenClawSSEKeepaliveBeforeFinalEnvelopeAndDone(t *tes
}
}
func TestHTTPHandlerGatewayOpenClawFiltersRawGatewayEventsAndKeepsFinalResult(t *testing.T) {
gateway := newAcpFakeOpenClawGateway(t)
defer gateway.Close()
gateway.largeGatewayPayloadBytes.Store(openClawGatewayMaxNotificationBytes * 2)
gateway.emitAgentDelta.Store(true)
t.Setenv("GATEWAY_RPC_URL", gateway.URL())
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
t.Setenv("BRIDGE_CONFIG_PATH", filepath.Join(t.TempDir(), "missing-config.yaml"))
server := NewServer()
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
request, err := http.NewRequest(
http.MethodPost,
httpServer.URL+"/gateway/openclaw",
strings.NewReader(`{"jsonrpc":"2.0","id":"task-filter","method":"session.start","params":{"sessionId":"session-filter","threadId":"thread-filter","taskPrompt":"make artifact","workingDirectory":"`+t.TempDir()+`"}}`),
)
if err != nil {
t.Fatalf("build request: %v", err)
}
request.Header.Set("Content-Type", "application/json")
request.Header.Set("Accept", "text/event-stream")
request.Header.Set("Authorization", "Bearer bridge-test-token")
response, err := http.DefaultClient.Do(request)
if err != nil {
t.Fatalf("send request: %v", err)
}
defer response.Body.Close()
body, err := io.ReadAll(response.Body)
if err != nil {
t.Fatalf("read response: %v", err)
}
bodyText := string(body)
if response.StatusCode != http.StatusOK {
t.Fatalf("expected 200, got %d: %s", response.StatusCode, bodyText)
}
if len(body) >= openClawGatewayMaxNotificationBytes {
t.Fatalf("expected compact gateway SSE body, got %d bytes", len(body))
}
for _, rawMethod := range []string{
"xworkmate.gateway.push",
"xworkmate.gateway.snapshot",
"xworkmate.gateway.log",
"largeIgnored",
} {
if strings.Contains(bodyText, rawMethod) {
t.Fatalf("expected raw gateway event %q to be filtered from SSE body: %s", rawMethod, bodyText)
}
}
events := strings.Split(strings.TrimSpace(bodyText), "\n\n")
if len(events) < 4 {
t.Fatalf("expected accepted, session.update, final envelope, and done events, got %q", bodyText)
}
if events[len(events)-1] != "data: [DONE]" {
t.Fatalf("expected done event, got %q", events[len(events)-1])
}
var sawAccepted bool
var sawDelta bool
var sawFinal bool
for _, event := range events[:len(events)-1] {
if !strings.HasPrefix(event, "data: ") {
t.Fatalf("expected data event, got %q", event)
}
var envelope map[string]any
if err := json.Unmarshal([]byte(strings.TrimPrefix(event, "data: ")), &envelope); err != nil {
t.Fatalf("decode event %q: %v", event, err)
}
switch envelope["method"] {
case "xworkmate.bridge.accepted":
sawAccepted = true
case "session.update":
params := shared.AsMap(envelope["params"])
if params["type"] == "delta" && params["delta"] == "streamed delta" {
sawDelta = true
if got := params["sessionId"]; got != "session-filter" {
t.Fatalf("expected session-filter session update, got %#v", params)
}
if got := params["threadId"]; got != "thread-filter" {
t.Fatalf("expected thread-filter session update, got %#v", params)
}
}
}
if envelope["id"] == "task-filter" {
sawFinal = true
result := shared.AsMap(envelope["result"])
if got := result["resolvedGatewayProviderId"]; got != "openclaw" {
t.Fatalf("expected openclaw final result, got %#v", result)
}
if !strings.Contains(bodyText, openClawArtifactDownloadPath) {
t.Fatalf("expected normalized artifact download URL in final result, got %s", bodyText)
}
}
}
if !sawAccepted {
t.Fatalf("expected accepted event, got %q", bodyText)
}
if !sawDelta {
t.Fatalf("expected compact session.update delta, got %q", bodyText)
}
if !sawFinal {
t.Fatalf("expected final result envelope, got %q", bodyText)
}
if got := gateway.Methods(); !sameMethods(got, []string{"connect", "xworkmate.artifacts.prepare", "chat.send", "agent.wait", "xworkmate.artifacts.export"}) {
t.Fatalf("expected artifact workflow methods to stay unchanged, got %#v", got)
}
}
func TestHTTPHandlerGatewayOpenClawAllowsOnlyTaskSubmitMethods(t *testing.T) {
t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token")
t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")