[dds] Support running scripts with DAP in debug mode

This starts the app paused, connects t the VM service and resumes. It handles dart:developer log() events, but no other debugging functionality yet (for ex. breakpoints, stepping).

Change-Id: Ib50680c775da5d13df95771eec62e77a4af75a08
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/201566
Reviewed-by: Ben Konyi <bkonyi@google.com>
Reviewed-by: Brian Wilkerson <brianwilkerson@google.com>
Commit-Queue: Ben Konyi <bkonyi@google.com>

pkg/dds/lib/src/dap/adapters/dart.dart

@@ -3,8 +3,13 @@
 // BSD-style license that can be found in the LICENSE file.
 
 import 'dart:async';
+import 'dart:io';
+
+import 'package:collection/collection.dart';
+import 'package:vm_service/vm_service.dart' as vm;
 
 import '../base_debug_adapter.dart';
+import '../isolate_manager.dart';
 import '../logging.dart';
 import '../protocol_generated.dart';
 import '../protocol_stream.dart';
@@ -47,12 +52,35 @@ import '../protocol_stream.dart';
 /// to then send a `stackTraceRequest` or `scopesRequest` to get variables).
 abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
     extends BaseDebugAdapter<T> {
-  late T args;
+  late final T args;
   final _debuggerInitializedCompleter = Completer<void>();
   final _configurationDoneCompleter = Completer<void>();
 
+  /// Managers VM Isolates and their events, including fanning out any requests
+  /// to set breakpoints etc. from the client to all Isolates.
+  late IsolateManager _isolateManager;
+
+  /// All active VM Service subscriptions.
+  ///
+  /// TODO(dantup): This may be changed to use StreamManager as part of using
+  /// DDS in this process.
+  final _subscriptions = <StreamSubscription<vm.Event>>[];
+
+  /// The VM service of the app being debugged.
+  ///
+  /// `null` if the session is running in noDebug mode of the connection has not
+  /// yet been made.
+  vm.VmServiceInterface? vmService;
+
+  /// Whether the current debug session is an attach request (as opposed to a
+  /// launch request). Not available until after launchRequest or attachRequest
+  /// have been called.
+  late final bool isAttach;
+
   DartDebugAdapter(ByteStreamServerChannel channel, Logger? logger)
-      : super(channel, logger);
+      : super(channel, logger) {
+    _isolateManager = IsolateManager(this);
+  }
 
   /// Completes when the debugger initialization has completed. Used to delay
   /// processing isolate events while initialization is still running to avoid
@@ -60,6 +88,33 @@ abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
   /// processed its initial paused state).
   Future<void> get debuggerInitialized => _debuggerInitializedCompleter.future;
 
+  /// attachRequest is called by the client when it wants us to to attach to
+  /// an existing app. This will only be called once (and only one of this or
+  /// launchRequest will be called).
+  @override
+  FutureOr<void> attachRequest(
+    Request request,
+    T args,
+    void Function(void) sendResponse,
+  ) async {
+    this.args = args;
+    isAttach = true;
+
+    // Don't start launching until configurationDone.
+    if (!_configurationDoneCompleter.isCompleted) {
+      logger?.call('Waiting for configurationDone request...');
+      await _configurationDoneCompleter.future;
+    }
+
+    // TODO(dantup): Implement attach support.
+    throw UnimplementedError();
+
+    // Delegate to the sub-class to attach to the process.
+    // await attachImpl();
+    //
+    // sendResponse(null);
+  }
+
   /// configurationDone is called by the client when it has finished sending
   /// any initial configuration (such as breakpoints and exception pause
   /// settings).
@@ -77,6 +132,97 @@ abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
     sendResponse(null);
   }
 
