/// Same-language Dart stress / benchmark harness.
///
/// 고성능 병렬 운용 기준선 Milestone의 lang-baseline Task에서, TypeScript stress.ts와 같은 출력
/// 계약(ROW|/SKIP|/SUMMARY|)으로 Dart same-language TCP/WS baseline을 남긴다. 절대 성능 합격선은
/// 고정하지 않고 local 환경 baseline(throughput, p50/p95/p99 latency, peak RSS)을 기록하며, 안정성
/// 합격선만 hard fail로 둔다: timeout 0, nonce mismatch 0, response type mismatch 0, connection별
/// FIFO 위반 0, 종료 후 pending leak 0.
///
/// 실행: dart run bench/stress.dart [--mode=quick|full] [--transport=tcp,ws] [--profiles=a,b]
library;

import 'dart:async';
import 'dart:io';

import 'package:protobuf/protobuf.dart';
import 'package:proto_socket/proto_socket.dart';

const _host = '127.0.0.1';
const _language = 'Dart';
const _wsPath = '/';
const _allProfiles = [
  'roundtrip',
  'burst',
  'sustained',
  'parallel',
  'slow-mix',
  'payload'
];

// payload size matrix 축. quick은 작은 샘플(1KB/64KB), full은 Milestone 기준(1KB/64KB/1MB)을 측정한다.
const _payloadSeed = '0123456789abcdef';
const _payloadSizesQuick = [1024, 65536];
const _payloadSizesFull = [1024, 65536, 1048576];

// Dart heartbeat는 이제 interval 또는 wait가 0이면 "비활성"이다(다른 언어와 동일). baseline
// 측정 중에는 heartbeat timer churn이 끼지 않도록 0으로 비활성화한다.
const _heartbeatSeconds = 0;

int _rowsEmitted = 0;
int _totalViolations = 0;

final _parserMap = <String, GeneratedMessage Function(List<int>)>{
  TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
};

/// 안정성 위반 카운터. 0이 아니면 해당 row는 FAIL이고 프로세스도 non-zero exit한다.
class _Stability {
  int timeouts = 0;
  int nonceMismatch = 0;
  int typeMismatch = 0;
  int fifoViolations = 0;
  int pendingLeak = 0;

  int violations() =>
      timeouts + nonceMismatch + typeMismatch + fifoViolations + pendingLeak;
}

class _TcpClient extends ProtobufClient {
  _TcpClient(Socket socket)
      : super(socket, _heartbeatSeconds, _heartbeatSeconds, _parserMap);
}

class _WsClient extends WsProtobufClient {
  _WsClient(WebSocket ws)
      : super(ws, _heartbeatSeconds, _heartbeatSeconds, _parserMap);
}

class _TcpServer extends ProtobufServer {
  final void Function(Communicator) onConnected;
  _TcpServer(int port, this.onConnected)
      : super(_host, port, (socket) => _TcpClient(socket));
  @override
  void onClientConnected(ProtobufClient client) => onConnected(client);
}

class _WsServer extends WsProtobufServer {
  final void Function(Communicator) onConnected;
  _WsServer(int port, this.onConnected)
      : super(_host, port, (ws) => _WsClient(ws));
  @override
  void onClientConnected(WsProtobufClient client) => onConnected(client);
}

class _ClientHandle {
  final Communicator comm;
  final Future<void> Function() close;
  _ClientHandle(this.comm, this.close);
}

abstract class _ServerHandle {
  Future<void> start();
  Future<void> stop();
}

class _ServerHandleImpl implements _ServerHandle {
  final Future Function() _start;
  final Future Function() _stop;
  _ServerHandleImpl(this._start, this._stop);
  @override
  Future<void> start() async => _start();
  @override
  Future<void> stop() async => _stop();
}

void _log(String line) => stderr.writeln(line);

double _memMb() => ProcessInfo.currentRss / (1024 * 1024);

Future<int> _freePort() async {
  final socket = await ServerSocket.bind(_host, 0);
  final port = socket.port;
  await socket.close();
  return port;
}

