Refactor project structure and add Close() method

- Add Close() method to Communicator for proper resource cleanup - Move examples from go/example/ to go/examples/ - Move test files to go/test/ directory - Add cross-test files for Dart-Go interop testing - Add comprehensive test suite (tcp, ws, tls, heartbeat tests) - Add crosstest directory for multi-language testing - Add SKILL documentation for adding crosstest languages
2026-04-11 09:54:03 +09:00 · 2026-04-11 09:54:03 +09:00 · e48722d811
commit e48722d811
parent 14e4cd56d0
16 changed files with 1463 additions and 380 deletions
--- a/dart/crosstest/dart_go.dart
+++ b/dart/crosstest/dart_go.dart
@ -0,0 +1,232 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:toki_socket/toki_socket.dart';
+
+const _host = '127.0.0.1';
+const _tcpPort = 29090;
+const _wsPort = 29092;
+const _heartbeatIntervalSeconds = 30;
+const _heartbeatWaitSeconds = 10;
+const _serverObservationWindow = Duration(milliseconds: 200);
+const _processTimeout = Duration(seconds: 20);
+final _repoRoot = File.fromUri(Platform.script).parent.parent.parent.path;
+final _goDir = Directory('$_repoRoot/go').path;
+
+class _DartTcpClient extends ProtobufClient {
+  _DartTcpClient(Socket socket)
+      : super(socket, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
+          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
+        });
+}
+
+class _DartWsClient extends WsProtobufClient {
+  _DartWsClient(WebSocket ws)
+      : super(ws, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
+          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
+        });
+}
+
+class _DartTcpServer extends ProtobufServer {
+  final void Function(ProtobufClient) onConnected;
+
+  _DartTcpServer(int port, this.onConnected)
+      : super(_host, port, (socket) => _DartTcpClient(socket));
+
+  @override
+  void onClientConnected(ProtobufClient client) => onConnected(client);
+}
+
+class _DartWsServer extends WsProtobufServer {
+  final void Function(WsProtobufClient) onConnected;
+
+  _DartWsServer(int port, this.onConnected)
+      : super(_host, port, (ws) => _DartWsClient(ws));
+
+  @override
+  void onClientConnected(WsProtobufClient client) => onConnected(client);
+}
+
+Future<void> main() async {
+  print(
+      'INFO typeName dart=${TestData.getDefault().info_.qualifiedMessageName}');
+  try {
+    await _runTcp();
+    await _runWs();
+    print('PASS all dart-server/go-client crosstests passed');
+  } catch (error) {
+    stderr.writeln('FAIL crosstest error=$error');
+    exitCode = 1;
+  }
+}
+
+Future<void> _runTcp() async {
+  await _runTcpSendPush();
+  await Future<void>.delayed(const Duration(milliseconds: 150));
+  await _runTcpRequests();
+}
+
+Future<void> _runWs() async {
+  await _runWsSendPush();
+  await Future<void>.delayed(const Duration(milliseconds: 150));
+  await _runWsRequests();
+}
+
+Future<void> _runTcpSendPush() async {
+  final received = Completer<bool>();
+  final server = _DartTcpServer(_tcpPort, (client) {
+    client.addListener<TestData>((data) {
+      print('SERVER_RECEIVED index=${data.index} message=${data.message}');
+      final valid = data.index == 101 && data.message == 'fire from go client';
+      if (!received.isCompleted) {
+        received.complete(valid);
+      }
+      if (valid) {
+        unawaited(client.send(TestData()
+          ..index = 200
+          ..message = 'push from dart server'));
+      }
+    });
+  });
+  await _withServer(server.start, server.stop, () async {
+    await _runGoClient('tcp', _tcpPort, 'send-push', {'1', '2'});
+    // The subprocess only proves the client observed PASS lines. This short
+    // wait also confirms the server-side listener recorded the expected data.
+    final ok = await received.future
+        .timeout(_serverObservationWindow, onTimeout: () => false);
+    if (!ok) {
+      throw StateError('TCP send-push server did not receive expected data');
+    }
+  });
+}
+
+Future<void> _runTcpRequests() async {
+  final server = _DartTcpServer(_tcpPort, (client) {
+    client.addRequestListener<TestData, TestData>((req) async {
+      return TestData()
+        ..index = req.index * 2
+        ..message = 'echo: ${req.message}';
+    });
+  });
+  await _withServer(
+    server.start,
+    server.stop,
+    () => _runGoClient('tcp', _tcpPort, 'requests', {'3', '4'}),
+  );
+}
+
+Future<void> _runWsSendPush() async {
+  final received = Completer<bool>();
+  final server = _DartWsServer(_wsPort, (client) {
+    client.addListener<TestData>((data) {
+      print('SERVER_RECEIVED index=${data.index} message=${data.message}');
+      final valid = data.index == 101 && data.message == 'fire from go client';
+      if (!received.isCompleted) {
+        received.complete(valid);
+      }
+      if (valid) {
+        unawaited(client.send(TestData()
+          ..index = 200
+          ..message = 'push from dart server'));
+      }
+    });
+  });
+  await _withServer(server.start, server.stop, () async {
+    await _runGoClient('ws', _wsPort, 'send-push', {'1', '2'});
+    // The subprocess only proves the client observed PASS lines. This short
+    // wait also confirms the server-side listener recorded the expected data.
+    final ok = await received.future
+        .timeout(_serverObservationWindow, onTimeout: () => false);
+    if (!ok) {
+      throw StateError('WS send-push server did not receive expected data');
+    }
+  });
+}
+
+Future<void> _runWsRequests() async {
+  final server = _DartWsServer(_wsPort, (client) {
+    client.addRequestListener<TestData, TestData>((req) async {
+      return TestData()
+        ..index = req.index * 2
+        ..message = 'echo: ${req.message}';
+    });
+  });
+  await _withServer(
+    server.start,
+    server.stop,
+    () => _runGoClient('ws', _wsPort, 'requests', {'3', '4'}),
+  );
+}
+
+Future<void> _withServer(Future<void> Function() start,
+    Future<void> Function() stop, Future<void> Function() body) async {
+  await start();
+  try {
+    await body();
+  } finally {
+    await stop();
+  }
+}
+
+Future<void> _runGoClient(
+    String mode, int port, String phase, Set<String> expectedScenarios) async {
+  final process = await Process.start(
+    'go',
+    [
+      'run',
+      './crosstest/dart_go_client',
+      '--mode=$mode',
+      '--port=$port',
+      '--phase=$phase',
+    ],
+    workingDirectory: _goDir,
+  );
+
+  final resultLines = <String>[];
+  final stdoutDone = process.stdout
+      .transform(utf8.decoder)
+      .transform(const LineSplitter())
+      .listen((line) {
+    print(line);
+    if (line.startsWith('PASS ') || line.startsWith('FAIL ')) {
+      resultLines.add(line);
+    }
+  }).asFuture<void>();
+
+  final stderrDone = process.stderr
+      .transform(utf8.decoder)
+      .transform(const LineSplitter())
+      .listen(stderr.writeln)
+      .asFuture<void>();
+
+  final code = await process.exitCode.timeout(_processTimeout, onTimeout: () {
+    process.kill(ProcessSignal.sigkill);
+    return -1;
+  });
+  await Future.wait([stdoutDone, stderrDone]);
+
+  _validateResultLines(
+      'go-client $mode/$phase', code, resultLines, expectedScenarios);
+}
+
+void _validateResultLines(
+  String label,
+  int exitCode,
+  List<String> lines,
+  Set<String> expectedScenarios,
+) {
+  final failed = lines.where((line) => line.startsWith('FAIL ')).toList();
+  final passed = <String>{};
+  for (final line in lines.where((line) => line.startsWith('PASS '))) {
+    final match = RegExp(r'scenario=([^ ]+)').firstMatch(line);
+    if (match != null) {
+      passed.add(match.group(1)!);
+    }
+  }
+  final missing = expectedScenarios.difference(passed);
+  if (exitCode != 0 || failed.isNotEmpty || missing.isNotEmpty) {
+    throw StateError(
+        '$label failed exitCode=$exitCode failed=$failed missing=$missing');
+  }
+}
--- a/dart/crosstest/go_dart_client.dart
+++ b/dart/crosstest/go_dart_client.dart
@ -0,0 +1,192 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:toki_socket/toki_socket.dart';
+
+const _host = '127.0.0.1';
+const _wsPath = '/';
+const _heartbeatIntervalSeconds = 30;
+const _heartbeatWaitSeconds = 10;
+const _connectWindow = Duration(seconds: 3);
+const _requestWindow = Duration(seconds: 2);
+
+class _TcpClient extends ProtobufClient {
+  _TcpClient(Socket socket)
+      : super(socket, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
+          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
+        });
+}
+
+class _WsClient extends WsProtobufClient {
+  _WsClient(WebSocket ws)
+      : super(ws, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
+          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
+        });
+}
+
+class _ClientHandle {
+  final Communicator client;
+  final Future<void> Function() close;
+
+  _ClientHandle(this.client, this.close);
+}
+
+Future<void> main(List<String> args) async {
+  final mode = _argValue(args, 'mode') ?? 'tcp';
+  final phase = _argValue(args, 'phase') ?? 'send-push';
+  final port = int.tryParse(_argValue(args, 'port') ?? '');
+
+  print(
+      'INFO typeName dart=${TestData.getDefault().info_.qualifiedMessageName}');
+
+  if (port == null) {
+    _fail('setup', 'port is required');
+    exitCode = 1;
+    return;
+  }
+
+  late _ClientHandle handle;
+  try {
+    handle = await _connectWithRetry(mode, port);
+  } catch (error) {
+    _fail('setup', error.toString());
+    exitCode = 1;
+    return;
+  }
+
+  var ok = false;
+  try {
+    switch (phase) {
+      case 'send-push':
+        ok = await _runSendPush(handle.client);
+        break;
+      case 'requests':
+        ok = await _runRequests(handle.client);
+        break;
+      default:
+        _fail('setup', 'unknown phase "$phase"');
+    }
+  } finally {
+    await handle.close();
+  }
+
+  if (!ok) {
+    exitCode = 1;
+  }
+}
+
+Future<_ClientHandle> _connectWithRetry(String mode, int port) async {
+  final deadline = DateTime.now().add(_connectWindow);
+  Object? lastError;
+
+  while (DateTime.now().isBefore(deadline)) {
+    try {
+      switch (mode) {
+        case 'tcp':
+          final socket = await ProtobufClient.connect(_host, port)
+              .timeout(const Duration(milliseconds: 300));
+          final client = _TcpClient(socket);
+          return _ClientHandle(client, () async => client.dispose());
+        case 'ws':
+          final ws = await WsProtobufClient.connect(_host, port, path: _wsPath)
+              .timeout(const Duration(milliseconds: 300));
+          final client = _WsClient(ws);
+          return _ClientHandle(client, () async => client.dispose());
+        default:
+          throw ArgumentError('unknown mode "$mode"');
+      }
+    } catch (error) {
+      lastError = error;
+      await Future<void>.delayed(const Duration(milliseconds: 100));
+    }
+  }
+
+  throw TimeoutException(
+      'connect $mode:$port timed out: $lastError', _connectWindow);
+}
+
+Future<bool> _runSendPush(Communicator client) async {
+  final pushCompleter = Completer<TestData>();
+  client.addListener<TestData>((data) {
+    if (!pushCompleter.isCompleted) {
+      pushCompleter.complete(data);
+    }
+  });
+
+  await client.send(TestData()
+    ..index = 101
+    ..message = 'fire from dart client');
+  _pass('1', 'fire-and-forget sent');
+
+  try {
+    final push = await pushCompleter.future.timeout(_requestWindow);
+    if (push.index != 200 || push.message != 'push from go server') {
+      _fail(
+          '2', 'unexpected push index=${push.index} message="${push.message}"');
+      return false;
+    }
+    _pass('2', 'received push from go server');
+    return true;
+  } catch (error) {
+    _fail('2', error.toString());
+    return false;
+  }
+}
+
+Future<bool> _runRequests(Communicator client) async {
+  try {
+    final single = await client
+        .sendRequest<TestData, TestData>(TestData()
+          ..index = 21
+          ..message = 'single request from dart')
+        .timeout(_requestWindow);
+    if (single.index != 42 ||
+        single.message != 'echo: single request from dart') {
+      _fail('3',
+          'unexpected response index=${single.index} message="${single.message}"');
+      return false;
+    }
+    _pass('3', 'single request response matched');
+
+    final futures = <Future<void>>[];
+    for (var i = 0; i < 5; i++) {
+      futures.add(() async {
+        final index = 30 + i;
+        final message = 'multi request $i from dart';
+        final res = await client
+            .sendRequest<TestData, TestData>(TestData()
+              ..index = index
+              ..message = message)
+            .timeout(_requestWindow);
+        if (res.index != index * 2 || res.message != 'echo: $message') {
+          throw StateError(
+              'request $i got index=${res.index} message="${res.message}"');
+        }
+      }());
+    }
+    await Future.wait(futures);
+    _pass('4', 'concurrent request responses matched');
+    return true;
+  } catch (error) {
+    _fail('4', error.toString());
+    return false;
+  }
+}
+
+String? _argValue(List<String> args, String name) {
+  final prefix = '--$name=';
+  for (final arg in args) {
+    if (arg.startsWith(prefix)) {
+      return arg.substring(prefix.length);
+    }
+  }
+  return null;
+}
+
+void _pass(String scenario, String detail) {
+  print('PASS scenario=$scenario detail=$detail');
+}
+
+void _fail(String scenario, String error) {
+  print('FAIL scenario=$scenario error=$error');
+}
--- a/go/communicator.go
+++ b/go/communicator.go
@ -98,6 +98,14 @@ func (c *Communicator) shutdown() {
 	})
 }

