oto/apps/runner/test/oto_agent_registration_test.dart

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

import 'package:test/test.dart';
import 'package:oto/oto/agent/agent_config.dart';
import 'package:oto/oto/agent/agent_runner.dart';
import 'package:oto/oto/agent/edge_registration_client.dart' show EdgeEndpoint;
import 'package:oto/oto/agent/registration_client.dart';
import 'package:oto/oto/agent/oto_server_job_client.dart';
import 'package:oto/oto/agent/remote_run_executor.dart';
import 'package:oto/oto/agent/runner_capability_provider.dart';
import 'package:oto/oto/commands/runner_command_capability_provider.dart';
import 'package:oto/oto/application.dart';
import 'package:oto/oto/core/build_result.dart';
import 'package:oto/oto/core/execution_context.dart';
import 'package:oto/oto/agent/oto/runner.pb.dart' as oto;

class FakeAgentSession implements AgentSession {
  @override
  final RegistrationResult result;
  bool closed = false;

  FakeAgentSession(this.result);

  @override
  Future<void> close() async {
    closed = true;
  }
}

class FakeRegistrationClient extends RegistrationClient {
  final RegistrationResult Function(AgentConfig config) _onRegister;
  FakeAgentSession? lastSession;

  FakeRegistrationClient(this._onRegister);

  @override
  Future<AgentSession> openSession(AgentConfig config) async {
    final session = FakeAgentSession(_onRegister(config));
    lastSession = session;
    return session;
  }
}

class FakeRunnerCapabilityProvider implements RunnerCapabilityProvider {
  final RunnerCapabilitySnapshot _snapshot;

  FakeRunnerCapabilityProvider(this._snapshot);

  @override
  RunnerCapabilitySnapshot snapshot() => _snapshot;
}