+  /// Connects to the VM Service at [uri] and initializes debugging.
+  ///
+  /// This method will be called by sub-classes when they are ready to start
+  /// a debug session and may provide a URI given by the user (in the case
+  /// of attach) or from something like a vm-service-info file or Flutter
+  /// app.debugPort message.
+  ///
+  /// The URI protocol will be changed to ws/wss but otherwise not normalised.
+  /// The caller should handle any other normalisation (such as adding /ws to
+  /// the end if required).
+  Future<void> connectDebugger(Uri uri) async {
+    // The VM Service library always expects the WebSockets URI so fix the
+    // scheme (http -> ws, https -> wss).
+    final isSecure = uri.isScheme('https') || uri.isScheme('wss');
+    uri = uri.replace(scheme: isSecure ? 'wss' : 'ws');
+
+    logger?.call('Connecting to debugger at $uri');
+    sendOutput('console', 'Connecting to VM Service at $uri\n');
+    final vmService =
+        await _vmServiceConnectUri(uri.toString(), logger: logger);
+    logger?.call('Connected to debugger at $uri!');
+
+    // TODO(dantup): VS Code currently depends on a custom dart.debuggerUris
+    // event to notify it of VM Services that become available (for example to
+    // register with the DevTools server). If this is still required, it will
+    // need implementing here (and also documented as a customisation and
+    // perhaps gated on a capability/argument).
+    this.vmService = vmService;
+
+    _subscriptions.addAll([
+      vmService.onIsolateEvent.listen(_handleIsolateEvent),
+      vmService.onDebugEvent.listen(_handleDebugEvent),
+      vmService.onLoggingEvent.listen(_handleLoggingEvent),
+      // TODO(dantup): Implement these.
+      // vmService.onExtensionEvent.listen(_handleExtensionEvent),
+      // vmService.onServiceEvent.listen(_handleServiceEvent),
+      // vmService.onStdoutEvent.listen(_handleStdoutEvent),
+      // vmService.onStderrEvent.listen(_handleStderrEvent),
+    ]);
+    await Future.wait([
+      vmService.streamListen(vm.EventStreams.kIsolate),
+      vmService.streamListen(vm.EventStreams.kDebug),
+      vmService.streamListen(vm.EventStreams.kLogging),
+      // vmService.streamListen(vm.EventStreams.kExtension),
+      // vmService.streamListen(vm.EventStreams.kService),
+      // vmService.streamListen(vm.EventStreams.kStdout),
+      // vmService.streamListen(vm.EventStreams.kStderr),
+    ]);
+
+    final vmInfo = await vmService.getVM();
+    logger?.call('Connected to ${vmInfo.name} on ${vmInfo.operatingSystem}');
+
+    // Let the subclass do any existing setup once we have a connection.
+    await debuggerConnected(vmInfo);
+
+    // Process any existing isolates that may have been created before the
+    // streams above were set up.
+    final existingIsolateRefs = vmInfo.isolates;
+    final existingIsolates = existingIsolateRefs != null
+        ? await Future.wait(existingIsolateRefs
+            .map((isolateRef) => isolateRef.id)
+            .whereNotNull()
+            .map(vmService.getIsolate))
+        : <vm.Isolate>[];
+    await Future.wait(existingIsolates.map((isolate) async {
+      // Isolates may have the "None" pauseEvent kind at startup, so infer it
+      // from the runnable field.
+      final pauseEventKind = isolate.runnable ?? false
+          ? vm.EventKind.kIsolateRunnable
+          : vm.EventKind.kIsolateStart;
+      await _isolateManager.registerIsolate(isolate, pauseEventKind);
+
+      // If the Isolate already has a Pause event we can give it to the
+      // IsolateManager to handle (if it's PausePostStart it will re-configure
+      // the isolate before resuming), otherwise we can just resume it (if it's
+      // runnable - otherwise we'll handle this when it becomes runnable in an
+      // event later).
+      if (isolate.pauseEvent?.kind?.startsWith('Pause') ?? false) {
+        await _isolateManager.handleEvent(isolate.pauseEvent!);
+      } else if (isolate.runnable == true) {
+        await _isolateManager.resumeIsolate(isolate);
+      }
+    }));
+
+    _debuggerInitializedCompleter.complete();
+  }
+
+  /// Overridden by sub-classes to perform any additional setup after the VM
+  /// Service is connected.
+  FutureOr<void> debuggerConnected(vm.VM vmInfo);
+
   /// Overridden by sub-classes to handle when the client sends a
   /// `disconnectRequest` (a forceful request to shut down).
   FutureOr<void> disconnectImpl();