+func (c *Communicator) Close() error {
+	c.shutdown()
+	if c.transport == nil {
+		return nil
+	}
+	return c.transport.Close()
+}
+
 func (c *Communicator) writeLoop() {
 	for {
 		select {
--- a/go/crosstest/dart_go_client/main.go
+++ b/go/crosstest/dart_go_client/main.go
@ -0,0 +1,205 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"sync"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+const (
+	host          = "127.0.0.1"
+	wsPath        = "/"
+	connectWindow = 3 * time.Second
+	requestWindow = 2 * time.Second
+)
+
+type clientHandle struct {
+	communicator *toki.Communicator
+	send         func(proto.Message) error
+	close        func() error
+}
+
+func parserMap() toki.ParserMap {
+	return toki.ParserMap{
+		toki.TypeNameOf(&packets.TestData{}): func(b []byte) (proto.Message, error) {
+			m := &packets.TestData{}
+			return m, proto.Unmarshal(b, m)
+		},
+	}
+}
+
+func main() {
+	mode := flag.String("mode", "tcp", "transport mode: tcp or ws")
+	port := flag.Int("port", 0, "server port")
+	phase := flag.String("phase", "send-push", "test phase: send-push or requests")
+	flag.Parse()
+
+	fmt.Printf("INFO typeName go=%s\n", toki.TypeNameOf(&packets.TestData{}))
+
+	if *port == 0 {
+		fail("setup", "port is required")
+		os.Exit(1)
+	}
+
+	client, err := dialWithRetry(*mode, *port)
+	if err != nil {
+		fail("setup", err.Error())
+		os.Exit(1)
+	}
+	defer client.close()
+
+	var ok bool
+	switch *phase {
+	case "send-push":
+		ok = runSendPush(client)
+	case "requests":
+		ok = runRequests(client)
+	default:
+		fail("setup", fmt.Sprintf("unknown phase %q", *phase))
+		ok = false
+	}
+	if !ok {
+		os.Exit(1)
+	}
+}
+
+func dialWithRetry(mode string, port int) (*clientHandle, error) {
+	deadline := time.Now().Add(connectWindow)
+	var lastErr error
+	for time.Now().Before(deadline) {
+		ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+		handle, err := dial(ctx, mode, port)
+		cancel()
+		if err == nil {
+			return handle, nil
+		}
+		lastErr = err
+		time.Sleep(100 * time.Millisecond)
+	}
+	return nil, fmt.Errorf("connect %s:%d timed out: %w", mode, port, lastErr)
+}
+
+func dial(ctx context.Context, mode string, port int) (*clientHandle, error) {
+	switch mode {
+	case "tcp":
+		client, err := toki.DialTcp(ctx, host, port, 0, 0, parserMap())
+		if err != nil {
+			return nil, err
+		}
+		return &clientHandle{
+			communicator: &client.Communicator,
+			send:         client.Send,
+			close:        client.Close,
+		}, nil
+	case "ws":
+		client, err := toki.DialWsWithHeartbeat(ctx, host, port, wsPath, 0, 0, parserMap())
+		if err != nil {
+			return nil, err
+		}
+		return &clientHandle{
+			communicator: &client.Communicator,
+			send:         client.Send,
+			close:        client.Close,
+		}, nil
+	default:
+		return nil, fmt.Errorf("unknown mode %q", mode)
+	}
+}
+
+func runSendPush(client *clientHandle) bool {
+	pushCh := make(chan *packets.TestData, 1)
+	toki.AddListenerTyped[*packets.TestData](client.communicator, func(msg *packets.TestData) {
+		pushCh <- msg
+	})
+
+	err := client.send(&packets.TestData{
+		Index:   101,
+		Message: "fire from go client",
+	})
+	if err != nil {
+		fail("1", err.Error())
+		return false
+	}
+	pass("1", "fire-and-forget sent")
+
+	select {
+	case msg := <-pushCh:
+		if msg.GetIndex() != 200 || msg.GetMessage() != "push from dart server" {
+			fail("2", fmt.Sprintf("unexpected push index=%d message=%q", msg.GetIndex(), msg.GetMessage()))
+			return false
+		}
+		pass("2", "received push from dart server")
+		return true
+	case <-time.After(requestWindow):
+		fail("2", "timeout waiting for server push")
+		return false
+	}
+}
+
+func runRequests(client *clientHandle) bool {
+	res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
+		client.communicator,
+		&packets.TestData{Index: 21, Message: "single request from go"},
+		requestWindow,
+	)
+	if err != nil {
+		fail("3", err.Error())
+		return false
+	}
+	if res.GetIndex() != 42 || res.GetMessage() != "echo: single request from go" {
+		fail("3", fmt.Sprintf("unexpected response index=%d message=%q", res.GetIndex(), res.GetMessage()))
+		return false
+	}
+	pass("3", "single request response matched")
+
+	const count = 5
+	var wg sync.WaitGroup
+	errCh := make(chan error, count)
+	for i := 0; i < count; i++ {
+		i := i
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			index := int32(30 + i)
+			message := fmt.Sprintf("multi request %d from go", i)
+			res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
+				client.communicator,
+				&packets.TestData{Index: index, Message: message},
+				requestWindow,
+			)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			if res.GetIndex() != index*2 || res.GetMessage() != "echo: "+message {
+				errCh <- fmt.Errorf("request %d got index=%d message=%q", i, res.GetIndex(), res.GetMessage())
+			}
+		}()
+	}
+	wg.Wait()
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			fail("4", err.Error())
+			return false
+		}
+	}
+	pass("4", "concurrent request responses matched")
+	return true
+}
+
+func pass(scenario, detail string) {
+	fmt.Printf("PASS scenario=%s detail=%s\n", scenario, detail)
+}
+
+func fail(scenario, detail string) {
+	fmt.Printf("FAIL scenario=%s error=%s\n", scenario, detail)
+}
--- a/go/crosstest/go_dart.go
+++ b/go/crosstest/go_dart.go
@ -0,0 +1,331 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"sync"
+	"time"
+
+	"google.golang.org/protobuf/proto"
+	"nhooyr.io/websocket"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+const (
+	host                    = "127.0.0.1"
+	goDartTCPPort           = 29190
+	goDartWSPort            = 29192
+	wsPath                  = "/"
+	processTimeout          = 20 * time.Second
+	serverObservationWindow = 200 * time.Millisecond
+)
+
+func parserMap() toki.ParserMap {
+	return toki.ParserMap{
+		toki.TypeNameOf(&packets.TestData{}): func(b []byte) (proto.Message, error) {
+			m := &packets.TestData{}
+			return m, proto.Unmarshal(b, m)
+		},
+	}
+}
+
+func main() {
+	fmt.Printf("INFO typeName go=%s\n", toki.TypeNameOf(&packets.TestData{}))
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "FAIL crosstest error=%v\n", err)
+		os.Exit(1)
+	}
+	fmt.Println("PASS all go-server/dart-client crosstests passed")
+}
+
+func run() error {
+	if err := runTCPSendPush(); err != nil {
+		return err
+	}
+	time.Sleep(150 * time.Millisecond)
+	if err := runTCPRequests(); err != nil {
+		return err
+	}
+	time.Sleep(150 * time.Millisecond)
+	if err := runWSSendPush(); err != nil {
+		return err
+	}
+	time.Sleep(150 * time.Millisecond)
+	return runWSRequests()
+}
+
+func runTCPSendPush() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	received := make(chan bool, 1)
+	server := toki.NewTcpServer(host, goDartTCPPort, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, parserMap())
+	})
+	server.OnClientConnected = func(client *toki.TcpClient) {
+		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
+			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
+			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from dart client"
+			select {
+			case received <- valid:
+			default:
+			}
+			if valid {
+				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
+			}
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		return err
+	}
+	defer server.Stop()
+
+	if err := runDartClient("tcp", goDartTCPPort, "send-push", map[string]bool{"1": true, "2": true}); err != nil {
+		return err
+	}
+	// The subprocess only proves the client observed PASS lines. This short
+	// wait also confirms the server-side listener recorded the expected data.
+	select {
+	case ok := <-received:
+		if !ok {
+			return errors.New("TCP send-push server received unexpected data")
+		}
+	case <-time.After(serverObservationWindow):
+		return errors.New("TCP send-push server did not receive expected data")
+	}
+	return nil
+}
+
+func runTCPRequests() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := toki.NewTcpServer(host, goDartTCPPort, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, parserMap())
+	})
+	server.OnClientConnected = func(client *toki.TcpClient) {
+		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
+			return &packets.TestData{
+				Index:   req.GetIndex() * 2,
+				Message: "echo: " + req.GetMessage(),
+			}, nil
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		return err
+	}
+	defer server.Stop()
+
+	return runDartClient("tcp", goDartTCPPort, "requests", map[string]bool{"3": true, "4": true})
+}
+
+func runWSSendPush() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	received := make(chan bool, 1)
+	server := toki.NewWsServer(host, goDartWSPort, wsPath, func(conn *websocket.Conn) *toki.WsClient {
+		return toki.NewWsClient(conn, 0, 0, parserMap())
+	})
+	server.OnClientConnected = func(client *toki.WsClient) {
+		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(data *packets.TestData) {
+			fmt.Printf("SERVER_RECEIVED index=%d message=%s\n", data.GetIndex(), data.GetMessage())
+			valid := data.GetIndex() == 101 && data.GetMessage() == "fire from dart client"
+			select {
+			case received <- valid:
+			default:
+			}
+			if valid {
+				_ = client.Send(&packets.TestData{Index: 200, Message: "push from go server"})
+			}
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		return err
+	}
+	defer server.Stop()
+
+	if err := runDartClient("ws", goDartWSPort, "send-push", map[string]bool{"1": true, "2": true}); err != nil {
+		return err
+	}
+	// The subprocess only proves the client observed PASS lines. This short
+	// wait also confirms the server-side listener recorded the expected data.
+	select {
+	case ok := <-received:
+		if !ok {
+			return errors.New("WS send-push server received unexpected data")
+		}
+	case <-time.After(serverObservationWindow):
+		return errors.New("WS send-push server did not receive expected data")
+	}
+	return nil
+}
+
+func runWSRequests() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := toki.NewWsServer(host, goDartWSPort, wsPath, func(conn *websocket.Conn) *toki.WsClient {
+		return toki.NewWsClient(conn, 0, 0, parserMap())
+	})
+	server.OnClientConnected = func(client *toki.WsClient) {
+		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
+			return &packets.TestData{
+				Index:   req.GetIndex() * 2,
+				Message: "echo: " + req.GetMessage(),
+			}, nil
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		return err
+	}
+	defer server.Stop()
+
+	return runDartClient("ws", goDartWSPort, "requests", map[string]bool{"3": true, "4": true})
+}
+
+func runDartClient(mode string, port int, phase string, expected map[string]bool) error {
+	ctx, cancel := context.WithTimeout(context.Background(), processTimeout)
+	defer cancel()
+
+	dartDir, err := dartPackageDir()
+	if err != nil {
+		return err
+	}
+
+	cmd := exec.CommandContext(ctx, "dart", "run", "crosstest/go_dart_client.dart",
+		"--mode="+mode,
+		fmt.Sprintf("--port=%d", port),
+		"--phase="+phase,
+	)
+	cmd.Dir = dartDir
+
+	stdoutPipe, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	stderrPipe, err := cmd.StderrPipe()
+	if err != nil {
+		return err
+	}
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+
+	var mu sync.Mutex
+	var resultLines []string
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go scanLines(&wg, stdoutPipe, os.Stdout, &mu, &resultLines)
+	go scanLines(&wg, stderrPipe, os.Stderr, nil, nil)
+
+	waitErr := cmd.Wait()
+	wg.Wait()
+	if ctx.Err() == context.DeadlineExceeded {
+		return fmt.Errorf("dart client %s/%s timed out", mode, phase)
+	}
+	if err := validateResultLines("dart-client "+mode+"/"+phase, waitErr, resultLines, expected); err != nil {
+		return err
+	}
+	return nil
+}
+
+func dartPackageDir() (string, error) {
+	candidates := make([]string, 0, 3)
+
+	_, filename, _, ok := runtime.Caller(0)
+	if ok {
+		repoRoot := filepath.Dir(filepath.Dir(filepath.Dir(filename)))
+		candidates = append(candidates, filepath.Join(repoRoot, "dart"))
+	}
+	if wd, err := os.Getwd(); err == nil {
+		candidates = append(candidates, findDartPackageCandidates(wd)...)
+	}
+	if executable, err := os.Executable(); err == nil {
+		candidates = append(candidates, findDartPackageCandidates(filepath.Dir(executable))...)
+	}
+
+	for _, candidate := range candidates {
+		if isDartPackageDir(candidate) {
+			return candidate, nil
+		}
+	}
+	return "", fmt.Errorf("cannot resolve dart package directory from candidates %v", candidates)
+}
+
+func findDartPackageCandidates(start string) []string {
+	candidates := make([]string, 0)
+	for dir := start; ; dir = filepath.Dir(dir) {
+		candidates = append(candidates, filepath.Join(dir, "dart"))
+		if filepath.Base(dir) == "dart" {
+			candidates = append(candidates, dir)
+		}
+		parent := filepath.Dir(dir)
+		if parent == dir {
+			return candidates
+		}
+	}
+}
+
+func isDartPackageDir(dir string) bool {
+	info, err := os.Stat(filepath.Join(dir, "pubspec.yaml"))
+	return err == nil && !info.IsDir()
+}
+
+func scanLines(wg *sync.WaitGroup, reader io.Reader, writer *os.File, mu *sync.Mutex, resultLines *[]string) {
+	defer wg.Done()
+	scanner := bufio.NewScanner(reader)
+	for scanner.Scan() {
+		line := scanner.Text()
+		fmt.Fprintln(writer, line)
+		if mu != nil && (startsWithResult(line)) {
+			mu.Lock()
+			*resultLines = append(*resultLines, line)
+			mu.Unlock()
+		}
+	}
+}
+
+func validateResultLines(label string, waitErr error, lines []string, expected map[string]bool) error {
+	failed := make([]string, 0)
+	passed := make(map[string]bool)
+	re := regexp.MustCompile(`scenario=([^ ]+)`)
+	for _, line := range lines {
+		if len(line) >= 5 && line[:5] == "FAIL " {
+			failed = append(failed, line)
+			continue
+		}
+		if len(line) >= 5 && line[:5] == "PASS " {
+			match := re.FindStringSubmatch(line)
+			if len(match) == 2 {
+				passed[match[1]] = true
+			}
+		}
+	}
+
+	missing := make([]string, 0)
+	for scenario := range expected {
+		if !passed[scenario] {
+			missing = append(missing, scenario)
+		}
+	}
+	if waitErr != nil || len(failed) > 0 || len(missing) > 0 {
+		return fmt.Errorf("%s failed waitErr=%v failed=%v missing=%v", label, waitErr, failed, missing)
+	}
+	return nil
+}
+
+func startsWithResult(line string) bool {
+	return (len(line) >= 5 && line[:5] == "PASS ") || (len(line) >= 5 && line[:5] == "FAIL ")
+}
--- a/go/examples/tcp_echo/main.go
+++ b/go/examples/tcp_echo/main.go
--- a/go/examples/ws_echo/main.go
+++ b/go/examples/ws_echo/main.go
--- a/go/socket_test.go
+++ b/go/socket_test.go
@ -1,368 +0,0 @@
-package toki_socket
-
-import (
-	"context"
-	"crypto/ecdsa"
-	"crypto/elliptic"
-	"crypto/rand"
-	"crypto/tls"
-	"crypto/x509"
-	"crypto/x509/pkix"
-	"encoding/binary"
-	"encoding/pem"
-	"io"
-	"math/big"
-	"net"
-	"testing"
-	"time"
-
-	"google.golang.org/protobuf/proto"
-	"nhooyr.io/websocket"
-
-	"toki-labs.com/toki_socket/go/packets"
-)
-
-func TestTcpRequestResponse(t *testing.T) {
-	port := freePort(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	server := NewTcpServer("127.0.0.1", port, func(conn net.Conn) *TcpClient {
-		return NewTcpClient(conn, 0, 0, testParserMap())
-	})
-	server.OnClientConnected = func(client *TcpClient) {
-		AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
-			return &packets.TestData{
-				Index:   req.GetIndex() * 2,
-				Message: "echo: " + req.GetMessage(),
-			}, nil
-		})
-	}
-	if err := server.Start(ctx); err != nil {
-		t.Fatal(err)
-	}
-	defer server.Stop()
-
-	client, err := DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer client.Close()
-
-	res, err := SendRequestTyped[*packets.TestData, *packets.TestData](
-		&client.Communicator,
-		&packets.TestData{Index: 21, Message: "hello"},
-		2*time.Second,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if res.GetIndex() != 42 || res.GetMessage() != "echo: hello" {
-		t.Fatalf("unexpected response: %v", res)
-	}
-}
-
-func TestTcpBroadcast(t *testing.T) {
-	port := freePort(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	server := NewTcpServer("127.0.0.1", port, func(conn net.Conn) *TcpClient {
-		return NewTcpClient(conn, 0, 0, testParserMap())
-	})
-	if err := server.Start(ctx); err != nil {
-		t.Fatal(err)
-	}
-	defer server.Stop()
-
-	client, err := DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer client.Close()
-
-	received := make(chan *packets.TestData, 1)
-	AddListenerTyped[*packets.TestData](&client.Communicator, func(m *packets.TestData) {
-		received <- m
-	})
-
-	for deadline := time.Now().Add(time.Second); time.Now().Before(deadline) && len(server.Clients()) == 0; {
-		time.Sleep(time.Millisecond)
-	}
-	if err := server.Broadcast(&packets.TestData{Index: 9, Message: "broadcast"}); err != nil {
-		t.Fatal(err)
-	}
-
-	select {
-	case msg := <-received:
-		if msg.GetMessage() != "broadcast" {
-			t.Fatalf("unexpected broadcast: %v", msg)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("broadcast timed out")
-	}
-}
-
-func TestTcpBroadcastAttemptsAllClients(t *testing.T) {
-	badConn, badPeer := net.Pipe()
-	badClient := NewTcpClient(badConn, 0, 0, testParserMap())
-	_ = badClient.Close()
-	_ = badPeer.Close()
-
-	goodConn, goodPeer := net.Pipe()
-	defer goodPeer.Close()
-	goodClient := NewTcpClient(goodConn, 0, 0, testParserMap())
-	defer goodClient.Close()
-
-	received := make(chan *packets.PacketBase, 1)
-	go func() {
-		base, err := readTCPPacket(goodPeer)
-		if err == nil {
-			received <- base
-		}
-	}()
-
-	server := &TcpServer{clients: []*TcpClient{badClient, goodClient}}
-	err := server.Broadcast(&packets.TestData{Index: 9, Message: "best effort"})
-	if err == nil {
-		t.Fatal("expected broadcast to report the failed client")
-	}
-
-	select {
-	case base := <-received:
-		if base.GetTypeName() != TypeNameOf(&packets.TestData{}) {
-			t.Fatalf("unexpected packet type: %s", base.GetTypeName())
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("broadcast did not attempt the healthy client")
-	}
-}
-
-func TestWsRequestResponse(t *testing.T) {
-	port := freePort(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	server := NewWsServer("127.0.0.1", port, "/", func(conn *websocket.Conn) *WsClient {
-		return NewWsClient(conn, 0, 0, testParserMap())
-	})
-	server.OnClientConnected = func(client *WsClient) {
-		AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
-			return &packets.TestData{
-				Index:   req.GetIndex() * 2,
-				Message: "echo: " + req.GetMessage(),
-			}, nil
-		})
-	}
-	if err := server.Start(ctx); err != nil {
-		t.Fatal(err)
-	}
-	defer server.Stop()
-
-	client, err := DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer client.Close()
-
-	res, err := SendRequestTyped[*packets.TestData, *packets.TestData](
-		&client.Communicator,
-		&packets.TestData{Index: 21, Message: "hello ws"},
-		2*time.Second,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if res.GetIndex() != 42 || res.GetMessage() != "echo: hello ws" {
-		t.Fatalf("unexpected response: %v", res)
-	}
-}
-
-func TestTcpSendReceive(t *testing.T) {
-	port := freePort(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	received := make(chan *packets.TestData, 1)
-	server := NewTcpServer("127.0.0.1", port, func(conn net.Conn) *TcpClient {
-		return NewTcpClient(conn, 0, 0, testParserMap())
-	})
-	server.OnClientConnected = func(client *TcpClient) {
-		AddListenerTyped[*packets.TestData](&client.Communicator, func(m *packets.TestData) {
-			received <- m
-		})
-	}
-	if err := server.Start(ctx); err != nil {
-		t.Fatal(err)
-	}
-	defer server.Stop()
-
-	client, err := DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer client.Close()
-
-	if err := client.Send(&packets.TestData{Index: 42, Message: "hello toki-socket"}); err != nil {
-		t.Fatal(err)
-	}
-
-	select {
-	case msg := <-received:
-		if msg.GetIndex() != 42 {
-			t.Fatalf("unexpected message: %v", msg)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("receive timed out")
-	}
-}
-
-func TestHeartbeatDisconnectsWithoutResponse(t *testing.T) {
-	clientConn, peerConn := net.Pipe()
-	defer peerConn.Close()
-
-	go func() {
-		_, _ = io.Copy(io.Discard, peerConn)
-	}()
-
-	client := NewTcpClient(clientConn, 1, 1, testParserMap())
-	defer client.Close()
-
-	disconnected := make(chan struct{}, 1)
-	client.AddDisconnectListener(func(*TcpClient) {
-		disconnected <- struct{}{}
-	})
-
-	select {
-	case <-disconnected:
-	case <-time.After(3500 * time.Millisecond):
-		t.Fatal("heartbeat timeout did not disconnect client")
-	}
-	if client.IsAlive() {
-		t.Fatal("client is still marked alive after heartbeat timeout")
-	}
-}
-
-func TestTypeNameMatchesDartConvention(t *testing.T) {
-	if got := TypeNameOf(&packets.TestData{}); got != "TestData" {
-		t.Fatalf("type name = %q, want TestData", got)
-	}
-}
-
-func freePort(t *testing.T) int {
-	t.Helper()
-	ln, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer ln.Close()
-	return ln.Addr().(*net.TCPAddr).Port
-}
-
-func readTCPPacket(conn net.Conn) (*packets.PacketBase, error) {
-	header := make([]byte, 4)
-	if _, err := io.ReadFull(conn, header); err != nil {
-		return nil, err
-	}
-	length := binary.BigEndian.Uint32(header)
-	packetBytes := make([]byte, int(length))
-	if _, err := io.ReadFull(conn, packetBytes); err != nil {
-		return nil, err
-	}
-	base := &packets.PacketBase{}
-	if err := proto.Unmarshal(packetBytes, base); err != nil {
-		return nil, err
-	}
-	return base, nil
-}
-
-// generateSelfSignedCert creates an in-memory self-signed certificate valid
-// for 127.0.0.1. Returns the server tls.Config and a client tls.Config that
-// trusts the generated certificate.
-func generateSelfSignedCert(t *testing.T) (serverCfg, clientCfg *tls.Config) {
-	t.Helper()
-
-	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	tmpl := &x509.Certificate{
-		SerialNumber: big.NewInt(1),
-		Subject:      pkix.Name{CommonName: "toki-socket-test"},
-		NotBefore:    time.Now().Add(-time.Hour),
-		NotAfter:     time.Now().Add(time.Hour),
-		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
-	}
-
-	certDER, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &priv.PublicKey, priv)
-	if err != nil {
-		t.Fatal(err)
-	}
-	privDER, err := x509.MarshalECPrivateKey(priv)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
-	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: privDER})
-
-	cert, err := tls.X509KeyPair(certPEM, keyPEM)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	serverCfg = &tls.Config{Certificates: []tls.Certificate{cert}}
-
-	pool := x509.NewCertPool()
-	parsed, err := x509.ParseCertificate(certDER)
-	if err != nil {
-		t.Fatal(err)
-	}
-	pool.AddCert(parsed)
-	clientCfg = &tls.Config{RootCAs: pool, ServerName: "127.0.0.1"}
-
-	return serverCfg, clientCfg
-}
-
-func TestTLSTcp(t *testing.T) {
-	port := freePort(t)
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	serverTLS, clientTLS := generateSelfSignedCert(t)
-
-	server := NewTcpServerTLS("127.0.0.1", port, serverTLS, func(conn net.Conn) *TcpClient {
-		return NewTcpClient(conn, 0, 0, testParserMap())
-	})
-	server.OnClientConnected = func(client *TcpClient) {
-		AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
-			return &packets.TestData{
-				Index:   req.GetIndex() * 2,
-				Message: "tls-echo: " + req.GetMessage(),
-			}, nil
-		})
-	}
-	if err := server.Start(ctx); err != nil {
-		t.Fatal(err)
-	}
-	defer server.Stop()
-
-	client, err := DialTcpTLS(ctx, "127.0.0.1", port, clientTLS, 0, 0, testParserMap())
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer client.Close()
-
-	res, err := SendRequestTyped[*packets.TestData, *packets.TestData](
-		&client.Communicator,
-		&packets.TestData{Index: 7, Message: "hello tls"},
-		2*time.Second,
-	)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if res.GetIndex() != 14 || res.GetMessage() != "tls-echo: hello tls" {
-		t.Fatalf("unexpected response: %v", res)
-	}
-}
--- a/go/test/communicator_test.go
+++ b/go/test/communicator_test.go
@ -1,4 +1,4 @@
-package toki_socket
+package toki_socket_test

 import (
 	"sync"
@ -7,6 +7,7 @@ import (

 	"google.golang.org/protobuf/proto"

+	toki "toki-labs.com/toki_socket/go"
 	"toki-labs.com/toki_socket/go/packets"
 )

@ -32,9 +33,9 @@ func (f *fakeTransport) sent() []*packets.PacketBase {
 	return append([]*packets.PacketBase{}, f.packets...)
 }

-func testParserMap() ParserMap {
-	return ParserMap{
-		TypeNameOf(&packets.TestData{}): func(b []byte) (proto.Message, error) {
+func testParserMap() toki.ParserMap {
+	return toki.ParserMap{
+		toki.TypeNameOf(&packets.TestData{}): func(b []byte) (proto.Message, error) {
 			m := &packets.TestData{}
 			return m, proto.Unmarshal(b, m)
 		},
@ -43,12 +44,12 @@ func testParserMap() ParserMap {

 func TestSendRequestTypeMismatch(t *testing.T) {
 	transport := &fakeTransport{}
-	communicator := NewCommunicator(transport, testParserMap())
-	defer communicator.shutdown()
+	communicator := toki.NewCommunicator(transport, testParserMap())
+	defer communicator.Close()

 	errCh := make(chan error, 1)
 	go func() {
-		_, err := SendRequestTyped[*packets.TestData, *packets.TestData](
+		_, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
 			communicator,
 			&packets.TestData{Index: 1, Message: "hello"},
 			time.Second,
@ -73,7 +74,7 @@ func TestSendRequestTypeMismatch(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	communicator.OnReceivedData(TypeNameOf(&packets.HeartBeat{}), hb, 0, requestNonce)
+	communicator.OnReceivedData(toki.TypeNameOf(&packets.HeartBeat{}), hb, 0, requestNonce)

 	err = <-errCh
 	if err == nil {
@ -83,17 +84,17 @@ func TestSendRequestTypeMismatch(t *testing.T) {

 func TestListenerAndRequestListenerConflict(t *testing.T) {
 	transport := &fakeTransport{}
-	communicator := NewCommunicator(transport, testParserMap())
-	defer communicator.shutdown()
+	communicator := toki.NewCommunicator(transport, testParserMap())
+	defer communicator.Close()

-	AddListenerTyped[*packets.TestData](communicator, func(*packets.TestData) {})
+	toki.AddListenerTyped[*packets.TestData](communicator, func(*packets.TestData) {})

 	defer func() {
 		if recover() == nil {
 			t.Fatal("expected panic")
 		}
 	}()
-	AddRequestListenerTyped[*packets.TestData, *packets.TestData](communicator, func(req *packets.TestData) (*packets.TestData, error) {
+	toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](communicator, func(req *packets.TestData) (*packets.TestData, error) {
 		return req, nil
 	})
 }
--- a/go/test/heartbeat_test.go
+++ b/go/test/heartbeat_test.go
@ -0,0 +1,36 @@
+package toki_socket_test
+
+import (
+	"io"
+	"net"
+	"testing"
+	"time"
+
+	toki "toki-labs.com/toki_socket/go"
+)
+
+func TestHeartbeatDisconnectsWithoutResponse(t *testing.T) {
+	clientConn, peerConn := net.Pipe()
+	defer peerConn.Close()
+
+	go func() {
+		_, _ = io.Copy(io.Discard, peerConn)
+	}()
+
+	client := toki.NewTcpClient(clientConn, 1, 1, testParserMap())
+	defer client.Close()
+
+	disconnected := make(chan struct{}, 1)
+	client.AddDisconnectListener(func(*toki.TcpClient) {
+		disconnected <- struct{}{}
+	})
+
+	select {
+	case <-disconnected:
+	case <-time.After(3500 * time.Millisecond):
+		t.Fatal("heartbeat timeout did not disconnect client")
+	}
+	if client.IsAlive() {
+		t.Fatal("client is still marked alive after heartbeat timeout")
+	}
+}
--- a/go/test/tcp_test.go
+++ b/go/test/tcp_test.go
@ -0,0 +1,131 @@
+package toki_socket_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+func TestTcpRequestResponse(t *testing.T) {
+	port := freePort(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, testParserMap())
+	})
+	server.OnClientConnected = func(client *toki.TcpClient) {
+		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
+			return &packets.TestData{
+				Index:   req.GetIndex() * 2,
+				Message: "echo: " + req.GetMessage(),
+			}, nil
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer server.Stop()
+
+	client, err := toki.DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
+		&client.Communicator,
+		&packets.TestData{Index: 21, Message: "hello"},
+		2*time.Second,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.GetIndex() != 42 || res.GetMessage() != "echo: hello" {
+		t.Fatalf("unexpected response: %v", res)
+	}
+}
+
+func TestTcpBroadcast(t *testing.T) {
+	port := freePort(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, testParserMap())
+	})
+	if err := server.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer server.Stop()
+
+	client, err := toki.DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	received := make(chan *packets.TestData, 1)
+	toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(m *packets.TestData) {
+		received <- m
+	})
+
+	for deadline := time.Now().Add(time.Second); time.Now().Before(deadline) && len(server.Clients()) == 0; {
+		time.Sleep(time.Millisecond)
+	}
+	if err := server.Broadcast(&packets.TestData{Index: 9, Message: "broadcast"}); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case msg := <-received:
+		if msg.GetMessage() != "broadcast" {
+			t.Fatalf("unexpected broadcast: %v", msg)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("broadcast timed out")
+	}
+}
+
+func TestTcpSendReceive(t *testing.T) {
+	port := freePort(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	received := make(chan *packets.TestData, 1)
+	server := toki.NewTcpServer("127.0.0.1", port, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, testParserMap())
+	})
+	server.OnClientConnected = func(client *toki.TcpClient) {
+		toki.AddListenerTyped[*packets.TestData](&client.Communicator, func(m *packets.TestData) {
+			received <- m
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer server.Stop()
+
+	client, err := toki.DialTcp(ctx, "127.0.0.1", port, 0, 0, testParserMap())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	if err := client.Send(&packets.TestData{Index: 42, Message: "hello toki-socket"}); err != nil {
+		t.Fatal(err)
+	}
+
+	select {
+	case msg := <-received:
+		if msg.GetIndex() != 42 {
+			t.Fatalf("unexpected message: %v", msg)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("receive timed out")
+	}
+}
--- a/go/test/test_helpers_test.go
+++ b/go/test/test_helpers_test.go
@ -0,0 +1,83 @@
+package toki_socket_test
+
+import (
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"net"
+	"testing"
+	"time"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+func TestTypeNameMatchesDartConvention(t *testing.T) {
+	if got := toki.TypeNameOf(&packets.TestData{}); got != "TestData" {
+		t.Fatalf("type name = %q, want TestData", got)
+	}
+}
+
+func freePort(t *testing.T) int {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	return ln.Addr().(*net.TCPAddr).Port
+}
+
+// generateSelfSignedCert creates an in-memory self-signed certificate valid
+// for 127.0.0.1. Returns the server tls.Config and a client tls.Config that
+// trusts the generated certificate.
+func generateSelfSignedCert(t *testing.T) (serverCfg, clientCfg *tls.Config) {
+	t.Helper()
+
+	priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tmpl := &x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject:      pkix.Name{CommonName: "toki-socket-test"},
+		NotBefore:    time.Now().Add(-time.Hour),
+		NotAfter:     time.Now().Add(time.Hour),
+		IPAddresses:  []net.IP{net.ParseIP("127.0.0.1")},
+	}
+
+	certDER, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &priv.PublicKey, priv)
+	if err != nil {
+		t.Fatal(err)
+	}
+	privDER, err := x509.MarshalECPrivateKey(priv)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
+	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: privDER})
+
+	cert, err := tls.X509KeyPair(certPEM, keyPEM)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	serverCfg = &tls.Config{Certificates: []tls.Certificate{cert}}
+
+	pool := x509.NewCertPool()
+	parsed, err := x509.ParseCertificate(certDER)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pool.AddCert(parsed)
+	clientCfg = &tls.Config{RootCAs: pool, ServerName: "127.0.0.1"}
+
+	return serverCfg, clientCfg
+}
--- a/go/test/tls_test.go
+++ b/go/test/tls_test.go
@ -0,0 +1,53 @@
+package toki_socket_test
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+func TestTLSTcp(t *testing.T) {
+	port := freePort(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	serverTLS, clientTLS := generateSelfSignedCert(t)
+
+	server := toki.NewTcpServerTLS("127.0.0.1", port, serverTLS, func(conn net.Conn) *toki.TcpClient {
+		return toki.NewTcpClient(conn, 0, 0, testParserMap())
+	})
+	server.OnClientConnected = func(client *toki.TcpClient) {
+		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
+			return &packets.TestData{
+				Index:   req.GetIndex() * 2,
+				Message: "tls-echo: " + req.GetMessage(),
+			}, nil
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer server.Stop()
+
+	client, err := toki.DialTcpTLS(ctx, "127.0.0.1", port, clientTLS, 0, 0, testParserMap())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
+		&client.Communicator,
+		&packets.TestData{Index: 7, Message: "hello tls"},
+		2*time.Second,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.GetIndex() != 14 || res.GetMessage() != "tls-echo: hello tls" {
+		t.Fatalf("unexpected response: %v", res)
+	}
+}
--- a/go/test/ws_test.go
+++ b/go/test/ws_test.go
@ -0,0 +1,52 @@
+package toki_socket_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"nhooyr.io/websocket"
+
+	toki "toki-labs.com/toki_socket/go"
+	"toki-labs.com/toki_socket/go/packets"
+)
+
+func TestWsRequestResponse(t *testing.T) {
+	port := freePort(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := toki.NewWsServer("127.0.0.1", port, "/", func(conn *websocket.Conn) *toki.WsClient {
+		return toki.NewWsClient(conn, 0, 0, testParserMap())
+	})
+	server.OnClientConnected = func(client *toki.WsClient) {
+		toki.AddRequestListenerTyped[*packets.TestData, *packets.TestData](&client.Communicator, func(req *packets.TestData) (*packets.TestData, error) {
+			return &packets.TestData{
+				Index:   req.GetIndex() * 2,
+				Message: "echo: " + req.GetMessage(),
+			}, nil
+		})
+	}
+	if err := server.Start(ctx); err != nil {
+		t.Fatal(err)
+	}
+	defer server.Stop()
+
+	client, err := toki.DialWsWithHeartbeat(ctx, "127.0.0.1", port, "/", 0, 0, testParserMap())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer client.Close()
+
+	res, err := toki.SendRequestTyped[*packets.TestData, *packets.TestData](
+		&client.Communicator,
+		&packets.TestData{Index: 21, Message: "hello ws"},
+		2*time.Second,
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if res.GetIndex() != 42 || res.GetMessage() != "echo: hello ws" {
+		t.Fatalf("unexpected response: %v", res)
+	}
+}
--- a/skills/add-crosstest-language/SKILL.md
+++ b/skills/add-crosstest-language/SKILL.md
@ -0,0 +1,123 @@
+---
+name: add-crosstest-language
+description: Create or extend toki_socket cross-language integration tests for a new language implementation. Use when adding folders, orchestrators, subprocess clients, or review notes for Dart/Go/other language protocol compatibility tests, especially tests that must verify TCP and WebSocket send, push, request-response, concurrent request nonce mapping, and protobuf typeName compatibility without cluttering the repository root.
+---
+
+# Add Crosstest Language
+
+## Overview
+
+Use this skill to add cross-language compatibility tests for a new `toki_socket` language implementation. Keep crosstest code inside each language package, preserve the root for shared documentation, and match the Dart/Go baseline behavior.
+
+## Folder Rules
+
+Place each language-specific runner or helper under that language's `crosstest` folder.
+
+Use this naming pattern:
+
+```text
+<server-language>/crosstest/<server>_<client>.<ext>
+<client-language>/crosstest/<server>_<client>_client.<ext-or-main-dir>
+```
+
+Examples from the baseline:
+
+```text
+dart/crosstest/dart_go.dart
+go/crosstest/dart_go_client/main.go
+go/crosstest/go_dart.go
+dart/crosstest/go_dart_client.dart
+```
+
+Avoid root-level runners such as `dart-go.dart` or `go-dart.go` unless the user explicitly asks for them. Do not add root-level package metadata (`pubspec.yaml`, `go.mod`, etc.) only to run crosstests. Run crosstests from the owning language package instead.
+
+Keep only shared review notes or planning docs at the repository root. When changing paths from a plan, record why in the review note.
+
+## Test Shape
+
+Create one orchestrator in the server language and one subprocess client helper in the client language.
+
+The orchestrator must:
+
+- Start a TCP server and run the client helper as a subprocess.
+- Start a WebSocket server and run the client helper as a subprocess.
+- Parse subprocess stdout for `PASS` and `FAIL` lines.
+- Fail if any expected scenario is missing, any `FAIL` line appears, or the subprocess exits non-zero.
+- Stop servers even when a subprocess fails.
+
+The subprocess client must:
+
+- Connect to the requested mode and port.
+- Print `INFO typeName <language>=TestData` before scenarios.
+- Print exactly one result line per scenario in this format:
+
+```text
+PASS scenario=N detail=...
+FAIL scenario=N error=...
+```
+
+## Required Scenarios
+
+Run these scenarios for both TCP and WebSocket:
+
+| Scenario | Requirement |
+| --- | --- |
+| 1 | Client sends fire-and-forget `TestData`; server validates `index` and `message`. |
+| 2 | Server pushes `TestData(index=200, message=push from ... server)` to the client. |
+| 3 | Client `sendRequest` receives `index=req.index*2` and `message=echo: req.message`. |
+| 4 | Multiple concurrent `sendRequest` calls verify nonce/responseNonce routing. |
+
+Split each protocol into separate `send-push` and `requests` phases when the language implementation forbids registering a normal listener and request listener for the same protobuf type at the same time. This is required for the current Dart and Go communicators.
+
+## Protocol Details
+
+Use the generated `TestData` protobuf from each language package. The current proto has no `package` declaration, so both sides must use the simple type name:
+
+```text
+Dart: TestData.getDefault().info_.qualifiedMessageName -> TestData
+Go:   toki.TypeNameOf(&packets.TestData{})              -> TestData
+```
+
+Log the type name from both sides during each crosstest run.
+
+Use fixed non-conflicting ports per language pair and protocol. Prefer a new documented range instead of reusing unit-test ports. If extending the Dart/Go baseline, keep:
+
+```text
+Dart server / Go client TCP: 29090
+Dart server / Go client WS:  29092
+Go server / Dart client TCP: 29190
+Go server / Dart client WS:  29192
+```
+
+## Path Robustness
+
+Resolve subprocess paths from the orchestrator source location or from the language package root. Do not assume the user starts the command from the repository root.
+
+Prefer commands like:
+
+```bash
+(cd dart && dart run crosstest/dart_go.dart)
+(cd go && go run ./crosstest)
+```
+
+For additional languages, provide the equivalent package-local command in the review note.
+
+## Verification
+
+After implementation, run:
+
+- The new cross-language command in both directions.
+- The language package's existing unit tests.
+- The language package's formatter/analyzer/compiler checks.
+
+If the local environment needs special PATH, cache, or sandbox settings, report that separately in the final answer and review note without baking machine-specific paths into the source.
+
+## Review Note
+
+Leave or update a root-level review note when the task requests reviewer guidance. Include:
+
+- Added files and their roles.
+- Path decision and any deviation from an earlier plan.
+- Scenario coverage.
+- Exact package-local commands used for validation.
+- Any environment-specific caveats.
--- a/skills/add-crosstest-language/agents/openai.yaml
+++ b/skills/add-crosstest-language/agents/openai.yaml
@ -0,0 +1,4 @@
+interface:
+  display_name: "Add Crosstest Language"
+  short_description: "Add language crosstests consistently"
+  default_prompt: "Use $add-crosstest-language to add cross-language integration tests for a new toki_socket implementation."