From 96eabb7248f18cfee65e0e2b2ab7ee960aa98d79 Mon Sep 17 00:00:00 2001 From: Haitao Pan Date: Fri, 8 May 2026 11:46:47 +0800 Subject: [PATCH] fix: keep ACP SSE task streams alive --- internal/acp/http_handler.go | 42 ++++++++++++++++ internal/acp/routing_test.go | 4 ++ internal/acp/web_contract_test.go | 83 +++++++++++++++++++++++++++++++ 3 files changed, 129 insertions(+) diff --git a/internal/acp/http_handler.go b/internal/acp/http_handler.go index 2f4b78e..da1aba9 100644 --- a/internal/acp/http_handler.go +++ b/internal/acp/http_handler.go @@ -11,10 +11,13 @@ import ( "strings" "sync" "sync/atomic" + "time" "xworkmate-bridge/internal/shared" ) +var httpSSEKeepaliveInterval = 20 * time.Second + func (s *Server) Handler() http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { @@ -194,15 +197,21 @@ func (s *Server) handleRPCWithTransform( } streamWriter := newSafeSSEStream(r.Context(), w) + stopKeepalive := func() {} writeNotification := func(message map[string]any) { if !stream { return } streamWriter.write(message) } + if stream { + stopKeepalive = streamWriter.startKeepalive(httpSSEKeepaliveInterval) + } + defer stopKeepalive() defer streamWriter.close() response, rpcErr := s.handleRequest(request, writeNotification) + stopKeepalive() if request.ID == nil { if stream { streamWriter.done() @@ -259,6 +268,39 @@ func (s *safeSSEStream) done() bool { }) } +func (s *safeSSEStream) startKeepalive(interval time.Duration) func() { + if s == nil || interval <= 0 { + return func() {} + } + done := make(chan struct{}) + var stopOnce sync.Once + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + if !s.write(map[string]any{ + "jsonrpc": "2.0", + "method": "xworkmate.bridge.keepalive", + "params": map[string]any{ + "intervalMs": interval.Milliseconds(), + }, + }) { + return + } + case <-done: + return + } + } + }() + return func() { + stopOnce.Do(func() { + close(done) + }) + } +} + func (s *safeSSEStream) close() { s.closed.Store(true) } diff --git a/internal/acp/routing_test.go b/internal/acp/routing_test.go index f131f90..45ad24b 100644 --- a/internal/acp/routing_test.go +++ b/internal/acp/routing_test.go @@ -1642,6 +1642,7 @@ type acpFakeOpenClawGateway struct { artifactReadFailures atomic.Int32 closeNextChatSend atomic.Bool alwaysCloseChatSend atomic.Bool + agentWaitDelayMs atomic.Int64 lastConnectClient atomic.Value lastArtifactExportParams atomic.Value lastAgentWaitParams atomic.Value @@ -1796,6 +1797,9 @@ func newAcpFakeOpenClawGateway(t *testing.T) *acpFakeOpenClawGateway { }) case "agent.wait": fake.agentWaitCount.Add(1) + if delayMs := fake.agentWaitDelayMs.Load(); delayMs > 0 { + time.Sleep(time.Duration(delayMs) * time.Millisecond) + } params := shared.AsMap(frame["params"]) fake.lastAgentWaitParams.Store(params) runID := strings.TrimSpace(shared.StringArg(params, "runId", "fake-run")) diff --git a/internal/acp/web_contract_test.go b/internal/acp/web_contract_test.go index 915949b..53b9080 100644 --- a/internal/acp/web_contract_test.go +++ b/internal/acp/web_contract_test.go @@ -3,12 +3,14 @@ package acp import ( "context" "encoding/json" + "io" "net/http" "net/http/httptest" "path/filepath" "reflect" "strings" "testing" + "time" "xworkmate-bridge/internal/shared" ) @@ -185,6 +187,87 @@ func TestHTTPHandlerRPCSSEWritesFinalEnvelopeAndDone(t *testing.T) { } } +func TestHTTPHandlerGatewayOpenClawSSEKeepaliveBeforeFinalEnvelopeAndDone(t *testing.T) { + gateway := newAcpFakeOpenClawGateway(t) + defer gateway.Close() + gateway.agentWaitDelayMs.Store(50) + + 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")) + previousInterval := httpSSEKeepaliveInterval + httpSSEKeepaliveInterval = 10 * time.Millisecond + t.Cleanup(func() { + httpSSEKeepaliveInterval = previousInterval + }) + + 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-keepalive","method":"session.start","params":{"sessionId":"s1","threadId":"t1","taskPrompt":"Reply pong","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) + } + if response.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", response.StatusCode, string(body)) + } + if contentType := response.Header.Get("Content-Type"); !strings.Contains(contentType, "text/event-stream") { + t.Fatalf("expected event-stream content type, got %q", contentType) + } + + events := strings.Split(strings.TrimSpace(string(body)), "\n\n") + if len(events) < 3 { + t.Fatalf("expected keepalive, final envelope, and done events, got %q", string(body)) + } + if events[len(events)-1] != "data: [DONE]" { + t.Fatalf("expected done event, got %q", events[len(events)-1]) + } + var sawKeepaliveBeforeFinal 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) + } + if envelope["method"] == "xworkmate.bridge.keepalive" && !sawFinal { + sawKeepaliveBeforeFinal = true + } + if envelope["id"] == "task-keepalive" { + sawFinal = true + if _, ok := envelope["result"].(map[string]any); !ok { + t.Fatalf("expected result envelope, got %#v", envelope) + } + } + } + if !sawKeepaliveBeforeFinal { + t.Fatalf("expected keepalive event before final envelope, got %q", string(body)) + } + if !sawFinal { + t.Fatalf("expected final task envelope, got %q", string(body)) + } +} + func TestHTTPHandlerGatewayOpenClawAllowsOnlyTaskSubmitMethods(t *testing.T) { t.Setenv("BRIDGE_AUTH_TOKEN", "bridge-test-token") t.Setenv("BRIDGE_CONFIG_PATH", "../../example/config.yaml")