oto/test/oto_iop_connection_smoke_test.dart

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

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

import 'support/iop/runtime.pb.dart' as iop;

const _host = '127.0.0.1';
const _token = 'oto-smoke-token';
const _nodeId = 'oto-smoke-node';
const _nodeAlias = 'oto-smoke';

class _OtoIopSmokeClient extends ProtobufClient {
  _OtoIopSmokeClient(Socket socket)
      : super(socket, 30, 45, {
          iop.RegisterResponse.getDefault().info_.qualifiedMessageName:
              iop.RegisterResponse.fromBuffer,
          iop.RunRequest.getDefault().info_.qualifiedMessageName:
              iop.RunRequest.fromBuffer,
          iop.CancelRequest.getDefault().info_.qualifiedMessageName:
              iop.CancelRequest.fromBuffer,
          iop.NodeCommandRequest.getDefault().info_.qualifiedMessageName:
              iop.NodeCommandRequest.fromBuffer,
          iop.EdgeNodeEvent.getDefault().info_.qualifiedMessageName:
              iop.EdgeNodeEvent.fromBuffer,
        });
}

void main() {
  test('OTO Dart proto-socket client registers with iop Edge', () async {
    final edgePort = await _freePort();
    final metricsPort = await _freePort();
    final workDir = await Directory.systemTemp.createTemp('oto-iop-smoke-');
    final config = File('${workDir.path}/edge.yaml');
    await config
        .writeAsString(_edgeConfig(edgePort, metricsPort, workDir.path));

    final edge = await Process.start(
      'go',
      [
        'run',
        './apps/edge/cmd/edge',
        'serve',
        '--config',
        config.path,
      ],
      workingDirectory: '../iop',
      environment: {
        'GOCACHE': '${workDir.path}/gocache',
      },
    );

    final output = StringBuffer();
    final stdoutSub = edge.stdout
        .transform(systemEncoding.decoder)
        .listen(output.write, onError: output.write);
    final stderrSub = edge.stderr
        .transform(systemEncoding.decoder)
        .listen(output.write, onError: output.write);

    _OtoIopSmokeClient? client;
    try {
      await _waitForPort(_host, edgePort, edge, output);
      final socket = await ProtobufClient.connect(_host, edgePort);
      client = _OtoIopSmokeClient(socket);

      final response =
          await client.sendRequest<iop.RegisterRequest, iop.RegisterResponse>(
        iop.RegisterRequest()..token = _token,
        timeout: const Duration(seconds: 5),
      );

      expect(response.accepted, isTrue);
      expect(response.nodeId, _nodeId);
      expect(response.alias, _nodeAlias);
      expect(response.config.runtime.concurrency, 1);
      expect(
          response.config.runtime.workspaceRoot, '${workDir.path}/workspace');
    } finally {
      await client?.close();
      edge.kill(ProcessSignal.sigterm);
      await edge.exitCode.timeout(
        const Duration(seconds: 5),
        onTimeout: () {
          edge.kill(ProcessSignal.sigkill);
          return edge.exitCode;
        },
      );
      await stdoutSub.cancel();
      await stderrSub.cancel();
      await workDir.delete(recursive: true);
    }
  }, timeout: const Timeout(Duration(seconds: 45)));
}

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

Future<void> _waitForPort(
  String host,
  int port,
  Process process,
  StringBuffer output,
) async {
  final deadline = DateTime.now().add(const Duration(seconds: 20));
  while (DateTime.now().isBefore(deadline)) {
    final exit = await _tryExitCode(process);
    if (exit != null) {
      fail('iop Edge exited before listening (code $exit):\n$output');
    }
    try {
      final socket = await Socket.connect(
        host,
        port,
        timeout: const Duration(milliseconds: 200),
      );
      await socket.close();
      return;
    } catch (_) {
      await Future<void>.delayed(const Duration(milliseconds: 100));
    }
  }
  fail('timed out waiting for iop Edge on $host:$port:\n$output');
}

Future<int?> _tryExitCode(Process process) {
  return process.exitCode
      .timeout(
        Duration.zero,
        onTimeout: () => -1,
      )
      .then((code) => code == -1 ? null : code);
}

String _edgeConfig(int edgePort, int metricsPort, String root) {
  return '''
edge:
  id: "oto-smoke-edge"
  name: "OTO Smoke Edge"

server:
  listen: "$_host:$edgePort"

tls:
  enabled: false

logging:
  level: "error"
  pretty: false

metrics:
  port: $metricsPort

openai:
  enabled: false

a2a:
  enabled: false

console:
  adapter: "oto"
  target: "pipeline"
  session_id: "default"
  background: false
  timeout_sec: 30

nodes:
  - id: "$_nodeId"
    alias: "$_nodeAlias"
    token: "$_token"
    adapters:
      cli:
        enabled: false
    runtime:
      concurrency: 1
      workspace_root: "$root/workspace"
''';
}