diff --git a/apps/mobile_app/lib/core/network/agent_api_client.dart b/apps/mobile_app/lib/core/network/agent_api_client.dart index fbd5945..82eaccd 100644 --- a/apps/mobile_app/lib/core/network/agent_api_client.dart +++ b/apps/mobile_app/lib/core/network/agent_api_client.dart @@ -1,7 +1,17 @@ import 'package:dio/dio.dart'; class AgentApiClient { - AgentApiClient(this.baseUri, {Dio? dio}) : _dio = dio ?? Dio(); + AgentApiClient(this.baseUri, {Dio? dio}) : _dio = dio ?? Dio() { + _dio.options = _dio.options.copyWith( + connectTimeout: _dio.options.connectTimeout ?? _defaultConnectTimeout, + sendTimeout: _dio.options.sendTimeout ?? _defaultSendTimeout, + receiveTimeout: _dio.options.receiveTimeout ?? _defaultReceiveTimeout, + ); + } + + static const Duration _defaultConnectTimeout = Duration(seconds: 8); + static const Duration _defaultSendTimeout = Duration(seconds: 10); + static const Duration _defaultReceiveTimeout = Duration(seconds: 20); final Uri baseUri; final Dio _dio; diff --git a/apps/mobile_app/lib/core/network/agent_error_formatter.dart b/apps/mobile_app/lib/core/network/agent_error_formatter.dart index f50050d..6189cba 100644 --- a/apps/mobile_app/lib/core/network/agent_error_formatter.dart +++ b/apps/mobile_app/lib/core/network/agent_error_formatter.dart @@ -14,5 +14,47 @@ String formatAgentError(Object error, {String fallback = 'Request failed.'}) { } } + if (error case DioException dioError) { + final origin = _describeAgentOrigin(dioError.requestOptions); + switch (dioError.type) { + case DioExceptionType.connectionTimeout: + case DioExceptionType.connectionError: + return 'Cannot reach the agent${origin == null ? '' : ' at $origin'}. ' + 'Check that the desktop agent is running and that the agent URL is correct, then try again.'; + case DioExceptionType.badCertificate: + return 'The certificate presented by the agent${origin == null ? '' : ' at $origin'} ' + 'is not trusted on this device. Trust the certificate or use HTTP for local testing.'; + case DioExceptionType.receiveTimeout: + return 'The agent${origin == null ? '' : ' at $origin'} took too long to respond. ' + 'Check that it is still running and try again.'; + case DioExceptionType.sendTimeout: + return 'Sending the request to the agent${origin == null ? '' : ' at $origin'} took too long. ' + 'Check the connection and try again.'; + case DioExceptionType.badResponse: + case DioExceptionType.cancel: + case DioExceptionType.unknown: + break; + } + } + return '$fallback $error'; } + +String? _describeAgentOrigin(RequestOptions requestOptions) { + final uri = requestOptions.uri; + if (uri.hasScheme && uri.host.isNotEmpty) { + return uri.origin; + } + + final baseUrl = requestOptions.baseUrl; + if (baseUrl.isEmpty) { + return null; + } + + final parsed = Uri.tryParse(baseUrl); + if (parsed == null || !parsed.hasScheme || parsed.host.isEmpty) { + return null; + } + + return parsed.origin; +} diff --git a/apps/mobile_app/lib/features/projects/project_detail_page.dart b/apps/mobile_app/lib/features/projects/project_detail_page.dart index 8e2c987..b80c7f7 100644 --- a/apps/mobile_app/lib/features/projects/project_detail_page.dart +++ b/apps/mobile_app/lib/features/projects/project_detail_page.dart @@ -257,7 +257,12 @@ class _ProjectDetailPageState extends ConsumerState { return Center( child: Padding( padding: const EdgeInsets.all(24), - child: Text('Failed to load project: ${snapshot.error}'), + child: Text( + formatAgentError( + snapshot.error!, + fallback: 'Failed to load project.', + ), + ), ), ); } diff --git a/apps/mobile_app/lib/features/terminal/terminal_session_coordinator.dart b/apps/mobile_app/lib/features/terminal/terminal_session_coordinator.dart index bb0cfdb..bac6bb2 100644 --- a/apps/mobile_app/lib/features/terminal/terminal_session_coordinator.dart +++ b/apps/mobile_app/lib/features/terminal/terminal_session_coordinator.dart @@ -47,6 +47,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { ); static const Duration reconnectDelay = Duration(seconds: 1); static const int initialHistoryLineCount = 200; + static const int pendingInputCharacterLimit = 2048; final TerminalInteractionController controller; final AgentApiClient apiClient; @@ -64,7 +65,11 @@ class TerminalSessionCoordinator extends ChangeNotifier { int _historyLineCount = initialHistoryLineCount; bool _isLoadingOlderHistory = false; bool _isDisposed = false; + bool _connectionAttemptInProgress = false; + bool _reconnectPending = false; String _connectionStatus = 'Connecting...'; + final List _pendingInputs = []; + int _pendingInputCharacterCount = 0; bool get isLoadingOlderHistory => _isLoadingOlderHistory; @@ -72,6 +77,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { Future start({bool isReconnect = false}) async { _cancelPendingReconnect(); + _reconnectPending = false; if (_isDisposed) { return; @@ -100,6 +106,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { final socketSession = sessionFactory(baseUri: baseUri, session: session); _socketSession = socketSession; + _connectionAttemptInProgress = true; try { await socketSession.connect( @@ -123,6 +130,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { 'socket.resize.send', '${viewport.columns}x${viewport.rows}', ); + _flushPendingInputs(socketSession); controller.markConnected(); _connectionStatus = 'Attached to ${session.name}'; diagnosticLog?.add('socket.attach.ack', session.sessionId); @@ -137,6 +145,8 @@ class TerminalSessionCoordinator extends ChangeNotifier { diagnosticLog?.add('socket.connect.error', '$error'); notifyListeners(); _scheduleReconnect(); + } finally { + _connectionAttemptInProgress = false; } } @@ -148,6 +158,15 @@ class TerminalSessionCoordinator extends ChangeNotifier { void sendInput(String input) { final socketSession = _socketSession; if (socketSession == null) { + if (_shouldBufferInput(input)) { + _bufferInput(input); + diagnosticLog?.add( + 'socket.input.buffer', + 'reason=no-session input=${_formatInputForDiagnostics(input)}', + ); + return; + } + diagnosticLog?.add( 'socket.input.skip', 'reason=no-session input=${_formatInputForDiagnostics(input)}', @@ -163,10 +182,18 @@ class TerminalSessionCoordinator extends ChangeNotifier { _formatInputForDiagnostics(input), ); case TerminalSocketDispatchResult.noTransport: - diagnosticLog?.add( - 'socket.input.skip', - 'reason=no-transport input=${_formatInputForDiagnostics(input)}', - ); + if (_shouldBufferInput(input)) { + _bufferInput(input); + diagnosticLog?.add( + 'socket.input.buffer', + 'reason=no-transport input=${_formatInputForDiagnostics(input)}', + ); + } else { + diagnosticLog?.add( + 'socket.input.skip', + 'reason=no-transport input=${_formatInputForDiagnostics(input)}', + ); + } case TerminalSocketDispatchResult.emptyInput: diagnosticLog?.add('socket.input.skip', 'reason=empty-input'); } @@ -212,6 +239,8 @@ class TerminalSessionCoordinator extends ChangeNotifier { Future close() async { _isDisposed = true; _cancelPendingReconnect(); + _pendingInputs.clear(); + _pendingInputCharacterCount = 0; await _closeActiveSession(); } @@ -245,6 +274,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { void _scheduleReconnect() { _cancelPendingReconnect(); + _reconnectPending = true; controller.markReconnecting(); _connectionStatus = 'Connection lost. Reconnecting...'; diagnosticLog?.add('socket.disconnect', session.sessionId); @@ -261,6 +291,7 @@ class TerminalSessionCoordinator extends ChangeNotifier { void _cancelPendingReconnect() { _cancelReconnect?.call(); _cancelReconnect = null; + _reconnectPending = false; } void _disposeActiveSessionInBackground() { @@ -292,4 +323,53 @@ class TerminalSessionCoordinator extends ChangeNotifier { static String _formatInputForDiagnostics(String input) { return input.replaceAll('\r', r'\r').replaceAll('\n', r'\n'); } + + bool _shouldBufferInput(String input) { + if (_isDisposed || input.isEmpty) { + return false; + } + + return _connectionAttemptInProgress || _reconnectPending; + } + + void _bufferInput(String input) { + if (input.isEmpty) { + return; + } + + _pendingInputs.add(input); + _pendingInputCharacterCount += input.length; + while (_pendingInputCharacterCount > pendingInputCharacterLimit && + _pendingInputs.isNotEmpty) { + final removed = _pendingInputs.removeAt(0); + _pendingInputCharacterCount -= removed.length; + } + } + + void _flushPendingInputs(TerminalSocketSession socketSession) { + if (_pendingInputs.isEmpty) { + return; + } + + final pendingInputs = List.of(_pendingInputs); + _pendingInputs.clear(); + _pendingInputCharacterCount = 0; + + for (var index = 0; index < pendingInputs.length; index += 1) { + final input = pendingInputs[index]; + final result = socketSession.sendInput(input); + if (result == TerminalSocketDispatchResult.sent) { + diagnosticLog?.add('socket.input.flush', _formatInputForDiagnostics(input)); + continue; + } + + if (result == TerminalSocketDispatchResult.noTransport) { + for (final remainingInput in pendingInputs.skip(index)) { + _bufferInput(remainingInput); + } + } + + break; + } + } } diff --git a/apps/mobile_app/lib/features/terminal/terminal_socket_session.dart b/apps/mobile_app/lib/features/terminal/terminal_socket_session.dart index b98e90c..dc48d61 100644 --- a/apps/mobile_app/lib/features/terminal/terminal_socket_session.dart +++ b/apps/mobile_app/lib/features/terminal/terminal_socket_session.dart @@ -85,11 +85,20 @@ class TerminalSocketSession { onFrame(message); }, onError: (error, stackTrace) { + final wasAttached = _isAttached; + _handleTransportClosed(transport); if (!attachedCompleter.isCompleted) { attachedCompleter.completeError(error, stackTrace); + return; + } + + if (wasAttached) { + onDisconnected?.call(); } }, onDone: () { + final wasAttached = _isAttached; + _handleTransportClosed(transport); if (!attachedCompleter.isCompleted) { attachedCompleter.completeError( StateError('Terminal socket closed before attach acknowledgement.'), @@ -97,7 +106,9 @@ class TerminalSocketSession { return; } - onDisconnected?.call(); + if (wasAttached) { + onDisconnected?.call(); + } }, ); @@ -121,8 +132,13 @@ class TerminalSocketSession { return TerminalSocketDispatchResult.noTransport; } - transport.send(jsonEncode(socketClient.buildInputMessage(input))); - return TerminalSocketDispatchResult.sent; + try { + transport.send(jsonEncode(socketClient.buildInputMessage(input))); + return TerminalSocketDispatchResult.sent; + } catch (_) { + _handleTransportClosed(transport); + return TerminalSocketDispatchResult.noTransport; + } } void sendResize(int columns, int rows) { @@ -131,15 +147,17 @@ class TerminalSocketSession { return; } - transport.send(jsonEncode(socketClient.buildResizeMessage(columns, rows))); + try { + transport.send(jsonEncode(socketClient.buildResizeMessage(columns, rows))); + } catch (_) { + _handleTransportClosed(transport); + } } Future dispose() async { final subscription = _subscription; final transport = _transport; - _subscription = null; - _transport = null; - _isAttached = false; + _handleTransportClosed(transport); await subscription?.cancel(); try { await transport?.close(); @@ -156,6 +174,21 @@ class TerminalSocketSession { return false; } + + void _handleTransportClosed(TerminalSocketTransport? transport) { + if (transport == null) { + _subscription = null; + _transport = null; + _isAttached = false; + return; + } + + if (identical(_transport, transport)) { + _subscription = null; + _transport = null; + _isAttached = false; + } + } } enum TerminalSocketDispatchResult { sent, noTransport, emptyInput } diff --git a/apps/mobile_app/test/core/network/agent_api_client_test.dart b/apps/mobile_app/test/core/network/agent_api_client_test.dart index 26f9dde..fba5440 100644 --- a/apps/mobile_app/test/core/network/agent_api_client_test.dart +++ b/apps/mobile_app/test/core/network/agent_api_client_test.dart @@ -142,6 +142,18 @@ void main() { expect(history['sessionId'], 'abc'); expect(history['hasMoreAbove'], isTrue); }); + + test('applies explicit default request timeouts', () async { + final adapter = _FakeHttpClientAdapter(jsonEncode(const [])); + final dio = Dio()..httpClientAdapter = adapter; + final client = AgentApiClient(Uri.parse('https://host:9443'), dio: dio); + + await client.listSessions(); + + expect(adapter.lastOptions?.connectTimeout, const Duration(seconds: 8)); + expect(adapter.lastOptions?.sendTimeout, const Duration(seconds: 10)); + expect(adapter.lastOptions?.receiveTimeout, const Duration(seconds: 20)); + }); } class _FakeHttpClientAdapter implements HttpClientAdapter { diff --git a/apps/mobile_app/test/core/network/agent_error_formatter_test.dart b/apps/mobile_app/test/core/network/agent_error_formatter_test.dart new file mode 100644 index 0000000..f9481f2 --- /dev/null +++ b/apps/mobile_app/test/core/network/agent_error_formatter_test.dart @@ -0,0 +1,48 @@ +import 'dart:io'; + +import 'package:dio/dio.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:term_remote_ctl/core/network/agent_error_formatter.dart'; + +void main() { + test( + 'formats connection timeout with zero duration as a reachability hint', + () { + final error = DioException.connectionTimeout( + timeout: Duration.zero, + requestOptions: RequestOptions( + path: '/api/projects/project-1', + baseUrl: 'http://100.81.30.82:5067', + ), + error: const SocketException( + 'Operation timed out', + osError: OSError('Operation timed out', 60), + port: 5067, + ), + ); + + final message = formatAgentError( + error, + fallback: 'Failed to load project.', + ); + + expect(message, contains('Cannot reach the agent')); + expect(message, contains('http://100.81.30.82:5067')); + expect(message, isNot(contains('0:00:00'))); + }, + ); + + test('formats certificate failures into trust guidance', () { + final error = DioException.badCertificate( + requestOptions: RequestOptions( + path: '/api/projects', + baseUrl: 'https://host.example:9443', + ), + ); + + final message = formatAgentError(error, fallback: 'Failed to load project.'); + + expect(message, contains('certificate')); + expect(message, contains('https://host.example:9443')); + }); +} diff --git a/apps/mobile_app/test/features/terminal/terminal_session_coordinator_test.dart b/apps/mobile_app/test/features/terminal/terminal_session_coordinator_test.dart index 676c710..388b4fa 100644 --- a/apps/mobile_app/test/features/terminal/terminal_session_coordinator_test.dart +++ b/apps/mobile_app/test/features/terminal/terminal_session_coordinator_test.dart @@ -83,6 +83,43 @@ void main() { }, ); + test( + 'input typed while reconnect is pending is flushed after the next attach', + () async { + final controller = TerminalInteractionController(); + final apiClient = _FakeAgentApiClient(); + final sessionFactory = _FakeTerminalSessionFactory(); + final reconnectScheduler = _FakeReconnectScheduler(); + final session = Session( + sessionId: 'abc', + name: 'codex-main', + status: 'idle', + ); + final coordinator = TerminalSessionCoordinator( + controller: controller, + apiClient: apiClient, + session: session, + sessionFactory: sessionFactory.create, + onFrame: (_) {}, + viewportProvider: () => const TerminalViewport(columns: 80, rows: 24), + reconnectScheduler: reconnectScheduler.schedule, + ); + + await coordinator.start(); + final firstSession = sessionFactory.createdSessions.single; + + firstSession.disconnect(); + coordinator.sendInput('dir\r'); + + expect(firstSession.sentInputs, isEmpty); + + await reconnectScheduler.runPending(); + + expect(sessionFactory.createdSessions, hasLength(2)); + expect(sessionFactory.createdSessions.last.sentInputs, ['dir\r']); + }, + ); + test( 'loadOlderHistory increases the requested history window size', () async { @@ -251,10 +288,12 @@ class _FakeTerminalSocketSession extends TerminalSocketSession { final bool autoConnect; final resizeCalls = >[]; + final sentInputs = []; int disposeCount = 0; Completer _connectCompleter = Completer(); void Function(String frame)? _onFrame; void Function()? _onDisconnected; + bool _isDisconnected = false; @override Future connect({ @@ -274,6 +313,16 @@ class _FakeTerminalSocketSession extends TerminalSocketSession { resizeCalls.add([columns, rows]); } + @override + TerminalSocketDispatchResult sendInput(String input) { + if (_isDisconnected || !_connectCompleter.isCompleted) { + return TerminalSocketDispatchResult.noTransport; + } + + sentInputs.add(input); + return TerminalSocketDispatchResult.sent; + } + void completeConnect() { if (!_connectCompleter.isCompleted) { _connectCompleter.complete(); @@ -285,6 +334,7 @@ class _FakeTerminalSocketSession extends TerminalSocketSession { } void disconnect() { + _isDisconnected = true; _onDisconnected?.call(); } diff --git a/apps/mobile_app/test/features/terminal/terminal_socket_session_test.dart b/apps/mobile_app/test/features/terminal/terminal_socket_session_test.dart index 247fa12..8dd6e27 100644 --- a/apps/mobile_app/test/features/terminal/terminal_socket_session_test.dart +++ b/apps/mobile_app/test/features/terminal/terminal_socket_session_test.dart @@ -123,22 +123,49 @@ void main() { expect(disconnectCount, 1); }); + + test('sendInput reports noTransport after the socket closes', () async { + final transport = _FakeTerminalSocketTransport(); + final session = TerminalSocketSession( + sessionId: 'session-123', + socketClient: AgentSocketClient(Uri.parse('https://host:9443')), + transportFactory: (_) => transport, + ); + + final connectFuture = session.connect(onFrame: (_) {}); + await Future.delayed(Duration.zero); + transport.emit('{"type":"attached","sessionId":"session-123"}'); + await connectFuture; + + await transport.close(); + await Future.delayed(Duration.zero); + + expect( + session.sendInput('dir\r'), + TerminalSocketDispatchResult.noTransport, + ); + }); } class _FakeTerminalSocketTransport implements TerminalSocketTransport { final _incoming = StreamController.broadcast(); final sentMessages = []; + var _isClosed = false; @override Stream get stream => _incoming.stream; @override void send(String message) { + if (_isClosed) { + throw StateError('transport is closed'); + } sentMessages.add(message); } @override Future close() async { + _isClosed = true; await _incoming.close(); } diff --git a/apps/mobile_app/test/widget_test.dart b/apps/mobile_app/test/widget_test.dart index 388d3f8..e03466b 100644 --- a/apps/mobile_app/test/widget_test.dart +++ b/apps/mobile_app/test/widget_test.dart @@ -1,3 +1,5 @@ +import 'dart:io'; + import 'dart:async'; import 'package:dio/dio.dart'; @@ -971,6 +973,39 @@ void main() { }, ); + testWidgets( + 'project detail formats agent connection errors instead of dumping Dio text', + (tester) async { + final project = Project( + projectId: 'project-1', + name: 'codex-main', + workingDirectory: r'C:\repo\codex-main', + createdAtUtc: DateTime.parse('2026-03-30T10:00:00Z'), + updatedAtUtc: DateTime.parse('2026-03-30T10:00:00Z'), + ); + + await tester.pumpWidget( + ProviderScope( + overrides: [ + agentApiClientProvider.overrideWithValue(_FakeAgentApiClient()), + projectRepositoryProvider.overrideWithValue( + _FailingProjectRepository(), + ), + sessionRepositoryProvider.overrideWithValue( + _FakeSessionRepository(), + ), + ], + child: MaterialApp(home: ProjectDetailPage(project: project)), + ), + ); + await tester.pumpAndSettle(); + + expect(find.textContaining('Cannot reach the agent'), findsOneWidget); + expect(find.textContaining('0:00:00'), findsNothing); + expect(find.textContaining('DioException'), findsNothing); + }, + ); + testWidgets('session list deletes a session after confirmation', ( tester, ) async { @@ -1251,6 +1286,26 @@ class _FailingSessionRepository extends SessionRepository { } } +class _FailingProjectRepository extends ProjectRepository { + _FailingProjectRepository() : super(_FakeAgentApiClient()); + + @override + Future getProjectDetail(String projectId) async { + throw DioException.connectionTimeout( + timeout: Duration.zero, + requestOptions: RequestOptions( + path: '/api/projects/$projectId', + baseUrl: 'http://100.81.30.82:5067', + ), + error: SocketException( + 'Operation timed out', + osError: OSError('Operation timed out', 60), + port: 5067, + ), + ); + } +} + class _FakeAgentApiClient extends AgentApiClient { _FakeAgentApiClient() : super(Uri.parse('http://100.81.30.82:5067')); diff --git a/apps/windows_agent/src/TermRemoteCtl.Agent/Realtime/TerminalWebSocketHandler.cs b/apps/windows_agent/src/TermRemoteCtl.Agent/Realtime/TerminalWebSocketHandler.cs index 323975e..d2294ec 100644 --- a/apps/windows_agent/src/TermRemoteCtl.Agent/Realtime/TerminalWebSocketHandler.cs +++ b/apps/windows_agent/src/TermRemoteCtl.Agent/Realtime/TerminalWebSocketHandler.cs @@ -73,12 +73,11 @@ public static class TerminalWebSocketHandler } } - host.OutputReceived += HandleOutput; - try { - await SendJsonAsync(socket, new TerminalAttachResponse(sessionId), sendGate, context.RequestAborted).ConfigureAwait(false); var replay = registry.GetReplaySnapshot(sessionId); + host.OutputReceived += HandleOutput; + await SendJsonAsync(socket, new TerminalAttachResponse(sessionId), sendGate, context.RequestAborted).ConfigureAwait(false); if (!string.IsNullOrEmpty(replay)) { await SendTextAsync(socket, replay, sendGate, context.RequestAborted).ConfigureAwait(false); diff --git a/apps/windows_agent/tests/TermRemoteCtl.Agent.IntegrationTests/Realtime/TerminalWebSocketHandlerTests.cs b/apps/windows_agent/tests/TermRemoteCtl.Agent.IntegrationTests/Realtime/TerminalWebSocketHandlerTests.cs index a8e6ee9..ddda5ba 100644 --- a/apps/windows_agent/tests/TermRemoteCtl.Agent.IntegrationTests/Realtime/TerminalWebSocketHandlerTests.cs +++ b/apps/windows_agent/tests/TermRemoteCtl.Agent.IntegrationTests/Realtime/TerminalWebSocketHandlerTests.cs @@ -73,6 +73,37 @@ public sealed class TerminalWebSocketHandlerTests Assert.Equal("prompt> dir\r\nnext> ", replayFrame); } + [Fact] + public async Task Attach_Does_Not_Duplicate_Output_Produced_During_Replay_Boundary() + { + await using var fixture = new TerminalApiFixture(); + var registry = fixture.Services.GetRequiredService(); + fixture.TerminalHost.Registry = registry; + var session = registry.Create("Shell", DateTimeOffset.UtcNow); + await registry.AppendOutputAsync(session.SessionId, "prompt> dir\r\n", CancellationToken.None); + fixture.TerminalHost.EmitOutputOnNextSubscription(session.SessionId, "next> "); + + using WebSocket socket = await fixture.Server.CreateWebSocketClient().ConnectAsync( + new Uri($"ws://localhost/ws/terminal?sessionId={session.SessionId}"), + CancellationToken.None); + + var attachedFrame = await ReceiveTextAsync(socket, CancellationToken.None); + var attachedPayload = JsonSerializer.Deserialize( + attachedFrame, + new JsonSerializerOptions(JsonSerializerDefaults.Web)); + + Assert.NotNull(attachedPayload); + Assert.Equal("attached", attachedPayload!.Type); + + var replayFrame = await ReceiveTextAsync(socket, CancellationToken.None); + Assert.Equal("prompt> dir\r\n", replayFrame); + + var liveFrame = await ReceiveTextAsync(socket, CancellationToken.None); + Assert.Equal("next> ", liveFrame); + + await AssertNoAdditionalTextFrameAsync(socket, TimeSpan.FromMilliseconds(200)); + } + private static async Task ReceiveTextAsync(WebSocket socket, CancellationToken cancellationToken) { var buffer = new byte[4096]; @@ -110,12 +141,24 @@ public sealed class TerminalWebSocketHandlerTests throw new TimeoutException("Condition was not met."); } + private static async Task AssertNoAdditionalTextFrameAsync(WebSocket socket, TimeSpan timeout) + { + using var cts = new CancellationTokenSource(timeout); + await Assert.ThrowsAnyAsync( + async () => await ReceiveTextAsync(socket, cts.Token)); + } + private sealed class TerminalApiFixture : WebApplicationFactory { private readonly string _dataRoot = Path.Combine(Path.GetTempPath(), "TermRemoteCtl.Tests", Guid.NewGuid().ToString("N")); - private readonly TestTerminalSessionHost _terminalHost = new(); + private readonly TestTerminalSessionHost _terminalHost; private readonly RecordingTerminalDiagnosticsSink _diagnostics = new(); + public TerminalApiFixture() + { + _terminalHost = new TestTerminalSessionHost(); + } + public TestTerminalSessionHost TerminalHost => _terminalHost; public RecordingTerminalDiagnosticsSink Diagnostics => _diagnostics; @@ -137,7 +180,8 @@ public sealed class TerminalWebSocketHandlerTests builder.ConfigureServices(services => { services.RemoveAll(); - services.AddSingleton(_terminalHost); + services.AddSingleton(_terminalHost); + services.AddSingleton(sp => sp.GetRequiredService()); services.AddSingleton(_diagnostics); }); } @@ -167,8 +211,38 @@ public sealed class TerminalWebSocketHandlerTests private sealed class TestTerminalSessionHost : ISessionHost { private readonly List<(string Kind, string SessionId, string Value)> _inputs = new(); + private EventHandler? _outputReceived; + private readonly object _gate = new(); + private (string SessionId, string Chunk)? _pendingSubscriptionEmission; - public event EventHandler? OutputReceived; + public event EventHandler? OutputReceived + { + add + { + (string SessionId, string Chunk)? emission; + + lock (_gate) + { + _outputReceived += value; + emission = _pendingSubscriptionEmission; + _pendingSubscriptionEmission = null; + } + + if (emission is { } pending) + { + EmitOutput(pending.SessionId, pending.Chunk); + } + } + remove + { + lock (_gate) + { + _outputReceived -= value; + } + } + } + + public SessionRegistry? Registry { get; set; } public IReadOnlyList<(string Kind, string SessionId, string Value)> Inputs => _inputs; @@ -194,9 +268,18 @@ public sealed class TerminalWebSocketHandlerTests return Task.CompletedTask; } + public void EmitOutputOnNextSubscription(string sessionId, string chunk) + { + lock (_gate) + { + _pendingSubscriptionEmission = (sessionId, chunk); + } + } + public void EmitOutput(string sessionId, string chunk) { - OutputReceived?.Invoke(this, new TerminalOutputEventArgs(sessionId, chunk)); + Registry?.AppendOutputAsync(sessionId, chunk, CancellationToken.None).GetAwaiter().GetResult(); + _outputReceived?.Invoke(this, new TerminalOutputEventArgs(sessionId, chunk)); } }