feat: add Dart-Kotlin crosstest and improve WS protobuf server

- Add Dart-Kotlin crosstest implementation (dart_kotlin.dart, kotlin_dart.kt)
- Add Dart-Kotlin client implementation for Go (kotlin_dart_client.kt, dart_kotlin_client.dart)
- Update SKILL.md and openai.yaml for crosstest language support
- Improve ws_protobuf_server.dart with better error handling
- Fix crosstest client implementations for Go

agent-ops/skills/project/add-toki-socket-crosstest-language/SKILL.md

@@ -1,14 +1,33 @@
 ---
 name: add-toki-socket-crosstest-language
 version: 1.0.0
-description: Create new-language implementation scaffolding and extend toki_socket cross-language integration tests. Use when adding language package README/checklist templates, crosstest 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.
+description: Plan and create new-language implementation scaffolding and toki_socket language-matrix integration tests. Use when adding language package README/checklist templates, crosstest folders, server-row orchestrators, subprocess clients, or review notes for Dart/Go/other language protocol compatibility tests, especially tests that must verify every supported server/client language pair over TCP and WebSocket for send, push, request-response, concurrent request nonce mapping, and protobuf typeName compatibility without cluttering the repository root.
 ---
 
 # Add Toki Socket Crosstest Language
 
 ## Purpose
 
-Create the package-local scaffold for a new Toki Socket language implementation and add its cross-language compatibility tests. Keep implementation checklists and crosstest code inside the relevant language package or this skill's templates, preserve the repository root for shared docs, and match the Dart/Go baseline behavior.
+Create the package-local scaffold for a new Toki Socket language implementation and add or update the full language compatibility matrix. Keep implementation checklists and crosstest code inside the relevant language package or this skill's templates, preserve the repository root for shared docs, and match the Dart/Go baseline behavior.
+
+## Plan-First Workflow
+
+For implementation work, follow the `agent-ops/skills/common/plan/SKILL.md` loop before editing crosstest or language source files:
+
+```text
+plan skill -> agent-task/{task_name}/PLAN.md + CODE_REVIEW.md stub
+implementation -> code changes + filled CODE_REVIEW.md
+code-review skill -> archive or follow-up plan
+```
+
+Do not create plan files for casual analysis, status checks, or review-only requests unless the user explicitly asks for a plan. When creating a crosstest plan:
+
+- Use a short snake_case task name under `agent-task/`.
+- Read every orchestrator, helper, package test, and manifest that the change will touch before writing `PLAN.md`.
+- Include the language matrix as a server-row/client-cell table.
+- For each matrix gap, state whether it needs new crosstest code, existing same-language tests already cover it, or it is deferred with a reason.
+- Include exact package-local validation commands and expected outcomes.
+- Create the paired `CODE_REVIEW.md` stub so the implementation pass can record decisions and command output.
 
 ## Language Scaffold
 
@@ -40,14 +59,33 @@ dart/crosstest/go_dart_client.dart
 
 Avoid root-level runners and root-level package metadata unless explicitly requested. If a plan path changes, record why in the review note.
 
+## Language Matrix
+
+Test by server-language rows. For every supported language, run it once as the server, then run every supported language as a client against that server for TCP and WebSocket.
+
+```text
+server=dart:   client=dart, client=go, client=kotlin, ...
+server=go:     client=dart, client=go, client=kotlin, ...
+server=kotlin: client=dart, client=go, client=kotlin, ...
+```
+
+When adding a new language, update the matrix in both dimensions:
+
+- Existing servers must run the new language as a client.
+- The new language must run as a server for every supported client language.
+- Same-language cells (`server == client`) are part of the matrix. They may be satisfied by package-local same-language tests only when those tests cover the same TCP/WS scenario set; otherwise add package-local crosstest entries.
+- Do not mark a language Available until its matrix row and column pass, or deferred cells are explicitly documented with reasons.
+
+Pairwise files are still allowed, but verification must be reported as completed server-language rows so omissions are visible.
+
 ## Test Shape
 
