toki/oto
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(agent): legacy IOP 등록 경로를 정합성 있게 갱신한다

Browse source 
계측/테스트가 최신 코드 경로를 따라가도록 runner agent 등록 및 연결 검증 흐름을
동기화해 회귀 위험을 줄인다.
This commit is contained in:
toki 2026-06-10 19:00:23 +09:00
parent 67625af737
commit b81879b0dd
6 changed files with 54 additions and 487 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
agent-task/control_plane_phase_exit_followups/02_legacy_iop_boundary/CODE_REVIEW-cloud-G06.md
View file

@ -1,22 +0,0 @@
# Code review stub: runner registration legacy iop boundary split
## Review target
- Plan: `PLAN-cloud-G06.md`
- Status: pending implementation
## Reviewer focus
- Confirm default runner production imports are OTO Server only.
- Confirm legacy iop compatibility remains explicit and opt-in.
- Confirm type aliases or transitional names do not preserve the same dependency leak.
- Confirm tests still cover registration, job claim, reporting, cancellation, status, and self-update behavior.
## Required checks
- `cd apps/runner && dart analyze`
- `cd apps/runner && dart test`
## Notes
This item is a boundary cleanup, not a protocol rewrite.

96
agent-task/control_plane_phase_exit_followups/02_legacy_iop_boundary/PLAN-cloud-G06.md
View file

@ -1,96 +0,0 @@
# Runner registration legacy iop boundary split
## Metadata
- Status: planned
- Task group: `control_plane_phase_exit_followups`
- Plan id: `cloud-G06`
- Review stub: `CODE_REVIEW-cloud-G06.md`
- Roadmap target: first phase exit follow-up for `agent-roadmap/phase/independent-control-plane/PHASE.md`
## Problem
The default runner path has migrated toward OTO Server HTTP registration, but production runner code still exposes and imports legacy iop Edge registration concepts.
Examples:
- `apps/runner/lib/oto/agent/edge_registration_client.dart` contains both OTO Server HTTP clients and legacy proto-socket Edge registration.
- `apps/runner/lib/oto/agent/agent_runner.dart` depends on types named around `EdgeAgentSession`.
- The legacy iop smoke now runs only when `OTO_ENABLE_LEGACY_IOP_SMOKE=1`, but the production import boundary still makes legacy iop appear first-class.
This makes the first phase hard to close because "independent control plane" evidence is mixed with legacy Edge compatibility code.
## Rules read
- `agent-ops/rules/project/rules.md`
- `agent-ops/rules/project/domain/agent/rules.md`
- `agent-test/local/rules.md`
- `agent-test/local/agent-smoke.md`
## Current evidence
- `apps/runner/lib/oto/agent/edge_registration_client.dart`
- Contains `OtoServerRegistrationClient`, `OtoServerJobClient`, `RegistrationClient`, `EdgeRegistrationClient`, and session/result types together.
- Imports legacy iop generated proto/proto-socket packages.
- `apps/runner/lib/oto/agent/agent_runner.dart`
- The default runner imports `edge_registration_client.dart`.
- The default runner works against the OTO Server path but keeps legacy naming in its interface.
- `apps/runner/test/oto_agent_registration_test.dart`
- Already has broad coverage for OTO Server registration/job client behavior and fake runner sessions.
- `apps/runner/test/oto_iop_connection_smoke_test.dart`
- Is now opt-in via `OTO_ENABLE_LEGACY_IOP_SMOKE=1`.
## Scope
Separate the production OTO Server runner path from legacy iop Edge compatibility code without removing the legacy smoke itself.
Do not change protocol behavior, generated protobuf files, or enrollment semantics in this task.
## Proposed implementation
1. Introduce a neutral runner registration boundary.
- Move shared OTO Server types to a neutral file, for example `registration_client.dart`.
- Prefer names like `AgentSession`, `RegistrationClient`, `OtoServerRegistrationClient`, and `OtoServerJobClient`.
- Keep public API changes localized to runner internals and tests.
2. Isolate legacy iop Edge code.
- Keep `EdgeRegistrationClient` and proto-socket-only imports in a legacy-specific file or clearly marked compatibility section.
- Ensure normal runner imports do not pull legacy iop generated proto/proto-socket code.
- Keep `oto_iop_connection_smoke_test.dart` importing the legacy path explicitly.
3. Update default runner code and tests.
- `DefaultAgentRunner` should depend on the neutral interface/session type.
- Fake clients in `oto_agent_registration_test.dart` should use neutral naming.
- Existing OTO Server registration/job tests should remain behaviorally equivalent.
4. Update documentation only where it names the default path.
- Keep legacy smoke instructions opt-in.
- Do not add broad migration docs unless needed to explain the new boundary.
## Expected files
- `apps/runner/lib/oto/agent/edge_registration_client.dart`
- `apps/runner/lib/oto/agent/agent_runner.dart`
- Possible new or renamed file under `apps/runner/lib/oto/agent/`
- `apps/runner/test/oto_agent_registration_test.dart`
- `apps/runner/test/oto_iop_connection_smoke_test.dart`
- Possibly `apps/runner/test/oto_agent_migration_plan_test.dart`
## Verification
- `cd apps/runner && dart analyze`
- `cd apps/runner && dart test test/oto_agent_registration_test.dart`
- `cd apps/runner && dart test test/oto_agent_migration_plan_test.dart`
- `cd apps/runner && dart test`
## Risks
- Renaming exported types may break hidden or downstream imports.
- A file split can accidentally remove legacy smoke coverage if imports are not explicit.
- Keeping compatibility aliases may be useful for one transition step, but they should not hide the default production boundary.
## Done when
- The default runner path no longer imports legacy iop proto/proto-socket code.
- Legacy iop smoke remains available only as explicit migration evidence.
- Runner analysis and tests pass locally.

