diff --git a/internal/acp/http_handler.go b/internal/acp/http_handler.go index 2b29093..0bc43f4 100644 --- a/internal/acp/http_handler.go +++ b/internal/acp/http_handler.go @@ -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 diff --git a/internal/acp/orchestrator.go b/internal/acp/orchestrator.go index 443cf39..279fa90 100644 --- a/internal/acp/orchestrator.go +++ b/internal/acp/orchestrator.go @@ -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 diff --git a/internal/acp/routing_test.go b/internal/acp/routing_test.go index 45ad24b..6298b0b 100644 --- a/internal/acp/routing_test.go +++ b/internal/acp/routing_test.go @@ -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", diff --git a/internal/acp/web_contract_test.go b/internal/acp/web_contract_test.go index 706a631..ad01483 100644 --- a/internal/acp/web_contract_test.go +++ b/internal/acp/web_contract_test.go @@ -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")