int _testDataPayloadBytes(String message) => (TestData()
      ..index = 0
      ..message = message)
    .writeToBuffer()
    .length;

/// size 바이트 길이의 deterministic filler string을 만든다.
String _payloadFiller(int size) {
  if (size <= 0) return '';
  final reps = (size + _payloadSeed.length - 1) ~/ _payloadSeed.length;
  return (_payloadSeed * reps).substring(0, size);
}

/// payload 크기를 사람이 읽는 축 이름으로 바꾼다(1024 -> 1KB, 1048576 -> 1MB).
String _payloadLabel(int bytes) {
  if (bytes >= 1048576 && bytes % 1048576 == 0) return '${bytes ~/ 1048576}MB';
  if (bytes >= 1024 && bytes % 1024 == 0) return '${bytes ~/ 1024}KB';
  return '${bytes}B';
}

double _percentile(List<double> sorted, double p) {
  if (sorted.isEmpty) return 0;
  var rank = ((p / 100.0) * sorted.length).ceil() - 1;
  if (rank < 0) rank = 0;
  if (rank > sorted.length - 1) rank = sorted.length - 1;
  return sorted[rank];
}

void _classifyRequestError(String message, _Stability s) {
  if (message.contains('timeout') || message.contains('TimeoutException')) {
    s.timeouts += 1;
  } else if (message.contains('type mismatch')) {
    s.typeMismatch += 1;
  } else {
    s.nonceMismatch += 1;
  }
}

void _emitRow({
  required String profile,
  required String axis,
  required String transport,
  required int payloadBytes,
  required int clientCount,
  required int requests,
  required double throughput,
  required double p50,
  required double p95,
  required double p99,
  required double mem,
  required _Stability s,
}) {
  final violations = s.violations();
  final status = violations == 0 ? 'PASS' : 'FAIL';
  _totalViolations += violations;
  _rowsEmitted += 1;
  final fields = [
    'ROW',
    profile,
    axis,
    _language,
    transport,
    '$payloadBytes',
    '$clientCount',
    '$requests',
    throughput.toStringAsFixed(1),
    p50.toStringAsFixed(3),
    p95.toStringAsFixed(3),
    p99.toStringAsFixed(3),
    '${s.timeouts}',
    '${s.nonceMismatch}',
    '${s.typeMismatch}',
    '${s.fifoViolations}',
    '${s.pendingLeak}',
    '0', // queueBacklog: same-language baseline에는 inbound gateway가 없다.
    '0', // gatewayBacklog
    'off',
    mem.toStringAsFixed(1),
    status,
  ];
  print(fields.join('|'));
}

void _emitSkip(String profile, String axis, String transport, String reason) {
  print('SKIP|$profile|$axis|$_language|$transport|$reason');
}

void _summarize(
    String profile,
    String axis,
    String transport,
    List<double> latencies,
    double elapsedMs,
    int clientCount,
    _Stability s,
    double mem) {
  final sorted = List<double>.from(latencies)..sort();
  final throughput =
      elapsedMs > 0 ? latencies.length / elapsedMs * 1000.0 : 0.0;
  _emitRow(
    profile: profile,
    axis: axis,
    transport: transport,
    payloadBytes: _testDataPayloadBytes('req-0'),
    clientCount: clientCount,
    requests: latencies.length,
    throughput: throughput,
    p50: _percentile(sorted, 50),
    p95: _percentile(sorted, 95),
    p99: _percentile(sorted, 99),
    mem: mem,
    s: s,
  );
}

void _emitThroughput(
    String profile,
    String axis,
    String transport,
    int count,
    double elapsedMs,
    int clientCount,
    int payloadBytes,
    _Stability s,
    double mem) {
  final throughput = elapsedMs > 0 ? count / elapsedMs * 1000.0 : 0.0;
  _emitRow(
    profile: profile,
    axis: axis,
    transport: transport,
    payloadBytes: payloadBytes,
    clientCount: clientCount,
    requests: count,
    throughput: throughput,
    p50: 0,
    p95: 0,
    p99: 0,
    mem: mem,
    s: s,
  );
}

