Align translated ModCDP runtimes

Author: pirate

go/modcdp/client/ModCDPClient.go

@@ -331,7 +331,7 @@ type upstreamTransportClient interface {
 	Update(map[string]any)
 	Connect() error
 	Close() error
-	Send(command string, params map[string]any, sessionID string, timeout ...time.Duration) (map[string]any, error)
+	Send(command any, params map[string]any, sessionID string, timeout ...time.Duration) (map[string]any, error)
 	ConfigForLauncher() LauncherConfig
 	OnRecv(func(map[string]any)) func()
 	OnClose(func(error)) func()
@@ -479,7 +479,10 @@ func (c *ModCDPClient) Configure(config Config) *ModCDPClient {
 	if c.Upstream != nil {
 		c.Upstream.Update(map[string]any{"upstream_cdp_send_timeout_ms": c.Config.ClientConfig.ClientCDPSendTimeoutMS})
 	}
-	if config.Upstream.UpstreamWSCDPURL != "" || config.Upstream.UpstreamWSConnectErrorSettleTimeoutMS != 0 || config.Upstream.UpstreamCDPSendTimeoutMS != 0 {
+	if config.Upstream.UpstreamMode != "" || config.Upstream.UpstreamWSCDPURL != "" || config.Upstream.UpstreamWSConnectErrorSettleTimeoutMS != 0 || config.Upstream.UpstreamCDPSendTimeoutMS != 0 {
+		if config.Upstream.UpstreamMode != "" {
+			c.Config.Upstream.UpstreamMode = config.Upstream.UpstreamMode
+		}
 		if config.Upstream.UpstreamWSCDPURL != "" {
 			c.Config.Upstream.UpstreamWSCDPURL = config.Upstream.UpstreamWSCDPURL
 		}
@@ -738,7 +741,8 @@ func (c *ModCDPClient) ensureModCDPServerConfigured() error {
 
 func (c *ModCDPClient) upstreamTransportConfig() map[string]any {
 	return map[string]any{
-		"upstream_ws_cdp_url":                         c.Config.Upstream.UpstreamWSCDPURL,
+		"upstream_mode":       c.Config.Upstream.UpstreamMode,
+		"upstream_ws_cdp_url": c.Config.Upstream.UpstreamWSCDPURL,
 		"upstream_ws_connect_error_settle_timeout_ms": c.Config.Upstream.UpstreamWSConnectErrorSettleTimeoutMS,
 		"upstream_cdp_send_timeout_ms":                c.Config.Upstream.UpstreamCDPSendTimeoutMS,
 	}

go/modcdp/injector/ExtensionInjector.go

@@ -184,7 +184,7 @@ func (i *ExtensionInjector) Inject() (*ExtensionInjectionResult, error) {
 }
 
 func (i ExtensionInjector) readyExpression() string {
-	if i.Config.InjectorServiceWorkerReadyExpression == "" {
+	if i.Config.InjectorServiceWorkerReadyExpression == "" || i.Config.InjectorServiceWorkerReadyExpression == modcdpReadyExpression {
 		return modcdpReadyExpression
 	}
 	return fmt.Sprintf("(%s) && Boolean(%s)", modcdpReadyExpression, i.Config.InjectorServiceWorkerReadyExpression)

go/modcdp/injector/ExtensionInjector_test.go

@@ -48,3 +48,20 @@ func TestExtensionInjectorBaseInjectReportsTheSubclassName(t *testing.T) {
 		t.Fatalf("Inject error = %v", err)
 	}
 }
+
+func TestExtensionInjectorDoesNotWrapTheDefaultReadyExpressionTwice(t *testing.T) {
+	injector := NewExtensionInjector(InjectorConfig{
+		InjectorServiceWorkerReadyExpression: modcdpReadyExpression,
+	})
+
+	if injector.readyExpression() != modcdpReadyExpression {
+		t.Fatalf("readyExpression = %q", injector.readyExpression())
+	}
+
+	injector.Update(InjectorConfig{
+		InjectorServiceWorkerReadyExpression: "globalThis.ready === true",
+	})
+	if got := injector.readyExpression(); got != "("+modcdpReadyExpression+") && Boolean(globalThis.ready === true)" {
+		t.Fatalf("custom readyExpression = %q", got)
+	}
+}

go/modcdp/router/AutoSessionRouter.go

@@ -5,6 +5,8 @@
 package router
 
 import (
+	"crypto/rand"
+	"encoding/hex"
 	"fmt"
 	"strings"
 	"sync"
@@ -176,23 +178,23 @@ func (r *AutoSessionRouter) Send(method string, params map[string]any, requested
 	return r.upstream.Send(method, routedParams, sessionID)
 }
 
-func (r *AutoSessionRouter) AttachToTarget(targetID string) string {
+func (r *AutoSessionRouter) AttachToTarget(targetID string) (string, error) {
 	r.mu.Lock()
 	sessionID := r.SessionId_from_targetId[targetID]
 	r.mu.Unlock()
 	if sessionID != "" {
-		return sessionID
+		return sessionID, nil
 	}
 	attachedSessionID, err := r.upstream.AttachToTarget(targetID)
 	if err != nil {
-		return ""
+		return "", err
 	}
 	if attachedSessionID != "" {
 		r.mu.Lock()
 		r.recordTargetSession(targetID, attachedSessionID, r.Targets[targetID])
 		r.mu.Unlock()
 	}
-	return attachedSessionID
+	return attachedSessionID, nil
 }
 
 func (r *AutoSessionRouter) EnsureSessionForTarget(targetID string) (string, error) {
@@ -233,7 +235,10 @@ func (r *AutoSessionRouter) EnsureRouteForTarget(targetID string) (string, strin
 			return "", "", err
 		}
 	}
-	sessionID := r.AttachToTarget(resolvedTargetID)
+	sessionID, err := r.AttachToTarget(resolvedTargetID)
+	if err != nil {
+		return "", "", err
+	}
 	if sessionID == "" {
 		r.recordTargetSessionlessAttachment(resolvedTargetID)
 		return resolvedTargetID, "", nil
@@ -428,7 +433,7 @@ func (r *AutoSessionRouter) GetTopology(params map[string]any) (map[string]any,
 	if params == nil {
 		params = map[string]any{}
 	}
-	objectGroup := fmt.Sprintf("modcdp-topology-%d", time.Now().UnixMilli())
+	objectGroup := fmt.Sprintf("modcdp-topology-%d-%s", time.Now().UnixMilli(), randomHexSuffix(8))
 	targetResult, err := r.upstream.Send("Target.getTargets", map[string]any{}, "")
 	if err != nil {
 		return nil, err
@@ -1079,6 +1084,14 @@ func cloneStringMap(input map[string]string) map[string]string {
 	return output
 }
 
+func randomHexSuffix(bytesLen int) string {
+	buf := make([]byte, bytesLen)
+	if _, err := rand.Read(buf); err == nil {
+		return hex.EncodeToString(buf)
+	}
+	return fmt.Sprintf("%d", time.Now().UnixNano())
+}
+
 func intFromAny(value any) (int, bool) {
 	switch typed := value.(type) {
 	case int:

go/modcdp/router/AutoSessionRouter_test.go

@@ -101,6 +101,25 @@ func TestAutoSessionRouterTracksRealTargetSessionsAndExecutionContextsFromLiveCD
 	if !foundContext {
 		t.Fatalf("context id %d for session %s was not recorded", contextID, sessionID)
 	}
+	topologyOne, err := cdp.Router.GetTopology(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	objectGroupOne, _ := topologyOne["objectGroup"].(string)
+	if !regexp.MustCompile(`^modcdp-topology-\d+-[0-9a-f]+$`).MatchString(objectGroupOne) {
+		t.Fatalf("topology objectGroup = %q", objectGroupOne)
+	}
+	topologyTwo, err := cdp.Router.GetTopology(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	objectGroupTwo, _ := topologyTwo["objectGroup"].(string)
+	if objectGroupOne == objectGroupTwo {
+		t.Fatalf("topology objectGroup was reused: %q", objectGroupOne)
+	}
+	if _, _, err := cdp.Router.EnsureRouteForTarget("missing-target-id"); err == nil {
+		t.Fatal("EnsureRouteForTarget should return the attach error for an unknown target")
+	}
 
 	detachTarget(t, cdp, sessionID)
 	expectEventually(t, func() error {

go/modcdp/transport/UpstreamTransport.go

@@ -5,6 +5,7 @@
 package transport
 
 import (
+	"encoding/json"
 	"fmt"
 	"net"
 	"net/url"
@@ -126,15 +127,33 @@ func (e *UpstreamTransport) Close() error {
 	return nil
 }
 
-func (e *UpstreamTransport) Send(command string, params map[string]any, sessionID string, timeout ...time.Duration) (map[string]any, error) {
+func (e *UpstreamTransport) Send(command any, params map[string]any, sessionID string, timeout ...time.Duration) (map[string]any, error) {
+	method, ok := command.(string)
+	if !ok {
+		message, ok := command.(types.CdpCommandMessage)
+		if !ok {
+			return nil, fmt.Errorf("command must be a CDP method name or CdpCommandMessage")
+		}
+		payload := map[string]any{
+			"id":     message.ID,
+			"method": message.Method,
+		}
+		if message.Params != nil {
+			payload["params"] = message.Params
+		}
+		if message.SessionID != "" {
+			payload["sessionId"] = message.SessionID
+		}
+		return map[string]any{}, e.writeCommand(payload)
+	}
 	e.pendingMu.Lock()
 	e.nextID++
 	id := e.nextID
 	done := make(chan map[string]any, 1)
 	e.pending[id] = done
 	e.pendingMu.Unlock()
 
-	message := map[string]any{"id": id, "method": command, "params": params}
+	message := map[string]any{"id": id, "method": method, "params": params}
 	if sessionID != "" {
 		message["sessionId"] = sessionID
 	}
@@ -151,7 +170,7 @@ func (e *UpstreamTransport) Send(command string, params map[string]any, sessionI
 	if effectiveTimeout <= 0 {
 		response := <-done
 		if errObj, ok := response["error"].(map[string]any); ok {
-			return nil, fmt.Errorf("%s failed: %v", command, errObj["message"])
+			return nil, fmt.Errorf("%s failed: %v", method, errObj["message"])
 		}
 		if result, ok := response["result"].(map[string]any); ok {
 			return result, nil
@@ -163,10 +182,10 @@ func (e *UpstreamTransport) Send(command string, params map[string]any, sessionI
 		e.pendingMu.Lock()
 		delete(e.pending, id)
 		e.pendingMu.Unlock()
-		return nil, fmt.Errorf("%s timed out after %s", command, effectiveTimeout)
+		return nil, fmt.Errorf("%s timed out after %s", method, effectiveTimeout)
 	case response := <-done:
 		if errObj, ok := response["error"].(map[string]any); ok {
-			return nil, fmt.Errorf("%s failed: %v", command, errObj["message"])
+			return nil, fmt.Errorf("%s failed: %v", method, errObj["message"])
 		}
 		if result, ok := response["result"].(map[string]any); ok {
 			return result, nil
@@ -332,6 +351,50 @@ func (e *UpstreamTransport) EmitRecv(message map[string]any) {
 	e.emitRecv(message)
 }
 
+func (e *UpstreamTransport) parseAndEmitRecv(data []byte) error {
+	var message map[string]any
+	if err := json.Unmarshal(data, &message); err != nil {
+		return fmt.Errorf("invalid CDP message: %w", err)
+	}
+	if _, ok := commandID(message["id"]); ok {
+		if errObj, hasError := message["error"]; hasError && errObj != nil {
+			errorMap, ok := errObj.(map[string]any)
+			if !ok {
+				return fmt.Errorf("invalid CDP response error")
+			}
+			if message, ok := errorMap["message"].(string); !ok || message == "" {
+				return fmt.Errorf("invalid CDP response error message")
+			}
+		}
+		if sessionID, ok := message["sessionId"]; ok && sessionID != nil {
+			if _, ok := sessionID.(string); !ok {
+				return fmt.Errorf("invalid CDP response sessionId")
+			}
+		}
+		e.emitRecv(message)
+		return nil
+	}
+	if _, hasID := message["id"]; hasID {
+		return fmt.Errorf("invalid CDP response id")
+	}
+	method, _ := message["method"].(string)
+	if method == "" {
+		return fmt.Errorf("invalid CDP event method")
+	}
+	if params, ok := message["params"]; ok && params != nil {
+		if _, ok := params.(map[string]any); !ok {
+			return fmt.Errorf("invalid CDP event params")
+		}
+	}
+	if sessionID, ok := message["sessionId"]; ok && sessionID != nil {
+		if _, ok := sessionID.(string); !ok {
+			return fmt.Errorf("invalid CDP event sessionId")
+		}
+	}
+	e.emitRecv(message)
+	return nil
+}
+
 func (e *UpstreamTransport) emitUpstreamEvent(method string, payload map[string]any, targetID string, sessionID string) {
 	e.listenerMu.Lock()
 	listeners := append([]upstreamEventListener(nil), e.eventListeners[method]...)

go/modcdp/transport/UpstreamTransport_internal_test.go

@@ -0,0 +1,30 @@
+// MODCDP_TRANSLATE_TEST: GO-SPECIFIC INTERNAL COVERAGE.
+// The public translated tests cover behavior through user-facing transport APIs.
+package transport
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestUpstreamTransportRejectsInvalidIncomingMessages(t *testing.T) {
+	transport := NewUpstreamTransport(UpstreamTransportConfig{})
+
+	for _, testCase := range []struct {
+		name    string
+		message string
+		want    string
+	}{
+		{name: "json", message: "{", want: "invalid CDP message"},
+		{name: "response id", message: `{"id":"one","result":{}}`, want: "invalid CDP response id"},
+		{name: "event method", message: `{"params":{}}`, want: "invalid CDP event method"},
+		{name: "event params", message: `{"method":"Runtime.executionContextCreated","params":[]}`, want: "invalid CDP event params"},
+	} {
+		t.Run(testCase.name, func(t *testing.T) {
+			err := transport.parseAndEmitRecv([]byte(testCase.message))
+			if err == nil || !strings.Contains(err.Error(), testCase.want) {
+				t.Fatalf("parse error = %v", err)
+			}
+		})
+	}
+}

go/modcdp/transport/UpstreamTransport_test.go

@@ -4,21 +4,22 @@
 // - ./python/tests/test_UpstreamTransport.py
 // NO MOCKING, NO MONKEY PATCHING, NO SIMULATING, NO FAKING, NO SKIPPING ALLOWED.
 // USE REAL USER-FACING CODE PATHS WITH REAL BROWSERS, REAL CLASSES, REAL URLS, etc. Hard fail if keys or other env requirements are missing.
-package transport_test
+package transport
 
 import (
-	. "github.com/browserbase/modcdp/go/modcdp/transport"
 	"reflect"
 	"strings"
 	"testing"
+
+	"github.com/browserbase/modcdp/go/modcdp/types"
 )
 
 type testUpstreamTransport struct {
 	UpstreamTransport
 }
 
-func (t *testUpstreamTransport) emit(message map[string]any) {
-	t.EmitRecv(message)
+func (t *testUpstreamTransport) emit(message string) error {
+	return t.parseAndEmitRecv([]byte(message))
 }
 
 func TestOwnsSharedTransportConfigAndRecvCallbacks(t *testing.T) {
@@ -41,14 +42,20 @@ func TestOwnsSharedTransportConfigAndRecvCallbacks(t *testing.T) {
 	testTransport := &testUpstreamTransport{UpstreamTransport: NewUpstreamTransport(UpstreamTransportConfig{})}
 	parsed := []map[string]any{}
 	testTransport.OnRecv(func(message map[string]any) { parsed = append(parsed, message) })
-	testTransport.emit(map[string]any{"id": 1, "result": map[string]any{"ok": true}})
-	testTransport.emit(map[string]any{"id": 2, "result": true})
-	testTransport.emit(map[string]any{"id": 3, "result": 0})
-	testTransport.emit(map[string]any{"method": "Runtime.executionContextCreated", "params": map[string]any{}})
+	for _, message := range []string{
+		`{"id":1,"result":{"ok":true}}`,
+		`{"id":2,"result":true}`,
+		`{"id":3,"result":0}`,
+		`{"method":"Runtime.executionContextCreated","params":{}}`,
+	} {
+		if err := testTransport.emit(message); err != nil {
+			t.Fatal(err)
+		}
+	}
 	expected := []map[string]any{
-		{"id": 1, "result": map[string]any{"ok": true}},
-		{"id": 2, "result": true},
-		{"id": 3, "result": 0},
+		{"id": float64(1), "result": map[string]any{"ok": true}},
+		{"id": float64(2), "result": true},
+		{"id": float64(3), "result": float64(0)},
 		{"method": "Runtime.executionContextCreated", "params": map[string]any{}},
 	}
 	if !reflect.DeepEqual(parsed, expected) {
@@ -62,7 +69,7 @@ func TestOwnsSharedTransportConfigAndRecvCallbacks(t *testing.T) {
 	if err := transport.Connect(); err == nil || !strings.Contains(err.Error(), "Connect is not implemented") {
 		t.Fatalf("connect error = %v", err)
 	}
-	if _, err := transport.Send("Browser.getVersion", map[string]any{}, ""); err == nil || !strings.Contains(err.Error(), "send is not implemented") {
+	if _, err := transport.Send(types.CdpCommandMessage{ID: 1, Method: "Browser.getVersion", Params: map[string]any{}}, nil, ""); err == nil || !strings.Contains(err.Error(), "send is not implemented") {
 		t.Fatalf("send error = %v", err)
 	}
 }