@@ -159,6 +305,7 @@ abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
     void Function(void) sendResponse,
   ) async {
     this.args = args;
+    isAttach = false;
 
     // Don't start launching until configurationDone.
     if (!_configurationDoneCompleter.isCompleted) {
@@ -178,6 +325,18 @@ abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
     sendEvent(OutputEventBody(category: category, output: message));
   }
 
+  /// Sends an OutputEvent for [message], prefixed with [prefix] and with [message]
+  /// indented to after the prefix.
+  ///
+  /// Assumes the output is in full lines and will always include a terminating
+  /// newline.
+  void sendPrefixedOutput(String category, String prefix, String message) {
+    final indentString = ' ' * prefix.length;
+    final indentedMessage =
+        message.trimRight().split('\n').join('\n$indentString');
+    sendOutput(category, '$prefix$indentedMessage\n');
+  }
+
   /// Overridden by sub-classes to handle when the client sends a
   /// `terminateRequest` (a request for a graceful shut down).
   FutureOr<void> terminateImpl();
@@ -199,6 +358,84 @@ abstract class DartDebugAdapter<T extends DartLaunchRequestArguments>
     terminateImpl();
     sendResponse(null);
   }
+
+  void _handleDebugEvent(vm.Event event) {
+    _isolateManager.handleEvent(event);
+  }
+
+  void _handleIsolateEvent(vm.Event event) {
+    _isolateManager.handleEvent(event);
+  }
+
+  /// Handles a dart:developer log() event, sending output to the client.
+  Future<void> _handleLoggingEvent(vm.Event event) async {
+    final record = event.logRecord;
+    final thread = _isolateManager.threadForIsolate(event.isolate);
+    if (record == null || thread == null) {
+      return;
+    }
+
+    /// Helper to convert to InstanceRef to a String, taking into account
+    /// [vm.InstanceKind.kNull] which is the type for the unused fields of a
+    /// log event.
+    FutureOr<String?> asString(vm.InstanceRef? ref) {
+      if (ref == null || ref.kind == vm.InstanceKind.kNull) {
+        return null;
+      }
+      // TODO(dantup): This should handle truncation and complex types.
+      return ref.valueAsString;
+    }
+
+    var loggerName = await asString(record.loggerName);
+    if (loggerName?.isEmpty ?? true) {
+      loggerName = 'log';
+    }
+    final message = await asString(record.message);
+    final error = await asString(record.error);
+    final stack = await asString(record.stackTrace);
+
+    final prefix = '[$loggerName] ';
+
+    if (message != null) {
+      sendPrefixedOutput('stdout', prefix, '$message\n');
+    }
+    if (error != null) {
+      sendPrefixedOutput('stderr', prefix, '$error\n');
+    }
+    if (stack != null) {
+      sendPrefixedOutput('stderr', prefix, '$stack\n');
+    }
+  }
+
+  /// A wrapper around the same name function from package:vm_service that
+  /// allows logging all traffic over the VM Service.
+  Future<vm.VmService> _vmServiceConnectUri(
+    String wsUri, {
+    Logger? logger,
+  }) async {
+    final socket = await WebSocket.connect(wsUri);
+    final controller = StreamController();
+    final streamClosedCompleter = Completer();
+
+    socket.listen(
+      (data) {
+        logger?.call('<== [VM] $data');
+        controller.add(data);
+      },
+      onDone: () => streamClosedCompleter.complete(),
+    );
+
+    return vm.VmService(
+      controller.stream,
+      (String message) {
+        logger?.call('==> [VM] $message');
+        socket.add(message);
+      },
+      log: logger != null ? VmServiceLogger(logger) : null,
+      disposeHandler: () => socket.close(),
+      streamClosed: streamClosedCompleter.future,
+    );
+  }
 }
 
 /// An implementation of [LaunchRequestArguments] that includes all fields used

pkg/dds/lib/src/dap/adapters/dart_cli.dart

@@ -6,7 +6,9 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 
+import 'package:path/path.dart' as path;
 import 'package:pedantic/pedantic.dart';
+import 'package:vm_service/vm_service.dart' as vm;
 
 import '../logging.dart';
 import '../protocol_generated.dart';
@@ -17,18 +19,53 @@ import 'dart.dart';
 class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
   Process? _process;
 