_ServerHandle _makeServer(
    String transport, int port, void Function(Communicator) onConnected) {
  if (transport == 'tcp') {
    final server = _TcpServer(port, onConnected);
    return _ServerHandleImpl(server.start, server.stop);
  } else if (transport == 'ws') {
    final server = _WsServer(port, onConnected);
    return _ServerHandleImpl(server.start, server.stop);
  }
  throw ArgumentError('unknown transport "$transport"');
}

// request-response 축에서는 client가 concurrency C로 동시에 send하므로 frame 도착 순서가 결정적이지
// 않다. 따라서 여기서는 FIFO를 측정하지 않고 응답 정확성/timeout만 본다. per-connection FIFO는 단일
// connection 순차 send인 burst 축에서 검증한다.
_ServerHandle _makeEchoServer(String transport, int port) {
  return _makeServer(transport, port, (comm) {
    comm.addRequestListener<TestData, TestData>((req) async {
      return TestData()
        ..index = req.index * 2
        ..message = 'echo:${req.message}';
    });
  });
}

_ServerHandle _makeSlowMixServer(
    String transport, int port, int slowEvery, Duration delay) {
  return _makeServer(transport, port, (comm) {
    comm.addRequestListener<TestData, TestData>((req) async {
      if ((req.index + 1) % slowEvery == 0) {
        await Future<void>.delayed(delay);
      }
      return TestData()
        ..index = req.index * 2
        ..message = 'echo:${req.message}';
    });
  });
}

Future<_ClientHandle> _dial(String transport, int port) async {
  if (transport == 'tcp') {
    final socket = await ProtobufClient.connect(_host, port)
        .timeout(const Duration(milliseconds: 300));
    final client = _TcpClient(socket);
    return _ClientHandle(client, client.close);
  } else if (transport == 'ws') {
    final ws = await WsProtobufClient.connect(_host, port, path: _wsPath)
        .timeout(const Duration(milliseconds: 300));
    final client = _WsClient(ws);
    return _ClientHandle(client, client.close);
  }
  throw ArgumentError('unknown transport "$transport"');
}

Future<_ClientHandle> _dialWithRetry(String transport, int port) async {
  final deadline = DateTime.now().add(const Duration(seconds: 3));
  Object? lastError;
  while (DateTime.now().isBefore(deadline)) {
    try {
      return await _dial(transport, port);
    } catch (error) {
      lastError = error;
      await Future<void>.delayed(const Duration(milliseconds: 50));
    }
  }
  throw TimeoutException('connect $transport:$port timed out: $lastError');
}

Future<List<double>> _runRequestLoad(_ClientHandle handle, int total,
    int concurrency, Duration timeout, _Stability s) async {
  final latencies = <double>[];
  var issued = 0;
  var nextIndex = 0;
  while (issued < total) {
    final batchSize =
        concurrency < total - issued ? concurrency : total - issued;
    final futures = <Future<void>>[];
    for (var i = 0; i < batchSize; i++) {
      final index = nextIndex++;
      futures.add(() async {
        final sw = Stopwatch()..start();
        try {
          final res = await handle.comm
              .sendRequest<TestData, TestData>(TestData()
                ..index = index
                ..message = 'req-$index')
              .timeout(timeout);
          latencies.add(sw.elapsedMicroseconds / 1000.0);
          if (res.index != index * 2 || res.message != 'echo:req-$index') {
            s.nonceMismatch += 1;
          }
        } catch (error) {
          _classifyRequestError(error.toString(), s);
        }
      }());
    }
    issued += batchSize;
    await Future.wait(futures);
  }
  return latencies;
}

