oto/apps/runner/test/oto_iop_connection_smoke_test.dart

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

import 'package:test/test.dart';
import 'package:oto/oto/agent/agent_config.dart';
import 'package:oto/oto/agent/edge_registration_client.dart';

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

/// iop repo root used to run the Edge under test.
///
/// Defaults to the workspace sibling `iop` checkout. From this package at
/// `apps/runner`, that sibling sits three levels up at `../../../iop`, but
/// `IOP_REPO_ROOT` can override it so this smoke runs against any iop checkout
/// (for example the `iop2oto` workspace clone).
String _iopRepoRoot() {
  final override = Platform.environment['IOP_REPO_ROOT'];
  if (override != null && override.trim().isNotEmpty) {
    return override;
  }
  return '../../../iop';
}

void main() {
  test(
    'OTO Dart proto-socket client registers with iop Edge and goes online',
    () async {
      final iopRepoRoot = _iopRepoRoot();
      final edgePort = await _freePort();
      final metricsPort = await _freePort();
      final bootstrapPort = await _freePort();
      final workDir = await Directory.systemTemp.createTemp('oto-iop-smoke-');
      // Reuse a stable build cache across runs so the Edge `go run` stays warm
      // and a fresh checkout still completes its first cold build in time.
      final goCache = '${Directory.systemTemp.path}/oto-iop-smoke-gocache';
      // The Edge writes its application (zap) logs to this file, not stderr, so
      // the smoke tails it for the node.online lifecycle marker.
      final edgeLog = File('${workDir.path}/edge.log');
      final config = File('${workDir.path}/edge.yaml');
      await config.writeAsString(_edgeConfig(
        edgePort: edgePort,
        metricsPort: metricsPort,
        bootstrapPort: bootstrapPort,
        logPath: edgeLog.path,
        root: workDir.path,
      ));

      final edge = await Process.start(
        'go',
        [
          'run',
          './apps/edge/cmd/edge',
          'serve',
          '--config',
          config.path,
        ],
        workingDirectory: iopRepoRoot,
        environment: {
          'GOCACHE': 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);

      // ignore: avoid_print
      print('[oto-smoke] iop repo root: $iopRepoRoot');

      EdgeAgentSession? session;
      try {
        await _waitForPort(_host, edgePort, edge, output);

        final agentConfig = AgentConfig(
          agent: const AgentIdentityConfig(
            id: _nodeId,
            alias: _nodeAlias,
            enrollmentToken: _token,
          ),
          edge: EdgeConnectionConfig(
            url: '$_host:$edgePort',
          ),
          runtime: AgentRuntimeConfig(
            installDir: '${workDir.path}/install',
            workspaceRoot: '${workDir.path}/workspace',
            logDir: '${workDir.path}/log',
          ),
        );

        // openSession keeps the connection alive after registration so the
        // proto-socket heartbeat reaches the Edge and flips the node online.
        session = await EdgeRegistrationClient().openSession(agentConfig);
        final result = session.result;

        expect(result.accepted, isTrue);
        expect(result.nodeId, _nodeId);
        expect(result.alias, _nodeAlias);
        expect(result.runtimeConfig, isNotNull);
        expect(result.runtimeConfig!['concurrency'], 1);
        expect(
          result.runtimeConfig!['workspaceRoot'],
          '${workDir.path}/workspace',
        );

        // ignore: avoid_print
        print('[oto-smoke] registration accepted '
            'node=${result.nodeId} alias=${result.alias}');

        // After accepted, wait for the Edge to record the first heartbeat and
        // transition the oto-agent node to online (node.online event). The
        // first heartbeat arrives on the proto-socket heartbeat interval
        // (~30s), so allow margin beyond it. The Edge emits this through its
        // application log file.
        final onlineLine = await _waitForLogMarker(
          'node.online',
          edgeLog,
          edge,
          output,
          const Duration(seconds: 60),
        );

        // ignore: avoid_print
        print('[oto-smoke] online evidence: ${onlineLine.trim()}');
      } finally {
        await session?.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: 180)),
  );
}

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 {
  // Generous deadline so a cold `go run` build of the Edge still finishes.
  final deadline = DateTime.now().add(const Duration(seconds: 120));
  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');
}

/// Polls the Edge [logFile] until [marker] appears and returns the matching
/// log line. The captured process [output] (Fx/stderr) is only used for error
/// context. Fails if the Edge exits early or the [timeout] elapses.
Future<String> _waitForLogMarker(
  String marker,
  File logFile,
  Process process,
  StringBuffer output,
  Duration timeout,
) async {
  final deadline = DateTime.now().add(timeout);
  while (DateTime.now().isBefore(deadline)) {
    if (logFile.existsSync()) {
      final text = logFile.readAsStringSync();
      final idx = text.indexOf(marker);
      if (idx >= 0) {
        final lineStart = text.lastIndexOf('\n', idx) + 1;
        var lineEnd = text.indexOf('\n', idx);
        if (lineEnd < 0) {
          lineEnd = text.length;
        }
        return text.substring(lineStart, lineEnd);
      }
    }
    final exit = await _tryExitCode(process);
    if (exit != null) {
      fail('iop Edge exited before "$marker" (code $exit):\n$output');
    }
    await Future<void>.delayed(const Duration(milliseconds: 100));
  }
  final logTail =
      logFile.existsSync() ? logFile.readAsStringSync() : '(no log)';
  fail('timed out waiting for "$marker" in Edge log:\n'
      'edge log:\n$logTail\nedge stderr:\n$output');
}

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

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

server:
  listen: "$_host:$edgePort"

tls:
  enabled: false

logging:
  level: "debug"
  pretty: false
  path: "$logPath"

metrics:
  port: $metricsPort

bootstrap:
  listen: "$_host:$bootstrapPort"

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"
    agent_kind: "oto-agent"
    adapters:
      cli:
        enabled: false
    runtime:
      concurrency: 1
      workspace_root: "$root/workspace"
''';
}