Announce engine closure on disposal and terminal reconnect failure

.changes/buffer-wait-typed-error

@@ -0,0 +1 @@
+patch type="fixed" "Data stream writes on a closed engine fail with UnexpectedConnectionState instead of a plain string error"

.changes/engine-terminal-closure

@@ -0,0 +1 @@
+patch type="fixed" "Announce engine closure before room disposal callbacks and when reconnection gives up, and reject data channel buffer waits started after closure"

lib/src/core/engine.dart

@@ -242,14 +242,20 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
     _setUpSignalListeners();
 
     onDispose(() async {
-      _isClosed = true;
       await cleanUp();
       await events.dispose();
       await _signalListener.dispose();
       _reliableReceivedState.dispose();
     });
   }
 
+  @override
+  Future<bool> dispose() {
+    // Disposal callbacks run in reverse registration order and may await engine work.
+    markClosed();
+    return super.dispose();
+  }
+
   Future<void> connect(
     String url,
     String token, {
@@ -417,14 +423,19 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
   }
 
   Future<void> waitForBufferStatusLow(Reliability kind) async {
+    // A wait started after closure cannot observe the closing event.
+    if (_isClosed || isDisposed) {
+      throw UnexpectedConnectionState('Engine disconnected');
+    }
+
     final Completer<void> completer = Completer();
 
     if (isBufferStatusLow(kind) == true) {
       completer.complete();
     } else {
       onClosing() {
         if (!completer.isCompleted) {
-          completer.completeError('Engine disconnected');
+          completer.completeError(UnexpectedConnectionState('Engine disconnected'));
         }
       }
 
@@ -1151,7 +1162,7 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
 
     if (_reconnectAttempts >= _reconnectCount) {
       logger.fine('reconnectAttempts exceeded, disconnecting...');
-      _isClosed = true;
+      markClosed();
       await cleanUp();
 
       events.emit(
@@ -1263,6 +1274,7 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
         unawaited(handleReconnect(ClientDisconnectReason.reconnectRetry));
       } else {
         logger.fine('attemptReconnect: disconnecting...');
+        markClosed();
         // clean up before emitting, room's EngineDisconnectedEvent handler
         // drops the event while fullReconnectOnNext is still true and
         // cleanUp() is what resets it
@@ -1677,11 +1689,19 @@ class Engine extends Disposable with EventsEmittable<EngineEvent> {
       events.emit(EngineRoomMovedEvent(response: event.response));
     });
 
+  // Announce terminal closure before tearing down transports or event listeners.
+  // A full reconnect also disposes transports, but must keep the engine open.
+  @internal
+  void markClosed() {
+    if (_isClosed) return;
+    _isClosed = true;
+    events.emit(EngineClosingEvent());
+  }
+
   Future<void> disconnect({
     DisconnectReason reason = DisconnectReason.clientInitiated,
   }) async {
-    _isClosed = true;
-    events.emit(EngineClosingEvent());
+    markClosed();
     if (connectionState == ConnectionState.connected) {
       await signalClient.sendLeave();
     } else {

lib/src/core/room.dart

@@ -241,6 +241,13 @@ class Room extends DisposableChangeNotifier with EventsEmittable<RoomEvent> {
     });
   }
 
+  @override
+  Future<bool> dispose() {
+    // Room cleanup tears down transports before Engine.dispose() is reached.
+    engine.markClosed();
+    return super.dispose();
+  }
+
   /// prepareConnection should be called as soon as the page is loaded, in order
   /// to speed up the connection attempt. This function will
   /// - perform DNS resolution and pre-warm the DNS cache

test/core/disconnect_during_reconnect_test.dart

@@ -12,9 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// disconnect() closes the engine for good, even when a full reconnect is
-// already tearing down the old connection and is about to connect again, and
-// the room can connect again afterwards.
+// disconnect() and Room.dispose() close the engine for good, even when a full
+// reconnect is already tearing down the old connection and is about to connect
+// again, and the room can connect again after disconnect().
 
 @Timeout(Duration(seconds: 10))
 library;
@@ -93,4 +93,35 @@ void main() {
     await container.connectRoom();
     expect(room.connectionState, ConnectionState.connected);
   });
+
+  test('disposing the room during a full reconnect does not connect again', () async {
+    final restartPaused = Completer<void>();
+    final resumeRestart = Completer<void>();
+    container.engine.publisher!.onDispose(() async {
+      restartPaused.complete();
+      await resumeRestart.future;
+    });
+    final disposePaused = Completer<void>();
+    final resumeDispose = Completer<void>();
+    container.engine.subscriber!.onDispose(() async {
+      disposePaused.complete();
+      await resumeDispose.future;
+    });
+
+    container.engine.fullReconnectOnNext = true;
+    final droppedHandlers = ws.handlers;
+    ws.onDispose();
+    await restartPaused.future;
+
+    // The room's teardown reaches the subscriber while the full reconnect still holds the publisher.
+    final disposing = room.dispose();
+    await disposePaused.future;
+    resumeRestart.complete();
+    // Give a reconnect that was not stopped time to open its socket.
+    await Future<void>.delayed(const Duration(milliseconds: 500));
+    resumeDispose.complete();
+    await disposing;
+
+    expect(identical(ws.handlers, droppedHandlers), isTrue, reason: 'the full reconnect opened a new signal socket');
+  });
 }

test/core/engine_closing_test.dart

@@ -0,0 +1,179 @@
+// Copyright 2026 LiveKit, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+@Timeout(Duration(seconds: 5))
+library;
+
+import 'dart:async';
+
+import 'package:flutter_test/flutter_test.dart';
+
+import 'package:livekit_client/livekit_client.dart';
+import 'package:livekit_client/src/core/engine.dart';
+import 'package:livekit_client/src/internal/events.dart';
+import 'package:livekit_client/src/types/internal.dart';
+import '../mock/e2e_container.dart';
+import '../mock/peerconnection_mock.dart';
+
+void main() {
+  TestWidgetsFlutterBinding.ensureInitialized();
+
+  late E2EContainer container;
+  late List<EngineEvent> events;
+
+  setUp(() async {
+    resetMockDataChannels();
+    container = E2EContainer();
+    await container.connectRoom();
+    events = [];
+    container.engine.events.listen(events.add);
+  });
+
+  tearDown(() async {
+    await container.dispose();
+  });
+
+  test('disposing the room announces engine closure', () async {
+    await container.room.dispose();
+
+    expect(events.whereType<EngineClosingEvent>(), hasLength(1));
+    expect(container.engine.isClosed, isTrue);
+  });
+
+  test('room closure is announced while the first transport teardown is blocked', () async {
+    final teardownStarted = Completer<void>();
+    final releaseTeardown = Completer<void>();
+    container.engine.publisher!.onDispose(() async {
+      teardownStarted.complete();
+      await releaseTeardown.future;
+    });
+
+    final disposing = container.room.dispose();
+    await teardownStarted.future;
+    final closedDuringTeardown = container.engine.isClosed;
+    final closingEventsDuringTeardown = events.whereType<EngineClosingEvent>().length;
+    final emitterDisposedDuringTeardown = container.engine.events.isDisposed;
+    releaseTeardown.complete();
+    await disposing;
+
+    expect(closedDuringTeardown, isTrue);
+    expect(closingEventsDuringTeardown, 1);
+    expect(emitterDisposedDuringTeardown, isFalse);
+  });
+
+  for (final owner in ['room', 'engine']) {
+    test('$owner is marked closed before later disposal callbacks run', () async {
+      final disposable = owner == 'room' ? container.room : container.engine;
+      bool? closedInCallback;
+      disposable.onDispose(() async {
+        closedInCallback = container.engine.isClosed;
+      });
+
+      if (owner == 'room') {
+        await container.room.dispose();
+      } else {
+        await container.engine.dispose();
+      }
+
+      expect(closedInCallback, isTrue);
+      expect(events.whereType<EngineClosingEvent>(), hasLength(1));
+    });
+  }
+
+  test('engine closure is delivered while disposing with its emitter still alive', () async {
+    bool? engineDisposedWhenDelivered;
+    bool? emitterDisposedWhenDelivered;
+    container.engine.events.once<EngineClosingEvent>((_) {
+      engineDisposedWhenDelivered = container.engine.isDisposed;
+      emitterDisposedWhenDelivered = container.engine.events.isDisposed;
+    });
+
+    await container.engine.dispose();
+
+    expect(engineDisposedWhenDelivered, isTrue);
+    expect(emitterDisposedWhenDelivered, isFalse);
+    expect(container.engine.events.isDisposed, isTrue);
+  });
+
+  test('an unrecoverable reconnect announces closure before disconnection', () async {
+    container.wsConnector.connectError = CertificatePinningException(
+      'Certificate pin mismatch',
+      host: 'www.example.com',
+    );
+    final disconnected = container.engine.events.waitFor<EngineDisconnectedEvent>(
+      duration: const Duration(seconds: 1),
+    );
+
+    await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry);
+    expect((await disconnected).reason, DisconnectReason.signalingConnectionFailure);
+
+    expect(container.engine.isClosed, isTrue);
+    expect(events.whereType<EngineClosingEvent>(), hasLength(1));
+    expect(
+      events.indexWhere((event) => event is EngineClosingEvent),
+      lessThan(events.indexWhere((event) => event is EngineDisconnectedEvent)),
+    );
+  });
+
+  test('exhausting reconnect attempts announces engine closure', () async {
+    // A generic resume failure stays recoverable without escalating to a full reconnect.
+    container.wsConnector.connectError = Exception('temporary connection failure');
+    final disconnected = container.engine.events.waitFor<EngineDisconnectedEvent>(
+      duration: const Duration(seconds: 1),
+    );
+
+    // Drive the attempts directly rather than waiting through the retry delays.
+    for (var i = 0; i < defaultRetryDelaysInMs.length; i++) {
+      await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry);
+    }
+    expect((await disconnected).reason, DisconnectReason.reconnectAttemptsExceeded);
+
+    expect(container.engine.isClosed, isTrue);
+    expect(events.whereType<EngineClosingEvent>(), hasLength(1));
+    expect(
+      events.indexWhere((event) => event is EngineClosingEvent),
+      lessThan(events.indexWhere((event) => event is EngineDisconnectedEvent)),
+    );
+  });
+
+  test('a recoverable reconnect does not announce engine closure', () async {
+    container.wsConnector.connectError = Exception('temporary connection failure');
+
+    await container.engine.attemptReconnect(ClientDisconnectReason.reconnectRetry);
+
+    expect(container.engine.isClosed, isFalse);
+    expect(events.whereType<EngineClosingEvent>(), isEmpty);
+    expect(events.whereType<EngineDisconnectedEvent>(), isEmpty);
+  });
+
+  test('a buffer wait started after disposal rejects even when the channel reports low', () async {
+    await container.room.dispose();
+
+    await expectLater(
+      container.engine.waitForBufferStatusLow(Reliability.reliable),
+      throwsA(isA<UnexpectedConnectionState>()),
+    );
+  });
+
+  test('disconnect followed by disposal announces closure only once', () async {
+    await container.engine.disconnect();
+    await expectLater(
+      container.engine.waitForBufferStatusLow(Reliability.reliable),
+      throwsA(isA<UnexpectedConnectionState>()),
+    );
+    await container.room.dispose();
+
+    expect(events.whereType<EngineClosingEvent>(), hasLength(1));
+  });
+}