Future<void> _profileRoundtrip(String transport, String mode) async {
  const concurrencies = [1, 16, 64, 256];
  final batches = mode == 'quick' ? 2 : 20;
  final timeout = Duration(seconds: mode == 'quick' ? 5 : 15);
  _log(
      '[roundtrip] transport=$transport mode=$mode concurrencies=1,16,64,256 batches/level=$batches');
  for (final concurrency in concurrencies) {
    final s = _Stability();
    final total = concurrency * batches;
    final port = await _freePort();
    final server = _makeEchoServer(transport, port);
    await server.start();
    try {
      final handle = await _dialWithRetry(transport, port);
      try {
        final sw = Stopwatch()..start();
        final latencies =
            await _runRequestLoad(handle, total, concurrency, timeout, s);
        _summarize('roundtrip', 'concurrency=$concurrency', transport,
            latencies, sw.elapsedMicroseconds / 1000.0, 1, s, _memMb());
      } finally {
        await handle.close();
        if (handle.comm.isAlive) s.pendingLeak += 1;
      }
    } catch (error) {
      s.timeouts += 1;
      _log('[roundtrip] concurrency=$concurrency error=$error');
      _summarize('roundtrip', 'concurrency=$concurrency', transport, [], 0, 1,
          s, _memMb());
    } finally {
      await server.stop();
    }
    _log(
        '[roundtrip] transport=$transport concurrency=$concurrency done violations=${s.violations()}');
  }
}

Future<void> _profileBurst(String transport, String mode) async {
  final counts = mode == 'quick' ? [200] : [1000, 10000, 100000];
  _log('[burst] transport=$transport mode=$mode counts=$counts');
  for (final count in counts) {
    final s = _Stability();
    var received = 0;
    var lastIndex = -1;
    final done = Completer<void>();
    final port = await _freePort();
    final server = _makeServer(transport, port, (comm) {
      comm.addListener<TestData>((data) {
        if (data.index <= lastIndex) s.fifoViolations += 1;
        lastIndex = data.index;
        received += 1;
        if (received >= count && !done.isCompleted) done.complete();
      });
    });
    await server.start();
    _ClientHandle? handle;
    try {
      handle = await _dialWithRetry(transport, port);
      final sw = Stopwatch()..start();
      for (var i = 0; i < count; i++) {
        await handle.comm.send(TestData()
          ..index = i
          ..message = 'b-$i');
      }
      try {
        await done.future.timeout(Duration(seconds: mode == 'quick' ? 10 : 60));
      } on TimeoutException {
        s.timeouts += 1;
        _log('[burst] count=$count dispatch timeout received=$received');
      }
      if (received != count) s.pendingLeak += 1;
      _emitThroughput(
          'burst',
          'count=$count',
          transport,
          count,
          sw.elapsedMicroseconds / 1000.0,
          1,
          _testDataPayloadBytes('b-0'),
          s,
          _memMb());
    } catch (error) {
      s.timeouts += 1;
      _log('[burst] count=$count error=$error');
      _emitThroughput('burst', 'count=$count', transport, received, 0, 1,
          _testDataPayloadBytes('b-0'), s, _memMb());
    } finally {
      if (handle != null) {
        await handle.close();
        if (handle.comm.isAlive) s.pendingLeak += 1;
      }
      await server.stop();
    }
    _log(
        '[burst] transport=$transport count=$count received=$received violations=${s.violations()}');
  }
}