void main() {
  const validBootstrapYaml = '''
agent:
  id: "agent-123"
  alias: "my-agent"
  enrollment_token: "token-456"
edge:
  url: "127.0.0.1:8080"
runtime:
  install_dir: "/usr/bin"
  workspace_root: "/var/oto"
  log_dir: "/var/log/oto"
''';

  late AgentConfig validConfig;

  setUp(() {
    validConfig = AgentConfig.fromYamlContent(validBootstrapYaml);
  });

  group('EdgeEndpoint', () {
    test('EdgeEndpoint parses host and port from config edge url', () {
      final endpoint1 = EdgeEndpoint.parse('127.0.0.1:8080');
      expect(endpoint1.host, '127.0.0.1');
      expect(endpoint1.port, 8080);

      final endpoint2 = EdgeEndpoint.parse('http://edge-server:9000');
      expect(endpoint2.host, 'edge-server');
      expect(endpoint2.port, 9000);

      final endpoint3 = EdgeEndpoint.parse('https://secure-edge');
      expect(endpoint3.host, 'secure-edge');
      expect(endpoint3.port, 443);

      final endpoint4 = EdgeEndpoint.parse('edge-no-port');
      expect(endpoint4.host, 'edge-no-port');
      expect(endpoint4.port, 80);
    });
  });

  group('EdgeRegistrationClient.register one-shot', () {
    test('register returns session result and closes the session', () async {
      final fakeClient = FakeRegistrationClient((config) {
        return RegistrationResult.accepted('node-123', 'alias-123', {
          'concurrency': 1,
        });
      });

      final result = await fakeClient.register(validConfig);

      expect(result.accepted, isTrue);
      expect(result.nodeId, 'node-123');
      expect(fakeClient.lastSession, isNotNull);
      expect(fakeClient.lastSession!.closed, isTrue);
    });
  });

  group('OtoServerRegistrationClient (Compatibility Fallback HTTP)', () {
    test('builds OTO Server registration request from config', () {
      final client = OtoServerRegistrationClient(
        commandTypes: ['Shell', 'Git'],
      );

      final request = client.buildRegisterRunnerRequest(validConfig);

      expect(request.enrollmentToken, 'token-456');
      expect(request.runnerId, 'agent-123');
      expect(request.alias, 'my-agent');
      expect(request.protocolVersion, otoRunnerProtocolVersion);
      expect(request.capability.name, otoRunnerCapabilityName);
      expect(request.commandCatalog.commandTypes, ['Shell', 'Git']);
    });

    test('builds registration request from injected capability provider', () {
      final client = OtoServerRegistrationClient(
        capabilityProvider: FakeRunnerCapabilityProvider(
          RunnerCapabilitySnapshot(
            name: 'custom-runner',
            version: '2.0.0',
            commandTypes: ['Print', 'ShellFile'],
          ),
        ),
      );

      final request = client.buildRegisterRunnerRequest(validConfig);

      expect(request.capability.name, 'custom-runner');
      expect(request.capability.version, '2.0.0');
      expect(request.commandCatalog.commandTypes, ['Print', 'ShellFile']);
    });

    test(
      'builds registration request using CommandCatalogRunnerCapabilityProvider',
      () {
        final client = OtoServerRegistrationClient(
          capabilityProvider: const CommandCatalogRunnerCapabilityProvider(),
        );

        final request = client.buildRegisterRunnerRequest(validConfig);

        expect(request.enrollmentToken, 'token-456');
        expect(request.runnerId, 'agent-123');
        expect(request.capability.name, otoRunnerCapabilityName);
        expect(request.capability.version, otoRunnerCapabilityVersion);
        expect(request.commandCatalog.commandTypes, isNotEmpty);
        expect(request.commandCatalog.commandTypes, contains('Shell'));
      },
    );

    test(
      'posts registration payload using CommandCatalogRunnerCapabilityProvider',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        final captured = Completer<Map<String, dynamic>>();
        final serverSub = server.listen((request) async {
          if (request.uri.path == '/api/v1/runners/register') {
            final body = await utf8.decoder.bind(request).join();
            captured.complete(jsonDecode(body) as Map<String, dynamic>);
            request.response
              ..headers.contentType = ContentType.json
              ..write(jsonEncode({'accepted': true, 'runner_id': 'agent-123'}));
          } else {
            request.response
              ..headers.contentType = ContentType.json
              ..write(jsonEncode({'success': true}));
          }
          await request.response.close();
        });

        try {
          final config = AgentConfig(
            agent: validConfig.agent,
            server: ServerConnectionConfig(
              url: 'http://${server.address.host}:${server.port}',
            ),
            runtime: validConfig.runtime,
          );
          final client = OtoServerRegistrationClient(
            capabilityProvider: const CommandCatalogRunnerCapabilityProvider(),
          );

          final result = await client.register(config);

          expect(result.accepted, isTrue);
          final payload = await captured.future;
          expect(payload['capability'], containsPair('name', 'oto-runner'));
          expect(
            (payload['command_catalog']['command_types'] as List<dynamic>),
            isNotEmpty,
          );
          expect(
            (payload['command_catalog']['command_types'] as List<dynamic>),
            contains('Shell'),
          );
        } finally {
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test('posts registration request to OTO Server endpoint', () async {
      final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
      final captured = Completer<Map<String, dynamic>>();
      final serverSub = server.listen((request) async {
        expect(request.method, 'POST');
        if (request.uri.path == '/api/v1/runners/register') {
          final body = await utf8.decoder.bind(request).join();
          captured.complete(jsonDecode(body) as Map<String, dynamic>);
          request.response
            ..headers.contentType = ContentType.json
            ..write(
              jsonEncode({
                'accepted': true,
                'runner_id': 'agent-123',
                'alias': 'my-agent',
              }),
            );
        } else {
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'success': true}));
        }
        await request.response.close();
      });

      try {
        final config = AgentConfig(
          agent: validConfig.agent,
          server: ServerConnectionConfig(
            url: 'http://${server.address.host}:${server.port}',
          ),
          runtime: validConfig.runtime,
        );
        final client = OtoServerRegistrationClient(
          commandTypes: ['Shell', 'Git'],
        );

        final result = await client.register(config);

        expect(result.accepted, isTrue);
        expect(result.runnerId, 'agent-123');
        final payload = await captured.future;
        expect(payload['enrollment_token'], 'token-456');
        expect(payload['runner_id'], 'agent-123');
        expect(payload['alias'], 'my-agent');
        expect(payload['protocol_version'], otoRunnerProtocolVersion);
        expect(payload['capability'], containsPair('name', 'oto-runner'));
        expect(
          payload['capability'],
          containsPair('version', otoRunnerCapabilityVersion),
        );
        expect(
          payload['command_catalog'],
          containsPair('command_types', ['Shell', 'Git']),
        );
      } finally {
        await serverSub.cancel();
        await server.close(force: true);
      }
    });

    test(
      'registration client handles incompatible runner error response',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        final serverSub = server.listen((request) async {
          request.response
            ..headers.contentType = ContentType.json
            ..write(
              jsonEncode({
                'accepted': false,
                'reject_reason':
                    'unsupported protocol version: "oto.runner.v2"',
                'error': {
                  'code': 'incompatible_runner',
                  'message': 'unsupported protocol version: "oto.runner.v2"',
                },
              }),
            );
          await request.response.close();
        });

        try {
          final config = AgentConfig(
            agent: validConfig.agent,
            server: ServerConnectionConfig(
              url: 'http://${server.address.host}:${server.port}',
            ),
            runtime: validConfig.runtime,
          );
          final client = OtoServerRegistrationClient(
            commandTypes: ['Shell', 'Git'],
          );

          final result = await client.register(config);
          expect(result.accepted, isFalse);
          expect(
            result.rejectReason,
            'unsupported protocol version: "oto.runner.v2"',
          );
        } finally {
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test('sends first heartbeat and disconnects on close', () async {
      final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
      final registerReceived = Completer<void>();
      final heartbeatReceived = Completer<void>();
      final disconnectReceived = Completer<void>();

      final serverSub = server.listen((request) async {
        if (request.method == 'POST' &&
            request.uri.path == '/api/v1/runners/register') {
          registerReceived.complete();
          request.response
            ..headers.contentType = ContentType.json
            ..write(
              jsonEncode({
                'accepted': true,
                'runner_id': 'agent-123',
                'alias': 'my-agent',
              }),
            );
        } else if (request.method == 'POST' &&
            request.uri.path == '/api/v1/runners/agent-123/heartbeat') {
          heartbeatReceived.complete();
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'success': true}));
        } else if (request.method == 'POST' &&
            request.uri.path == '/api/v1/runners/agent-123/disconnect') {
          disconnectReceived.complete();
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'success': true}));
        }
        await request.response.close();
      });

      try {
        final config = AgentConfig(
          agent: validConfig.agent,
          server: ServerConnectionConfig(
            url: 'http://${server.address.host}:${server.port}',
          ),
          runtime: validConfig.runtime,
        );
        final client = OtoServerRegistrationClient(
          commandTypes: ['Shell', 'Git'],
          heartbeatInterval: const Duration(milliseconds: 100),
        );

        final session = await client.openSession(config);

        expect(session, isA<OtoServerJobSession>());

        // First heartbeat should be sent automatically
        await heartbeatReceived.future.timeout(const Duration(seconds: 2));

        // Close session
        await session.close();

        // Disconnect should be sent
        await disconnectReceived.future.timeout(const Duration(seconds: 2));
      } finally {
        await serverSub.cancel();
        await server.close(force: true);
      }
    });

    test(
      'does not send heartbeat or disconnect on rejected registration',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        var heartbeatReceived = false;
        var disconnectReceived = false;

        final serverSub = server.listen((request) async {
          if (request.method == 'POST' &&
              request.uri.path == '/api/v1/runners/register') {
            request.response
              ..headers.contentType = ContentType.json
              ..write(
                jsonEncode({
                  'accepted': false,
                  'reject_reason': 'Invalid token',
                }),
              );
          } else if (request.uri.path.contains('heartbeat')) {
            heartbeatReceived = true;
            request.response
              ..headers.contentType = ContentType.json
              ..write(jsonEncode({'success': true}));
          } else if (request.uri.path.contains('disconnect')) {
            disconnectReceived = true;
            request.response
              ..headers.contentType = ContentType.json
              ..write(jsonEncode({'success': true}));
          }
          await request.response.close();
        });

        try {
          final config = AgentConfig(
            agent: validConfig.agent,
            server: ServerConnectionConfig(
              url: 'http://${server.address.host}:${server.port}',
            ),
            runtime: validConfig.runtime,
          );
          final client = OtoServerRegistrationClient(
            commandTypes: ['Shell', 'Git'],
            heartbeatInterval: const Duration(milliseconds: 50),
          );

          final session = await client.openSession(config);
          expect(session.result.accepted, isFalse);

          // Wait a short duration to ensure no heartbeat is sent
          await Future<void>.delayed(const Duration(milliseconds: 200));
          expect(heartbeatReceived, isFalse);

          await session.close();
          // Wait a short duration to ensure no disconnect is sent
          await Future<void>.delayed(const Duration(milliseconds: 100));
          expect(disconnectReceived, isFalse);
        } finally {
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test('session close cancels future heartbeats', () async {
      final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
      var heartbeatCount = 0;

      final serverSub = server.listen((request) async {
        if (request.method == 'POST' &&
            request.uri.path == '/api/v1/runners/register') {
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'accepted': true, 'runner_id': 'agent-123'}));
        } else if (request.uri.path.contains('heartbeat')) {
          heartbeatCount++;
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'success': true}));
        } else if (request.uri.path.contains('disconnect')) {
          request.response
            ..headers.contentType = ContentType.json
            ..write(jsonEncode({'success': true}));
        }
        await request.response.close();
      });

      try {
        final config = AgentConfig(
          agent: validConfig.agent,
          server: ServerConnectionConfig(
            url: 'http://${server.address.host}:${server.port}',
          ),
          runtime: validConfig.runtime,
        );
        final client = OtoServerRegistrationClient(
          commandTypes: ['Shell', 'Git'],
          heartbeatInterval: const Duration(milliseconds: 50),
        );

        final session = await client.openSession(config);

        // Wait for first heartbeat(s)
        await Future<void>.delayed(const Duration(milliseconds: 150));
        final countBeforeClose = heartbeatCount;
        expect(countBeforeClose, greaterThan(0));

        await session.close();

        // Wait longer to see if any more heartbeats occur
        await Future<void>.delayed(const Duration(milliseconds: 150));
        expect(heartbeatCount, equals(countBeforeClose));
      } finally {
        await serverSub.cancel();
        await server.close(force: true);
      }
    });

    test('job client claims jobs and reports build results', () async {
      final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
      final claimReceived = Completer<Map<String, dynamic>>();
      final reportReceived = Completer<Map<String, dynamic>>();
      final logReceived = Completer<Map<String, dynamic>>();
      final artifactReceived = Completer<Map<String, dynamic>>();

      final serverSub = server.listen((request) async {
        final bodyText = await utf8.decoder.bind(request).join();
        final body = bodyText.isEmpty
            ? <String, dynamic>{}
            : jsonDecode(bodyText) as Map<String, dynamic>;
        request.response.headers.contentType = ContentType.json;
        if (request.method == 'POST' &&
            request.uri.path == '/api/v1/runners/agent-123/jobs/claim') {
          claimReceived.complete(body);
          request.response.write(
            jsonEncode({
              'accepted': true,
              'runner_id': 'agent-123',
              'job_id': 'job-123',
              'execution_id': 'exec-123',
              'state': 'running',
              'run_request': {
                'runner_id': 'agent-123',
                'job_id': 'job-123',
                'execution_id': 'exec-123',
                'pipeline_yaml': 'commands:\n  - type: Shell',
                'variables': {'FLAVOR': 'release'},
                'command_types': ['Shell', 'Git'],
              },
            }),
          );
        } else if (request.method == 'POST' &&
            request.uri.path ==
                '/api/v1/runners/agent-123/executions/exec-123/report') {
          reportReceived.complete(body);
          request.response.write(
            jsonEncode({
              'accepted': true,
              'runner_id': 'agent-123',
              'job_id': 'job-123',
              'execution_id': 'exec-123',
              'state': 'succeeded',
            }),
          );
        } else if (request.method == 'POST' &&
            request.uri.path ==
                '/api/v1/runners/agent-123/executions/exec-123/logs') {
          logReceived.complete(body);
          request.response.statusCode = HttpStatus.created;
          request.response.write(jsonEncode({'accepted': true}));
        } else if (request.method == 'POST' &&
            request.uri.path ==
                '/api/v1/runners/agent-123/executions/exec-123/artifacts') {
          artifactReceived.complete(body);
          request.response.statusCode = HttpStatus.created;
          request.response.write(jsonEncode({'accepted': true}));
        } else {
          request.response.statusCode = HttpStatus.notFound;
          request.response.write(jsonEncode({'error': 'not found'}));
        }
        await request.response.close();
      });

      final client = OtoServerJobClient(
        serverUrl: 'http://${server.address.host}:${server.port}',
        runnerId: 'agent-123',
      );

      try {
        final claim = await client.claimJob(
          jobId: 'job-123',
          executionId: 'exec-123',
        );
        expect(claim.accepted, isTrue);
        expect(claim.state, 'running');
        expect(claim.runRequest, isNotNull);
        expect(claim.runRequest!.jobId, 'job-123');
        expect(claim.runRequest!.executionId, 'exec-123');
        expect(claim.runRequest!.runnerId, 'agent-123');
        expect(claim.runRequest!.pipelineYaml, 'commands:\n  - type: Shell');
        expect(claim.runRequest!.variables['FLAVOR'], 'release');
        expect(claim.runRequest!.commandTypes, ['Shell', 'Git']);

        final result = BuildResult.success(
          stepEvents: [
            StepEvent(
              stepId: 0,
              workflowIndex: 0,
              stepType: 'command',
              event: 'completed',
              timestamp: '2026-06-05T12:00:00Z',
              commandId: 'build',
              commandType: 'Shell',
            ),
          ],
        );
        final report = await client.reportExecution(
          jobId: 'job-123',
          executionId: 'exec-123',
          result: result,
        );
        expect(report.accepted, isTrue);
        expect(report.state, 'succeeded');

        await client.appendLog(
          executionId: 'exec-123',
          line: 'build completed',
        );
        await client.reportArtifact(
          executionId: 'exec-123',
          name: 'binary',
          path: '/dist/app',
        );

        final claimPayload = await claimReceived.future;
        expect(claimPayload['runner_id'], 'agent-123');
        expect(claimPayload['job_id'], 'job-123');
        expect(claimPayload['execution_id'], 'exec-123');

        final reportPayload = await reportReceived.future;
        expect(reportPayload['success'], isTrue);
        expect(reportPayload['exit_code'], 0);
        expect(reportPayload['message'], 'Build completed successfully.');
        expect(reportPayload['execution_result'], isA<Map<String, dynamic>>());
        final stepEvents = reportPayload['step_events'] as List<dynamic>;
        expect(stepEvents, hasLength(1));
        expect(stepEvents.first, containsPair('commandId', 'build'));

        final logPayload = await logReceived.future;
        expect(logPayload['line'], 'build completed');

        final artifactPayload = await artifactReceived.future;
        expect(artifactPayload['name'], 'binary');
        expect(artifactPayload['path'], '/dist/app');
      } finally {
        client.close();
        await serverSub.cancel();
        await server.close(force: true);
      }
    });

    test(
      'OTO runner protocol exposes run, cancel, status, self-update message fields',
      () {
        final runRequest = oto.RunRequest()
          ..runnerId = 'agent-123'
          ..jobId = 'job-123'
          ..executionId = 'exec-123'
          ..pipelineYamlPath = '/path/to/pipeline.yaml'
          ..pipelineYaml = 'commands:\n  - type: Shell'
          ..variables.addAll({'foo': 'bar'})
          ..commandTypes.addAll(['Shell', 'Git']);

        expect(runRequest.runnerId, 'agent-123');
        expect(runRequest.jobId, 'job-123');
        expect(runRequest.executionId, 'exec-123');
        expect(runRequest.pipelineYamlPath, '/path/to/pipeline.yaml');
        expect(runRequest.pipelineYaml, 'commands:\n  - type: Shell');
        expect(runRequest.variables['foo'], 'bar');
        expect(runRequest.commandTypes, ['Shell', 'Git']);

        final claimResponse = oto.JobClaimResponse()
          ..accepted = true
          ..runRequest = runRequest;
        expect(claimResponse.runRequest.jobId, 'job-123');

        final cancelRequest = oto.CancelRunRequest()
          ..runnerId = 'agent-123'
          ..executionId = 'exec-123'
          ..reason = 'user cancelled';
        expect(cancelRequest.reason, 'user cancelled');

        final cancelResponse = oto.CancelRunResponse()
          ..success = false
          ..error = (oto.ProtocolError()
            ..code = 'CANCEL_FAILED'
            ..message = 'execution not found');
        expect(cancelResponse.error.code, 'CANCEL_FAILED');
        expect(cancelResponse.error.message, 'execution not found');

        final statusRequest = oto.RunnerStatusRequest()..runnerId = 'agent-123';
        expect(statusRequest.runnerId, 'agent-123');

        final statusResponse = oto.RunnerStatusResponse()
          ..runnerId = 'agent-123'
          ..status = 'RUNNING'
          ..currentExecutionId = 'exec-123';
        expect(statusResponse.status, 'RUNNING');

        final selfUpdateRequest = oto.SelfUpdateRequest()
          ..runnerId = 'agent-123'
          ..version = 'v2.0.0'
          ..downloadUrl = 'http://example.com/download';
        expect(selfUpdateRequest.version, 'v2.0.0');

        final selfUpdateResponse = oto.SelfUpdateResponse()..success = true;
        expect(selfUpdateResponse.success, isTrue);
      },
    );

    test(
      'registration rejects enrollment_token_invalid without exposing token value',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        const secretToken = 'super-secret-enrollment-token-9876';

        final serverSub = server.listen((request) async {
          request.response
            ..headers.contentType = ContentType.json
            ..write(
              jsonEncode({
                'accepted': false,
                'reject_reason': 'enrollment_token_invalid',
                'error': {
                  'code': 'enrollment_token_invalid',
                  'message': 'invalid or expired enrollment token',
                },
              }),
            );
          await request.response.close();
        });

        try {
          final config = AgentConfig(
            agent: const AgentIdentityConfig(
              id: 'test-agent',
              enrollmentToken: secretToken,
            ),
            server: ServerConnectionConfig(
              url: 'http://${server.address.host}:${server.port}',
            ),
            runtime: validConfig.runtime,
          );
          final client = OtoServerRegistrationClient(commandTypes: ['Shell']);

          final result = await client.register(config);
          expect(result.accepted, isFalse);
          // fromOtoJson picks reject_reason before error.message
          expect(result.rejectReason, 'enrollment_token_invalid');
          // enrollment token value must not appear in rejection message
          expect(result.rejectReason, isNot(contains(secretToken)));
        } finally {
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test(
      'registration client fallback error parsing from ProtocolError map object',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        final serverSub = server.listen((request) async {
          request.response
            ..headers.contentType = ContentType.json
            ..write(
              jsonEncode({
                'accepted': false,
                'error': {
                  'code': 'REGISTRATION_FAILED',
                  'message': 'Failed to register via protocol error',
                },
              }),
            );
          await request.response.close();
        });

        try {
          final config = AgentConfig(
            agent: validConfig.agent,
            server: ServerConnectionConfig(
              url: 'http://${server.address.host}:${server.port}',
            ),
            runtime: validConfig.runtime,
          );
          final client = OtoServerRegistrationClient(
            commandTypes: ['Shell', 'Git'],
          );

          final result = await client.register(config);
          expect(result.accepted, isFalse);
          expect(result.rejectReason, 'Failed to register via protocol error');
        } finally {
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test(
      'job client fallback error parsing from ProtocolError map object',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        final serverSub = server.listen((request) async {
          request.response.headers.contentType = ContentType.json;
          if (request.uri.path.contains('/jobs/claim')) {
            request.response.write(
              jsonEncode({
                'accepted': false,
                'error': {
                  'code': 'CLAIM_FAILED',
                  'message': 'Failed to claim job via protocol error',
                },
              }),
            );
          } else if (request.uri.path.contains('/report')) {
            request.response.write(
              jsonEncode({
                'accepted': false,
                'error': {
                  'code': 'REPORT_FAILED',
                  'message': 'Failed to report execution via protocol error',
                },
              }),
            );
          } else if (request.uri.path.contains('/logs')) {
            request.response.statusCode = HttpStatus.created;
            request.response.write(
              jsonEncode({
                'accepted': false,
                'error': {
                  'code': 'LOG_FAILED',
                  'message': 'Failed to append log via protocol error',
                },
              }),
            );
          } else if (request.uri.path.contains('/artifacts')) {
            request.response.statusCode = HttpStatus.created;
            request.response.write(
              jsonEncode({
                'accepted': false,
                'error': {
                  'code': 'ARTIFACT_FAILED',
                  'message': 'Failed to report artifact via protocol error',
                },
              }),
            );
          }
          await request.response.close();
        });

        final client = OtoServerJobClient(
          serverUrl: 'http://${server.address.host}:${server.port}',
          runnerId: 'agent-123',
        );

        try {
          final claim = await client.claimJob(
            jobId: 'job-123',
            executionId: 'exec-123',
          );
          expect(claim.accepted, isFalse);
          expect(claim.errorMessage, 'Failed to claim job via protocol error');

          final report = await client.reportExecution(
            jobId: 'job-123',
            executionId: 'exec-123',
            result: BuildResult.failure('fail', StackTrace.empty),
          );
          expect(report.accepted, isFalse);
          expect(
            report.errorMessage,
            'Failed to report execution via protocol error',
          );

          await expectLater(
            client.appendLog(executionId: 'exec-123', line: 'test log'),
            throwsA(
              isA<HttpException>().having(
                (e) => e.message,
                'message',
                'Failed to append log via protocol error',
              ),
            ),
          );

          await expectLater(
            client.reportArtifact(
              executionId: 'exec-123',
              name: 'art',
              path: 'path',
            ),
            throwsA(
              isA<HttpException>().having(
                (e) => e.message,
                'message',
                'Failed to report artifact via protocol error',
              ),
            ),
          );
        } finally {
          client.close();
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );

    test(
      'job client cancels, fetches status, and requests self-update',
      () async {
        final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0);
        final cancelReceived = Completer<Map<String, dynamic>>();
        final statusReceived = Completer<void>();
        final updateReceived = Completer<Map<String, dynamic>>();

        final serverSub = server.listen((request) async {
          final bodyText = await utf8.decoder.bind(request).join();
          final body = bodyText.isEmpty
              ? <String, dynamic>{}
              : jsonDecode(bodyText) as Map<String, dynamic>;
          request.response.headers.contentType = ContentType.json;

          if (request.method == 'POST' &&
              request.uri.path ==
                  '/api/v1/runners/agent-123/executions/exec-123/cancel') {
            cancelReceived.complete(body);
            request.response.write(jsonEncode({'success': true}));
          } else if (request.method == 'GET' &&
              request.uri.path == '/api/v1/runners/agent-123/status') {
            statusReceived.complete();
            request.response.write(
              jsonEncode({
                'accepted': true,
                'runner_id': 'agent-123',
                'status': 'online',
                'current_job_id': 'job-123',
                'current_execution_id': 'exec-123',
                'message': 'running smoothly',
              }),
            );
          } else if (request.method == 'POST' &&
              request.uri.path == '/api/v1/runners/agent-123/self-update') {
            updateReceived.complete(body);
            request.response.write(
              jsonEncode({
                'success': true,
                'accepted': true,
                'deferred': false,
                'target_version': 'v2.0.0',
                'message': 'update scheduled',
              }),
            );
          } else {
            request.response.statusCode = HttpStatus.notFound;
            request.response.write(jsonEncode({'error': 'not found'}));
          }
          await request.response.close();
        });

        final client = OtoServerJobClient(
          serverUrl: 'http://${server.address.host}:${server.port}',
          runnerId: 'agent-123',
        );

        try {
          final cancelRes = await client.cancelRun(
            executionId: 'exec-123',
            reason: 'user cancel request',
          );
          expect(cancelRes.success, isTrue);

          final statusRes = await client.fetchRunnerStatus();
          expect(statusRes.accepted, isTrue);
          expect(statusRes.runnerId, 'agent-123');
          expect(statusRes.status, 'online');
          expect(statusRes.currentJobId, 'job-123');
          expect(statusRes.currentExecutionId, 'exec-123');
          expect(statusRes.message, 'running smoothly');

          final updateRes = await client.requestSelfUpdate(
            version: 'v2.0.0',
            downloadUrl: 'https://example.com/binary',
          );
          expect(updateRes.success, isTrue);
          expect(updateRes.accepted, isTrue);
          expect(updateRes.deferred, isFalse);
          expect(updateRes.targetVersion, 'v2.0.0');
          expect(updateRes.message, 'update scheduled');

          final cancelPayload = await cancelReceived.future;
          expect(cancelPayload['runner_id'], 'agent-123');
          expect(cancelPayload['execution_id'], 'exec-123');
          expect(cancelPayload['reason'], 'user cancel request');

          final updatePayload = await updateReceived.future;
          expect(updatePayload['runner_id'], 'agent-123');
          expect(updatePayload['version'], 'v2.0.0');
          expect(updatePayload['download_url'], 'https://example.com/binary');
        } finally {
          client.close();
          await serverSub.cancel();
          await server.close(force: true);
        }
      },
    );
  });

  group('AgentRunner', () {
    test('AgentRunner maps config token to register request', () async {
      AgentConfig? capturedConfig;
      final fakeClient = FakeRegistrationClient((config) {
        capturedConfig = config;
        return RegistrationResult.accepted('node-123', 'alias-123', {
          'concurrency': 1,
        });
      });

      final runner = DefaultAgentRunner(
        client: fakeClient,
        onLog: (msg) {},
        shutdownFuture: Future.value(),
      );
      await runner.run(validConfig);

      expect(capturedConfig, isNotNull);
      expect(capturedConfig!.agent.enrollmentToken, 'token-456');
    });

    test('AgentRunner keeps session open until shutdown then closes', () async {
      final fakeClient = FakeRegistrationClient((config) {
        return RegistrationResult.accepted('node-123', 'alias-123', {
          'concurrency': 1,
        });
      });

      final shutdown = Completer<void>();
      final runner = DefaultAgentRunner(
        client: fakeClient,
        onLog: (msg) {},
        shutdownFuture: shutdown.future,
      );

      final runFuture = runner.run(validConfig);
      // 종료 신호 전에는 session이 살아 있어야 한다.
      await Future<void>.delayed(Duration.zero);
      expect(fakeClient.lastSession, isNotNull);
      expect(fakeClient.lastSession!.closed, isFalse);

      shutdown.complete();
      await runFuture;
      expect(fakeClient.lastSession!.closed, isTrue);
    });

    test('AgentRunner runs injected job loop after registration', () async {
      final fakeClient = FakeRegistrationClient((config) {
        return RegistrationResult.accepted('node-123', 'alias-123', {
          'concurrency': 1,
        });
      });
      var jobLoopRan = false;

      final runner = DefaultAgentRunner(
        client: fakeClient,
        onLog: (msg) {},
        runJobs: (config, session) async {
          jobLoopRan = true;
          expect(config.agent.id, 'agent-123');
          expect(session.result.accepted, isTrue);
        },
        shutdownFuture: Future.value(),
      );

      await runner.run(validConfig);

      expect(jobLoopRan, isTrue);
    });

    test(
      'AgentRunner reports rejected registration and closes session',
      () async {
        final fakeClient = FakeRegistrationClient((config) {
          return RegistrationResult.rejected('Invalid token');
        });

        final runner = DefaultAgentRunner(
          client: fakeClient,
          onLog: (msg) {},
          shutdownFuture: Future.value(),
        );
        await expectLater(
          () => runner.run(validConfig),
          throwsA(
            isA<RegistrationException>().having(
              (e) => e.message,
              'message',
              'Invalid token',
            ),
          ),
        );
        expect(fakeClient.lastSession, isNotNull);
        expect(fakeClient.lastSession!.closed, isTrue);
      },
    );
  });

  group('RunRequestParser', () {
    test('parses inline YAML with variables', () {
      final parser = RunRequestParser(workspaceRoot: '/tmp/workspace');
      final request = oto.RunRequest()
        ..pipelineYaml = 'commands:\n  - type: Shell'
        ..variables['MY_VAR'] = 'my_value';
      final result = parser.parse(request);
      expect(result.yamlContent, 'commands:\n  - type: Shell');
      expect(result.variables, {'MY_VAR': 'my_value'});
    });

    test('rejects path traversal in pipelineYamlPath', () {
      final parser = RunRequestParser(workspaceRoot: '/var/workspace');
      final request = oto.RunRequest()
        ..pipelineYamlPath = '../../../etc/passwd';
      expect(() => parser.parse(request), throwsA(isA<ArgumentError>()));
    });

    test('accepts canonical path that stays within workspaceRoot', () {
      final tempDir = Directory.systemTemp.createTempSync('oto_parse_');
      addTearDown(() => tempDir.delete(recursive: true));
      final yamlFile = File('${tempDir.path}/pipeline.yaml');
      yamlFile.writeAsStringSync('commands:\n  - type: Print');
      final parser = RunRequestParser(workspaceRoot: tempDir.path);
      final request = oto.RunRequest()..pipelineYamlPath = 'pipeline.yaml';
      final result = parser.parse(request);
      expect(result.yamlContent, 'commands:\n  - type: Print');
    });

    test(
      'rejects sibling-prefix path escape (e.g. /var/workspace2 vs /var/workspace)',
      () {
        final parser = RunRequestParser(workspaceRoot: '/var/workspace');
        final request = oto.RunRequest()
          ..pipelineYamlPath = '/var/workspace2/evil.yaml';
        expect(() => parser.parse(request), throwsA(isA<ArgumentError>()));
      },
    );
  });

  group('RemoteRunExecutor.mergeVariablesIntoYaml', () {
    // The merged document is JSON, which Application.build re-parses with the
    // same parser used here. Asserting on the parsed property map proves the
    // structured merge instead of relying on fragile string layout.
    Map<String, dynamic> mergedProperty(
      String yaml,
      Map<String, String> variables,
    ) {
      final merged = RemoteRunExecutor.mergeVariablesIntoYaml(yaml, variables);
      final root = Application.getMapFromYamlA(merged);
      expect(root, isNotNull, reason: 'merged document must parse as a map');
      final property = root!['property'];
      expect(
        property,
        isA<Map>(),
        reason: 'merged document must have property map',
      );
      return Map<String, dynamic>.from(property as Map);
    }

    test('block-style property is overridden by remote variables', () {
      const yaml = '''
property:
  workspace: .
  FLAVOR: original
commands:
  - type: Print
''';
      final property = mergedProperty(yaml, {'FLAVOR': 'release'});
      expect(property['FLAVOR'], 'release');
      // Untouched existing entries are preserved.
      expect(property['workspace'], '.');
    });

    test('missing property section is created from remote variables', () {
      const yaml = '''
commands:
  - type: Print
''';
      final property = mergedProperty(yaml, {'FLAVOR': 'release'});
      expect(property['FLAVOR'], 'release');
    });

    test('inline property map is overridden and stays valid YAML', () {
      // The previous string-injection merge produced invalid YAML for this
      // valid inline-map form; the structured merge must keep it parseable.
      const yaml = '''
property: {workspace: ., FLAVOR: original}
commands:
  - type: Print
''';
      final property = mergedProperty(yaml, {'FLAVOR': 'release'});
      expect(property['FLAVOR'], 'release');
      expect(property['workspace'], '.');
    });

    test('empty variables leave the YAML untouched', () {
      const yaml = 'property:\n  workspace: .\ncommands:\n  - type: Print\n';
      expect(RemoteRunExecutor.mergeVariablesIntoYaml(yaml, const {}), yaml);
    });

    test('invalid non-map property is left untouched for build validation', () {
      // A present-but-non-map property is invalid. The merge must not coerce it
      // into a clean map (which would bypass Application._validateBuildMap); it
      // returns the original YAML so build validation still rejects it.
      const yaml = '''
property:
  - bad
commands:
  - type: Print
''';
      final merged = RemoteRunExecutor.mergeVariablesIntoYaml(yaml, {
        'FLAVOR': 'release',
      });
      expect(merged, yaml);
      // Confirm the preserved property is still the invalid (non-map) form.
      final root = Application.getMapFromYamlA(merged);
      expect(root!['property'], isA<List>());
    });
  });

  group('RemoteRunExecutor.parseArtifactDeclarations', () {
    test('parses valid artifact declarations from property.artifacts', () {
      const yaml = '''
property:
  workspace: .
  artifacts:
    - name: smoke-report
      path: dist/report.json
commands:
  - command: Print
    id: hello
    param:
      message: hi
pipeline:
  id: main
  workflow:
    - exe: hello
''';
      final artifacts = RemoteRunExecutor.parseArtifactDeclarations(
        yaml,
        workspaceRoot: '/tmp/workspace',
      );
      expect(artifacts, hasLength(1));
      expect(artifacts.first.name, 'smoke-report');
      expect(artifacts.first.path, 'dist/report.json');
    });

    test('returns empty list when no artifacts declared', () {
      const yaml = '''
property:
  workspace: .
commands:
  - type: Print
''';
      expect(
        RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/tmp/workspace',
        ),
        isEmpty,
      );
    });

    test('rejects artifact with missing name', () {
      const yaml = '''
property:
  artifacts:
    - path: dist/out.txt
commands:
  - type: Print
''';
      expect(
        () => RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/tmp/workspace',
        ),
        throwsA(isA<ArgumentError>()),
      );
    });

    test('rejects artifact with empty path', () {
      const yaml = '''
property:
  artifacts:
    - name: report
      path: ""
commands:
  - type: Print
''';
      expect(
        () => RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/tmp/workspace',
        ),
        throwsA(isA<ArgumentError>()),
      );
    });

    test('rejects non-list artifacts declaration', () {
      const yaml = '''
property:
  artifacts: not-a-list
commands:
  - type: Print
''';
      expect(
        () => RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/tmp/workspace',
        ),
        throwsA(isA<ArgumentError>()),
      );
    });

    test('rejects artifact path that escapes the workspace', () {
      const yaml = '''
property:
  artifacts:
    - name: evil
      path: ../../etc/passwd
commands:
  - type: Print
''';
      expect(
        () => RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/var/workspace',
        ),
        throwsA(isA<ArgumentError>()),
      );
    });

    test('rejects sibling-prefix absolute path escape', () {
      const yaml = '''
property:
  artifacts:
    - name: evil
      path: /var/workspace2/out.txt
commands:
  - type: Print
''';
      expect(
        () => RemoteRunExecutor.parseArtifactDeclarations(
          yaml,
          workspaceRoot: '/var/workspace',
        ),
        throwsA(isA<ArgumentError>()),
      );
    });
  });

  group('DefaultAgentRunner shutdown-aware loop', () {
    test(
      'no-job polling exits promptly when shutdownFuture completes',
      () async {
        // Set up a FakeOtoServerJobSession that always returns no-job.
        final shutdown = Completer<void>();
        final fakeRegistrationClient = _FakeJobRegistrationClient(
          result: RegistrationResult.accepted('node-1', null, {}),
          jobClient: _SingleResponseJobClient([]),
          onShutdown: shutdown,
        );

        final runner = DefaultAgentRunner(
          client: fakeRegistrationClient,
          onLog: (_) {},
          shutdownFuture: shutdown.future,
        );

        // Complete shutdown after a short delay.
        Future.delayed(const Duration(milliseconds: 50), shutdown.complete);

        final start = DateTime.now();
        await runner.run(validConfig);
        final elapsed = DateTime.now().difference(start);

        // Should exit well before the 3-second polling delay.
        expect(elapsed.inMilliseconds, lessThan(1000));
        expect(fakeRegistrationClient.lastSession!.closed, isTrue);
      },
    );

    test(
      'claimNextJob is invoked by compatibility polling loop and queued job is executed',
      () async {
        final shutdown = Completer<void>();
        var jobExecuted = false;
        final fakeRegistrationClient = _FakeJobRegistrationClient(
          result: RegistrationResult.accepted('node-1', null, {}),
          jobClient: _SingleResponseJobClient([
            JobClaimResult(
              accepted: true,
              jobId: 'job-loop-1',
              executionId: 'exec-loop-1',
              state: 'running',
              runRequest: (oto.RunRequest()
                ..pipelineYaml =
                    'property:\n  workspace: .\ncommands:\n  - command: Print\n    id: hello\n    param:\n      message: loop-test\npipeline:\n  id: p\n  workflow:\n    - exe: hello'),
            ),
          ]),
          onShutdown: shutdown,
          onJobExecuted: () {
            jobExecuted = true;
            shutdown.complete();
          },
        );

        final runner = DefaultAgentRunner(
          client: fakeRegistrationClient,
          onLog: (_) {},
          shutdownFuture: shutdown.future,
        );

        await runner.run(validConfig);
        expect(jobExecuted, isTrue);
      },
    );
  });
}

