From 051ceea3b2e54608a6d465952564945b892e3c78 Mon Sep 17 00:00:00 2001 From: Jeroen Schweitzer Date: Fri, 12 Jun 2026 03:04:58 +0200 Subject: [PATCH] route DaemonClient through a DaemonTransport seam (T-331) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The UI's backend client connected straight to the workspace unix socket, hard-coding the local shape. It now talks JSON-lines through a DaemonTransport (new lib/src/ipc/transport.dart, Flutter-free), with LocalSocketTransport reproducing today's connect byte-for-byte — zero behavior change, proven by the untouched client test suite plus new seam tests driving the client over an in-memory transport. This is the slot the SSH-remote backend (T-329/Q-23) plugs into: request correlation, reconnect/backoff, and event forwarding live above the seam and won't change when the endpoint is remote. main.dart's swapIpcServer becomes swapBackend per the same plan. Co-Authored-By: Claude Fable 5 --- .pql/changelog/ticket_history/2026-06.sql | 1 + .pql/changelog/tickets/2026-06.sql | 1 + lib/clide.dart | 1 + lib/kernel/src/facade.dart | 2 +- lib/kernel/src/ipc/client.dart | 99 ++++++++++++----------- lib/main.dart | 20 ++--- lib/src/ipc/transport.dart | 73 +++++++++++++++++ test/helpers/fake_ipc.dart | 2 +- test/ipc/transport_test.dart | 73 +++++++++++++++++ test/kernel/src/ipc/client_test.dart | 98 +++++++++++++++++++++- 10 files changed, 311 insertions(+), 59 deletions(-) create mode 100644 lib/src/ipc/transport.dart create mode 100644 test/ipc/transport_test.dart diff --git a/.pql/changelog/ticket_history/2026-06.sql b/.pql/changelog/ticket_history/2026-06.sql index bfc7220e..0a5da1d0 100644 --- a/.pql/changelog/ticket_history/2026-06.sql +++ b/.pql/changelog/ticket_history/2026-06.sql @@ -4146,3 +4146,4 @@ Plan: DECSTR (`CSI ! p`, soft reset) is a separate, smaller follow-up — same intermediates mechanism, maps to a subset of the existing reset paths; file separately if wanted. Done when: claude/vim cursor-shape changes (insert vs normal mode) render as bar vs block in the terminal pane.', NULL, '2026-06-12 00:52:40', '2026-06-12 00:52:40', '2026-06-12 00:52:40', NULL, 'fd38b932757c81de62d693d72b883448', 2) ON CONFLICT(hash) DO NOTHING; +INSERT INTO ticket_history (ticket_record_id, field, old_value, new_value, changed_by, changed_at, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FB3DKQQJ583944DG8561VQ3G', 'status', 'backlog', 'done', NULL, '2026-06-12 01:04:52', '2026-06-12 01:04:52', '2026-06-12 01:04:52', NULL, '66a692e143b6a4be1b4845825e9e16ad', 2) ON CONFLICT(hash) DO NOTHING; diff --git a/.pql/changelog/tickets/2026-06.sql b/.pql/changelog/tickets/2026-06.sql index efd32c0a..a2d7091a 100644 --- a/.pql/changelog/tickets/2026-06.sql +++ b/.pql/changelog/tickets/2026-06.sql @@ -4429,3 +4429,4 @@ Plan: DECSTR (`CSI ! p`, soft reset) is a separate, smaller follow-up — same intermediates mechanism, maps to a subset of the existing reset paths; file separately if wanted. Done when: claude/vim cursor-shape changes (insert vs normal mode) render as bar vs block in the terminal pane.', 'backlog', 'low', NULL, NULL, NULL, '2026-06-12 00:52:25', '2026-06-12 00:52:40', NULL, '5c07c49f5ad28c005a77ba26434e66d7', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); +INSERT INTO tickets (record_id, type, parent_record_id, title, description, status, priority, assigned_to, team, decision_ref, created_at, updated_at, deleted_at, hash, canonical_version) VALUES ('06FB3DKQQJ583944DG8561VQ3G', 'story', '06FB3DHCTP001YCHFP39XER0ZM', 'Phase 1 — backend transport seam (DaemonTransport, prove locally)', 'Model-independent backbone. Make the local app talk to its backend through a DaemonTransport abstraction with ZERO behavior change, so a remote transport can slot in later. New lib/src/ipc/transport.dart: DaemonTransport interface + LocalTransport (wraps today''s unix-socket connect). DaemonClient takes a transport instead of a hard-coded socket path; its reconnect loop generalizes to ''re-establish via transport''. lib/main.dart: swapIpcServer (~:119) becomes swapBackend — local binds the in-process IpcServer as today; seam ready for a remote branch. Wire protocol unchanged — JSON-lines IpcRequest/IpcResponse/IpcEvent (envelope.dart) already serialize over IPC; event stream, tail --events, events --since cursor-pull ride the same stream untouched. Verify: make test-core (IPC) + testmode skill — local path behaves identically through the seam, no regression. Depends on Phase 0 (T-330).', 'done', 'high', NULL, NULL, NULL, '2026-06-10 13:26:25', '2026-06-12 01:04:52', NULL, 'c36521dfda73828837c32d44eb2f6a4e', 2) ON CONFLICT(record_id) DO UPDATE SET type=excluded.type, parent_record_id=excluded.parent_record_id, title=excluded.title, description=excluded.description, status=excluded.status, priority=excluded.priority, assigned_to=excluded.assigned_to, team=excluded.team, decision_ref=excluded.decision_ref, updated_at=excluded.updated_at, deleted_at=excluded.deleted_at, hash=excluded.hash, canonical_version=excluded.canonical_version WHERE excluded.updated_at > tickets.updated_at OR (excluded.updated_at = tickets.updated_at AND excluded.hash > tickets.hash); diff --git a/lib/clide.dart b/lib/clide.dart index 188cec1d..8f46a7fa 100644 --- a/lib/clide.dart +++ b/lib/clide.dart @@ -24,6 +24,7 @@ export 'src/pql/client.dart' show PqlClient, PqlException; export 'src/ipc/envelope.dart'; export 'src/ipc/paths.dart'; export 'src/ipc/schema_v1.dart'; +export 'src/ipc/transport.dart' show DaemonTransport, DaemonConnection, LocalSocketTransport; export 'src/panes/event_sink.dart'; export 'src/panes/pane.dart' show Pane, PaneKind; export 'src/util/value_stream.dart' show ValueStream; diff --git a/lib/kernel/src/facade.dart b/lib/kernel/src/facade.dart index 4cad6b30..40dded4a 100644 --- a/lib/kernel/src/facade.dart +++ b/lib/kernel/src/facade.dart @@ -191,7 +191,7 @@ class KernelServices { isolateClient ?? (daemonClientFactory != null ? daemonClientFactory(log, events, arrangement, panels) - : DaemonClient( + : DaemonClient.unixSocket( // Legacy socket-client fallback — kept until T-127 // replaces it with the in-process socket loopback. // Today nothing in production hits this branch diff --git a/lib/kernel/src/ipc/client.dart b/lib/kernel/src/ipc/client.dart index f278cd4e..c0e8b7e9 100644 --- a/lib/kernel/src/ipc/client.dart +++ b/lib/kernel/src/ipc/client.dart @@ -1,6 +1,4 @@ import 'dart:async'; -import 'dart:convert'; -import 'dart:io'; import 'dart:math' as math; import 'package:clide/clide.dart'; @@ -10,14 +8,24 @@ import 'package:clide/kernel/src/log.dart'; import 'package:flutter/foundation.dart'; class DaemonClient extends ChangeNotifier { - DaemonClient({required String socketPath, required Logger log, required DaemonBus events}) : _socketPath = socketPath, _log = log, _events = events; + /// Connects through [transport] (T-331). The local app passes a + /// [LocalSocketTransport]; a remote workspace will pass an SSH-backed + /// transport without this class changing. + DaemonClient({required DaemonTransport transport, required Logger log, required DaemonBus events}) : _transport = transport, _log = log, _events = events; - String _socketPath; - String get socketPath => _socketPath; + /// Convenience for the local unix-socket path — today's only + /// production shape. + DaemonClient.unixSocket({required String socketPath, required Logger log, required DaemonBus events}) + : this(transport: LocalSocketTransport(socketPath), log: log, events: events); + + DaemonTransport _transport; + + /// The backend endpoint description — the unix socket path locally. + String get socketPath => _transport.endpoint; final Logger _log; final DaemonBus _events; - Socket? _socket; + DaemonConnection? _conn; bool _connected = false; bool _disposed = false; bool _started = false; @@ -48,30 +56,33 @@ class DaemonClient extends ChangeNotifier { _started = false; _reconnectTimer?.cancel(); _reconnectTimer = null; - final s = _socket; - _socket = null; - await s?.close(); + final c = _conn; + _conn = null; + await c?.close(); _failPending('client stopped'); _wakeConnectWaiters(); _setConnected(false); } - /// Point the client at a different socket path and reconnect. + /// Point the client at a different local socket path and reconnect. /// Used on project switch — the workspace-derived socket path /// (D-70) changes when the user opens a different project, so the - /// client follows. Cancels the reconnect timer, closes the live - /// socket (failing in-flight requests with `disconnect`), updates - /// the path, and re-arms the connect loop. Idempotent if the new - /// path equals the current one. - Future reconnectAt(String newPath) async { - if (newPath == _socketPath && _connected) return; - _socketPath = newPath; + /// client follows. Sugar over [reconnectWith]. + Future reconnectAt(String newPath) => reconnectWith(LocalSocketTransport(newPath)); + + /// Swap the backend transport and reconnect. Cancels the reconnect + /// timer, closes the live connection (failing in-flight requests with + /// `disconnect`), swaps the transport, and re-arms the connect loop. + /// Idempotent if the new endpoint equals the current connected one. + Future reconnectWith(DaemonTransport transport) async { + if (transport.endpoint == _transport.endpoint && _connected) return; + _transport = transport; _reconnectTimer?.cancel(); _reconnectTimer = null; - final s = _socket; - _socket = null; - await s?.close(); - _failPending('socket path changed'); + final c = _conn; + _conn = null; + await c?.close(); + _failPending('backend endpoint changed'); _setConnected(false); _disposed = false; _started = true; @@ -80,7 +91,7 @@ class DaemonClient extends ChangeNotifier { } Future request(String cmd, {Map args = const {}}) async { - if (!_connected || _socket == null) { + if (!_connected || _conn == null) { // A connection attempt is in flight (startup or reconnect) — wait // for it rather than failing instantly, so queries issued during // the startup window don't get a spurious not-connected error. @@ -88,7 +99,7 @@ class DaemonClient extends ChangeNotifier { if (_started && !_disposed) { await _awaitConnected(_connectWait); } - if (!_connected || _socket == null) { + if (!_connected || _conn == null) { return IpcResponse.err( id: '', error: IpcError(code: IpcExitCode.toolError, kind: IpcErrorKind.toolError, message: 'daemon not connected'), @@ -99,7 +110,7 @@ class DaemonClient extends ChangeNotifier { final completer = Completer(); _pending[id] = completer; final req = IpcRequest(id: id, cmd: cmd, args: args); - _socket!.writeln(req.encode()); + _conn!.writeLine(req.encode()); return completer.future; } @@ -126,30 +137,25 @@ class DaemonClient extends ChangeNotifier { } Future _connect() async { - // Already connected? Don't open a second socket. Guards against + // Already connected? Don't open a second connection. Guards against // racing connect attempts (e.g. start() arming the reconnect loop - // while swapIpcServer's reconnectAt connects on first boot). + // while swapBackend's reconnectAt connects on first boot). if (_disposed || _connected) return; try { - final addr = InternetAddress(_socketPath, type: InternetAddressType.unix); - final socket = await Socket.connect(addr, 0); - _socket = socket; + final conn = await _transport.open(); + _conn = conn; _backoff = const Duration(milliseconds: 200); _setConnected(true); - _log.info('ipc', 'connected to $_socketPath'); - socket - .cast>() - .transform(utf8.decoder) - .transform(const LineSplitter()) - .listen( - _handleLine, - onDone: _handleDisconnect, - onError: (Object e) { - _log.warn('ipc', 'socket error', error: e); - _handleDisconnect(); - }, - cancelOnError: true, - ); + _log.info('ipc', 'connected to ${_transport.endpoint}'); + conn.lines.listen( + _handleLine, + onDone: _handleDisconnect, + onError: (Object e) { + _log.warn('ipc', 'socket error', error: e); + _handleDisconnect(); + }, + cancelOnError: true, + ); } catch (e) { _log.debug('ipc', 'connect failed ($e); retry in ${_backoff.inMilliseconds}ms'); _scheduleReconnect(); @@ -175,7 +181,7 @@ class DaemonClient extends ChangeNotifier { } void _handleDisconnect() { - _socket = null; + _conn = null; _failPending('daemon disconnected'); _setConnected(false); _scheduleReconnect(); @@ -218,8 +224,9 @@ class DaemonClient extends ChangeNotifier { _disposed = true; _started = false; _reconnectTimer?.cancel(); - unawaited(_socket?.close()); - _socket = null; + final c = _conn; + if (c != null) unawaited(c.close()); + _conn = null; _failPending('client disposed'); _wakeConnectWaiters(); super.dispose(); diff --git a/lib/main.dart b/lib/main.dart index b2a3ec83..2923aae3 100644 --- a/lib/main.dart +++ b/lib/main.dart @@ -132,8 +132,8 @@ Future main() async { McpServer? mcpServer; final ipcLog = Logger(); - // IPC-server swaps must run one-at-a-time — see the swapIpcServer wrapper - // below doSwapIpcServer for why. (T-352) + // Backend swaps must run one-at-a-time — see the swapBackend wrapper + // below doSwapBackend for why. (T-352) Future swapChain = Future.value(); // Teardown of the service set behind the currently-served dispatcher @@ -142,7 +142,7 @@ Future main() async { // workspace's watchers into the new one's bus (T-367). Future Function()? activeSubsystemTeardown; - Future doSwapIpcServer(DaemonDispatcher dispatcher, Future Function() teardown, Directory workRoot) async { + Future doSwapBackend(DaemonDispatcher dispatcher, Future Function() teardown, Directory workRoot) async { if (kIsWeb) return; // Already serving this exact workspace? Reuse the live server. // The startup factory binds the launch CWD, then the project-open @@ -216,8 +216,8 @@ Future main() async { // load (stale/global pql.db) yet working after a manual refresh. Chaining // every swap makes them apply in call order; the repo swap is issued last // and therefore wins. (T-352) - Future swapIpcServer(DaemonDispatcher dispatcher, Future Function() teardown, Directory workRoot) { - final next = swapChain.then((_) => doSwapIpcServer(dispatcher, teardown, workRoot)); + Future swapBackend(DaemonDispatcher dispatcher, Future Function() teardown, Directory workRoot) { + final next = swapChain.then((_) => doSwapBackend(dispatcher, teardown, workRoot)); // A failed swap must not break the chain for the next one. swapChain = next.catchError((Object _) {}); return next; @@ -350,20 +350,20 @@ Future main() async { final workRoot = startupWorkRoot; final (dispatcher, teardown) = buildDispatcher(events, toolchain, workRoot, arrangement, panels); // Build the client at the workspace's socket path. The - // server is started below (swapIpcServer) which the + // server is started below (swapBackend) which the // client will then auto-connect to via its reconnect // loop. autoStartDaemonClient:false means we own the // lifecycle here. - final client = DaemonClient(socketPath: workspaceSocketPath(workRoot.path), log: log, events: events); + final client = DaemonClient.unixSocket(socketPath: workspaceSocketPath(workRoot.path), log: log, events: events); ipcClient = client; // start() synchronously marks the client "connecting" (so // requests issued during the startup window park for the // socket instead of failing) and arms the reconnect loop. - // swapIpcServer then binds the server and reconnectAt makes + // swapBackend then binds the server and reconnectAt makes // the connect immediate. _connect's already-connected guard // keeps these two paths from opening a second socket. unawaited(client.start()); - unawaited(swapIpcServer(dispatcher, teardown, workRoot)); + unawaited(swapBackend(dispatcher, teardown, workRoot)); return client; }, onProjectOpen: kIsWeb @@ -374,7 +374,7 @@ Future main() async { final panels = kernelPanels; if (bus == null || arrangement == null || panels == null) return; final (dispatcher, teardown) = buildDispatcher(bus, toolchain, Directory(path), arrangement, panels); - await swapIpcServer(dispatcher, teardown, Directory(path)); + await swapBackend(dispatcher, teardown, Directory(path)); }, ); // Expose the reader nav to the `clide status` snapshot (T-221). Boot diff --git a/lib/src/ipc/transport.dart b/lib/src/ipc/transport.dart new file mode 100644 index 00000000..4e55166c --- /dev/null +++ b/lib/src/ipc/transport.dart @@ -0,0 +1,73 @@ +/// DaemonTransport (T-331): the seam between the local app and its +/// backend. The UI's [DaemonClient] talks JSON-lines through a +/// [DaemonTransport] instead of a hard-coded unix-socket connect, so a +/// remote transport (SSH-tunnelled agent socket or ssh-exec channel, +/// T-329/Q-23) can slot in without touching the client's correlation, +/// reconnect, or event-forwarding logic. +/// +/// The wire protocol is unchanged either way: one JSON envelope +/// (IpcRequest/IpcResponse/IpcEvent, see envelope.dart) per line. +/// +/// Kept Flutter-free — this file runs under plain `dart test`. +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +/// How the app reaches its backend. Implementations own endpoint +/// resolution + connection establishment; the caller owns retry policy +/// (the client's backoff loop calls [open] again after a failure). +abstract interface class DaemonTransport { + /// Stable, human-readable endpoint description — the unix socket path + /// locally, a `ssh://host/path` form remotely. Used for logs, status + /// surfaces, and same-endpoint reconnect short-circuits. + String get endpoint; + + /// Establish one connection. Throws on failure (caller retries). + Future open(); +} + +/// One live backend connection carrying JSON-lines both ways. +abstract interface class DaemonConnection { + /// Incoming lines, one JSON envelope each. Done/error signals the + /// connection dropped. + Stream get lines; + + /// Send one JSON envelope line (the newline is appended here). + void writeLine(String line); + + Future close(); +} + +/// Today's path: connect to the workspace-derived unix domain socket +/// (D-70) the in-process IpcServer is bound to. +class LocalSocketTransport implements DaemonTransport { + LocalSocketTransport(this.socketPath); + + final String socketPath; + + @override + String get endpoint => socketPath; + + @override + Future open() async { + final addr = InternetAddress(socketPath, type: InternetAddressType.unix); + return _SocketConnection(await Socket.connect(addr, 0)); + } +} + +class _SocketConnection implements DaemonConnection { + _SocketConnection(this._socket); + + final Socket _socket; + + @override + Stream get lines => _socket.cast>().transform(utf8.decoder).transform(const LineSplitter()); + + @override + void writeLine(String line) => _socket.writeln(line); + + @override + Future close() => _socket.close(); +} diff --git a/test/helpers/fake_ipc.dart b/test/helpers/fake_ipc.dart index 337cf9e5..1e607dfb 100644 --- a/test/helpers/fake_ipc.dart +++ b/test/helpers/fake_ipc.dart @@ -4,7 +4,7 @@ import 'package:clide/kernel/kernel.dart'; /// A DaemonClient that doesn't actually open a socket. Use in tests /// that need a connected-state observable but not a real daemon. class FakeDaemonClient extends DaemonClient { - FakeDaemonClient({required super.log, required super.events}) : super(socketPath: '/dev/null/fake-clide.sock'); + FakeDaemonClient({required super.log, required super.events}) : super.unixSocket(socketPath: '/dev/null/fake-clide.sock'); bool _fakeConnected = false; final Map Function(Map)> _stubs = {}; diff --git a/test/ipc/transport_test.dart b/test/ipc/transport_test.dart new file mode 100644 index 00000000..f860054f --- /dev/null +++ b/test/ipc/transport_test.dart @@ -0,0 +1,73 @@ +/// Tests for `lib/src/ipc/transport.dart` (T-331) — the DaemonTransport +/// seam. Runs under plain `dart test` (core suite): no Flutter imports. +library; + +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:clide/src/ipc/transport.dart'; +import 'package:test/test.dart'; + +void main() { + group('LocalSocketTransport', () { + late Directory dir; + late String path; + late ServerSocket server; + + setUp(() async { + dir = await Directory.systemTemp.createTemp('clide-transport-'); + path = '${dir.path}/sock'; + server = await ServerSocket.bind(InternetAddress(path, type: InternetAddressType.unix), 0); + }); + + tearDown(() async { + await server.close(); + await dir.delete(recursive: true); + }); + + test('endpoint reports the socket path', () { + expect(LocalSocketTransport(path).endpoint, path); + }); + + test('open connects; lines round-trip both directions', () async { + final accepted = Completer(); + server.listen((s) => accepted.complete(s)); + + final conn = await LocalSocketTransport(path).open(); + final serverSide = await accepted.future; + + // client -> server + final serverLines = serverSide.cast>().transform(utf8.decoder).transform(const LineSplitter()); + final firstLine = serverLines.first; + conn.writeLine('{"hello":1}'); + expect(await firstLine.timeout(const Duration(seconds: 2)), '{"hello":1}'); + + // server -> client + final clientLine = conn.lines.first; + serverSide.writeln('{"world":2}'); + expect(await clientLine.timeout(const Duration(seconds: 2)), '{"world":2}'); + + await conn.close(); + await serverSide.close(); + }); + + test('open throws when nothing is bound (caller owns retry)', () async { + final t = LocalSocketTransport('${dir.path}/no-such.sock'); + await expectLater(t.open(), throwsA(isA())); + }); + + test('lines closes when the server drops the connection', () async { + final accepted = Completer(); + server.listen((s) => accepted.complete(s)); + + final conn = await LocalSocketTransport(path).open(); + final serverSide = await accepted.future; + + final done = conn.lines.drain(); + await serverSide.close(); + await done.timeout(const Duration(seconds: 2)); + await conn.close(); + }); + }); +} diff --git a/test/kernel/src/ipc/client_test.dart b/test/kernel/src/ipc/client_test.dart index b238f3be..31e2fbe6 100644 --- a/test/kernel/src/ipc/client_test.dart +++ b/test/kernel/src/ipc/client_test.dart @@ -56,13 +56,50 @@ Future _tmpSocket() async { } DaemonClient _build(String socketPath, DaemonBus bus) { - return DaemonClient( + return DaemonClient.unixSocket( socketPath: socketPath, log: Logger(minLevel: LogLevel.error, sinks: const []), events: bus, ); } +/// In-memory transport (T-331): proves DaemonClient runs unmodified over +/// any [DaemonTransport], not just the unix socket — the seam the remote +/// backend (T-329) slots into. +class _MemoryTransport implements DaemonTransport { + final toClient = StreamController.broadcast(); + final fromClient = StreamController.broadcast(); + int opens = 0; + + @override + String get endpoint => 'memory://test'; + + @override + Future open() async { + opens++; + return _MemoryConnection(this); + } + + Future close() async { + await toClient.close(); + await fromClient.close(); + } +} + +class _MemoryConnection implements DaemonConnection { + _MemoryConnection(this._t); + final _MemoryTransport _t; + + @override + Stream get lines => _t.toClient.stream; + + @override + void writeLine(String line) => _t.fromClient.add(line); + + @override + Future close() async {} +} + void main() { group('DaemonClient — happy path', () { test('connect → request → matching response completes', () async { @@ -341,4 +378,63 @@ void main() { expect(resp.ok, isTrue); }); }); + + group('DaemonClient — transport seam (T-331)', () { + test('request/response round-trips over a non-socket transport', () async { + final transport = _MemoryTransport(); + addTearDown(transport.close); + final bus = DaemonBus(); + addTearDown(bus.dispose); + final client = DaemonClient( + transport: transport, + log: Logger(minLevel: LogLevel.error, sinks: const []), + events: bus, + ); + addTearDown(client.dispose); + + await client.start(); + expect(transport.opens, 1); + expect(client.isConnected, isTrue); + expect(client.socketPath, 'memory://test'); + + final lineFuture = transport.fromClient.stream.first; + final respFuture = client.request('ping', args: {'n': 1}); + final line = await lineFuture; + final req = IpcMessage.decode(line) as IpcRequest; + expect(req.cmd, 'ping'); + transport.toClient.add(IpcResponse.ok(id: req.id, data: const {'pong': true}).encode()); + final resp = await respFuture.timeout(const Duration(seconds: 2)); + expect(resp.ok, isTrue); + expect(resp.data['pong'], isTrue); + }); + + test('reconnectWith swaps from a socket transport to another transport', () async { + final path = await _tmpSocket(); + final daemon = _TestDaemon(path); + await daemon.start(); + addTearDown(daemon.close); + + final bus = DaemonBus(); + addTearDown(bus.dispose); + final client = _build(path, bus); + addTearDown(client.dispose); + await client.start(); + await daemon.waitForClient(); + await Future.delayed(const Duration(milliseconds: 50)); + expect(client.isConnected, isTrue); + + final transport = _MemoryTransport(); + addTearDown(transport.close); + await client.reconnectWith(transport); + expect(client.isConnected, isTrue); + expect(client.socketPath, 'memory://test'); + + // Requests now flow over the new transport, not the old socket. + final lineFuture = transport.fromClient.stream.first; + final respFuture = client.request('over-memory'); + final req = IpcMessage.decode(await lineFuture) as IpcRequest; + transport.toClient.add(IpcResponse.ok(id: req.id, data: const {}).encode()); + expect((await respFuture).ok, isTrue); + }); + }); }