Future<void> _profileSustained(String transport, String mode) async {
  final durationsMs = mode == 'quick' ? [2000] : [30000, 300000, 1800000];
  const concurrency = 16;
  const timeout = Duration(seconds: 15);
  _log(
      '[sustained] transport=$transport mode=$mode durations=${durationsMs.join(",")}ms concurrency=$concurrency');
  for (final durationMs in durationsMs) {
    final s = _Stability();
    var peak = _memMb();
    final port = await _freePort();
    final server = _makeEchoServer(transport, port);
    await server.start();
    try {
      final handle = await _dialWithRetry(transport, port);
      final latencies = <double>[];
      var nextIndex = 0;
      final deadline = DateTime.now().add(Duration(milliseconds: durationMs));
      try {
        while (DateTime.now().isBefore(deadline)) {
          final futures = <Future<void>>[];
          for (var i = 0; i < concurrency; i++) {
            final index = nextIndex++;
            futures.add(() async {
              final sw = Stopwatch()..start();
              try {
                final res = await handle.comm
                    .sendRequest<TestData, TestData>(TestData()
                      ..index = index
                      ..message = 's-$index')
                    .timeout(timeout);
                latencies.add(sw.elapsedMicroseconds / 1000.0);
                if (res.index != index * 2 || res.message != 'echo:s-$index') {
                  s.nonceMismatch += 1;
                }
              } catch (error) {
                _classifyRequestError(error.toString(), s);
              }
            }());
          }
          await Future.wait(futures);
          final cur = _memMb();
          if (cur > peak) peak = cur;
        }
        _summarize('sustained', 'duration=${durationMs}ms', transport,
            latencies, durationMs.toDouble(), 1, s, peak);
      } finally {
        await handle.close();
        if (handle.comm.isAlive) s.pendingLeak += 1;
      }
    } catch (error) {
      s.timeouts += 1;
      _log('[sustained] duration=${durationMs}ms error=$error');
      _summarize('sustained', 'duration=${durationMs}ms', transport, [], 0, 1,
          s, peak);
    } finally {
      await server.stop();
    }
    _log(
        '[sustained] transport=$transport duration=${durationMs}ms done peakMemMb=${peak.toStringAsFixed(1)} violations=${s.violations()}');
  }
}

Future<void> _profileParallel(String transport, String mode) async {
  final clientCounts = mode == 'quick' ? [4] : [16, 128, 512, 1024];
  final perClient = mode == 'quick' ? 50 : 500;
  const concurrency = 8;
  const timeout = Duration(seconds: 15);
  _log(
      '[parallel] transport=$transport mode=$mode clients=$clientCounts perClient=$perClient');
  for (final clientCount in clientCounts) {
    final s = _Stability();
    final port = await _freePort();
    final server = _makeEchoServer(transport, port);
    await server.start();
    final handles = <_ClientHandle>[];
    try {
      for (var i = 0; i < clientCount; i++) {
        try {
          handles.add(await _dialWithRetry(transport, port));
        } catch (error) {
          s.timeouts += 1;
          _log('[parallel] clients=$clientCount dial error=$error');
        }
      }
      final allLatencies = <double>[];
      final sw = Stopwatch()..start();
      await Future.wait(handles.map((handle) async {
        allLatencies.addAll(
            await _runRequestLoad(handle, perClient, concurrency, timeout, s));
      }));
      _summarize('parallel', 'clients=$clientCount', transport, allLatencies,
          sw.elapsedMicroseconds / 1000.0, clientCount, s, _memMb());
    } finally {
      for (final handle in handles) {
        await handle.close();
        if (handle.comm.isAlive) s.pendingLeak += 1;
      }
      await server.stop();
    }
    _log(
        '[parallel] transport=$transport clients=$clientCount done violations=${s.violations()}');
  }
}

