diff --git a/lib/src/mcp/mcp_server.dart b/lib/src/mcp/mcp_server.dart index 7d698ac5c..31b8291cf 100644 --- a/lib/src/mcp/mcp_server.dart +++ b/lib/src/mcp/mcp_server.dart @@ -484,7 +484,11 @@ Only one value can be selected. Future _handleCreate(CallToolRequest request) async { final args = request.arguments ?? {}; final cliArgs = _parseCreate(args); - return _runToolCommand(cliArgs, toolName: 'create'); + return _runToolCommand( + cliArgs, + toolName: 'create', + progressToken: request.meta?.progressToken, + ); } Future _handleTest(CallToolRequest request) async { @@ -494,6 +498,7 @@ Only one value can be selected. cliArgs, toolName: 'test', directory: args['directory'] as String?, + progressToken: request.meta?.progressToken, ); } @@ -504,6 +509,7 @@ Only one value can be selected. cliArgs, toolName: 'packages get', directory: args['directory'] as String?, + progressToken: request.meta?.progressToken, ); } @@ -532,6 +538,7 @@ Only one value can be selected. cliArgs, toolName: 'packages check licenses', directory: args['directory'] as String?, + progressToken: request.meta?.progressToken, ); } @@ -552,17 +559,49 @@ Only one value can be selected. /// The [Logger] is constructed *inside* the zone on purpose: mason captures /// `IOOverrides.current` at [Logger] construction time, so building it /// outside the zone would defeat the redirect. + /// + /// When [progressToken] is provided, `notifications/progress` messages are + /// emitted as the command runs. Each settled line captured from the + /// command's `stdout`/`stderr` becomes one progress notification, with a + /// monotonically-increasing `progress` counter. Progress is a + /// fire-and-forget UX signal on top of the final tool result (which stays + /// the source of truth). If the client did not supply a token, the server + /// stays silent, as required by the MCP spec. Future _runToolCommand( List args, { required String toolName, String? directory, + ProgressToken? progressToken, }) { return _lock.run(() async { final commandString = 'very_good ${args.join(' ')}'; final output = StringBuffer(); + // Monotonically-increasing counter for [ProgressNotification.progress]. + // The spec requires progress to strictly increase across notifications + // for the same token, even when the total is unknown. + var progressCounter = 0; + void reportProgress(String message) { + if (progressToken == null) return; + progressCounter++; + notifyProgress( + ProgressNotification( + progressToken: progressToken, + progress: progressCounter, + message: message, + ), + ); + } + + if (progressToken != null) { + reportProgress('Starting "$toolName"...'); + } + Future runCaptured(Future Function(Logger logger) body) { - final sink = CapturingStdout(output); + final sink = CapturingStdout( + output, + onLine: progressToken == null ? null : reportProgress, + ); return IOOverrides.runZoned( () => body(Logger()), stdout: () => sink, @@ -640,38 +679,61 @@ Only one value can be selected. /// through `stdout`/`stderr`, including progress spinners) away from the real /// stdout shared with the MCP JSON-RPC stream. It reports no terminal so mason /// emits plain, animation-free lines. +/// +/// When `onLine` is provided, it is invoked with the sanitized, non-empty text +/// of each settled line. Boundary detection is delegated to [LineSplitter], +/// which treats a bare `\r` as a boundary too (so spinner redraws, which +/// rewrite one line in place using `\r`, still surface each intermediate +/// state) and buffers partial lines across separate writes. @visibleForTesting class CapturingStdout implements Stdout { /// Creates a [CapturingStdout] that appends all writes to [_buffer]. - CapturingStdout(this._buffer); + /// + /// If `onLine` is non-null, it is called with each settled line's sanitized, + /// non-empty text (see [sanitizeCommandOutput]). + CapturingStdout(this._buffer, {void Function(String line)? onLine}) + : _lineSink = onLine == null + ? null + : const LineSplitter().startChunkedConversion(_LineSink(onLine)); final StringBuffer _buffer; + /// Feeds writes through [LineSplitter] to detect settled lines. `null` when + /// no `onLine` callback was supplied, so writes skip boundary detection. + final Sink? _lineSink; + @override Encoding encoding = utf8; @override String lineTerminator = '\n'; + /// Appends [text] to the capture buffer and, when wired, feeds it through + /// the line-boundary detector. + void _append(String text) { + _buffer.write(text); + _lineSink?.add(text); + } + @override - void write(Object? object) => _buffer.write(object ?? 'null'); + void write(Object? object) => _append(object?.toString() ?? 'null'); @override - void writeln([Object? object = '']) => _buffer.writeln(object ?? ''); + void writeln([Object? object = '']) => _append('${object ?? ''}\n'); @override void writeAll(Iterable objects, [String separator = '']) => - _buffer.writeAll(objects, separator); + _append(objects.map((o) => o.toString()).join(separator)); @override - void writeCharCode(int charCode) => _buffer.writeCharCode(charCode); + void writeCharCode(int charCode) => _append(String.fromCharCode(charCode)); @override void add(List data) { try { - _buffer.write(encoding.decode(data)); + _append(encoding.decode(data)); } on FormatException { - _buffer.write(String.fromCharCodes(data)); + _append(String.fromCharCodes(data)); } } @@ -685,7 +747,7 @@ class CapturingStdout implements Stdout { Future flush() async {} @override - Future close() async {} + Future close() async => _lineSink?.close(); @override Future get done => Future.value(); @@ -707,6 +769,24 @@ class CapturingStdout implements Stdout { IOSink get nonBlocking => this; } +/// Forwards each line [LineSplitter] settles to [onLine], dropping +/// blank/whitespace-only lines and stripping ANSI escapes via +/// [sanitizeCommandOutput]. +class _LineSink implements Sink { + _LineSink(this.onLine); + + final void Function(String line) onLine; + + @override + void add(String data) { + final settled = sanitizeCommandOutput(data).trim(); + if (settled.isNotEmpty) onLine(settled); + } + + @override + void close() {} +} + /// Matches a CSI ANSI escape sequence (colors, cursor moves, line erases). final _ansiEscape = RegExp(r'\x1B\[[0-?]*[ -/]*[@-~]'); diff --git a/test/src/mcp/mcp_server_test.dart b/test/src/mcp/mcp_server_test.dart index 6c36ce9a3..6ac1e231c 100644 --- a/test/src/mcp/mcp_server_test.dart +++ b/test/src/mcp/mcp_server_test.dart @@ -833,6 +833,178 @@ void main() { }); }); + group('progress notifications', () { + /// Collects every `notifications/progress` frame the server writes to the + /// channel for the lifetime of the returned canceller. + ({List> notifications, Future Function() stop}) + captureProgressNotifications() { + final notifications = >[]; + final subscription = serverResponses.listen((event) { + if (event['method'] == 'notifications/progress') { + notifications.add( + (event['params'] as Map).cast(), + ); + } + }); + return ( + notifications: notifications, + stop: subscription.cancel, + ); + } + + test('are not sent when the client omits a progressToken', () async { + when(() => mockCommandRunner.run(any())).thenAnswer((_) async { + stdout.writeln('some progress line'); + return ExitCode.success.code; + }); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest( + CallToolRequest.methodName, + _params(CallToolRequest(name: 'test', arguments: const {})), + ); + + expect(captured.notifications, isEmpty); + }); + + test( + 'are emitted with monotonic progress when a token is supplied', + () async { + when(() => mockCommandRunner.run(any())).thenAnswer((_) async { + stdout + ..writeln('Optimizing tests...') + ..writeln('00:01 +1: some_test') + ..writeln('All tests passed!'); + return ExitCode.success.code; + }); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest(CallToolRequest.methodName, { + 'name': 'test', + 'arguments': {}, + '_meta': {'progressToken': 'tkn-1'}, + }); + + expect(captured.notifications, isNotEmpty); + for (final n in captured.notifications) { + expect(n['progressToken'], equals('tkn-1')); + } + + final values = captured.notifications + .map((n) => (n['progress'] as num).toDouble()) + .toList(); + for (var i = 1; i < values.length; i++) { + expect( + values[i], + greaterThan(values[i - 1]), + reason: 'progress must strictly increase per MCP spec', + ); + } + + final messages = captured.notifications + .map((n) => n['message'] as String?) + .toList(); + expect(messages.first, contains('Starting "test"')); + expect(messages, contains('Optimizing tests...')); + expect(messages, contains('All tests passed!')); + }, + ); + + test('accepts integer progress tokens', () async { + when(() => mockCommandRunner.run(any())).thenAnswer((_) async { + stdout.writeln('a line'); + return ExitCode.success.code; + }); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest(CallToolRequest.methodName, { + 'name': 'packages_get', + 'arguments': {}, + '_meta': {'progressToken': 42}, + }); + + expect(captured.notifications, isNotEmpty); + for (final n in captured.notifications) { + expect(n['progressToken'], equals(42)); + } + }); + + test( + r'flushes each spinner redraw (\r boundary) as its own notification', + () async { + when(() => mockCommandRunner.run(any())).thenAnswer((_) async { + // A single line rewritten in place three times before settling. + stdout.write('00:01 +1\r00:02 +5\r00:03 +9\rAll passed\n'); + return ExitCode.success.code; + }); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest(CallToolRequest.methodName, { + 'name': 'test', + 'arguments': {}, + '_meta': {'progressToken': 'redraw'}, + }); + + final messages = captured.notifications + .map((n) => n['message'] as String) + .toList(); + expect(messages, contains('00:01 +1')); + expect(messages, contains('00:02 +5')); + expect(messages, contains('00:03 +9')); + expect(messages, contains('All passed')); + }, + ); + + test('are also emitted for packages_check_licenses', () async { + final licenseDir = Directory.systemTemp.createTempSync('vgmcp_lic_'); + addTearDown(() => licenseDir.deleteSync(recursive: true)); + + when(() => mockCommandRunner.run(any())).thenAnswer((_) async { + stdout.writeln('checking licenses for package_a'); + return ExitCode.success.code; + }); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest(CallToolRequest.methodName, { + 'name': 'packages_check_licenses', + 'arguments': {'directory': licenseDir.path}, + '_meta': {'progressToken': 'lic'}, + }); + + expect(captured.notifications, isNotEmpty); + expect( + captured.notifications.map((n) => n['message'] as String).toList(), + contains('checking licenses for package_a'), + ); + }); + + test('skips a run whose captured output is blank', () async { + when( + () => mockCommandRunner.run(any()), + ).thenAnswer((_) async => ExitCode.success.code); + final captured = captureProgressNotifications(); + addTearDown(captured.stop); + + await sendRequest(CallToolRequest.methodName, { + 'name': 'test', + 'arguments': {}, + '_meta': {'progressToken': 'silent'}, + }); + + // Only the initial "Starting …" notification, no per-line ones. + expect(captured.notifications, hasLength(1)); + expect( + captured.notifications.single['message'], + contains('Starting "test"'), + ); + }); + }); + group('working directory', () { test( 'serializes overlapping runs so each keeps its own directory', @@ -965,6 +1137,66 @@ void main() { expect(buffer.toString(), equals('xy')); }); + group('onLine callback', () { + test(r'is invoked once per \n-terminated line with sanitized text', () { + final lines = []; + final sink = CapturingStdout(StringBuffer(), onLine: lines.add) + ..write('hello\nworld\n'); + + expect(lines, equals(['hello', 'world'])); + // The next boundary flushes whatever has been written since. + sink.writeln('again'); + expect(lines, equals(['hello', 'world', 'again'])); + }); + + test(r'treats each \r as a settled-line boundary', () { + final lines = []; + CapturingStdout( + StringBuffer(), + onLine: lines.add, + ).write('tick 1\rtick 2\rdone\n'); + expect(lines, equals(['tick 1', 'tick 2', 'done'])); + }); + + test('strips ANSI escapes from the line delivered to the callback', () { + final lines = []; + CapturingStdout( + StringBuffer(), + onLine: lines.add, + ).writeln('\x1B[31mred\x1B[0m alert'); + expect(lines, equals(['red alert'])); + }); + + test('does not emit for blank/whitespace-only chunks', () { + final lines = []; + CapturingStdout(StringBuffer(), onLine: lines.add) + ..writeln() + ..writeln(' ') + ..writeln('real'); + expect(lines, equals(['real'])); + }); + + test('buffers across writes until the next boundary', () { + final lines = []; + final sink = CapturingStdout(StringBuffer(), onLine: lines.add) + ..write('hel') + ..write('lo ') + ..write('world'); + expect(lines, isEmpty); + sink.write('!\n'); + expect(lines, equals(['hello world!'])); + }); + + test('flushes the final unterminated line on close', () async { + final lines = []; + final sink = CapturingStdout(StringBuffer(), onLine: lines.add) + ..write('done'); + expect(lines, isEmpty); + await sink.close(); + expect(lines, equals(['done'])); + }); + }); + test('reports no terminal and tolerates sink lifecycle calls', () async { expect(capturing.hasTerminal, isFalse); expect(capturing.supportsAnsiEscapes, isFalse);