diff --git a/packages/celest_auth/example/lib/main.dart b/packages/celest_auth/example/lib/main.dart
index 8bb4f52d..5eccfc89 100644
--- a/packages/celest_auth/example/lib/main.dart
+++ b/packages/celest_auth/example/lib/main.dart
@@ -13,7 +13,7 @@ final auth = CelestAuth(
   httpClient: http.Client(),
 );
 
-final class CelestAuth extends AuthImpl with Passkeys, Email {
+final class CelestAuth extends AuthImpl with Passkeys, EmailProvider {
   CelestAuth({required super.baseUri, required super.httpClient});
 }
 
diff --git a/packages/celest_auth/lib/src/auth.dart b/packages/celest_auth/lib/src/auth.dart
index 2825b3de..94f84bad 100644
--- a/packages/celest_auth/lib/src/auth.dart
+++ b/packages/celest_auth/lib/src/auth.dart
@@ -1,10 +1,13 @@
+import 'dart:async';
+
 import 'package:celest_auth/src/flows/auth_flow.dart';
 import 'package:celest_auth/src/platform/auth_platform.dart';
+import 'package:celest_auth/src/state/auth_state.dart';
 import 'package:celest_core/celest_core.dart';
 // ignore: implementation_imports
-import 'package:celest_core/src/storage/secure/secure_storage.dart';
+import 'package:celest_core/src/storage/local/local_storage.dart';
 // ignore: implementation_imports
-import 'package:celest_core/src/storage/storage.dart';
+import 'package:celest_core/src/storage/secure/secure_storage.dart';
 import 'package:http/http.dart' as http;
 import 'package:meta/meta.dart';
 
@@ -13,7 +16,10 @@ import 'package:meta/meta.dart';
 ///
 /// Generated Celest clients extend this class and mix in the various
 /// [AuthFlow]s supported by the backend.
-abstract interface class Auth {}
+abstract interface class Auth {
+  AuthState get authState;
+  Stream<AuthState> get authStateChanges;
+}
 
 abstract base class AuthImpl implements Auth {
   AuthImpl({
@@ -23,19 +29,76 @@ abstract base class AuthImpl implements Auth {
     init();
   }
 
+  @override
+  AuthState get authState => _authState;
+  AuthState _authState =
+      const Unauthenticated(); // TODO(dnys1): const AuthInitializing();
+
+  @override
+  Stream<AuthState> get authStateChanges => _authStateController.stream;
+
+  final StreamController<AuthState> _authStateController =
+      StreamController.broadcast(sync: true);
+
+  late final StreamSubscription<AuthState> _authStateSubscription;
+  StreamSubscription<AuthFlowState>? _authFlowSubscription;
+
   /// Initializes Celest Auth.
   ///
   /// Must be called before any other getters or methods are accessed.
   @mustCallSuper
-  void init() {}
+  void init() {
+    _authStateSubscription =
+        authStateChanges.listen((state) => _authState = state);
+  }
+
+  Future<StreamSink<FlowState>>
+      requestFlow<FlowState extends AuthFlowState>() async {
+    switch (_authState) {
+      case AuthInitializing():
+        await authStateChanges.firstWhere(
+          (state) => state is! AuthInitializing,
+        );
+        return requestFlow();
+      case AuthFlowState():
+        throw Exception('Auth flow already in progress');
+      case Authenticated():
+        throw Exception(
+          'User is already authenticated. Sign out before continuing.',
+        );
+      case Unauthenticated() || NeedsReauthentication():
+        assert(_authFlowSubscription == null);
+        final previousState = _authState;
+        final controller = StreamController<FlowState>(
+          onCancel: () => _authStateController.add(previousState),
+        );
+        _authFlowSubscription = controller.stream.listen(
+          (state) => _authStateController.add(state),
+          onError: (error, stackTrace) {
+            // TODO(dnys1)
+          },
+          onDone: () => _authFlowSubscription = null,
+          cancelOnError: true,
+        );
+        return controller.sink;
+    }
+  }
 
   final Uri baseUri;
   final http.Client httpClient;
-  final Storage secureStorage = SecureStorage();
+
+  final LocalStorage localStorage = LocalStorage();
+  final SecureStorage secureStorage = SecureStorage();
 
   late final AuthClient protocol = AuthClient(
     baseUri: baseUri,
     httpClient: httpClient,
   );
   late final AuthPlatform platform = AuthPlatform(protocol: protocol);
+
+  Future<void> close() async {
+    await _authStateSubscription.cancel();
+    await _authFlowSubscription?.cancel();
+    await _authStateController.close();
+  }
 }
diff --git a/packages/celest_auth/lib/src/flows/auth_flow.dart b/packages/celest_auth/lib/src/flows/auth_flow.dart
index b09ed371..bdda1a33 100644
--- a/packages/celest_auth/lib/src/flows/auth_flow.dart
+++ b/packages/celest_auth/lib/src/flows/auth_flow.dart
@@ -4,4 +4,6 @@ library;
 import 'package:meta/meta.dart';
 
 /// Base type for all authentication flows.
-abstract interface class AuthFlow {}
+abstract interface class AuthFlow {
+  void cancel();
+}
diff --git a/packages/celest_auth/lib/src/flows/email_flow.dart b/packages/celest_auth/lib/src/flows/email_flow.dart
index f7d00fe1..aef78db9 100644
--- a/packages/celest_auth/lib/src/flows/email_flow.dart
+++ b/packages/celest_auth/lib/src/flows/email_flow.dart
@@ -2,51 +2,78 @@ import 'dart:async';
 
 import 'package:celest_auth/src/auth.dart';
 import 'package:celest_auth/src/flows/auth_flow.dart';
+import 'package:celest_auth/src/state/auth_state.dart';
 import 'package:celest_core/celest_core.dart';
 import 'package:state_notifier/state_notifier.dart';
 
-base mixin Email on AuthImpl {
-  late final EmailFlow email = EmailFlow(this);
+base mixin EmailProvider on AuthImpl {
+  late final Email email = Email(this);
 }
 
-final class EmailFlow implements AuthFlow {
-  EmailFlow(this._hub);
+final class Email {
+  Email(this._hub);
 
   final AuthImpl _hub;
-  EmailProtocol get _protocol => _hub.protocol.email;
 
   Future<EmailSignUpNeedsVerification> signUp({
     required String email,
   }) async {
-    final parameters = await _protocol.sendOtp(
-      request: OtpSendRequest(email: email),
-    );
-    return EmailSignUpNeedsVerification(
-      flow: this,
-      email: email,
-      parameters: parameters,
-    );
+    final flowSink = await _hub.requestFlow<EmailFlowState>();
+    final flow = EmailFlow._(_hub, _EmailFlowController(flowSink));
+    return flow._signUp(email: email);
+  }
+}
+
+final class EmailFlow implements AuthFlow {
+  EmailFlow._(this._hub, this._flowController);
+
+  final AuthImpl _hub;
+  final _EmailFlowController _flowController;
+
+  EmailProtocol get _protocol => _hub.protocol.email;
+
+  Future<EmailSignUpNeedsVerification> _signUp({
+    required String email,
+  }) {
+    return _flowController.capture(() async {
+      final parameters = await _protocol.sendOtp(
+        request: OtpSendRequest(email: email),
+      );
+      return EmailSignUpNeedsVerification(
+        flow: this,
+        email: email,
+        parameters: parameters,
+      );
+    });
   }
 
-  Future<User> _verifyOtp({
+  Future<EmailAuthenticated> _verifyOtp({
     required String email,
     required String otp,
-  }) async {
-    final user = await _protocol.verifyOtp(
-      verification: OtpVerifyRequest(email: email, otp: otp),
-    );
-    _hub.secureStorage.write('cork', user.cork);
-    return user.user;
+  }) {
+    return _flowController.capture(() async {
+      final user = await _protocol.verifyOtp(
+        verification: OtpVerifyRequest(email: email, otp: otp),
+      );
+      _hub.secureStorage.write('cork', user.cork);
+      _hub.localStorage.write('userId', user.user.userId);
+      return EmailAuthenticated(flow: this, user: user.user);
+    });
+  }
+
+  @override
+  void cancel() {
+    // TODO(dnys1)
+    throw UnimplementedError();
   }
 }
 
-final class EmailSignUpNeedsVerification {
+final class EmailSignUpNeedsVerification extends EmailFlowState {
   EmailSignUpNeedsVerification({
-    required EmailFlow flow,
+    required super.flow,
     required this.email,
     required OtpParameters parameters,
-  })  : _flow = flow,
-        _parameters = parameters {
+  }) : _parameters = parameters {
     _resendCountdown =
         Stream<void>.periodic(const Duration(seconds: 1)).listen((_) {
       final resendIn = canResendIn;
@@ -54,7 +81,6 @@ final class EmailSignUpNeedsVerification {
     });
   }
 
-  final EmailFlow _flow;
   final String email;
   OtpParameters _parameters;
 
@@ -72,7 +98,7 @@ final class EmailSignUpNeedsVerification {
     _resendCountdown?.pause();
     _resendEligible.setState(false);
     try {
-      _parameters = await _flow._protocol.resendOtp(
+      _parameters = await flow._protocol.resendOtp(
         request: OtpSendRequest(email: email),
       );
     } finally {
@@ -82,9 +108,9 @@ final class EmailSignUpNeedsVerification {
   }
 
   Future<User> verifyOtp(String otp) async {
-    final user = await _flow._verifyOtp(email: email, otp: otp);
+    final authenticated = await flow._verifyOtp(email: email, otp: otp);
     _close();
-    return user;
+    return authenticated.user;
   }
 
   void _close() {
@@ -101,3 +127,18 @@ final class _ResendEligibleNotifier extends StateNotifier<(bool, Duration)> {
     this.state = (state, canResendIn ?? this.state.$2);
   }
 }
+
+extension type _EmailFlowController(StreamSink<EmailFlowState> _sink) {
+  Future<R> capture<R extends EmailFlowState>(
+    Future<R> Function() action,
+  ) async {
+    try {
+      final result = await action();
+      _sink.add(result);
+      return result;
+    } on Object catch (e, st) {
+      _sink.addError(e, st);
+      rethrow;
+    }
+  }
+}
diff --git a/packages/celest_auth/lib/src/flows/passkey_flow.dart b/packages/celest_auth/lib/src/flows/passkey_flow.dart
index 054d3d4f..8b341233 100644
--- a/packages/celest_auth/lib/src/flows/passkey_flow.dart
+++ b/packages/celest_auth/lib/src/flows/passkey_flow.dart
@@ -39,5 +39,6 @@ final class PasskeyFlow implements AuthFlow {
   }
 
   /// Cancels the in-progress passkey operation, if any.
+  @override
   void cancel() => _platform.cancel();
 }
diff --git a/packages/celest_auth/lib/src/state/auth_state.dart b/packages/celest_auth/lib/src/state/auth_state.dart
index 7ff9d406..26bcd028 100644
--- a/packages/celest_auth/lib/src/state/auth_state.dart
+++ b/packages/celest_auth/lib/src/state/auth_state.dart
@@ -1,3 +1,53 @@
+import 'package:celest_auth/src/flows/auth_flow.dart';
+import 'package:celest_auth/src/flows/email_flow.dart';
+import 'package:celest_core/celest_core.dart';
+
 sealed class AuthState {
   const AuthState();
 }
+
+final class AuthInitializing extends AuthState {
+  const AuthInitializing();
+}
+
+final class Unauthenticated extends AuthState {
+  const Unauthenticated();
+}
+
+final class NeedsReauthentication extends AuthState {
+  const NeedsReauthentication({
+    required this.userId,
+  });
+
+  final String userId;
+}
+
+sealed class AuthFlowState<Flow extends AuthFlow> extends AuthState {
+  const AuthFlowState({
+    required this.flow,
+  });
+
+  final Flow flow;
+
+  void cancel() => flow.cancel();
+}
+
+class EmailFlowState extends AuthFlowState<EmailFlow> {
+  const EmailFlowState({required super.flow});
+}
+
+final class EmailAuthenticated extends EmailFlowState implements Authenticated {
+  const EmailAuthenticated({
+    required super.flow,
+    required this.user,
+  });
+
+  @override
+  final User user;
+}
+
+final class Authenticated extends AuthState {
+  const Authenticated(this.user);
+
+  final User user;
+}