+  /// The location of the vm-service-info file (if debugging).
+  ///
+  /// This may be provided by the user (eg. if attaching) or generated by the DA.
+  File? _vmServiceInfoFile;
+
+  /// A watcher for [_vmServiceInfoFile] to detect when the VM writes the service
+  /// info file.
+  ///
+  /// Should be cancelled once the file has been successfully read.
+  StreamSubscription<FileSystemEvent>? _vmServiceInfoFileWatcher;
+
+  /// Process IDs to terminate during shutdown.
+  ///
+  /// This may be populated with pids from the VM Service to ensure we clean up
+  /// properly where signals may not be passed through the shell to the
+  /// underlying VM process.
+  /// https://github.com/Dart-Code/Dart-Code/issues/907
+  final pidsToTerminate = <int>{};
+
   @override
   final parseLaunchArgs = DartLaunchRequestArguments.fromJson;
 
   DartCliDebugAdapter(ByteStreamServerChannel channel, [Logger? logger])
       : super(channel, logger);
 
+  FutureOr<void> debuggerConnected(vm.VM vmInfo) {
+    if (!isAttach) {
+      // Capture the PID from the VM Service so that we can terminate it when
+      // cleaning up. Terminating the process might not be enough as it could be
+      // just a shell script (eg. pub on Windows) and may not pass the
+      // signal on correctly.
+      // See: https://github.com/Dart-Code/Dart-Code/issues/907
+      final pid = vmInfo.pid;
+      if (pid != null) {
+        pidsToTerminate.add(pid);
+      }
+    }
+  }
+
   /// Called by [disconnectRequest] to request that we forcefully shut down the
   /// app being run (or in the case of an attach, disconnect).
   FutureOr<void> disconnectImpl() {
-    // TODO(dantup): In Dart-Code DAP, we first try again with SIGINT and wait
+    // TODO(dantup): In Dart-Code DAP, we first try again with sigint and wait
     // for a few seconds before sending sigkill.
-    _process?.kill(ProcessSignal.sigkill);
+    pidsToTerminate.forEach(
+      (pid) => Process.killPid(pid, ProcessSignal.sigkill),
+    );
   }
 
   /// Called by [launchRequest] to request that we actually start the app to be