8
apps/runner/lib/oto/agent/agent_runner.dart
View file

@ -2,7 +2,7 @@ import 'dart:async';
import 'dart:io';
import 'package:oto/cli/cli.dart';
import 'package:oto/oto/agent/agent_config.dart';
import 'package:oto/oto/agent/edge_registration_client.dart';
import 'package:oto/oto/agent/registration_client.dart';
import 'package:oto/oto/agent/remote_run_executor.dart';
abstract class AgentRunner {
@ -20,14 +20,14 @@ class RegistrationException implements Exception {
class DefaultAgentRunner implements AgentRunner {
final RegistrationClient _client;
final void Function(String)? _onLog;
final Future<void> Function(AgentConfig, EdgeAgentSession)? _runJobs;
final Future<void> Function(AgentConfig, AgentSession)? _runJobs;
// Injected shutdown future for tests; null means use real OS signals.
final Future<void>? _shutdownFuture;
DefaultAgentRunner({
RegistrationClient? client,
void Function(String)? onLog,
Future<void> Function(AgentConfig, EdgeAgentSession)? runJobs,
Future<void> Function(AgentConfig, AgentSession)? runJobs,
Future<void>? shutdownFuture,
}) : _client = client ?? OtoServerRegistrationClient(),
_onLog = onLog,
@ -38,7 +38,7 @@ class DefaultAgentRunner implements AgentRunner {
Future<void> run(AgentConfig config) async {
_log('Starting registration with OTO Server at "${config.server.url}"...');
final EdgeAgentSession session;
final AgentSession session;
try {
session = await _client.openSession(config);
} on RegistrationException {

384
apps/runner/lib/oto/agent/edge_registration_client.dart
View file

@ -1,19 +1,25 @@
import 'dart:async';
import 'dart:convert';
// Legacy iop Edge registration client.
//
// This file contains the proto-socket based EdgeRegistrationClient used by the
// legacy iop Edge compatibility path. Normal production runner code should
// import registration_client.dart instead.
//
// The legacy iop smoke test (oto_iop_connection_smoke_test.dart) imports this
// file explicitly so that it remains available as migration evidence without
// pulling legacy iop proto/proto-socket code into the default runner path.
import 'dart:io';
import 'package:http/http.dart' as http;
import 'package:proto_socket/proto_socket.dart';
import 'package:oto/oto/agent/agent_config.dart';
import 'package:oto/oto/agent/iop/runtime.pb.dart' as iop;
import 'package:oto/oto/agent/oto/runner.pb.dart' as oto;
import 'package:oto/oto/agent/oto_server_job_client.dart';
import 'package:oto/oto/commands/command.dart';
import 'package:oto/oto/commands/command_registry.dart';
import 'package:oto/oto/agent/registration_client.dart';
const otoRunnerProtocolVersion = 'oto.runner.v1';
const otoRunnerCapabilityName = 'oto-runner';
const otoRunnerCapabilityVersion = '1.0.0';
export 'package:oto/oto/agent/registration_client.dart'
show AgentSession, RegistrationResult, RegistrationClient;
// Legacy alias for code that still uses the old name.
typedef EdgeAgentSession = AgentSession;
class EdgeEndpoint {
final String host;
@ -43,109 +49,24 @@ class EdgeEndpoint {
}
}
class RegistrationResult {
final bool accepted;
final String? rejectReason;
final String? nodeId;
final String? alias;
final Map<String, dynamic>? runtimeConfig;
RegistrationResult._({
required this.accepted,
this.rejectReason,
this.nodeId,
this.alias,
this.runtimeConfig,
});
String? get runnerId => nodeId;
factory RegistrationResult.accepted(
String nodeId,
String? alias,
Map<String, dynamic> runtimeConfig,
) {
return RegistrationResult._(
accepted: true,
nodeId: nodeId,
alias: alias,
runtimeConfig: runtimeConfig,
);
RegistrationResult _registrationResultFromIopResponse(
iop.RegisterResponse response,
) {
if (!response.accepted) {
return RegistrationResult.rejected(response.reason);
}
factory RegistrationResult.rejected(String reason) {
return RegistrationResult._(accepted: false, rejectReason: reason);
final runtimeMap = <String, dynamic>{};
if (response.hasConfig() && response.config.hasRuntime()) {
runtimeMap['concurrency'] = response.config.runtime.concurrency.toInt();
runtimeMap['workspaceRoot'] = response.config.runtime.workspaceRoot;
}
factory RegistrationResult.fromResponse(iop.RegisterResponse response) {
if (!response.accepted) {
return RegistrationResult.rejected(response.reason);
}
final runtimeMap = <String, dynamic>{};
if (response.hasConfig() && response.config.hasRuntime()) {
runtimeMap['concurrency'] = response.config.runtime.concurrency.toInt();
runtimeMap['workspaceRoot'] = response.config.runtime.workspaceRoot;
}
return RegistrationResult.accepted(
response.nodeId,
response.alias.isEmpty ? null : response.alias,
runtimeMap,
);
}
factory RegistrationResult.fromOtoResponse(
oto.RegisterRunnerResponse response,
) {
if (!response.accepted) {
final reason = response.rejectReason.isNotEmpty
? response.rejectReason
: (response.hasError() ? response.error.message : 'rejected');
return RegistrationResult.rejected(reason);
}
return RegistrationResult.accepted(
response.runnerId,
response.alias.isEmpty ? null : response.alias,
const {},
);
}
factory RegistrationResult.fromOtoJson(Map<String, dynamic> json) {
final accepted = json['accepted'] == true;
if (!accepted) {
final err = json['error'];
final errMessageFromObj = (err is Map) ? err['message'] : null;
return RegistrationResult.rejected(
(json['reject_reason'] ??
json['rejectReason'] ??
errMessageFromObj ??
'rejected')
.toString(),
);
}
return RegistrationResult.accepted(
(json['runner_id'] ?? json['runnerId'] ?? '').toString(),
_emptyToNull((json['alias'] ?? '').toString()),
const {},
);
}
static String? _emptyToNull(String value) => value.isEmpty ? null : value;
}
abstract class RegistrationClient {
Future<EdgeAgentSession> openSession(AgentConfig config);
Future<RegistrationResult> register(AgentConfig config) async {
final session = await openSession(config);
try {
return session.result;
} finally {
await session.close();
}
}
return RegistrationResult.accepted(
response.nodeId,
response.alias.isEmpty ? null : response.alias,
runtimeMap,
);
}
class _OtoIopClient extends ProtobufClient {
@ -164,24 +85,7 @@ class _OtoIopClient extends ProtobufClient {
});
}
/// Edge에 등록한 뒤에도 연결을 유지하는 agent session.
///
/// registration accepted 이후 최초 heartbeat까지 살아 있어야 하는 OTO agent를
/// 위해, connection을 닫지 않고 보존한다. session 소유자는 종료 시 [close]를
/// 호출해 transport를 정리할 책임이 있다.
abstract class EdgeAgentSession {
/// register 요청에 대한 Edge 응답 결과.
RegistrationResult get result;
/// 유지 중인 Edge 연결을 닫는다. 여러 번 호출해도 안전하다.
Future<void> close();
}
abstract class OtoServerJobSession implements EdgeAgentSession {
OtoServerJobClient get jobs;
}
class _OtoEdgeAgentSession implements EdgeAgentSession {
class _OtoEdgeAgentSession implements AgentSession {
final _OtoIopClient _client;
@override
@ -193,234 +97,14 @@ class _OtoEdgeAgentSession implements EdgeAgentSession {
Future<void> close() => _client.close();
}
class OtoServerRegistrationClient extends RegistrationClient {
final http.Client? _client;
final List<String>? _commandTypes;
final Duration _heartbeatInterval;
OtoServerRegistrationClient({
http.Client? client,
Iterable<String>? commandTypes,
Duration heartbeatInterval = const Duration(seconds: 30),
}) : _client = client,
_commandTypes = commandTypes == null
? null
: List.unmodifiable(commandTypes),
_heartbeatInterval = heartbeatInterval;
@override
Future<EdgeAgentSession> openSession(AgentConfig config) async {
final client = _client ?? http.Client();
final shouldCloseClient = _client == null;
try {
final request = buildRegisterRunnerRequest(config);
final response = await client
.post(
_registerUri(config.server.url),
headers: {'content-type': 'application/json'},
body: jsonEncode(_registerRequestJson(request)),
)
.timeout(const Duration(seconds: 5));
if (response.statusCode < 200 || response.statusCode >= 300) {
throw HttpException(
'Registration request failed with HTTP ${response.statusCode}',
);
}
final decoded = jsonDecode(response.body);
if (decoded is! Map<String, dynamic>) {
throw const FormatException('Invalid registration response body');
}
final result = RegistrationResult.fromOtoJson(decoded);
return _OtoServerAgentSession(
result,
client,
shouldCloseClient,
config.server.url,
config.agent.id,
heartbeatInterval: _heartbeatInterval,
);
} catch (_) {
if (shouldCloseClient) {
client.close();
}
rethrow;
}
}
oto.RegisterRunnerRequest buildRegisterRunnerRequest(AgentConfig config) {
final request = oto.RegisterRunnerRequest()
..enrollmentToken = config.agent.enrollmentToken
..runnerId = config.agent.id
..alias = config.agent.alias ?? ''
..protocolVersion = otoRunnerProtocolVersion
..capability = (oto.RunnerCapability()
..name = otoRunnerCapabilityName
..version = otoRunnerCapabilityVersion)
..commandCatalog = (oto.CommandCatalogSummary()
..commandTypes.addAll(_commandTypes ?? _defaultCommandTypes()));
return request;
}
static Uri _registerUri(String serverUrl) {
var raw = serverUrl;
if (!raw.contains('://')) {
raw = 'http://$raw';
}
final uri = Uri.parse(raw);
if (uri.host.isEmpty) {
throw FormatException('Invalid server url: "$serverUrl"');
}
return uri.replace(path: _joinPath(uri.path, '/api/v1/runners/register'));
}
static Uri heartbeatUri(String serverUrl, String runnerId) {
var raw = serverUrl;
if (!raw.contains('://')) {
raw = 'http://$raw';
}
final uri = Uri.parse(raw);
if (uri.host.isEmpty) {
throw FormatException('Invalid server url: "$serverUrl"');
}
return uri.replace(
path: _joinPath(uri.path, '/api/v1/runners/$runnerId/heartbeat'),
);
}
static Uri disconnectUri(String serverUrl, String runnerId) {
var raw = serverUrl;
if (!raw.contains('://')) {
raw = 'http://$raw';
}
final uri = Uri.parse(raw);
if (uri.host.isEmpty) {
throw FormatException('Invalid server url: "$serverUrl"');
}
return uri.replace(
path: _joinPath(uri.path, '/api/v1/runners/$runnerId/disconnect'),
);
}
static String _joinPath(String basePath, String endpointPath) {
final base = basePath.endsWith('/')
? basePath.substring(0, basePath.length - 1)
: basePath;
return '$base$endpointPath';
}
static Map<String, dynamic> _registerRequestJson(
oto.RegisterRunnerRequest request,
) => {
'enrollment_token': request.enrollmentToken,
'runner_id': request.runnerId,
'alias': request.alias,
'protocol_version': request.protocolVersion,
'capability': {
'name': request.capability.name,
'version': request.capability.version,
},
'command_catalog': {
'command_types': request.commandCatalog.commandTypes.toList(),
},
};
List<String> _defaultCommandTypes() {
registerAllCommands();
return Command.catalogRows.map((row) => row['type']!).toList();
}
}
class _OtoServerAgentSession implements OtoServerJobSession {
final http.Client _client;
final bool _shouldCloseClient;
final String _serverUrl;
final String _runnerId;
Timer? _heartbeatTimer;
bool _closed = false;
@override
final RegistrationResult result;
@override
late final OtoServerJobClient jobs;
_OtoServerAgentSession(
this.result,
this._client,
this._shouldCloseClient,
this._serverUrl,
this._runnerId, {
required Duration heartbeatInterval,
}) {
jobs = OtoServerJobClient(
serverUrl: _serverUrl,
runnerId: _runnerId,
client: _client,
);
if (result.accepted) {
_heartbeatTimer = Timer.periodic(
heartbeatInterval,
(_) => _sendHeartbeat(),
);
scheduleMicrotask(() => _sendHeartbeat());
}
}
Future<void> _sendHeartbeat() async {
if (_closed) return;
try {
final uri = OtoServerRegistrationClient.heartbeatUri(
_serverUrl,
_runnerId,
);
final body = jsonEncode({
'runner_id': _runnerId,
'status': 1, // HEARTBEAT_STATUS_HEALTHY
});
await _client
.post(uri, headers: {'content-type': 'application/json'}, body: body)
.timeout(const Duration(seconds: 5));
} catch (_) {
// Ignore background errors
}
}
@override
Future<void> close() async {
if (_closed) return;
_closed = true;
_heartbeatTimer?.cancel();
_heartbeatTimer = null;
if (result.accepted) {
try {
final uri = OtoServerRegistrationClient.disconnectUri(
_serverUrl,
_runnerId,
);
await _client
.post(uri, headers: {'content-type': 'application/json'})
.timeout(const Duration(seconds: 5));
} catch (_) {
// Ignore disconnect errors
}
}
if (_shouldCloseClient) {
_client.close();
}
}
}
class EdgeRegistrationClient extends RegistrationClient {
/// Edge에 연결해 register 요청을 보낸 뒤, 연결을 유지한 채 session을 돌려준다.
///
/// 반환된 [EdgeAgentSession] 소유자는 종료 시 반드시 [EdgeAgentSession.close]를
/// 반환된 [AgentSession] 소유자는 종료 시 반드시 [AgentSession.close]를
/// 호출해야 한다. 연결 단계에서 실패하면 내부에서 transport를 닫고 예외를
/// 다시 던지므로 leak이 생기지 않는다.
@override
Future<EdgeAgentSession> openSession(AgentConfig config) async {
Future<AgentSession> openSession(AgentConfig config) async {
final endpoint = EdgeEndpoint.parse(config.edge.url);
final socket = await ProtobufClient.connect(endpoint.host, endpoint.port);
final client = _OtoIopClient(socket);
@ -432,7 +116,7 @@ class EdgeRegistrationClient extends RegistrationClient {
);
return _OtoEdgeAgentSession(
client,
RegistrationResult.fromResponse(response),
_registrationResultFromIopResponse(response),
);
} catch (_) {
await client.close();

29
apps/runner/test/oto_agent_registration_test.dart
View file

@ -5,7 +5,8 @@ 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';
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/application.dart';
@ -13,12 +14,12 @@ 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 FakeEdgeAgentSession implements EdgeAgentSession {
class FakeAgentSession implements AgentSession {
@override
final RegistrationResult result;
bool closed = false;
FakeEdgeAgentSession(this.result);
FakeAgentSession(this.result);
@override
Future<void> close() async {
@ -26,15 +27,15 @@ class FakeEdgeAgentSession implements EdgeAgentSession {
}
}
class FakeEdgeRegistrationClient extends RegistrationClient {
class FakeRegistrationClient extends RegistrationClient {
final RegistrationResult Function(AgentConfig config) _onRegister;
FakeEdgeAgentSession? lastSession;
FakeAgentSession? lastSession;
FakeEdgeRegistrationClient(this._onRegister);
FakeRegistrationClient(this._onRegister);
@override
Future<EdgeAgentSession> openSession(AgentConfig config) async {
final session = FakeEdgeAgentSession(_onRegister(config));
Future<AgentSession> openSession(AgentConfig config) async {
final session = FakeAgentSession(_onRegister(config));
lastSession = session;
return session;
}
@ -82,7 +83,7 @@ runtime:
group('EdgeRegistrationClient.register one-shot', () {
test('register returns session result and closes the session', () async {
final fakeClient = FakeEdgeRegistrationClient((config) {
final fakeClient = FakeRegistrationClient((config) {
return RegistrationResult.accepted('node-123', 'alias-123', {
'concurrency': 1,
});
@ -823,7 +824,7 @@ runtime:
group('AgentRunner', () {
test('AgentRunner maps config token to register request', () async {
AgentConfig? capturedConfig;
final fakeClient = FakeEdgeRegistrationClient((config) {
final fakeClient = FakeRegistrationClient((config) {
capturedConfig = config;
return RegistrationResult.accepted('node-123', 'alias-123', {
'concurrency': 1,
@ -842,7 +843,7 @@ runtime:
});
test('AgentRunner keeps session open until shutdown then closes', () async {
final fakeClient = FakeEdgeRegistrationClient((config) {
final fakeClient = FakeRegistrationClient((config) {
return RegistrationResult.accepted('node-123', 'alias-123', {
'concurrency': 1,
});
@ -867,7 +868,7 @@ runtime:
});
test('AgentRunner runs injected job loop after registration', () async {
final fakeClient = FakeEdgeRegistrationClient((config) {
final fakeClient = FakeRegistrationClient((config) {
return RegistrationResult.accepted('node-123', 'alias-123', {
'concurrency': 1,
});
@ -893,7 +894,7 @@ runtime:
test(
'AgentRunner reports rejected registration and closes session',
() async {
final fakeClient = FakeEdgeRegistrationClient((config) {
final fakeClient = FakeRegistrationClient((config) {
return RegistrationResult.rejected('Invalid token');
});
@ -1290,7 +1291,7 @@ class _FakeJobRegistrationClient extends RegistrationClient {
});
@override
Future<EdgeAgentSession> openSession(AgentConfig config) async {
Future<AgentSession> openSession(AgentConfig config) async {
jobClient._onExecuted = onJobExecuted;
final session = _FakeOtoServerJobSession(result, jobClient);
lastSession = session;

2
apps/runner/test/oto_server_connection_smoke_test.dart
View file

@ -5,7 +5,7 @@ import 'dart:io';
import 'package:http/http.dart' as http;
import 'package:test/test.dart';
import 'package:oto/oto/agent/agent_config.dart';
import 'package:oto/oto/agent/edge_registration_client.dart';
import 'package:oto/oto/agent/registration_client.dart';
import 'package:oto/oto/agent/remote_run_executor.dart';
import 'package:oto/oto/application.dart';
import 'package:oto/oto/core/build_result.dart';