diff --git a/pkg/perf_witness/analysis_options.yaml b/pkg/perf_witness/analysis_options.yaml
new file mode 100644
index 00000000000..0c512d97fde
--- /dev/null
+++ b/pkg/perf_witness/analysis_options.yaml
@@ -0,0 +1,11 @@
+include: package:lints/recommended.yaml
+
+analyzer:
+  exclude:
+    - lib/src/assets/**
+
+linter:
+  rules:
+    - directives_ordering
+    - prefer_final_locals
+    - sort_pub_dependencies
diff --git a/pkg/perf_witness/bin/recorder.dart b/pkg/perf_witness/bin/recorder.dart
new file mode 100644
index 00000000000..1c424970713
--- /dev/null
+++ b/pkg/perf_witness/bin/recorder.dart
@@ -0,0 +1,9 @@
+// Copyright (c) 2025, 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:perf_witness/recorder.dart' as recorder;
+
+Future<void> main(List<String> args) async {
+  await recorder.record(recorder.PerfWitnessRecorderConfig.fromArgs(args));
+}
diff --git a/pkg/perf_witness/example/example.dart b/pkg/perf_witness/example/example.dart
new file mode 100644
index 00000000000..82d4f76a437
--- /dev/null
+++ b/pkg/perf_witness/example/example.dart
@@ -0,0 +1,42 @@
+// Copyright (c) 2025, 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.
+
+// Simple binary which continuously does some busy work and generates
+// timeline events.
+
+import 'dart:async';
+import 'dart:developer';
+
+import 'package:perf_witness/server.dart';
+import 'package:perf_witness/src/async_span.dart';
+
+int fib(int i) {
+  if (i < 2) return 1;
+  return fib(i - 1) + fib(i - 2);
+}
+
+Future<void> task(int id) async {
+  await AsyncSpan.run('task#$id', () async {
+    for (var i = 0; i < 10; i++) {
+      Timeline.timeSync('fib', () {
+        final sw = Stopwatch()..start();
+        while (sw.elapsedMilliseconds < 100) {
+          fib(10);
+        }
+      });
+      await Future.delayed(Duration(milliseconds: 100));
+    }
+  });
+}
+
+void main() async {
+  await PerfWitnessServer.start();
+  var id = 0;
+  while (true) {
+    await AsyncSpan.run('task-group-${id ~/ 2}', () async {
+      await Future.wait([task(id++), task(id++)]);
+      await Future.delayed(Duration(milliseconds: 100));
+    });
+  }
+}
diff --git a/pkg/perf_witness/lib/recorder.dart b/pkg/perf_witness/lib/recorder.dart
new file mode 100644
index 00000000000..bd19e439ecc
--- /dev/null
+++ b/pkg/perf_witness/lib/recorder.dart
@@ -0,0 +1,240 @@
+// Copyright (c) 2025, 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 'dart:developer';
+import 'dart:io' as io;
+
+import 'package:args/args.dart';
+import 'package:path/path.dart' as p;
+
+import 'src/common.dart';
+import 'src/json_rpc.dart';
+import 'src/process_info.dart';
+
+class PerfWitnessRecorderConfig {
+  final String? outputDir;
+  final String? tag;
+  final bool recordNewProcesses;
+  final bool enableAsyncSpans;
+  final bool enableProfiler;
+  final List<String> streams;
+
+  PerfWitnessRecorderConfig({
+    this.outputDir,
+    this.tag,
+    this.recordNewProcesses = false,
+    this.enableAsyncSpans = false,
+    this.enableProfiler = true,
+    this.streams = const [],
+  });
+
+  factory PerfWitnessRecorderConfig.fromParsedArgs(ArgResults args) {
+    var streams = args['streams'] as List<String>;
+    if (streams.contains('all')) {
+      streams = TimelineStream.values.map((s) => s.name).toList();
+    }
+    return PerfWitnessRecorderConfig(
+      outputDir: args['output-dir'] as String?,
+      tag: args['tag'] as String?,
+      recordNewProcesses: args['record-new-processes'] as bool,
+      enableAsyncSpans: args['enable-async-spans'] as bool,
+      enableProfiler: args['enable-profiler'] as bool,
+      streams: streams,
+    );
+  }
+
+  factory PerfWitnessRecorderConfig.fromArgs(List<String> args) {
+    final parsedArgs = configureArgParser().parse(args);
+    return PerfWitnessRecorderConfig.fromParsedArgs(parsedArgs);
+  }
+
+  static ArgParser configureArgParser([ArgParser? parser]) {
+    return (parser ?? ArgParser())
+      ..addOption('output-dir', abbr: 'o')
+      ..addOption('tag', help: 'Tag to filter processes by.')
+      ..addFlag(
+        'record-new-processes',
+        help: 'Record processes that start after the recorder.',
+        negatable: false,
+      )
+      ..addFlag(
+        'enable-async-spans',
+        help: 'Enable async spans.',
+        negatable: false,
+      )
+      ..addFlag(
+        'enable-profiler',
+        help: 'Enable profiler.',
+        negatable: true,
+        defaultsTo: true,
+      )
+      ..addMultiOption(
+        'streams',
+        help: 'Streams to record.',
+        allowed: [...TimelineStream.values.map((s) => s.name), 'all'],
+        defaultsTo: [TimelineStream.gc.name, TimelineStream.dart.name],
+      );
+  }
+}
+
+Future<void> record(PerfWitnessRecorderConfig config) async {
+  final io.Directory outputDir;
+  if (config.outputDir case final String outputDirPath) {
+    outputDir = io.Directory(outputDirPath);
+  } else {
+    outputDir = io.Directory.systemTemp.createTempSync('recording');
+  }
+
+  final sockets = getAllControlSockets();
+  final connections = (await Future.wait([
+    for (var s in sockets) Connection._tryConnectTo(s.socketPath),
+  ])).nonNulls.toList(growable: false);
+
+  print('Found ${connections.length} processes:');
+  for (final c in connections) {
+    print('  ${c.info}');
+  }
+
+  final matchedConnections = _closeNotMatching(connections, config.tag);
+  if (config.tag != null) {
+    print('Tag ${config.tag} matched ${matchedConnections.length} processes.');
+  }
+
+  print('... data will be written to $outputDir');
+
+  final sw = Stopwatch()..start();
+  await Future.wait([
+    for (var conn in matchedConnections)
+      conn.startRecording(outputDir.path, config: config),
+  ]);
+
+  bool recording = true;
+
+  JsonRpcServer? newProcessServer;
+  if (config.recordNewProcesses) {
+    if (recorderSocketPath case final path?) {
+      if (io.FileSystemEntity.typeSync(path) ==
+          io.FileSystemEntityType.unixDomainSock) {
+        print(
+          'Warning: Control socket $path already exists '
+          '(another recorder might be running).',
+        );
+      } else {
+        newProcessServer = JsonRpcServer(await UnixDomainSocket.bind(path), {
+          'process.announce': (requestor, params) async {
+            if (!recording) {
+              return null;
+            }
+
+            final info = ProcessInfo.fromJson(params as Map<String, Object?>);
+            print('New process announced: $info');
+            if (config.tag == null || info.tag == config.tag) {
+              try {
+                final conn = Connection._(info, requestor);
+                matchedConnections.add(conn);
+                await conn.startRecording(outputDir.path, config: config);
+              } catch (e) {
+                print('Failed to start recording: $e');
+              }
+            }
+            return null;
+          },
+        });
+        print('Listening for new processes on $path');
+      }
+    } else {
+      print(
+        'Warning: Unable to listen for new processes '
+        '(path to the control socket is null).',
+      );
+    }
+  }
+
+  if (matchedConnections.isNotEmpty || config.recordNewProcesses) {
+    await io.ProcessSignal.sigint.watch().first;
+    recording = false;
+    await Future.wait([
+      for (var conn in matchedConnections)
+        conn.stopRecording().catchError((e) {
+          print('Failed to stop recording of process ${conn.info.pid}: $e');
+        }),
+    ]);
+    print('Recorded for ${sw.elapsed}');
+  }
+
+  for (final conn in matchedConnections) {
+    conn.disconnect();
+  }
+  await newProcessServer?.close();
+}
+
+class Connection {
+  final ProcessInfo info;
+  final JsonRpcPeer _endpoint;
+
+  Connection._(this.info, this._endpoint);
+
+  Future<void> startRecording(
+    String outputDir, {
+    required PerfWitnessRecorderConfig config,
+  }) async {
+    await _endpoint.sendRequest('timeline.streamTo', {
+      'recorder': 'perfetto',
+      'path': p.join(outputDir, '${info.pid}.timeline'),
+      'enableProfiler': config.enableProfiler,
+      'enableAsyncSpans': config.enableAsyncSpans,
+      'streams': config.streams,
+    });
+  }
+
+  Future<void> stopRecording() async {
+    await _endpoint.sendRequest('timeline.stopStreaming');
+  }
+
+  void disconnect() async {
+    try {
+      await _endpoint.close();
+    } catch (_) {
+      // Ignore exceptions
+    }
+  }
+
+  static Future<Connection> connectTo(String controlSocketPath) async {
+    final client = jsonRpcPeerFromSocket(
+      await UnixDomainSocket.connect(controlSocketPath),
+    );
+    final info = ProcessInfo.fromJson(
+      await client.sendRequest('process.getInfo') as Map<String, Object?>,
+    );
+    return Connection._(info, client);
+  }
+
+  static Future<Connection?> _tryConnectTo(io.File controlSocket) async {
+    try {
+      return await Connection.connectTo(controlSocket.path);
+    } catch (_) {
+      try {
+        controlSocket.deleteSync(); // Likely stale file. Purge it.
+      } catch (_) {}
+      return null;
+    }
+  }
+}
+
+List<Connection> _closeNotMatching(List<Connection> v, String? tag) {
+  if (tag == null) {
+    return v.toList(growable: true);
+  }
+
+  final open = <Connection>[];
+  for (final c in v) {
+    if (c.info.tag == tag) {
+      open.add(c);
+      continue;
+    }
+    c.disconnect();
+  }
+  return open;
+}
diff --git a/pkg/perf_witness/lib/server.dart b/pkg/perf_witness/lib/server.dart
new file mode 100644
index 00000000000..66973c12eab
--- /dev/null
+++ b/pkg/perf_witness/lib/server.dart
@@ -0,0 +1,201 @@
+// Copyright (c) 2025, 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 'dart:developer' as developer;
+import 'dart:ffi' as ffi;
+import 'dart:io' as io;
+
+import 'package:ffi/ffi.dart' show calloc;
+import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+
+import 'src/common.dart';
+import 'src/json_rpc.dart';
+import 'src/process_info.dart';
+
+class PerfWitnessServer {
+  final String? _tag;
+  final String _controlSocketPath;
+  final String _recorderSocketPath;
+  final ffi.Pointer<ffi.Bool> _isRecordingTimelineWithAsyncSpans;
+
+  bool _isRecordingTimeline = false;
+
+  JsonRpcServer? _server;
+  json_rpc.Peer? _recorderConnection;
+
+  static ffi.Pointer<ffi.Bool>? _sharedIsRecordingTimelineWithAsyncSpans;
+
+  late final Map<String, JsonRpcMethod> _methods = {
+    'process.getInfo': _getProcessInfo,
+    'timeline.streamTo': _timelineStreamTo,
+    'timeline.stopStreaming': _timelineStopStreaming,
+    'process._isRecordingTimelineWithAsyncSpansAddr':
+        _isRecordingTimelineWithAsyncSpansAddr,
+  };
+
+  static PerfWitnessServer? _instance;
+
+  PerfWitnessServer._(
+    this._tag,
+    this._controlSocketPath,
+    this._recorderSocketPath,
+  ) : _isRecordingTimelineWithAsyncSpans = calloc(ffi.sizeOf<ffi.Bool>());
+
+  static Future<void> start({String? tag}) async {
+    if (_instance != null) {
+      return;
+    }
+
+    if (controlSocketPath case final socketPath?) {
+      if (io.FileSystemEntity.typeSync(socketPath) == .unixDomainSock) {
+        // Another isolate is already serving the process. We assume that
+        // server will remain open as long as the process is running.
+        // However we want to make sure that setting global settings (e.g.
+        // whether async spans are enabled or not) will affect all isolates
+        // not just the one that created the server.
+        final client = jsonRpcPeerFromSocket(
+          await UnixDomainSocket.connect(socketPath),
+        );
+        final {'address': int addr, 'pid': int pid} =
+            await client.sendRequest(
+                  'process._isRecordingTimelineWithAsyncSpansAddr',
+                )
+                as Map<String, dynamic>;
+        // Just double check that we are the very same process.
+        if (pid != io.pid) {
+          return;
+        }
+        _sharedIsRecordingTimelineWithAsyncSpans = .fromAddress(addr);
+        return;
+      }
+
+      _instance = PerfWitnessServer._(tag, socketPath, recorderSocketPath!);
+      await _instance!._start();
+    }
+  }
+
+  static Future<void> shutdown() async {
+    await _instance?._shutdown();
+    _instance = null;
+  }
+
+  static bool get isRecordingTimelineWithAsyncSpans {
+    return _sharedIsRecordingTimelineWithAsyncSpans?.value ?? false;
+  }
+
+  Future<void> _timelineStreamTo(
+    json_rpc.Peer requestor,
+    Map<String, Object?>? params,
+  ) async {
+    if (_isRecordingTimeline) {
+      throw StateError('Timeline is already being recorded');
+    }
+
+    final paramsObj = StreamTimelineToRequest(params ?? {});
+
+    final streams = paramsObj.streams
+        ?.map((s) => developer.TimelineStream.values.byName(s))
+        .toList();
+
+    final samplingIntervalUs = paramsObj.samplingInterval;
+    final samplingInterval = samplingIntervalUs != null
+        ? Duration(microseconds: samplingIntervalUs)
+        : const Duration(microseconds: 1000);
+
+    final enableAsyncSpans = paramsObj.enableAsyncSpans ?? false;
+
+    developer.NativeRuntime.streamTimelineTo(
+      developer.TimelineRecorder.values.byName(paramsObj.recorder),
+      path: paramsObj.path,
+      streams:
+          streams ??
+          const [developer.TimelineStream.dart, developer.TimelineStream.gc],
+      enableProfiler: paramsObj.enableProfiler ?? false,
+      samplingInterval: samplingInterval,
+    );
+    _isRecordingTimeline = true;
+    _isRecordingTimelineWithAsyncSpans.value = enableAsyncSpans;
+  }
+
+  Future<void> _timelineStopStreaming(
+    json_rpc.Peer requestor,
+    Map<String, Object?>? params,
+  ) async {
+    if (!_isRecordingTimeline) {
+      throw StateError('Timeline is not being recorded');
+    }
+
+    developer.NativeRuntime.stopStreamingTimeline();
+    _isRecordingTimeline = false;
+    _isRecordingTimelineWithAsyncSpans.value = false;
+  }
+
+  Future<Map<String, Object?>> _getProcessInfo(
+    json_rpc.Peer requestor,
+    Map<String, Object?>? params,
+  ) async {
+    return ProcessInfo.current(tag: _tag).toJson();
+  }
+
+  Future<Map<String, Object?>> _isRecordingTimelineWithAsyncSpansAddr(
+    json_rpc.Peer requestor,
+    Map<String, Object?>? params,
+  ) async {
+    return {
+      'address': _isRecordingTimelineWithAsyncSpans.address,
+      'pid': io.pid,
+    };
+  }
+
+  Future<void> _start() async {
+    _sharedIsRecordingTimelineWithAsyncSpans =
+        _isRecordingTimelineWithAsyncSpans;
+    _server = JsonRpcServer(
+      await UnixDomainSocket.bind(_controlSocketPath),
+      _methods,
+    );
+    await _announceProcessTo(
+      _recorderSocketPath,
+      _tag,
+    ).timeout(Duration(milliseconds: 100), onTimeout: () => Future.value());
+  }
+
+  Future<void> _announceProcessTo(String recorderPath, String? tag) async {
+    try {
+      _recorderConnection = jsonRpcPeerFromSocket(
+        await UnixDomainSocket.connect(recorderPath),
+        _methods,
+      );
+      await _recorderConnection!.sendRequest(
+        'process.announce',
+        ProcessInfo.current(tag: tag).toJson(),
+      );
+    } catch (e) {
+      // ignore, recorder might not be running.
+    }
+  }
+
+  Future<void> _shutdown() async {
+    _recorderConnection?.close();
+    await _server?.close();
+    if (io.FileSystemEntity.typeSync(_controlSocketPath) != .notFound) {
+      io.File(_controlSocketPath).deleteSync();
+    }
+    calloc.free(_isRecordingTimelineWithAsyncSpans);
+    if (_isRecordingTimeline) {
+      developer.NativeRuntime.stopStreamingTimeline();
+      _isRecordingTimeline = false;
+    }
+  }
+}
+
+extension type StreamTimelineToRequest(Map<String, Object?> json) {
+  String get recorder => json['recorder'] as String;
+  String? get path => json['path'] as String?;
+  List<String>? get streams => (json['streams'] as List?)?.cast<String>();
+  bool? get enableProfiler => json['enableProfiler'] as bool?;
+  int? get samplingInterval => json['samplingInterval'] as int?;
+  bool? get enableAsyncSpans => json['enableAsyncSpans'] as bool?;
+}
diff --git a/pkg/perf_witness/lib/src/async_span.dart b/pkg/perf_witness/lib/src/async_span.dart
new file mode 100644
index 00000000000..93b3206527d
--- /dev/null
+++ b/pkg/perf_witness/lib/src/async_span.dart
@@ -0,0 +1,183 @@
+// Copyright (c) 2025, 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 'dart:developer';
+
+import '../../server.dart';
+
+/// With synchronous execution the nesting between spans is naturally induced
+/// by the callstack. Consider:
+///
+/// ```dart
+/// Timeline.timeSync('a', () {
+///   work();
+///   Timeline.timeSync('b', () {
+///     work();
+///   });
+///   work();
+///   Timeline.timeSync('c', () {
+///     work();
+///   });
+///   work();
+/// })
+/// ```
+//
+/// This will created three spans `a`, `b` and `c` all properly nested. The time
+/// outside of `b` and `c` will be correctly attributed to `a`.
+//
+/// However the same is not easy to achieve for async computations. Compare:
+///
+/// ```
+/// void a() async {
+///   work();
+///   await b();
+///   work();
+///   await c();
+///   work();
+/// }
+/// ```
+//
+/// There is no functionality available in `dart:developer` which would allow
+/// to create proper span structure to automatically accurately capture the
+/// work done in `a`, `b` and `c`. The best you can do is to manually wrap
+/// synchronous parts of work into `timeSync`.
+///
+/// This class tries to help with this by creating a `Zone` which automatically
+/// does this - though result still might be confusing: completion of async task
+/// causes resumption of async task that awaits on the current task which creates
+/// inversely nested spans (e.g. if `b` is suspended and completes
+/// asynchronously you get span `a` nested inside span `b` - even though you
+/// would like an opposite picture or worst case you want these spans to be
+/// siblings).
+///
+/// Wrapping execution in a `Zone` is expensive so we only enable it if
+/// recorder requests it explicitly. When disabled we only emit timeline
+/// spans for the first synchronous portion of the computation and then
+/// a instantaneous span for the completion. This allows developer to
+/// estimate how long asynchronous action took - but it will not actually
+/// reveal when it was actively running on the stack.
+class AsyncSpan {
+  final String name;
+  final Map<String, Object?>? parameters;
+  final Flow _flow = Flow.begin();
+  bool issuedBegin = false;
+  int running = 0;
+
+  AsyncSpan._(this.name, {this.parameters});
+
+  static AsyncSpan of(Zone zone) => zone[AsyncSpan] as AsyncSpan;
+
+  static final _zoneSpecification = ZoneSpecification(
+    run: <R>(self, parent, zone, R Function() f) {
+      final span = AsyncSpan.of(self);
+
+      span.startSync();
+      try {
+        return parent.run(zone, f);
+      } finally {
+        span.finishSync();
+      }
+    },
+    runUnary: <R, T1>(self, parent, zone, R Function(T1) f, T1 a1) {
+      final span = AsyncSpan.of(self);
+      span.startSync();
+      try {
+        return parent.runUnary(zone, f, a1);
+      } finally {
+        span.finishSync();
+      }
+    },
+    runBinary:
+        <R, T1, T2>(self, parent, zone, R Function(T1, T2) f, T1 a1, T2 a2) {
+          final span = AsyncSpan.of(self);
+          span.startSync();
+          try {
+            return parent.runBinary(zone, f, a1, a2);
+          } finally {
+            span.finishSync();
+          }
+        },
+  );
+
+  static Future<R> run<R>(
+    String name,
+    Future<R> Function() action, {
+    Map<String, Object?>? parameters,
+  }) async {
+    if (PerfWitnessServer.isRecordingTimelineWithAsyncSpans) {
+      return AsyncSpan._create(name, parameters: parameters).run(action);
+    } else {
+      final Future<R> result;
+      final flow = Flow.begin();
+      try {
+        Timeline.startSync(name, flow: flow);
+        result = action();
+      } finally {
+        Timeline.finishSync();
+      }
+
+      try {
+        return await result;
+      } finally {
+        Timeline.startSync(name, flow: Flow.end(flow.id));
+        Timeline.finishSync();
+      }
+    }
+  }
+
+  static Future<R> runUnary<R, T>(
+    String name,
+    Future<R> Function(T) action,
+    T arg, {
+    Map<String, Object?>? parameters,
+  }) async {
+    if (PerfWitnessServer.isRecordingTimelineWithAsyncSpans) {
+      return AsyncSpan._create(
+        name,
+        parameters: parameters,
+      ).runUnary(action, arg);
+    } else {
+      final Future<R> result;
+      final flow = Flow.begin();
+      try {
+        Timeline.startSync(name, flow: flow);
+        result = action(arg);
+      } finally {
+        Timeline.finishSync();
+      }
+
+      try {
+        return await result;
+      } finally {
+        Timeline.startSync(name, flow: Flow.end(flow.id));
+        Timeline.finishSync();
+      }
+    }
+  }
+
+  static Zone _create(String name, {Map<String, Object?>? parameters}) =>
+      Zone.current.fork(
+        specification: _zoneSpecification,
+        zoneValues: {AsyncSpan: AsyncSpan._(name, parameters: parameters)},
+      );
+
+  void startSync() {
+    if (running == 0) {
+      Timeline.startSync(
+        name,
+        flow: issuedBegin ? Flow.step(_flow.id) : _flow,
+        arguments: issuedBegin ? null : parameters,
+      );
+      issuedBegin = true;
+    }
+    running++;
+  }
+
+  void finishSync() {
+    if (--running == 0) {
+      Timeline.finishSync();
+    }
+  }
+}
diff --git a/pkg/perf_witness/lib/src/common.dart b/pkg/perf_witness/lib/src/common.dart
new file mode 100644
index 00000000000..61768bf125c
--- /dev/null
+++ b/pkg/perf_witness/lib/src/common.dart
@@ -0,0 +1,75 @@
+// Copyright (c) 2025, 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:io' as io;
+
+import 'package:dart_data_home/dart_data_home.dart';
+import 'package:path/path.dart' as p;
+
+final String? _controlSocketsDirectory = () {
+  final dir = getDartDataHome('perf');
+  try {
+    // Ensure that directory exists.
+    io.Directory(dir).createSync(recursive: true);
+    return dir;
+  } catch (_) {
+    // Ignore any sort of exceptions.
+    return null;
+  }
+}();
+
+List<({int pid, io.File socketPath})> getAllControlSockets() {
+  if (_controlSocketsDirectory == null) {
+    return const [];
+  }
+
+  try {
+    final allPidFiles = io.Directory(
+      _controlSocketsDirectory!,
+    ).listSync().whereType<io.File>();
+    return [
+      for (var file in allPidFiles)
+        if (int.tryParse(p.basenameWithoutExtension(file.path)) case final pid?)
+          (pid: pid, socketPath: file),
+    ];
+  } catch (_) {
+    // Ignore
+    return [];
+  }
+}
+
+final String? controlSocketPath = () {
+  final dirPath = _controlSocketsDirectory;
+  if (dirPath == null) {
+    return null;
+  }
+  return p.join(dirPath, '${io.pid}');
+}();
+
+final String? recorderSocketPath = () {
+  final dirPath = _controlSocketsDirectory;
+  if (dirPath == null) {
+    return null;
+  }
+  return p.join(dirPath, 'rec');
+}();
+
+abstract class UnixDomainSocket {
+  static Future<io.Socket> connect(String path) => io.Socket.connect(
+    io.InternetAddress(path, type: io.InternetAddressType.unix),
+    0,
+  );
+
+  static Future<io.ServerSocket> bind(String path) {
+    if (io.FileSystemEntity.typeSync(path) !=
+        io.FileSystemEntityType.notFound) {
+      io.File(path).deleteSync();
+    }
+
+    return io.ServerSocket.bind(
+      io.InternetAddress(path, type: io.InternetAddressType.unix),
+      0,
+    );
+  }
+}
diff --git a/pkg/perf_witness/lib/src/json_rpc.dart b/pkg/perf_witness/lib/src/json_rpc.dart
new file mode 100644
index 00000000000..182a6b53d3c
--- /dev/null
+++ b/pkg/perf_witness/lib/src/json_rpc.dart
@@ -0,0 +1,71 @@
+// Copyright (c) 2025, 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 'dart:convert';
+import 'dart:io' as io;
+
+import 'package:json_rpc_2/json_rpc_2.dart' as json_rpc;
+import 'package:stream_channel/stream_channel.dart';
+
+typedef JsonRpcPeer = json_rpc.Peer;
+
+typedef JsonRpcException = json_rpc.RpcException;
+
+typedef JsonRpcMethod =
+    FutureOr<Object?> Function(
+      JsonRpcPeer requestor,
+      Map<String, Object?>? params,
+    );
+
+JsonRpcPeer jsonRpcPeerFromSocket(
+  io.Socket socket, [
+  Map<String, JsonRpcMethod>? methods,
+]) {
+  final lineChannel = StreamChannel<String>(
+    const LineSplitter().bind(utf8.decoder.bind(socket)),
+    StreamController<String>(sync: true, onCancel: socket.close)
+      ..stream.listen((line) {
+        socket.write(line);
+        socket.write('\n');
+      }),
+  );
+  final peer = json_rpc.Peer(lineChannel);
+  if (methods != null) {
+    for (final MapEntry(:key, :value) in methods.entries) {
+      peer.registerMethod(key, (json_rpc.Parameters params) {
+        return value(
+          peer,
+          params.value == null ? null : params.asMap.cast<String, Object?>(),
+        );
+      });
+    }
+  }
+
+  peer.listen().ignore();
+  return peer;
+}
+
+class JsonRpcServer {
+  final io.ServerSocket _serverSocket;
+  final _endpoints = <JsonRpcPeer>{};
+
+  JsonRpcServer(this._serverSocket, [Map<String, JsonRpcMethod>? methods]) {
+    _serverSocket.listen((client) {
+      final endpoint = jsonRpcPeerFromSocket(client, methods);
+      _endpoints.add(endpoint);
+      endpoint.done.whenComplete(() {
+        _endpoints.remove(endpoint);
+      });
+    });
+  }
+
+  /// Returns a list of currently connected endpoints.
+  List<JsonRpcPeer> get endpoints => _endpoints.toList();
+
+  Future<void> close() async {
+    await Future.wait(_endpoints.toList().map((e) => e.close()));
+    await _serverSocket.close();
+  }
+}
diff --git a/pkg/perf_witness/lib/src/process_info.dart b/pkg/perf_witness/lib/src/process_info.dart
new file mode 100644
index 00000000000..013eda9e66b
--- /dev/null
+++ b/pkg/perf_witness/lib/src/process_info.dart
@@ -0,0 +1,57 @@
+// Copyright (c) 2025, 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:io' as io;
+
+class ProcessInfo {
+  final int pid;
+  final String command;
+  final String script;
+  final String dartBinary;
+  final int rss;
+  final String? tag;
+
+  ProcessInfo({
+    required this.pid,
+    required this.command,
+    required this.script,
+    required this.dartBinary,
+    required this.rss,
+    this.tag,
+  });
+
+  ProcessInfo.current({String? tag})
+    : this(
+        pid: io.pid,
+        command: io.Platform.executableArguments.join(' '),
+        script: io.Platform.script.toFilePath(),
+        dartBinary: io.Platform.executable,
+        rss: io.ProcessInfo.currentRss,
+        tag: tag,
+      );
+
+  factory ProcessInfo.fromJson(Map<String, dynamic> json) => ProcessInfo(
+    pid: json['pid'] as int,
+    command: json['command'] as String,
+    script: json['script'] as String,
+    dartBinary: json['dartBinary'] as String,
+    rss: json['rss'] as int,
+    tag: json['tag'] as String?,
+  );
+
+  Map<String, dynamic> toJson() => {
+    'pid': pid,
+    'command': command,
+    'script': script,
+    'dartBinary': dartBinary,
+    'rss': rss,
+    if (tag != null) 'tag': tag,
+  };
+
+  @override
+  String toString() {
+    final tagString = tag != null ? ' (tag: $tag)' : '';
+    return '[PID: $pid, script: $script, RSS: $rss]$tagString';
+  }
+}
diff --git a/pkg/perf_witness/pubspec.yaml b/pkg/perf_witness/pubspec.yaml
new file mode 100644
index 00000000000..7be18787e6e
--- /dev/null
+++ b/pkg/perf_witness/pubspec.yaml
@@ -0,0 +1,26 @@
+name: perf_witness
+version: 0.0.1
+description: Shared performance observability infrastructure for Dart CLI tools
+repository: https://github.com/dart-lang/sdk/tree/main/pkg/perf_witness
+publish_to: none
+resolution: workspace
+
+environment:
+  sdk: ^3.11.0-edge
+
+dependencies:
+  args: any
+  dart_data_home: any
+  ffi: any
+  json_rpc_2: any
+  path: any
+  stream_channel: any
+
+# We use 'any' version constraints here as we get our package versions from
+# the dart-lang/sdk repo's DEPS file. Note that this is a special case; the
+# best practice for packages is to specify their compatible version ranges.
+# See also https://dart.dev/tools/pub/dependencies.
+dev_dependencies:
+  lints: any
+  test: any
+  vm_service_protos: any
diff --git a/pkg/perf_witness/test/common/busy_loop.dart b/pkg/perf_witness/test/common/busy_loop.dart
new file mode 100644
index 00000000000..f8a6d1e5b2b
--- /dev/null
+++ b/pkg/perf_witness/test/common/busy_loop.dart
@@ -0,0 +1,59 @@
+// Copyright (c) 2025, 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 'dart:io';
+import 'dart:isolate';
+
+import 'package:args/args.dart';
+import 'package:perf_witness/server.dart';
+import 'package:perf_witness/src/async_span.dart';
+
+final parser = ArgParser()
+  ..addOption('tag', abbr: 't', help: 'Tag for the process')
+  ..addFlag(
+    'start-isolate',
+    abbr: 'i',
+    help: 'Start test isolate',
+    defaultsTo: false,
+  );
+
+bool shouldStop = false;
+
+Future<void> busyLoop({required String name}) async {
+  print('[$name] BUSY LOOP READY');
+  while (!shouldStop) {
+    await AsyncSpan.run('sleep', () async {
+      print(
+        '[$name] AsyncSpan.create is nop: ${identical(Zone.current, Zone.root)}',
+      );
+      await Future.delayed(const Duration(milliseconds: 500));
+    });
+  }
+  print('done');
+}
+
+void main(List<String> args) async {
+  ProcessSignal.sigint.watch().listen((_) {
+    print('SIGINT received');
+    shouldStop = true;
+  });
+
+  final parsedArgs = parser.parse(args);
+  final tag = parsedArgs['tag'] as String?;
+  await PerfWitnessServer.start(tag: tag);
+  if (parsedArgs.flag('start-isolate')) {
+    Isolate.run(() async {
+      await PerfWitnessServer.start(tag: tag);
+      await busyLoop(name: 'child-isolate');
+    }).onError((e, s) {
+      print('Isolate error: $e');
+      print(s);
+      exit(1);
+    });
+  }
+  await busyLoop(name: 'main');
+  await PerfWitnessServer.shutdown();
+  exit(0);
+}
diff --git a/pkg/perf_witness/test/json_rpc_test.dart b/pkg/perf_witness/test/json_rpc_test.dart
new file mode 100644
index 00000000000..67d43fb4718
--- /dev/null
+++ b/pkg/perf_witness/test/json_rpc_test.dart
@@ -0,0 +1,94 @@
+// Copyright (c) 2025, 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 'dart:io' as io;
+
+import 'package:path/path.dart' as p;
+import 'package:perf_witness/src/common.dart';
+import 'package:perf_witness/src/json_rpc.dart';
+import 'package:test/test.dart';
+
+void main() {
+  group('JsonRpc', () {
+    late JsonRpcServer server;
+    late JsonRpcPeer client;
+    late String socketPath;
+
+    setUp(() async {
+      socketPath = p.join(
+        io.Directory.systemTemp.createTempSync().path,
+        'test.sock',
+      );
+      server = JsonRpcServer(await UnixDomainSocket.bind(socketPath), {
+        'testMethod': (requestor, params) => 'Hello, ${params!['name']}',
+        'errorMethod': (requestor, params) => throw 'Something went wrong',
+        'ping': (requestor, params) => 'pong',
+        'checkEndpoint': (requestor, params) {
+          expect(server.endpoints, contains(requestor));
+          return 'ok';
+        },
+      });
+      client =
+          jsonRpcPeerFromSocket(await UnixDomainSocket.connect(socketPath), {
+            'reverse': (requestor, params) =>
+                (params!['text'] as String).split('').reversed.join(),
+          });
+    });
+
+    tearDown(() async {
+      await client.close();
+      await server.close();
+      final file = io.File(socketPath);
+      if (file.existsSync()) {
+        file.deleteSync();
+      }
+    });
+
+    test('can make a successful request', () async {
+      final result = await client.sendRequest('testMethod', {'name': 'World'});
+      expect(result, 'Hello, World');
+    });
+
+    test('handles method not found', () async {
+      try {
+        await client.sendRequest('nonExistentMethod');
+        fail('Expected an error');
+      } catch (e) {
+        expect(e, isA<JsonRpcException>());
+        expect(
+          (e as JsonRpcException).message,
+          'Unknown method "nonExistentMethod".',
+        );
+      }
+    });
+
+    test('handles internal server error', () async {
+      try {
+        await client.sendRequest('errorMethod');
+        fail('Expected an error');
+      } catch (e) {
+        expect(e, isA<JsonRpcException>());
+        expect((e as JsonRpcException).message, 'Something went wrong');
+      }
+    });
+
+    test('can make bidirectional requests', () async {
+      // Client calls server
+      expect(await client.sendRequest('ping'), 'pong');
+
+      // Server calls client
+      // Wait for the server to accept the connection.
+      while (server.endpoints.isEmpty) {
+        await Future.delayed(const Duration(milliseconds: 10));
+      }
+      final endpoint = server.endpoints.first;
+      expect(await endpoint.sendRequest('reverse', {'text': 'hello'}), 'olleh');
+    });
+
+    test('method receives correct endpoint', () async {
+      expect(await client.sendRequest('checkEndpoint'), equals('ok'));
+    });
+  });
+}
diff --git a/pkg/perf_witness/test/recorder_server_test.dart b/pkg/perf_witness/test/recorder_server_test.dart
new file mode 100644
index 00000000000..95699291b0b
--- /dev/null
+++ b/pkg/perf_witness/test/recorder_server_test.dart
@@ -0,0 +1,567 @@
+// Copyright (c) 2025, 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 'dart:convert';
+import 'dart:io' as io;
+import 'dart:isolate';
+
+import 'package:path/path.dart' as p;
+import 'package:test/test.dart';
+import 'package:vm_service_protos/vm_service_protos.dart';
+
+final packageRoot = p.dirname(
+  p.dirname(
+    Isolate.resolvePackageUriSync(
+      Uri.parse('package:perf_witness/server.dart'),
+    )!.toFilePath(),
+  ),
+);
+
+final testsDir = p.join(packageRoot, 'test');
+final binDir = p.join(packageRoot, 'bin');
+
+Future<io.Process> runProcess(
+  String executable,
+  List<String> arguments, {
+  String tag = '',
+  String? waitFor,
+  Map<String, String>? environment,
+  List<String>? stdout,
+}) async {
+  final ready = Completer();
+
+  final process = await io.Process.start(
+    executable,
+    arguments,
+    environment: environment,
+  );
+  process.stdout.transform(Utf8Decoder()).transform(LineSplitter()).listen((
+    line,
+  ) {
+    if (waitFor != null && !ready.isCompleted && line.contains(waitFor)) {
+      ready.complete();
+    }
+    print('[$tag]stdout> $line');
+    stdout?.add(line);
+  });
+  process.stderr.transform(Utf8Decoder()).transform(LineSplitter()).listen((
+    line,
+  ) {
+    print('[$tag]stderr> $line');
+  });
+  if (waitFor != null) {
+    await ready.future;
+  }
+  return process;
+}
+
+class BusyLoopProcess {
+  final io.Process process;
+  final String tag;
+  final List<String> stdout;
+
+  BusyLoopProcess._(this.process, this.tag, this.stdout);
+
+  static Future<BusyLoopProcess> start(
+    String tag,
+    io.Directory tempDir, {
+    bool startIsolate = false,
+  }) async {
+    final stdout = <String>[];
+    return BusyLoopProcess._(
+      await runProcess(
+        io.Platform.executable,
+        [
+          'run',
+          p.join(testsDir, 'common', 'busy_loop.dart'),
+          '--tag',
+          tag,
+          if (startIsolate) '--start-isolate',
+        ],
+        tag: 'busy-loop($tag)',
+        waitFor: 'BUSY LOOP READY',
+        environment: {'DART_DATA_HOME': tempDir.path},
+        stdout: stdout,
+      ),
+      tag,
+      stdout,
+    );
+  }
+
+  int get pid => process.pid;
+
+  void kill() {
+    process.kill();
+  }
+}
+
+class RecorderProcess {
+  final io.Process process;
+
+  RecorderProcess._(this.process);
+
+  static Future<RecorderProcess> start(
+    io.Directory tempDir,
+    io.Directory outputDir, {
+    String? tag,
+    bool recordNewProcesses = false,
+    bool enableAsyncSpans = false,
+    bool enableProfiler = true,
+    List<String> streams = const ['dart', 'gc'],
+    String? waitFor,
+  }) async {
+    return RecorderProcess._(
+      await runProcess(
+        io.Platform.executable,
+        [
+          'run',
+          p.join(binDir, 'recorder.dart'),
+          '-o',
+          outputDir.path,
+          if (tag != null) ...['--tag', tag],
+          if (recordNewProcesses) '--record-new-processes',
+          if (enableAsyncSpans) '--enable-async-spans',
+          if (!enableProfiler) '--no-enable-profiler',
+          if (streams != const ['dart', 'gc']) ...[
+            '--streams',
+            streams.join(','),
+          ],
+        ],
+        tag: 'recorder',
+        environment: {'DART_DATA_HOME': tempDir.path},
+        waitFor: waitFor,
+      ),
+    );
+  }
+
+  Future<void> stop() async {
+    process.kill(io.ProcessSignal.sigint);
+    if (await process.exitCode case final int exitCode when exitCode != 0) {
+      throw Exception('Recorder process failed with exit code $exitCode');
+    }
+  }
+}
+
+void main() {
+  group('Recorder and Server', () {
+    late BusyLoopProcess busyLoopProcess;
+    late io.Directory tempDir;
+
+    setUp(() async {
+      tempDir = io.Directory.systemTemp.createTempSync();
+      busyLoopProcess = await BusyLoopProcess.start('busy-loop-tag', tempDir);
+    });
+
+    tearDown(() {
+      busyLoopProcess.kill();
+      tempDir.deleteSync(recursive: true);
+    });
+
+    test('end-to-end test with recorder script', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(tempDir, outputDir);
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      final timelineFiles = outputDir
+          .listSync()
+          .whereType<io.File>()
+          .where((file) => file.path.endsWith('.timeline'))
+          .toList();
+
+      final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
+      expect(
+        timelines,
+        equals(['${busyLoopProcess.pid}.timeline']),
+        reason: 'Expected timeline file to be created',
+      );
+
+      final trace = Trace()
+        ..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
+      expect(trace.packet, isNotEmpty);
+      expect(trace.packet.any((p) => p.hasPerfSample()), isTrue);
+      // Dart track should be enabled by default.
+      expect(extractSeenEvents(trace), containsAll(['sleep']));
+    });
+
+    test('end-to-end test with recorder script - early exit', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(tempDir, outputDir);
+      await Future.delayed(const Duration(seconds: 2));
+      busyLoopProcess.process.kill(io.ProcessSignal.sigint);
+      await busyLoopProcess.process.exitCode;
+      await recorder.stop();
+
+      final timelineFiles = outputDir
+          .listSync()
+          .whereType<io.File>()
+          .where((file) => file.path.endsWith('.timeline'))
+          .toList();
+
+      final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
+      expect(
+        timelines,
+        equals(['${busyLoopProcess.pid}.timeline']),
+        reason: 'Expected timeline file to be created',
+      );
+
+      final trace = Trace()
+        ..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
+      expect(trace.packet, isNotEmpty);
+      expect(trace.packet.any((p) => p.hasPerfSample()), isTrue);
+      // Dart track should be enabled by default.
+      expect(extractSeenEvents(trace), containsAll(['sleep']));
+    });
+
+    test('profiler can be disabled', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        enableProfiler: false,
+      );
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      final timelineFiles = outputDir
+          .listSync()
+          .whereType<io.File>()
+          .where((file) => file.path.endsWith('.timeline'))
+          .toList();
+
+      final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
+      expect(
+        timelines,
+        equals(['${busyLoopProcess.pid}.timeline']),
+        reason: 'Expected timeline file to be created',
+      );
+
+      final trace = Trace()
+        ..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
+      expect(trace.packet, isNotEmpty);
+      expect(trace.packet.any((p) => p.hasPerfSample()), isFalse);
+    });
+
+    test('streams can be configured', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        enableProfiler: false,
+        streams: ['isolate', 'compiler'],
+      );
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      final timelineFiles = outputDir
+          .listSync()
+          .whereType<io.File>()
+          .where((file) => file.path.endsWith('.timeline'))
+          .toList();
+
+      final timelines = timelineFiles.map((e) => p.basename(e.path)).toList();
+      expect(
+        timelines,
+        equals(['${busyLoopProcess.pid}.timeline']),
+        reason: 'Expected timeline file to be created',
+      );
+
+      final trace = Trace()
+        ..mergeFromBuffer(timelineFiles.first.readAsBytesSync());
+      expect(trace.packet, isNotEmpty);
+
+      expect(trace.packet.any((p) => p.hasPerfSample()), isFalse);
+      final seenEvents = extractSeenEvents(trace);
+      expect(seenEvents, containsAll(['HandleMessage', 'CompileFunction']));
+      // Dart trace is disabled.
+      expect(seenEvents, isNot(contains('sleep')));
+    });
+
+    test('tag filtering positive test', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        tag: 'busy-loop-tag',
+      );
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      final timelines = outputDir
+          .listSync()
+          .map((e) => p.basename(e.path))
+          .toList();
+      expect(
+        timelines,
+        equals(['${busyLoopProcess.pid}.timeline']),
+        reason: 'Expected timeline file to be created',
+      );
+    });
+
+    test('tag filtering negative test', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        tag: 'unmatched-tag',
+      );
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      final timelines = outputDir
+          .listSync()
+          .map((e) => p.basename(e.path))
+          .toList();
+      expect(
+        timelines,
+        isEmpty,
+        reason: 'Expected no timeline file to be created',
+      );
+    });
+
+    test('async spans are not activated by default', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      final busyLoopWithIsolate = await BusyLoopProcess.start(
+        'busy-loop-with-isolate-tag',
+        tempDir,
+        startIsolate: true,
+      );
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(tempDir, outputDir);
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      busyLoopWithIsolate.kill();
+
+      expect(
+        busyLoopProcess.stdout,
+        contains('[main] AsyncSpan.create is nop: true'),
+      );
+      expect(
+        busyLoopProcess.stdout,
+        isNot(contains('[main] AsyncSpan.create is nop: false')),
+      );
+      expect(
+        busyLoopProcess.stdout,
+        isNot(contains('[child-isolate] AsyncSpan.create is nop: true')),
+      );
+      expect(
+        busyLoopProcess.stdout,
+        isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
+      );
+
+      expect(
+        busyLoopWithIsolate.stdout,
+        contains('[main] AsyncSpan.create is nop: true'),
+      );
+      expect(
+        busyLoopWithIsolate.stdout,
+        isNot(contains('[main] AsyncSpan.create is nop: false')),
+      );
+      expect(
+        busyLoopWithIsolate.stdout,
+        contains('[child-isolate] AsyncSpan.create is nop: true'),
+      );
+      expect(
+        busyLoopWithIsolate.stdout,
+        isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
+      );
+
+      final timelines = outputDir
+          .listSync()
+          .map((e) => p.basename(e.path))
+          .toList();
+      expect(
+        timelines,
+        unorderedEquals([
+          '${busyLoopProcess.pid}.timeline',
+          '${busyLoopWithIsolate.pid}.timeline',
+        ]),
+        reason: 'Expected timeline file to be created',
+      );
+    });
+
+    test('async spans are activated when requested', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      final busyLoopWithIsolate = await BusyLoopProcess.start(
+        'busy-loop-with-isolate-tag',
+        tempDir,
+        startIsolate: true,
+      );
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        enableAsyncSpans: true,
+      );
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      busyLoopWithIsolate.kill();
+
+      expect(
+        busyLoopProcess.stdout,
+        contains('[main] AsyncSpan.create is nop: false'),
+      );
+      expect(
+        busyLoopProcess.stdout,
+        isNot(contains('[child-isolate] AsyncSpan.create is nop: true')),
+      );
+      expect(
+        busyLoopProcess.stdout,
+        isNot(contains('[child-isolate] AsyncSpan.create is nop: false')),
+      );
+
+      expect(
+        busyLoopWithIsolate.stdout,
+        contains('[main] AsyncSpan.create is nop: false'),
+      );
+      expect(
+        busyLoopWithIsolate.stdout,
+        contains('[child-isolate] AsyncSpan.create is nop: false'),
+      );
+
+      final timelines = outputDir
+          .listSync()
+          .map((e) => p.basename(e.path))
+          .toList();
+      expect(
+        timelines,
+        unorderedEquals([
+          '${busyLoopProcess.pid}.timeline',
+          '${busyLoopWithIsolate.pid}.timeline',
+        ]),
+        reason: 'Expected timeline file to be created',
+      );
+    });
+
+    test('record new processes - all', () async {
+      final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+      // Run the recorder in a separate process.
+      final recorder = await RecorderProcess.start(
+        tempDir,
+        outputDir,
+        recordNewProcesses: true,
+        waitFor: 'Listening for new processes',
+      );
+
+      // Start a new process that should be recorded.
+      final newProcess = await BusyLoopProcess.start(
+        'new-process-tag',
+        tempDir,
+      );
+
+      await Future.delayed(const Duration(seconds: 2));
+      await recorder.stop();
+
+      newProcess.kill();
+
+      final timelines = outputDir
+          .listSync()
+          .map((e) => p.basename(e.path))
+          .toList();
+      expect(
+        timelines,
+        unorderedEquals([
+          '${newProcess.pid}.timeline',
+          '${busyLoopProcess.pid}.timeline',
+        ]),
+      );
+    }, timeout: Timeout(Duration(seconds: 15)));
+
+    test(
+      'record new processes - specific tag',
+      () async {
+        final outputDir = io.Directory('${tempDir.path}/output')..createSync();
+
+        // Run the recorder in a separate process.
+        final recorder = await RecorderProcess.start(
+          tempDir,
+          outputDir,
+          tag: 'new-process-tag',
+          recordNewProcesses: true,
+        );
+
+        // Start a new process that should be recorded.
+        final newProcess = await BusyLoopProcess.start(
+          'new-process-tag',
+          tempDir,
+        );
+
+        // Start a new process that should NOT be recorded.
+        final ignoredProcess = await BusyLoopProcess.start(
+          'ignored-tag',
+          tempDir,
+        );
+
+        await Future.delayed(const Duration(seconds: 2));
+        await recorder.stop();
+
+        newProcess.kill();
+        ignoredProcess.kill();
+
+        final timelines = outputDir
+            .listSync()
+            .map((e) => p.basename(e.path))
+            .toList();
+        expect(timelines, unorderedEquals(['${newProcess.pid}.timeline']));
+      },
+      timeout: Timeout(Duration(seconds: 15)),
+    );
+  });
+}
+
+class IncrementalState {
+  final eventNames = <int, String>{};
+
+  void update(InternedData internedData) {
+    for (var eventName in internedData.eventNames) {
+      eventNames[eventName.iid.toInt()] = eventName.name;
+    }
+  }
+}
+
+Set<String> extractSeenEvents(Trace trace) {
+  var state = IncrementalState();
+  final seenEvents = <String>{};
+  for (var packet in trace.packet) {
+    if ((packet.sequenceFlags &
+            TracePacket_SequenceFlags.SEQ_INCREMENTAL_STATE_CLEARED.value) !=
+        0) {
+      state = IncrementalState();
+    }
+
+    if (packet.hasInternedData()) {
+      state.update(packet.internedData);
+    }
+
+    if (packet.hasTrackEvent()) {
+      final trackEvent = packet.trackEvent;
+      if (trackEvent.type == TrackEvent_Type.TYPE_SLICE_BEGIN) {
+        final name = state.eventNames[packet.trackEvent.nameIid.toInt()]!;
+        seenEvents.add(name);
+      }
+    }
+  }
+
+  return seenEvents;
+}
diff --git a/pubspec.yaml b/pubspec.yaml
index 9d606fb0f8e..1b06f25de1e 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -63,6 +63,7 @@ workspace:
   - pkg/native_compiler
   - pkg/native_stack_traces
   - pkg/node_preamble
+  - pkg/perf_witness
   - pkg/reload_test
   - pkg/scrape
   - pkg/server_plugin