-Create a server-language orchestrator plus a client-language subprocess helper.
+Create a server-language orchestrator plus client-language subprocess helpers. Prefer a server-row orchestrator that runs all clients for that server; a pairwise orchestrator is acceptable only when the review note lists the full matrix cells covered by all commands.
 
 The orchestrator must:
 
 - Start TCP and WebSocket servers.
-- Run the client helper for each protocol/phase.
+- Run each required client helper for each protocol/phase.
 - Parse subprocess stdout for `PASS` and `FAIL`.
 - Fail on missing expected scenario, any `FAIL`, or non-zero subprocess exit.
 - Stop servers even on subprocess failure.
@@ -126,7 +164,7 @@ For new languages, include equivalent commands in the review note.
 
 Run:
 
-- New cross-language command in both directions.
+- Full affected matrix rows and columns, not only the new pair in both directions.
 - Existing unit tests for each touched language package.
 - Formatter/analyzer/compiler checks for each touched language package.
 
@@ -138,6 +176,7 @@ When reviewer guidance is requested, include:
 
 - Added files and roles.
 - Path decisions and deviations.
+- Matrix coverage by server row and client cell, including same-language cells.
 - Scenario coverage, including secure transports or deferral reason.
 - Exact package-local validation commands.
 - Environment-specific caveats.

agent-ops/skills/project/add-toki-socket-crosstest-language/agents/openai.yaml

@@ -1,4 +1,4 @@
 interface:
   display_name: "Add Toki Socket Crosstest Language"
-  short_description: "Add Toki Socket crosstests"
-  default_prompt: "Use $add-toki-socket-crosstest-language to add cross-language integration tests for a new toki_socket implementation."
+  short_description: "Plan and add Toki Socket matrix crosstests"
+  default_prompt: "Use $add-toki-socket-crosstest-language to plan and add language-matrix integration tests for a new toki_socket implementation."

dart/crosstest/dart_kotlin.dart

@@ -0,0 +1,228 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:toki_socket/toki_socket.dart';
+
+const _host = '127.0.0.1';
+const _tcpPort = 29590;
+const _wsPort = 29592;
+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 _kotlinDir = Directory('$_repoRoot/kotlin').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/kotlin-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 kotlin 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 _runKotlinClient('tcp', _tcpPort, 'send-push', {'1', '2'});
+    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,
+    () => _runKotlinClient('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 kotlin 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 _runKotlinClient('ws', _wsPort, 'send-push', {'1', '2'});
+    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,
+    () => _runKotlinClient('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> _runKotlinClient(
+    String mode, int port, String phase, Set<String> expectedScenarios) async {
+  final process = await Process.start(
+    './gradlew',
+    [
+      'run',
+      '-PmainClass=com.tokilabs.toki_socket.crosstest.DartKotlinClientKt',
+      '--args=--mode=$mode --port=$port --phase=$phase',
+    ],
+    workingDirectory: _kotlinDir,
+  );
+
+  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(
+      'kotlin-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');
+  }
+}

dart/crosstest/kotlin_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, client.close);
+        case 'ws':
+          final ws = await WsProtobufClient.connect(_host, port, path: _wsPath)
+              .timeout(const Duration(milliseconds: 300));
+          final client = _WsClient(ws);
+          return _ClientHandle(client, client.close);
+        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 kotlin server') {
+      _fail(
+          '2', 'unexpected push index=${push.index} message="${push.message}"');
+      return false;
+    }
+    _pass('2', 'received push from kotlin 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');
+}

dart/lib/src/ws_protobuf_server.dart