/// 빠른/느린 request handler를 섞어 same-connection FIFO delay와 connection 간 nonce 독립성을 검증한다.
Future<void> _profileSlowMix(String transport, String mode) async {
  final clientCount = mode == 'quick' ? 4 : 16;
  final perClient = mode == 'quick' ? 20 : 200;
  final slowEvery = mode == 'quick' ? 5 : 10;
  final delayMs = mode == 'quick' ? 50 : 100;
  final timeout = Duration(seconds: mode == 'quick' ? 10 : 30);
  final axis = 'clients=$clientCount,slowEvery=$slowEvery,delayMs=$delayMs';
  final s = _Stability();
  var peak = _memMb();
  final latencies = <double>[];
  var elapsedMs = 0.0;
  _log('[slow-mix] transport=$transport mode=$mode $axis perClient=$perClient');

  final port = await _freePort();
  final server = _makeSlowMixServer(
      transport, port, slowEvery, Duration(milliseconds: delayMs));
  await server.start();
  final handles = <_ClientHandle>[];
  final clientIds = <int>[];
  try {
    for (var clientId = 0; clientId < clientCount; clientId++) {
      try {
        handles.add(await _dialWithRetry(transport, port));
        clientIds.add(clientId);
      } catch (error) {
        s.timeouts += 1;
        _log('[slow-mix] client=$clientId dial error=$error');
      }
    }
    final completedByClient = List<int>.filled(clientCount, -1);
    final sw = Stopwatch()..start();
    await Future.wait(List.generate(handles.length, (slot) async {
      final handle = handles[slot];
      final clientId = clientIds[slot];
      await Future.wait(List.generate(perClient, (index) async {
        final message = 'sm-$clientId-$index';
        final reqSw = Stopwatch()..start();
        try {
          final res = await handle.comm
              .sendRequest<TestData, TestData>(TestData()
                ..index = index
                ..message = message)
              .timeout(timeout);
          latencies.add(reqSw.elapsedMicroseconds / 1000.0);
          if (res.index != index * 2 || res.message != 'echo:$message') {
            s.nonceMismatch += 1;
          }
          if (index <= completedByClient[clientId]) {
            s.fifoViolations += 1;
          }
          completedByClient[clientId] = index;
        } catch (error) {
          _classifyRequestError(error.toString(), s);
        }
      }));
    }));
    final cur = _memMb();
    if (cur > peak) peak = cur;
    elapsedMs = sw.elapsedMicroseconds / 1000.0;
  } finally {
    for (final handle in handles) {
      await handle.close();
      if (handle.comm.isAlive) s.pendingLeak += 1;
    }
    await server.stop();
  }
  final sorted = List<double>.from(latencies)..sort();
  final throughput =
      elapsedMs > 0 ? latencies.length / elapsedMs * 1000.0 : 0.0;
  _emitRow(
    profile: 'slow-mix',
    axis: axis,
    transport: transport,
    payloadBytes: _testDataPayloadBytes('sm-0-0'),
    clientCount: clientCount,
    requests: latencies.length,
    throughput: throughput,
    p50: _percentile(sorted, 50),
    p95: _percentile(sorted, 95),
    p99: _percentile(sorted, 99),
    mem: peak,
    s: s,
  );
  _log('[slow-mix] transport=$transport clients=$clientCount done '
      'peakMemMb=${peak.toStringAsFixed(1)} violations=${s.violations()}');
}

/// payload size matrix. 동일 connection request-response를 1KB/64KB/1MB payload에서 측정한다.
/// message field를 deterministic filler로 채우고 실제 serialized payload bytes, payload별
/// p50/p95/p99 latency, throughput, peak RSS, stability counters를 결과 row에 남긴다.
Future<void> _profilePayload(String transport, String mode) async {
  final sizes = mode == 'quick' ? _payloadSizesQuick : _payloadSizesFull;
  final concurrency = mode == 'quick' ? 4 : 8;
  final requests = mode == 'quick' ? 20 : 100;
  final timeout = Duration(seconds: mode == 'quick' ? 10 : 30);
  _log('[payload] transport=$transport mode=$mode '
      'sizes=${sizes.map(_payloadLabel).toList()} concurrency=$concurrency requests/size=$requests');
  for (final size in sizes) {
    final s = _Stability();
    final filler = _payloadFiller(size);
    final expectedEcho = 'echo:$filler';
    final payloadBytes = _testDataPayloadBytes(filler);
    final axis = 'payload=${_payloadLabel(size)}';
    var peak = _memMb();
    final port = await _freePort();
    final server = _makeEchoServer(transport, port);
    await server.start();
    try {
      final handle = await _dialWithRetry(transport, port);
      try {
        final latencies = <double>[];
        var issued = 0;
        var nextIndex = 0;
        final sw = Stopwatch()..start();
        while (issued < requests) {
          final batchSize =
              concurrency < requests - issued ? concurrency : requests - issued;
          final futures = <Future<void>>[];
          for (var i = 0; i < batchSize; i++) {
            final index = nextIndex++;
            futures.add(() async {
              final reqSw = Stopwatch()..start();
              try {
                final res = await handle.comm
                    .sendRequest<TestData, TestData>(TestData()
                      ..index = index
                      ..message = filler)
                    .timeout(timeout);
                latencies.add(reqSw.elapsedMicroseconds / 1000.0);
                if (res.index != index * 2 || res.message != expectedEcho) {
                  s.nonceMismatch += 1;
                }
              } catch (error) {
                _classifyRequestError(error.toString(), s);
              }
            }());
          }
          issued += batchSize;
          await Future.wait(futures);
          final cur = _memMb();
          if (cur > peak) peak = cur;
        }
        final elapsedMs = sw.elapsedMicroseconds / 1000.0;
        final sorted = List<double>.from(latencies)..sort();
        final throughput =
            elapsedMs > 0 ? latencies.length / elapsedMs * 1000.0 : 0.0;
        _emitRow(
          profile: 'payload',
          axis: axis,
          transport: transport,
          payloadBytes: payloadBytes,
          clientCount: 1,
          requests: latencies.length,
          throughput: throughput,
          p50: _percentile(sorted, 50),
          p95: _percentile(sorted, 95),
          p99: _percentile(sorted, 99),
          mem: peak,
          s: s,
        );
      } finally {
        await handle.close();
        if (handle.comm.isAlive) s.pendingLeak += 1;
      }
    } catch (error) {
      s.timeouts += 1;
      _log('[payload] size=${_payloadLabel(size)} error=$error');
      _emitRow(
        profile: 'payload',
        axis: axis,
        transport: transport,
        payloadBytes: payloadBytes,
        clientCount: 1,
        requests: 0,
        throughput: 0,
        p50: 0,
        p95: 0,
        p99: 0,
        mem: peak,
        s: s,
      );
    } finally {
      await server.stop();
    }
    _log(
        '[payload] transport=$transport size=${_payloadLabel(size)} bytes=$payloadBytes '
        'peakMemMb=${peak.toStringAsFixed(1)} violations=${s.violations()}');
  }
}