@@ -38,8 +75,46 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
   /// breakpoints, and resume.
   Future<void> launchImpl() async {
     final vmPath = Platform.resolvedExecutable;
-    final vmArgs = <String>[]; // TODO(dantup): enable-asserts, debug, etc.
 
+    final debug = !(args.noDebug ?? false);
+    if (debug) {
+      // Create a temp folder for the VM to write the service-info-file into.
+      // Using tmpDir.createTempory() is flakey on Windows+Linux (at least
+      // on GitHub Actions) complaining the file does not exist when creating a
+      // watcher. Creating/watching a folder and writing the file into it seems
+      // to be reliable.
+      final serviceInfoFilePath = path.join(
+        Directory.systemTemp.createTempSync('dart-vm-service').path,
+        'vm.json',
+      );
+      _vmServiceInfoFile = File(serviceInfoFilePath);
+      _vmServiceInfoFileWatcher = _vmServiceInfoFile?.parent
+          .watch(events: FileSystemEvent.all)
+          .where((event) => event.path == _vmServiceInfoFile?.path)
+          .listen(
+            _handleVmServiceInfoEvent,
+            onError: (e) => logger?.call('Ignoring exception from watcher: $e'),
+          );
+    }
+
+    // TODO(dantup): Currently this just spawns the new VM and completely
+    // ignores DDS. Figure out how this will ultimately work - will we just wrap
+    // the call to initDebugger() with something that starts DDS?
+    final vmServiceInfoFile = _vmServiceInfoFile;
+    final vmArgs = <String>[
+      '--no-serve-devtools',
+      if (debug) ...[
+        '--enable-vm-service=${args.vmServicePort ?? 0}',
+        '--pause_isolates_on_start=true',
+      ],
+      if (debug && vmServiceInfoFile != null) ...[
+        '-DSILENT_OBSERVATORY=true',
+        '--write-service-info=${Uri.file(vmServiceInfoFile.path)}'
+      ],
+      // Default to asserts on, this seems like the most useful behaviour for
+      // editor-spawned debug sessions.
+      if (args.enableAsserts ?? true) '--enable-asserts'
+    ];
     final processArgs = [
       ...vmArgs,
       args.program,
@@ -53,6 +128,7 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
       workingDirectory: args.cwd,
     );
     _process = process;
+    pidsToTerminate.add(process.pid);
 
     process.stdout.listen(_handleStdout);
     process.stderr.listen(_handleStderr);
@@ -62,7 +138,9 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
   /// Called by [terminateRequest] to request that we gracefully shut down the
   /// app being run (or in the case of an attach, disconnect).
   FutureOr<void> terminateImpl() async {
-    _process?.kill(ProcessSignal.sigint);
+    pidsToTerminate.forEach(
+      (pid) => Process.killPid(pid, ProcessSignal.sigint),
+    );
     await _process?.exitCode;
   }
 
@@ -82,4 +160,24 @@ class DartCliDebugAdapter extends DartDebugAdapter<DartLaunchRequestArguments> {
   void _handleStdout(List<int> data) {
     sendOutput('stdout', utf8.decode(data));
   }
+
+  /// Handles file watcher events for the vm-service-info file and connects the
+  /// debugger.
+  ///
+  /// The vm-service-info file is written by the VM when we start the app/script
+  /// to debug and contains the VM Service URI. This allows us to access the
+  /// auth token without needing to have the URI printed to/scraped from stdout.
+  void _handleVmServiceInfoEvent(FileSystemEvent event) {
+    try {
+      final content = _vmServiceInfoFile!.readAsStringSync();
+      final json = jsonDecode(content);
+      final uri = Uri.parse(json['uri']);
+      unawaited(connectDebugger(uri));
+      _vmServiceInfoFileWatcher?.cancel();
+    } catch (e) {
+      // It's possible we tried to read the file before it was completely
+      // written so ignore and try again on the next event.
+      logger?.call('Ignoring error parsing vm-service-info file: $e');
+    }
+  }
 }

pkg/dds/lib/src/dap/base_debug_adapter.dart

@@ -34,6 +34,12 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments> {
   /// Dart CLI, Dart tests, Flutter, Flutter tests).
   TLaunchArgs Function(Map<String, Object?>) get parseLaunchArgs;
 
+  FutureOr<void> attachRequest(
+    Request request,
+    TLaunchArgs args,
+    void Function(void) sendResponse,
+  );
+
   FutureOr<void> configurationDoneRequest(
     Request request,
     ConfigurationDoneArguments? args,
@@ -164,6 +170,8 @@ abstract class BaseDebugAdapter<TLaunchArgs extends LaunchRequestArguments> {
       handle(request, initializeRequest, InitializeRequestArguments.fromJson);
     } else if (request.command == 'launch') {
       handle(request, launchRequest, parseLaunchArgs);
+    } else if (request.command == 'attach') {
+      handle(request, attachRequest, parseLaunchArgs);
     } else if (request.command == 'terminate') {
       handle(
         request,

pkg/dds/lib/src/dap/isolate_manager.dart

@@ -0,0 +1,269 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:vm_service/vm_service.dart' as vm;
+
+import 'adapters/dart.dart';
+import 'protocol_generated.dart';
+
+/// Manages state of Isolates (called Threads by the DAP protocol).
+///
+/// Handles incoming Isolate and Debug events to track the lifetime of isolates
+/// and updating breakpoints for each isolate as necessary.
+class IsolateManager {
+  // TODO(dantup): This class has a lot of overlap with the same-named class
+  //  in DDS. Review what can be shared.
+  final DartDebugAdapter _adapter;
+  final Map<String, Completer<void>> _isolateRegistrations = {};
+  final Map<String, ThreadInfo> _threadsByIsolateId = {};
+  final Map<int, ThreadInfo> _threadsByThreadId = {};
+  int _nextThreadNumber = 1;
+
+  IsolateManager(this._adapter);
+
+  /// A list of all current active isolates.
+  ///
+  /// When isolates exit, they will no longer be returned in this list, although
+  /// due to the async nature, it's not guaranteed that threads in this list have
+  /// not exited between accessing this list and trying to use the results.
+  List<ThreadInfo> get threads => _threadsByIsolateId.values.toList();
+
+  Future<T> getObject<T extends vm.Response>(
+      vm.IsolateRef isolate, vm.ObjRef object) async {
+    final res = await _adapter.vmService?.getObject(isolate.id!, object.id!);
+    return res as T;
+  }
+
+  /// Handles Isolate and Debug events
+  FutureOr<void> handleEvent(vm.Event event) async {
+    final isolateId = event.isolate?.id;
+    if (isolateId == null) {
+      return;
+    }
+
+    // Delay processing any events until the debugger initialization has
+    // finished running, as events may arrive (for ex. IsolateRunnable) while
+    // it's doing is own initialization that this may interfere with.
+    await _adapter.debuggerInitialized;
+
+    final eventKind = event.kind;
+    if (eventKind == vm.EventKind.kIsolateStart ||
+        eventKind == vm.EventKind.kIsolateRunnable) {
+      await registerIsolate(event.isolate!, eventKind!);
+    }
+
+    // Additionally, ensure the thread registration has completed before trying
+    // to process any other events. This is to cover the case where we are
+    // processing the above registerIsolate call in the handler for one isolate
+    // event but another one arrives and gets us here before the registration
+    // above (in the other event handler) has finished.
+    await _isolateRegistrations[isolateId]?.future;
+
+    if (eventKind == vm.EventKind.kIsolateExit) {
+      await _handleExit(event);
+    } else if (eventKind?.startsWith('Pause') ?? false) {
+      await _handlePause(event);
+    } else if (eventKind == vm.EventKind.kResume) {
+      await _handleResumed(event);
+    }
+  }
+
+  /// Registers a new isolate that exists at startup, or has subsequently been
+  /// created.
+  ///
+  /// New isolates will be configured with the correct pause-exception behaviour,
+  /// libraries will be marked as debuggable if appropriate, and breakpoints
+  /// sent.
+  FutureOr<void> registerIsolate(
+    vm.IsolateRef isolate,
+    String eventKind,
+  ) async {
+    // Ensure the completer is set up before doing any async work, so future
+    // events can wait on it.
+    final registrationCompleter =
+        _isolateRegistrations.putIfAbsent(isolate.id!, () => Completer<void>());
+
+    final info = _threadsByIsolateId.putIfAbsent(
+      isolate.id!,
+      () {
+        // The first time we see an isolate, start tracking it.
+        final info = ThreadInfo(_nextThreadNumber++, isolate);
+        _threadsByThreadId[info.threadId] = info;
+        // And notify the client about it.
+        _adapter.sendEvent(
+          ThreadEventBody(reason: 'started', threadId: info.threadId),
+        );
+        return info;
+      },
+    );
+
+    // If it's just become runnable (IsolateRunnable), configure the isolate
+    // by sending breakpoints etc.
+    if (eventKind == vm.EventKind.kIsolateRunnable && !info.runnable) {
+      info.runnable = true;
+      await _configureIsolate(isolate);
+      registrationCompleter.complete();
+    }
+  }
+
+  FutureOr<void> resumeIsolate(vm.IsolateRef isolateRef,
+      [String? resumeType]) async {
+    final isolateId = isolateRef.id;
+    if (isolateId == null) {
+      return;
+    }
+
+    final thread = _threadsByIsolateId[isolateId];
+    if (thread == null) {
+      return;
+    }
+
+    await resumeThread(thread.threadId);
+  }
+
+  /// Resumes (or steps) an isolate using its client [threadId].
+  ///
+  /// If the isolate is not paused, or already has a pending resume request
+  /// in-flight, a request will not be sent.
+  ///
+  /// If the isolate is paused at an async suspension and the [resumeType] is
+  /// [vm.StepOption.kOver], a [StepOption.kOverAsyncSuspension] step will be
+  /// sent instead.
+  Future<void> resumeThread(int threadId, [String? resumeType]) async {
+    final thread = _threadsByThreadId[threadId];
+    if (thread == null) {
+      throw 'Thread $threadId was not found';
+    }
+
+    // Check this thread hasn't already been resumed by another handler in the
+    // meantime (for example if the user performs a hot restart or something
+    // while we processing some previous events).
+    if (!thread.paused || thread.hasPendingResume) {
+      return;
+    }
+
+    // We always assume that a step when at an async suspension is intended to
+    // be an async step.
+    if (resumeType == vm.StepOption.kOver && thread.atAsyncSuspension) {
+      resumeType = vm.StepOption.kOverAsyncSuspension;
+    }
+
+    thread.hasPendingResume = true;
+    try {
+      await _adapter.vmService?.resume(thread.isolate.id!, step: resumeType);
+    } finally {
+      thread.hasPendingResume = false;
+    }
+  }
+
+  ThreadInfo? threadForIsolate(vm.IsolateRef? isolate) =>
+      isolate?.id != null ? _threadsByIsolateId[isolate!.id!] : null;
+
+  /// Configures a new isolate, setting it's exception-pause mode, which
+  /// libraries are debuggable, and sending all breakpoints.
+  FutureOr<void> _configureIsolate(vm.IsolateRef isolate) async {
+    // TODO(dantup): set library debuggable, exception pause mode, breakpoints
+  }
+
+  FutureOr<void> _handleExit(vm.Event event) {
+    final isolate = event.isolate!;
+    final isolateId = isolate.id!;
+    final thread = _threadsByIsolateId[isolateId];
+    if (thread != null) {
+      // Notify the client.
+      _adapter.sendEvent(
+        ThreadEventBody(reason: 'exited', threadId: thread.threadId),
+      );
+      _threadsByIsolateId.remove(isolateId);
+      _threadsByThreadId.remove(thread.threadId);
+    }
+  }
+
+  /// Handles a pause event.
+  ///
+  /// For [vm.EventKind.kPausePostRequest] which occurs after a restart, the isolate
+  /// will be re-configured (pause-exception behaviour, debuggable libraries,
+  /// breakpoints) and then resumed.
+  ///
+  /// For [vm.EventKind.kPauseStart], the isolate will be resumed.
+  ///
+  /// For breakpoints with conditions that are not met and for logpoints, the
+  /// isolate will be automatically resumed.
+  ///
+  /// For all other pause types, the isolate will remain paused and a
+  /// corresponding "Stopped" event sent to the editor.
+  FutureOr<void> _handlePause(vm.Event event) async {
+    final eventKind = event.kind;
+    final isolate = event.isolate!;
+    final thread = _threadsByIsolateId[isolate.id!];
+
+    if (thread == null) {
+      return;
+    }
+
+    thread.atAsyncSuspension = event.atAsyncSuspension ?? false;
+    thread.paused = true;
+    thread.pauseEvent = event;
+
+    // For PausePostRequest we need to re-send all breakpoints; this happens
+    // after a hot restart.
+    if (eventKind == vm.EventKind.kPausePostRequest) {
+      _configureIsolate(isolate);
+      await resumeThread(thread.threadId);
+    } else if (eventKind == vm.EventKind.kPauseStart) {
+      await resumeThread(thread.threadId);
+    } else {
+      // PauseExit, PauseBreakpoint, PauseInterrupted, PauseException
+      var reason = 'pause';
+
+      if (eventKind == vm.EventKind.kPauseBreakpoint &&
+          (event.pauseBreakpoints?.isNotEmpty ?? false)) {
+        reason = 'breakpoint';
+      } else if (eventKind == vm.EventKind.kPauseBreakpoint) {
+        reason = 'step';
+      } else if (eventKind == vm.EventKind.kPauseException) {
+        reason = 'exception';
+      }
+
+      // TODO(dantup): Store exception.
+
+      // Notify the client.
+      _adapter.sendEvent(
+        StoppedEventBody(reason: reason, threadId: thread.threadId),
+      );
+    }
+  }
+
+  /// Handles a resume event from the VM, updating our local state.
+  FutureOr<void> _handleResumed(vm.Event event) {
+    final isolate = event.isolate!;
+    final thread = _threadsByIsolateId[isolate.id!];
+    if (thread != null) {
+      thread.paused = false;
+      thread.pauseEvent = null;
+      thread.exceptionReference = null;
+    }
+  }
+}
+
+/// Holds state for a single Isolate/Thread.
+class ThreadInfo {
+  final vm.IsolateRef isolate;
+  final int threadId;
+  var runnable = false;
+  var atAsyncSuspension = false;
+  int? exceptionReference;
+  var paused = false;
+
+  // The most recent pauseEvent for this isolate.
+  vm.Event? pauseEvent;
+
+  /// Whether this isolate has an in-flight resume request that has not yet
+  /// been responded to.
+  var hasPendingResume = false;
+
+  ThreadInfo(this.threadId, this.isolate);
+}

pkg/dds/lib/src/dap/logging.dart

@@ -2,4 +2,19 @@
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
+import 'package:vm_service/vm_service.dart' as vm;
+
 typedef Logger = void Function(String);
+
+/// Wraps a [Logger] as a [vm/Log] to be passed to the VM Service library.
+class VmServiceLogger extends vm.Log {
+  final Logger _logger;
+
+  VmServiceLogger(this._logger);
+
+  @override
+  void severe(String message) => _logger.call('ERROR: $message');
+
+  @override
+  void warning(String message) => _logger.call('WARN: $message');
+}

pkg/dds/pubspec.yaml

@@ -12,6 +12,7 @@ environment:
 
 dependencies:
   async: ^2.4.1
+  collection: ^1.15.0
   devtools_shared: ^2.3.0
   json_rpc_2: ^3.0.0
   meta: ^1.1.8
@@ -29,7 +30,6 @@ dependencies:
 
 dev_dependencies:
   args: ^2.0.0
-  collection: ^1.15.0
   http: ^0.13.0
   test: ^1.0.0
   webdriver: ^3.0.0

pkg/dds/test/dap/integration/debug_logging_test.dart

@@ -0,0 +1,36 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:test/test.dart';
+
+import 'test_support.dart';
+
+main() {
+  testDap((dap) async {
+    group('debug mode', () {
+      test('prints messages from dart:developer log()', () async {
+        final testFile = dap.createTestFile(r'''
+import 'dart:developer';
+
+void main(List<String> args) async {
+  log('this is a test\nacross two lines');
+  log('this is a test', name: 'foo');
+}
+    ''');
+
+        var outputEvents = await dap.client.collectOutput(file: testFile);
+
+        // Skip the first line because it's the VM Service connection info.
+        final output = outputEvents.skip(1).map((e) => e.output).join();
+        expectLines(output, [
+          '[log] this is a test',
+          '      across two lines',
+          '[foo] this is a test',
+          '',
+          'Exited.',
+        ]);
+      });
+    });
+  });
+}

pkg/dds/test/dap/integration/debug_test.dart

@@ -0,0 +1,47 @@
+// Copyright (c) 2021, the Dart project authors.  Please see the AUTHORS file
+// for details. All rights reserved. Use of this source code is governed by a
+// BSD-style license that can be found in the LICENSE file.
+
+import 'package:test/test.dart';
+
+import 'test_support.dart';
+
+main() {
+  testDap((dap) async {
+    group('debug mode', () {
+      test('runs a simple script', () async {
+        final testFile = dap.createTestFile(r'''
+void main(List<String> args) async {
+  print('Hello!');
+  print('World!');
+  print('args: $args');
+}
+    ''');
+
+        var outputEvents = await dap.client.collectOutput(
+          launch: () => dap.client.launch(
+            testFile.path,
+            args: ['one', 'two'],
+          ),
+        );
+
+        // Expect a "console" output event that prints the URI of the VM Service
+        // the debugger connects to.
+        final vmConnection = outputEvents.first;
+        expect(vmConnection.output,
+            startsWith('Connecting to VM Service at ws://127.0.0.1:'));
+        expect(vmConnection.category, equals('console'));
+
+        // Expect the normal applications output.
+        final output = outputEvents.skip(1).map((e) => e.output).join();
+        expectLines(output, [
+          'Hello!',
+          'World!',
+          'args: [one, two]',
+          '',
+          'Exited.',
+        ]);
+      });
+    });
+  });
+}

pkg/dds/test/dap/integration/no_debug_test.dart

@@ -8,7 +8,7 @@ import 'test_support.dart';
 
 main() {
   testDap((dap) async {
-    group('noDebug', () {
+    group('noDebug mode', () {
       test('runs a simple script', () async {
         final testFile = dap.createTestFile(r'''
 void main(List<String> args) async {

pkg/dds/test/dap/integration/test_support.dart

@@ -81,7 +81,7 @@ class DapTestSession {
     // Since we don't get a signal from the DAP server when it's ready and we
     // just started it, add a short retry to connections.
     var attempt = 1;
-    while (attempt++ <= 5) {
+    while (attempt++ <= 20) {
       try {
         return await DapTestClient.connect(server.port);
       } catch (e) {
@@ -89,7 +89,8 @@ class DapTestSession {
       }
     }
 
-    throw 'Failed to connect to DAP server after $attempt attempts';
+    throw 'Failed to connect to DAP server on port ${server.port}'
+        ' after $attempt attempts. Did the server start correctly?';
   }
 
   /// Starts a DAP server that can be shared across tests.