class _SingleResponseJobClient extends OtoServerJobClient {
  final List<JobClaimResult> _responses;
  int _callCount = 0;
  void Function()? _onExecuted;

  _SingleResponseJobClient(this._responses)
    : super(serverUrl: 'http://localhost', runnerId: 'fake');

  @override
  Future<JobClaimResult> claimNextJob() async {
    if (_callCount < _responses.length) {
      return _responses[_callCount++];
    }
    return const JobClaimResult(
      accepted: false,
      jobId: '',
      executionId: '',
      state: '',
    );
  }

  @override
  Future<ExecutionReportResult> reportExecution({
    required String jobId,
    required String executionId,
    required BuildResult result,
  }) async {
    _onExecuted?.call();
    return ExecutionReportResult(
      accepted: true,
      jobId: jobId,
      executionId: executionId,
      state: 'succeeded',
    );
  }

  @override
  Future<void> appendLog({
    required String executionId,
    required String line,
  }) async {}
}

class _FakeOtoServerJobSession implements OtoServerJobSession {
  @override
  final RegistrationResult result;
  @override
  final OtoServerJobClient jobs;
  bool closed = false;

  _FakeOtoServerJobSession(this.result, this.jobs);

  @override
  Future<void> close() async {
    closed = true;
  }
}

class _FakeJobRegistrationClient extends RegistrationClient {
  final RegistrationResult result;
  final _SingleResponseJobClient jobClient;
  final Completer<void> onShutdown;
  final void Function()? onJobExecuted;
  _FakeOtoServerJobSession? lastSession;

  _FakeJobRegistrationClient({
    required this.result,
    required this.jobClient,
    required this.onShutdown,
    this.onJobExecuted,
  });

  @override
  Future<AgentSession> openSession(AgentConfig config) async {
    jobClient._onExecuted = onJobExecuted;
    final session = _FakeOtoServerJobSession(result, jobClient);
    lastSession = session;
    return session;
  }
}