final _runners = <String, Future<void> Function(String, String)>{
  'roundtrip': _profileRoundtrip,
  'burst': _profileBurst,
  'sustained': _profileSustained,
  'parallel': _profileParallel,
  'slow-mix': _profileSlowMix,
  'payload': _profilePayload,
};

List<String> _parseList(List<String> args, String prefix, List<String> def) {
  for (final arg in args) {
    if (arg.startsWith(prefix)) {
      final items = arg
          .substring(prefix.length)
          .split(',')
          .map((v) => v.trim())
          .where((v) => v.isNotEmpty)
          .toList();
      if (items.isNotEmpty) return items;
    }
  }
  return def;
}

Future<void> main(List<String> args) async {
  var mode = 'quick';
  for (final arg in args) {
    if (arg == '--full' || arg == '--mode=full') mode = 'full';
    if (arg == '--quick' || arg == '--mode=quick') mode = 'quick';
  }
  var transports = _parseList(args, '--transport=', ['tcp', 'ws']);
  transports = _parseList(args, '--transports=', transports);
  var profiles = _parseList(args, '--profiles=', _allProfiles);
  profiles = _parseList(args, '--profile=', profiles);

  _log('INFO stress harness language=$_language mode=$mode '
      'transports=${transports.join(',')} profiles=${profiles.join(',')} '
      'typeName=${TestData.getDefault().info_.qualifiedMessageName}');

  for (final transport in transports) {
    if (transport != 'tcp' && transport != 'ws') {
      _emitSkip('all', 'transport', transport,
          'unsupported transport for Dart same-language baseline');
      continue;
    }
    for (final profile in profiles) {
      final runner = _runners[profile];
      if (runner == null) {
        _emitSkip(profile, 'profile', transport,
            'profile not part of Dart same-language baseline (gateway is TypeScript-specific)');
        continue;
      }
      await runner(transport, mode);
    }
  }

  final status = _totalViolations == 0 ? 'PASS' : 'FAIL';
  print('SUMMARY|status=$status|language=$_language|mode=$mode|'
      'transports=${transports.join(',')}|profiles=${profiles.join(',')}|'
      'rows=$_rowsEmitted|stability_violations=$_totalViolations');
  if (status != 'PASS') {
    exitCode = 1;
  }
}