@@ -45,7 +45,10 @@ abstract class WsProtobufServer {
     _started = true;
     _server!.listen((HttpRequest request) async {
       if (WebSocketTransformer.isUpgradeRequest(request)) {
-        final ws = await WebSocketTransformer.upgrade(request);
+        final ws = await WebSocketTransformer.upgrade(
+          request,
+          compression: CompressionOptions.compressionOff,
+        );
         _onClientWebSocket(ws);
       }
     });

go/crosstest/kotlin_go_client/main.go

@@ -15,10 +15,11 @@ import (
 )
 
 const (
-	host          = "127.0.0.1"
-	wsPath        = "/"
-	connectWindow = 3 * time.Second
-	requestWindow = 2 * time.Second
+	host             = "127.0.0.1"
+	wsPath           = "/"
+	connectWindow    = 3 * time.Second
+	requestWindow    = 2 * time.Second
+	serverReadyDelay = 50 * time.Millisecond
 )
 
 type clientHandle struct {
@@ -55,6 +56,7 @@ func main() {
 		os.Exit(1)
 	}
 	defer client.close()
+	time.Sleep(serverReadyDelay)
 
 	var ok bool
 	switch *phase {

kotlin/crosstest/dart_kotlin_client.kt

@@ -0,0 +1,210 @@
+@file:JvmName("DartKotlinClientKt")
+
+package com.tokilabs.toki_socket.crosstest
+
+import com.tokilabs.toki_socket.DialTcp
+import com.tokilabs.toki_socket.DialWs
+import com.tokilabs.toki_socket.ParserMap
+import com.tokilabs.toki_socket.WsClient
+import com.tokilabs.toki_socket.TcpClient
+import com.tokilabs.toki_socket.addListenerTyped
+import com.tokilabs.toki_socket.sendRequestTyped
+import com.tokilabs.toki_socket.typeNameOf
+import com.tokilabs.toki_socket.packets.TestData
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlin.system.exitProcess
+
+private const val HOST = "127.0.0.1"
+private const val WS_PATH = "/"
+private const val CONNECT_WINDOW_MS = 3_000L
+private const val REQUEST_WINDOW_MS = 2_000L
+
+private interface DartClientHandle {
+    suspend fun send(data: TestData)
+
+    suspend fun close()
+
+    val communicator: com.tokilabs.toki_socket.Communicator
+}
+
+private class DartTcpHandle(
+    private val client: TcpClient,
+) : DartClientHandle {
+    override val communicator = client.communicator
+
+    override suspend fun send(data: TestData) {
+        client.send(data)
+    }
+
+    override suspend fun close() {
+        client.close()
+    }
+}
+
+private class DartWsHandle(
+    private val client: WsClient,
+) : DartClientHandle {
+    override val communicator = client.communicator
+
+    override suspend fun send(data: TestData) {
+        client.send(data)
+    }
+
+    override suspend fun close() {
+        client.close()
+    }
+}
+
+fun main(args: Array<String>) = runBlocking {
+    val mode = argValue(args, "mode") ?: "tcp"
+    val phase = argValue(args, "phase") ?: "send-push"
+    val port = argValue(args, "port")?.toIntOrNull()
+
+    println("INFO typeName kotlin=${typeNameOf<TestData>()}")
+
+    if (port == null) {
+        fail("setup", "port is required")
+        exitProcess(1)
+    }
+
+    val client = try {
+        connectWithRetry(mode, port)
+    } catch (error: Throwable) {
+        fail("setup", error.message ?: error.toString())
+        exitProcess(1)
+    }
+
+    val ok: Boolean = try {
+        when (phase) {
+            "send-push" -> runSendPush(client)
+            "requests" -> runRequests(client)
+            else -> {
+                fail("setup", "unknown phase $phase")
+                false
+            }
+        }
+    } finally {
+        client.close()
+    }
+
+    if (!ok) exitProcess(1)
+}
+
+private suspend fun connectWithRetry(mode: String, port: Int): DartClientHandle {
+    val deadline = System.nanoTime() + CONNECT_WINDOW_MS * 1_000_000L
+    var lastError: Throwable? = null
+    while (System.nanoTime() < deadline) {
+        try {
+            return when (mode) {
+                "tcp" -> DartTcpHandle(DialTcp(HOST, port, 0, 0, parserMap()))
+                "ws" -> DartWsHandle(DialWs(HOST, port, WS_PATH, 0, 0, parserMap()))
+                else -> error("unknown mode $mode")
+            }
+        } catch (error: Throwable) {
+            lastError = error
+            delay(100)
+        }
+    }
+    error("connect $mode:$port timed out: $lastError")
+}
+
+private suspend fun runSendPush(client: DartClientHandle): Boolean {
+    val push = CompletableDeferred<TestData>()
+    addListenerTyped<TestData>(client.communicator) {
+        push.complete(it)
+    }
+
+    return try {
+        client.send(
+            TestData.newBuilder()
+                .setIndex(101)
+                .setMessage("fire from kotlin client")
+                .build(),
+        )
+        pass("1", "fire-and-forget sent")
+
+        val msg = withTimeout(REQUEST_WINDOW_MS) { push.await() }
+        if (msg.index != 200 || msg.message != "push from dart server") {
+            fail("2", "unexpected push index=${msg.index} message=${msg.message}")
+            false
+        } else {
+            pass("2", "received push from dart server")
+            true
+        }
+    } catch (error: Throwable) {
+        fail("2", error.message ?: error.toString())
+        false
+    }
+}
+
+private suspend fun runRequests(client: DartClientHandle): Boolean =
+    try {
+        val single = sendRequestTyped<TestData, TestData>(
+            client.communicator,
+            TestData.newBuilder()
+                .setIndex(21)
+                .setMessage("single request from kotlin")
+                .build(),
+            REQUEST_WINDOW_MS,
+        )
+        if (single.index != 42 || single.message != "echo: single request from kotlin") {
+            fail("3", "unexpected response index=${single.index} message=${single.message}")
+            false
+        } else {
+            pass("3", "single request response matched")
+            runConcurrentRequests(client)
+        }
+    } catch (error: Throwable) {
+        fail("3", error.message ?: error.toString())
+        false
+    }
+
+private suspend fun runConcurrentRequests(client: DartClientHandle): Boolean =
+    coroutineScope {
+        val jobs = (0 until 5).map { i ->
+            async {
+                val index = 30 + i
+                val message = "multi request $i from kotlin"
+                val res = sendRequestTyped<TestData, TestData>(
+                    client.communicator,
+                    TestData.newBuilder()
+                        .setIndex(index)
+                        .setMessage(message)
+                        .build(),
+                    REQUEST_WINDOW_MS,
+                )
+                check(res.index == index * 2 && res.message == "echo: $message") {
+                    "request $i got index=${res.index} message=${res.message}"
+                }
+            }
+        }
+        try {
+            jobs.forEach { it.await() }
+            pass("4", "concurrent request responses matched")
+            true
+        } catch (error: Throwable) {
+            fail("4", error.message ?: error.toString())
+            false
+        }
+    }
+
+private fun parserMap(): ParserMap =
+    mapOf(typeNameOf<TestData>() to { TestData.parseFrom(it) })
+
+private fun argValue(args: Array<String>, name: String): String? {
+    val prefix = "--$name="
+    return args.firstOrNull { it.startsWith(prefix) }?.substring(prefix.length)
+}
+
+private fun pass(scenario: String, detail: String) {
+    println("PASS scenario=$scenario detail=$detail")
+}
+
+private fun fail(scenario: String, error: String) {
+    println("FAIL scenario=$scenario error=$error")
+}

kotlin/crosstest/kotlin_dart.kt

@@ -0,0 +1,235 @@
+@file:JvmName("KotlinDartKt")
+
+package com.tokilabs.toki_socket.crosstest
+
+import com.tokilabs.toki_socket.TcpClient
+import com.tokilabs.toki_socket.TcpServer
+import com.tokilabs.toki_socket.ParserMap
+import com.tokilabs.toki_socket.WsClient
+import com.tokilabs.toki_socket.WsServer
+import com.tokilabs.toki_socket.addListenerTyped
+import com.tokilabs.toki_socket.addRequestListenerTyped
+import com.tokilabs.toki_socket.typeNameOf
+import com.tokilabs.toki_socket.packets.TestData
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeoutOrNull
+import java.io.File
+import java.util.concurrent.TimeUnit
+import kotlin.system.exitProcess
+
+private const val HOST = "127.0.0.1"
+private const val TCP_PORT = 29490
+private const val WS_PORT = 29492
+private const val WS_PATH = "/"
+private const val PROCESS_TIMEOUT_MS = 20_000L
+private const val SERVER_OBSERVATION_MS = 200L
+
+fun main() = runBlocking {
+    println("INFO typeName kotlin=${typeNameOf<TestData>()}")
+    try {
+        runTcpSendPush()
+        delay(150)
+        runTcpRequests()
+        delay(150)
+        runWsSendPush()
+        delay(150)
+        runWsRequests()
+        println("PASS all kotlin-server/dart-client crosstests passed")
+    } catch (error: Throwable) {
+        System.err.println("FAIL crosstest error=${error.message ?: error}")
+        exitProcess(1)
+    }
+}
+
+private suspend fun runTcpSendPush() = coroutineScope {
+    val received = CompletableDeferred<Boolean>()
+    val server = TcpServer(HOST, TCP_PORT) { socket ->
+        TcpClient.fromSocket(socket, 0, 0, parserMap())
+    }
+    server.onClientConnected = { client ->
+        addListenerTyped<TestData>(client.communicator) { data ->
+            println("SERVER_RECEIVED index=${data.index} message=${data.message}")
+            val valid = data.index == 101 && data.message == "fire from dart client"
+            received.complete(valid)
+            if (valid) {
+                launch {
+                    client.send(
+                        TestData.newBuilder()
+                            .setIndex(200)
+                            .setMessage("push from kotlin server")
+                            .build(),
+                    )
+                }
+            }
+        }
+    }
+    withServer(server::start, server::stop) {
+        runDartClient("tcp", TCP_PORT, "send-push", setOf("1", "2"))
+        val ok = withTimeoutOrNull(SERVER_OBSERVATION_MS) { received.await() } ?: false
+        check(ok) { "TCP send-push server did not receive expected data" }
+    }
+}
+
+private suspend fun runTcpRequests() {
+    val server = TcpServer(HOST, TCP_PORT) { socket ->
+        TcpClient.fromSocket(socket, 0, 0, parserMap())
+    }
+    server.onClientConnected = { client ->
+        addRequestListenerTyped<TestData, TestData>(client.communicator) { req ->
+            TestData.newBuilder()
+                .setIndex(req.index * 2)
+                .setMessage("echo: ${req.message}")
+                .build()
+        }
+    }
+    withServer(server::start, server::stop) {
+        runDartClient("tcp", TCP_PORT, "requests", setOf("3", "4"))
+    }
+}
+
+private suspend fun runWsSendPush() = coroutineScope {
+    val received = CompletableDeferred<Boolean>()
+    val server = WsServer(HOST, WS_PORT, WS_PATH) { conn ->
+        WsClient.forServer(conn, 0, 0, parserMap())
+    }
+    server.onClientConnected = { client ->
+        addListenerTyped<TestData>(client.communicator) { data ->
+            println("SERVER_RECEIVED index=${data.index} message=${data.message}")
+            val valid = data.index == 101 && data.message == "fire from dart client"
+            received.complete(valid)
+            if (valid) {
+                launch {
+                    client.send(
+                        TestData.newBuilder()
+                            .setIndex(200)
+                            .setMessage("push from kotlin server")
+                            .build(),
+                    )
+                }
+            }
+        }
+    }
+    withServer(server::start, server::stop) {
+        runDartClient("ws", WS_PORT, "send-push", setOf("1", "2"))
+        val ok = withTimeoutOrNull(SERVER_OBSERVATION_MS) { received.await() } ?: false
+        check(ok) { "WS send-push server did not receive expected data" }
+    }
+}
+
+private suspend fun runWsRequests() {
+    val server = WsServer(HOST, WS_PORT, WS_PATH) { conn ->
+        WsClient.forServer(conn, 0, 0, parserMap())
+    }
+    server.onClientConnected = { client ->
+        addRequestListenerTyped<TestData, TestData>(client.communicator) { req ->
+            TestData.newBuilder()
+                .setIndex(req.index * 2)
+                .setMessage("echo: ${req.message}")
+                .build()
+        }
+    }
+    withServer(server::start, server::stop) {
+        runDartClient("ws", WS_PORT, "requests", setOf("3", "4"))
+    }
+}
+
+private suspend fun withServer(
+    start: () -> Unit,
+    stop: () -> Unit,
+    body: suspend () -> Unit,
+) {
+    start()
+    try {
+        body()
+    } finally {
+        stop()
+    }
+}
+
+private suspend fun runDartClient(
+    mode: String,
+    port: Int,
+    phase: String,
+    expectedScenarios: Set<String>,
+) = withContext(Dispatchers.IO) {
+    val process = ProcessBuilder(
+        "dart",
+        "run",
+        "crosstest/kotlin_dart_client.dart",
+        "--mode=$mode",
+        "--port=$port",
+        "--phase=$phase",
+    )
+        .directory(dartDir())
+        .redirectErrorStream(false)
+        .start()
+
+    val stdoutLines = mutableListOf<String>()
+    val stdoutThread = Thread {
+        process.inputStream.bufferedReader().forEachLine { line ->
+            println(line)
+            if (line.startsWith("PASS ") || line.startsWith("FAIL ")) {
+                stdoutLines.add(line)
+            }
+        }
+    }
+    val stderrThread = Thread {
+        process.errorStream.bufferedReader().forEachLine { line ->
+            System.err.println(line)
+        }
+    }
+    stdoutThread.start()
+    stderrThread.start()
+
+    val finished = process.waitFor(PROCESS_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+    if (!finished) {
+        process.destroyForcibly()
+    }
+    stdoutThread.join()
+    stderrThread.join()
+
+    validateResultLines(
+        "dart-client $mode/$phase",
+        if (finished) process.exitValue() else -1,
+        stdoutLines,
+        expectedScenarios,
+    )
+}
+
+private fun validateResultLines(
+    label: String,
+    exitCode: Int,
+    lines: List<String>,
+    expectedScenarios: Set<String>,
+) {
+    val failed = lines.filter { it.startsWith("FAIL ") }
+    val passed = lines
+        .filter { it.startsWith("PASS ") }
+        .mapNotNull { Regex("""scenario=([^ ]+)""").find(it)?.groupValues?.get(1) }
+        .toSet()
+    val missing = expectedScenarios - passed
+    check(exitCode == 0 && failed.isEmpty() && missing.isEmpty()) {
+        "$label failed exitCode=$exitCode failed=$failed missing=$missing"
+    }
+}
+
+private fun parserMap(): ParserMap =
+    mapOf(typeNameOf<TestData>() to { TestData.parseFrom(it) })
+
+private fun dartDir(): File {
+    var dir = File(System.getProperty("user.dir")).absoluteFile
+    while (dir.parentFile != null) {
+        val candidate = File(dir, "../dart").canonicalFile
+        if (File(candidate, "pubspec.yaml").isFile) return candidate
+        val direct = File(dir, "dart").canonicalFile
+        if (File(direct, "pubspec.yaml").isFile) return direct
+        dir = dir.parentFile
+    }
+    error("cannot resolve dart directory")
+}

kotlin/crosstest/kotlin_go.kt

@@ -12,10 +12,12 @@ import com.tokilabs.toki_socket.addRequestListenerTyped
 import com.tokilabs.toki_socket.typeNameOf
 import com.tokilabs.toki_socket.packets.TestData
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeoutOrNull
 import java.io.File
 import java.util.concurrent.TimeUnit
@@ -137,10 +139,10 @@ private suspend fun runWsRequests() {
     }
 }
 
-private inline fun withServer(
+private suspend fun withServer(
     start: () -> Unit,
     stop: () -> Unit,
-    body: () -> Unit,
+    body: suspend () -> Unit,
 ) {
     start()
     try {
@@ -150,12 +152,12 @@ private inline fun withServer(
     }
 }
 
-private fun runGoClient(
+private suspend fun runGoClient(
     mode: String,
     port: Int,
     phase: String,
     expectedScenarios: Set<String>,
-) {
+) = withContext(Dispatchers.IO) {
     val process = ProcessBuilder(
         "go",
         "run",