diff --git a/crates/jacquard-axum/tests/service_auth_tests.rs b/crates/jacquard-axum/tests/service_auth_tests.rs
index cddb97c03..eea796b87 100644
--- a/crates/jacquard-axum/tests/service_auth_tests.rs
+++ b/crates/jacquard-axum/tests/service_auth_tests.rs
@@ -15,7 +15,7 @@ use jacquard_axum::service_auth::{
 };
 use jacquard_common::{
     bos::BosStr,
-    deps::smol_str::SmolStr,
+    deps::smol_str::{SmolStr, format_smolstr},
     service_auth::JwtHeader,
     types::{
         did::Did,
@@ -114,7 +114,7 @@ fn create_test_did_doc(did: &str, public_key: &k256::ecdsa::VerifyingKey) -> Did
         id: Did::new_owned(did).unwrap(),
         also_known_as: None,
         verification_method: Some(vec![VerificationMethod {
-            id: SmolStr::from(format!("{}#atproto", did)),
+            id: format_smolstr!("{}#atproto", did),
             r#type: SmolStr::new_static("Multikey"),
             controller: Some(SmolStr::from(did)),
             public_key_multibase: Some(SmolStr::from(multibase_key)),
diff --git a/crates/jacquard-common/src/session.rs b/crates/jacquard-common/src/session.rs
index 980475b8d..62d683df3 100644
--- a/crates/jacquard-common/src/session.rs
+++ b/crates/jacquard-common/src/session.rs
@@ -1,20 +1,22 @@
 //! Generic session storage traits and utilities.
 
 use alloc::boxed::Box;
-#[cfg(feature = "std")]
-use alloc::string::ToString;
+use alloc::collections::BTreeMap;
+use alloc::string::String;
 use alloc::sync::Arc;
+use alloc::vec::Vec;
 use core::error::Error as StdError;
-#[cfg(feature = "std")]
-use core::fmt::Display;
+use core::fmt;
 use core::future::Future;
 use core::hash::Hash;
-use hashbrown::HashMap;
 #[cfg(feature = "std")]
 use miette::Diagnostic;
-use serde::Serialize;
-use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use smol_str::SmolStr;
+
+use crate::bos::{BosStr, DefaultStr};
+use crate::types::{did::Did, handle::Handle};
 
 #[cfg(feature = "std")]
 use std::path::{Path, PathBuf};
@@ -45,6 +47,109 @@ pub enum SessionStoreError {
     Other(#[from] Box<dyn StdError + Send + Sync>),
 }
 
+/// Shared storage key for app-password and OAuth sessions.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
+pub struct SessionKey {
+    /// Account DID.
+    pub did: Did,
+    /// Store-local session identifier.
+    pub session_id: SmolStr,
+}
+
+impl SessionKey {
+    /// Create a new session key.
+    pub fn new(did: Did, session_id: impl Into<SmolStr>) -> Self {
+        Self {
+            did,
+            session_id: session_id.into(),
+        }
+    }
+
+    /// Borrow the account DID.
+    pub fn did(&self) -> Did<&str> {
+        self.did.borrow()
+    }
+
+    /// Borrow the session identifier.
+    pub fn session_id(&self) -> &str {
+        self.session_id.as_str()
+    }
+}
+
+impl fmt::Display for SessionKey {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}/{}", self.did, self.session_id)
+    }
+}
+
+impl From<(Did, SmolStr)> for SessionKey {
+    fn from((did, session_id): (Did, SmolStr)) -> Self {
+        Self { did, session_id }
+    }
+}
+
+impl From<SessionKey> for (Did, SmolStr) {
+    fn from(key: SessionKey) -> Self {
+        (key.did, key.session_id)
+    }
+}
+
+/// Resolver-free hint for choosing a stored session.
+///
+/// Matching in `jacquard-common` is intentionally key-only and does not perform identity
+/// resolution. [`SessionHint::Handle`] cannot be matched from [`SessionKey`] values alone and
+/// returns no match in [`match_session_key`]; higher-level stores may add handle-aware matching
+/// when they have typed records containing handle metadata.
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub enum SessionHint<S: BosStr = DefaultStr> {
+    /// Use any available session.
+    Any,
+    /// Use the first session for the given DID.
+    Did(Did<S>),
+    /// Use a session for the given handle, if a higher-level matcher can resolve it.
+    Handle(Handle<S>),
+    /// Use this exact key.
+    Key(SessionKey),
+    /// Login/start-auth identifier that is not necessarily session-addressable.
+    ///
+    /// Examples include an email address, explicit PDS/entryway URL, or
+    /// application-specific login input. Default resolver-free selectors do not
+    /// match this as an existing session.
+    Identifier(S),
+}
+
+/// Match a session key using only resolver-free key data.
+pub fn match_session_key<I>(hint: &SessionHint, keys: I) -> Option<SessionKey>
+where
+    I: IntoIterator<Item = SessionKey>,
+{
+    match hint {
+        SessionHint::Any => keys.into_iter().next(),
+        SessionHint::Did(did) => keys.into_iter().find(|key| key.did == *did),
+        SessionHint::Handle(_) | SessionHint::Identifier(_) => None,
+        SessionHint::Key(target) => keys.into_iter().find(|key| key == target),
+    }
+}
+
+/// Selects a session from a hint, optionally returning richer implementation-specific data.
+///
+/// This trait is intentionally separate from [`SessionStore`]. Simple implementations may select
+/// by enumerating store keys and filtering, while database-backed or otherwise indexed
+/// implementations can resolve [`SessionHint::Key`] or [`SessionHint::Did`] without a full scan.
+/// Higher-level crates can also implement selectors that resolve [`SessionHint::Handle`] using an
+/// identity resolver and return metadata such as cached endpoints alongside the selected key.
+#[cfg_attr(not(target_arch = "wasm32"), trait_variant::make(Send))]
+pub trait SessionSelector<M>: Send + Sync {
+    /// Error returned by this selector.
+    type Error;
+
+    /// Select a matching session, if one exists.
+    fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> impl Future<Output = Result<Option<M>, Self::Error>>;
+}
+
 /// Pluggable storage for arbitrary session records.
 #[cfg_attr(not(target_arch = "wasm32"), trait_variant::make(Send))]
 pub trait SessionStore<K, T>: Send + Sync
@@ -58,21 +163,28 @@ where
     fn set(&self, key: K, session: T) -> impl Future<Output = Result<(), SessionStoreError>>;
     /// Delete the given session.
     fn del(&self, key: &K) -> impl Future<Output = Result<(), SessionStoreError>>;
+    /// List known session keys when the backend supports enumeration.
+    fn list_keys(&self) -> impl Future<Output = Result<Vec<K>, SessionStoreError>>
+    where
+        K: Clone,
+    {
+        async { Ok(Vec::new()) }
+    }
 }
 
 /// In-memory session store suitable for short-lived sessions and tests.
 #[derive(Clone)]
-pub struct MemorySessionStore<K, T>(Arc<RwLock<HashMap<K, T>>>);
+pub struct MemorySessionStore<K, T>(Arc<RwLock<BTreeMap<K, T>>>);
 
 impl<K, T> Default for MemorySessionStore<K, T> {
     fn default() -> Self {
-        Self(Arc::new(RwLock::new(HashMap::new())))
+        Self(Arc::new(RwLock::new(BTreeMap::new())))
     }
 }
 
 impl<K, T> SessionStore<K, T> for MemorySessionStore<K, T>
 where
-    K: Eq + Hash + Send + Sync,
+    K: Eq + Hash + Send + Sync + Ord,
     T: Clone + Send + Sync,
 {
     async fn get(&self, key: &K) -> Option<T> {
@@ -86,6 +198,24 @@ where
         self.0.write().await.remove(key);
         Ok(())
     }
+
+    async fn list_keys(&self) -> Result<Vec<K>, SessionStoreError>
+    where
+        K: Clone,
+    {
+        Ok(self.0.read().await.keys().cloned().collect())
+    }
+}
+
+impl<T> SessionSelector<SessionKey> for MemorySessionStore<SessionKey, T>
+where
+    T: Clone + Send + Sync,
+{
+    type Error = SessionStoreError;
+
+    async fn select_session(&self, hint: &SessionHint) -> Result<Option<SessionKey>, Self::Error> {
+        Ok(match_session_key(hint, self.list_keys().await?))
+    }
 }
 
 /// File-backed token store using a JSON file.
@@ -149,43 +279,138 @@ impl FileTokenStore {
 }
 
 #[cfg(feature = "std")]
-impl<K: Eq + Hash + Display + Send + Sync, T: Clone + Serialize + DeserializeOwned + Send + Sync>
-    SessionStore<K, T> for FileTokenStore
-{
-    /// Get the current session if present.
-    async fn get(&self, key: &K) -> Option<T> {
-        let file = std::fs::read_to_string(&self.path).ok()?;
-        let store: Value = serde_json::from_str(&file).ok()?;
-
-        let session = store.get(key.to_string())?;
-        serde_json::from_value(session.clone()).ok()
+impl FileTokenStore {
+    /// Read a JSON value by string key.
+    pub fn get_value(&self, key: &str) -> Result<Option<Value>, SessionStoreError> {
+        let file = std::fs::read_to_string(&self.path)?;
+        let store: Value = serde_json::from_str(&file)?;
+        Ok(store.get(key).cloned())
     }
-    /// Persist the given session.
-    async fn set(&self, key: K, session: T) -> Result<(), SessionStoreError> {
+
+    /// Insert or replace a JSON value by string key.
+    pub fn set_value(&self, key: impl Into<String>, value: Value) -> Result<(), SessionStoreError> {
         let file = std::fs::read_to_string(&self.path)?;
         let mut store: Value = serde_json::from_str(&file)?;
-        let key_string = key.to_string();
         if let Some(store) = store.as_object_mut() {
-            store.insert(key_string, serde_json::to_value(session.clone())?);
-
+            store.insert(key.into(), value);
             std::fs::write(&self.path, serde_json::to_string_pretty(&store)?)?;
             Ok(())
         } else {
             Err(SessionStoreError::Other("invalid store".into()))
         }
     }
-    /// Delete the given session.
-    async fn del(&self, key: &K) -> Result<(), SessionStoreError> {
+
+    /// Remove a JSON value by string key.
+    pub fn remove_value(&self, key: &str) -> Result<(), SessionStoreError> {
         let file = std::fs::read_to_string(&self.path)?;
         let mut store: Value = serde_json::from_str(&file)?;
-        let key_string = key.to_string();
         if let Some(store) = store.as_object_mut() {
-            store.remove(&key_string);
-
+            store.remove(key);
             std::fs::write(&self.path, serde_json::to_string_pretty(&store)?)?;
             Ok(())
         } else {
             Err(SessionStoreError::Other("invalid store".into()))
         }
     }
+
+    /// Return all JSON object entries in the store.
+    pub fn entries(&self) -> Result<Vec<(String, Value)>, SessionStoreError> {
+        let file = std::fs::read_to_string(&self.path)?;
+        let store: Value = serde_json::from_str(&file)?;
+        if let Some(store) = store.as_object() {
+            Ok(store
+                .iter()
+                .map(|(key, value)| (key.clone(), value.clone()))
+                .collect())
+        } else {
+            Err(SessionStoreError::Other("invalid store".into()))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloc::string::ToString;
+
+    #[test]
+    fn session_key_display_uses_slash_separator() {
+        let did = Did::new_static("did:plc:alice").unwrap();
+        let key = SessionKey::new(did, "session_1");
+        assert_eq!(key.to_string(), "did:plc:alice/session_1");
+    }
+
+    #[tokio::test]
+    async fn memory_store_lists_keys() {
+        let store = MemorySessionStore::<SessionKey, String>::default();
+        let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "session");
+        store.set(key.clone(), "value".to_string()).await.unwrap();
+        assert_eq!(store.list_keys().await.unwrap(), vec![key]);
+    }
+
+    struct EmptyStore;
+
+    impl SessionStore<SessionKey, String> for EmptyStore {
+        async fn get(&self, _key: &SessionKey) -> Option<String> {
+            None
+        }
+
+        async fn set(&self, _key: SessionKey, _session: String) -> Result<(), SessionStoreError> {
+            Ok(())
+        }
+
+        async fn del(&self, _key: &SessionKey) -> Result<(), SessionStoreError> {
+            Ok(())
+        }
+    }
+
+    #[tokio::test]
+    async fn default_list_keys_is_empty() {
+        assert!(EmptyStore.list_keys().await.unwrap().is_empty());
+    }
+
+    #[test]
+    fn match_session_key_is_resolver_free() {
+        let alice = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "a");
+        let bob = SessionKey::new(Did::new_static("did:plc:bob").unwrap(), "b");
+        let keys = vec![alice.clone(), bob.clone()];
+
+        assert_eq!(
+            match_session_key(&SessionHint::Any, keys.clone()),
+            Some(alice.clone())
+        );
+        assert_eq!(
+            match_session_key(&SessionHint::Did(bob.did.clone()), keys.clone()),
+            Some(bob.clone())
+        );
+        assert_eq!(
+            match_session_key(&SessionHint::Key(bob.clone()), keys.clone()),
+            Some(bob.clone())
+        );
+        assert_eq!(
+            match_session_key(
+                &SessionHint::Key(SessionKey::new(
+                    Did::new_static("did:plc:carol").unwrap(),
+                    "c",
+                )),
+                keys.clone(),
+            ),
+            None
+        );
+        assert_eq!(match_session_key(&SessionHint::Any, Vec::new()), None);
+        assert_eq!(
+            match_session_key(
+                &SessionHint::Handle(Handle::new_static("alice.example.com").unwrap()),
+                keys.clone(),
+            ),
+            None
+        );
+        assert_eq!(
+            match_session_key(
+                &SessionHint::Identifier(SmolStr::new("alice@example.com")),
+                keys
+            ),
+            None
+        );
+    }
 }
diff --git a/crates/jacquard-oauth/src/atproto.rs b/crates/jacquard-oauth/src/atproto.rs
index 78f263d72..9abf095ce 100644
--- a/crates/jacquard-oauth/src/atproto.rs
+++ b/crates/jacquard-oauth/src/atproto.rs
@@ -8,7 +8,7 @@ use crate::{
 use jacquard_common::deps::fluent_uri::Uri;
 use jacquard_common::{BosStr, IntoStatic};
 use serde::{Deserialize, Serialize};
-use smol_str::SmolStr;
+use smol_str::{SmolStr, ToSmolStr};
 use thiserror::Error;
 
 /// Errors that can occur when building AT Protocol OAuth client metadata.
@@ -248,7 +248,7 @@ where
         }
         let redir_str = redirect_uris.as_ref().map(|uris| {
             uris.iter()
-                .map(|u| SmolStr::from(u.as_str().trim_end_matches("/")))
+                .map(|u| u.as_str().trim_end_matches("/").to_smolstr())
                 .collect()
         });
         let query = serde_html_form::to_string(Parameters {
diff --git a/crates/jacquard-oauth/src/authstore.rs b/crates/jacquard-oauth/src/authstore.rs
index 5f188b8a3..43b475cc7 100644
--- a/crates/jacquard-oauth/src/authstore.rs
+++ b/crates/jacquard-oauth/src/authstore.rs
@@ -4,13 +4,117 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use jacquard_common::{
     bos::BosStr,
-    session::{SessionStore, SessionStoreError},
+    session::{SessionHint, SessionKey, SessionSelector, SessionStore, SessionStoreError},
     types::did::Did,
 };
+use jacquard_identity::resolver::IdentityResolver;
 use smol_str::{SmolStr, format_smolstr};
 
 use crate::session::{AuthRequestData, ClientSessionData};
 
+/// OAuth session lookup result with the matched key and session data.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct OAuthSessionMatch {
+    /// Matched session key.
+    pub key: SessionKey,
+    /// Stored OAuth client session data for the matched key.
+    pub session: ClientSessionData,
+}
+
+/// Resolver-backed OAuth session selector.
+///
+/// This adapter keeps selection pluggable: callers can depend on [`SessionSelector`] while stores
+/// with better indexing can provide their own selector implementation.
+pub struct OAuthSessionSelector<'a, S, R> {
+    store: &'a S,
+    resolver: &'a R,
+}
+
+impl<'a, S, R> OAuthSessionSelector<'a, S, R> {
+    /// Create a selector over an OAuth auth store and identity resolver.
+    pub fn new(store: &'a S, resolver: &'a R) -> Self {
+        Self { store, resolver }
+    }
+}
+
+impl<S, R> SessionSelector<OAuthSessionMatch> for OAuthSessionSelector<'_, S, R>
+where
+    S: ClientAuthStore + SessionSelector<OAuthSessionMatch, Error = SessionStoreError> + Sync,
+    R: IdentityResolver + Sync,
+{
+    type Error = SessionStoreError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<OAuthSessionMatch>, Self::Error> {
+        if let Some(matched) = self.store.select_session(hint).await? {
+            return Ok(Some(matched));
+        }
+
+        let SessionHint::Handle(handle) = hint else {
+            return Ok(None);
+        };
+
+        let did = self
+            .resolver
+            .resolve_handle(handle)
+            .await
+            .map_err(|e| SessionStoreError::Other(Box::new(e)))?;
+        self.store.select_session(&SessionHint::Did(did)).await
+    }
+}
+
+/// Resolve a [`SessionHint`] against an OAuth [`ClientAuthStore`].
+///
+/// Exact key lookup avoids enumeration. `Any`, `Did`, and `Handle` use
+/// [`ClientAuthStore::list_session_keys`] as the generic fallback; stores that need more efficient
+/// indexed lookup can add specialized APIs later without changing the common key type.
+pub async fn resolve_oauth_session_hint<S, R>(
+    store: &S,
+    resolver: &R,
+    hint: &SessionHint,
+) -> Result<Option<OAuthSessionMatch>, SessionStoreError>
+where
+    S: ClientAuthStore + SessionSelector<OAuthSessionMatch, Error = SessionStoreError> + Sync,
+    R: IdentityResolver + Sync,
+{
+    OAuthSessionSelector::new(store, resolver)
+        .select_session(hint)
+        .await
+}
+
+async fn oauth_match_for_did<S, D>(
+    store: &S,
+    did: &Did<D>,
+) -> Result<Option<OAuthSessionMatch>, SessionStoreError>
+where
+    S: ClientAuthStore,
+    D: BosStr + Send + Sync,
+{
+    for key in store.list_session_keys().await? {
+        if key.did.as_str() == did.as_ref() {
+            if let Some(matched) = oauth_match_for_key(store, key).await? {
+                return Ok(Some(matched));
+            }
+        }
+    }
+    Ok(None)
+}
+
+async fn oauth_match_for_key<S>(
+    store: &S,
+    key: SessionKey,
+) -> Result<Option<OAuthSessionMatch>, SessionStoreError>
+where
+    S: ClientAuthStore,
+{
+    Ok(store
+        .get_session(&key.did, key.session_id.as_str())
+        .await?
+        .map(|session| OAuthSessionMatch { key, session }))
+}
+
 /// Persistent storage backend for OAuth client sessions and in-flight authorization requests.
 ///
 /// Implementors are responsible for durably storing two categories of data:
@@ -56,6 +160,13 @@ pub trait ClientAuthStore {
         &self,
         state: &str,
     ) -> impl Future<Output = Result<(), SessionStoreError>>;
+
+    /// List active OAuth session keys when the backend supports enumeration.
+    fn list_session_keys(
+        &self,
+    ) -> impl Future<Output = Result<Vec<SessionKey>, SessionStoreError>> {
+        async { Ok(Vec::new()) }
+    }
 }
 
 /// An in-memory implementation of [`ClientAuthStore`], suitable for testing and single-process
@@ -81,12 +192,12 @@ impl ClientAuthStore for MemoryAuthStore {
         did: &Did<D>,
         session_id: &str,
     ) -> Result<Option<ClientSessionData>, SessionStoreError> {
-        let key = format_smolstr!("{}_{}", did, session_id);
+        let key = format_smolstr!("{}/{}", did, session_id);
         Ok(self.sessions.get(&key).map(|v| v.clone()))
     }
 
     async fn upsert_session(&self, session: ClientSessionData) -> Result<(), SessionStoreError> {
-        let key = format_smolstr!("{}_{}", session.account_did, session.session_id);
+        let key = format_smolstr!("{}/{}", session.account_did, session.session_id);
         self.sessions.insert(key, session);
         Ok(())
     }
@@ -96,7 +207,7 @@ impl ClientAuthStore for MemoryAuthStore {
         did: &Did<D>,
         session_id: &str,
     ) -> Result<(), SessionStoreError> {
-        let key = format_smolstr!("{}_{}", did, session_id);
+        let key = format_smolstr!("{}/{}", did, session_id);
         self.sessions.remove(&key);
         Ok(())
     }
@@ -121,14 +232,47 @@ impl ClientAuthStore for MemoryAuthStore {
         self.auth_reqs.remove(state);
         Ok(())
     }
+
+    async fn list_session_keys(&self) -> Result<Vec<SessionKey>, SessionStoreError> {
+        let mut sessions = self
+            .sessions
+            .iter()
+            .map(|entry| {
+                let session = entry.value();
+                SessionKey::new(session.account_did.clone(), session.session_id.clone())
+            })
+            .collect::<Vec<_>>();
+        sessions.sort();
+        Ok(sessions)
+    }
 }
 
-impl<T: ClientAuthStore + Send + Sync> SessionStore<(Did, SmolStr), ClientSessionData> for Arc<T> {
+impl SessionSelector<OAuthSessionMatch> for MemoryAuthStore {
+    type Error = SessionStoreError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<OAuthSessionMatch>, Self::Error> {
+        match hint {
+            SessionHint::Any => {
+                let Some(key) = self.list_session_keys().await?.into_iter().next() else {
+                    return Ok(None);
+                };
+                oauth_match_for_key(self, key).await
+            }
+            SessionHint::Did(did) => oauth_match_for_did(self, did).await,
+            SessionHint::Handle(_) | SessionHint::Identifier(_) => Ok(None),
+            SessionHint::Key(key) => oauth_match_for_key(self, key.clone()).await,
+        }
+    }
+}
+
+impl<T: ClientAuthStore + Send + Sync> SessionStore<SessionKey, ClientSessionData> for Arc<T> {
     /// Get the current session if present.
-    async fn get(&self, key: &(Did, SmolStr)) -> Option<ClientSessionData> {
-        let (did, session_id) = key;
+    async fn get(&self, key: &SessionKey) -> Option<ClientSessionData> {
         self.as_ref()
-            .get_session(did, session_id)
+            .get_session(&key.did, key.session_id.as_str())
             .await
             .ok()
             .flatten()
@@ -136,14 +280,195 @@ impl<T: ClientAuthStore + Send + Sync> SessionStore<(Did, SmolStr), ClientSessio
     /// Persist the given session.
     async fn set(
         &self,
-        _key: (Did, SmolStr),
+        _key: SessionKey,
         session: ClientSessionData,
     ) -> Result<(), SessionStoreError> {
         self.as_ref().upsert_session(session).await
     }
     /// Delete the given session.
-    async fn del(&self, key: &(Did, SmolStr)) -> Result<(), SessionStoreError> {
-        let (did, session_id) = key;
-        self.as_ref().delete_session(did, session_id).await
+    async fn del(&self, key: &SessionKey) -> Result<(), SessionStoreError> {
+        self.as_ref()
+            .delete_session(&key.did, key.session_id.as_str())
+            .await
+    }
+
+    async fn list_keys(&self) -> Result<Vec<SessionKey>, SessionStoreError> {
+        self.as_ref().list_session_keys().await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use jacquard_common::deps::fluent_uri::Uri;
+
+    use crate::scopes::Scopes;
+    use crate::session::DpopClientData;
+    use crate::types::{OAuthTokenType, TokenSet};
+
+    fn client_session(did: &'static str, session_id: &'static str) -> ClientSessionData {
+        let account_did = Did::new_static(did).unwrap();
+        ClientSessionData {
+            account_did: account_did.clone(),
+            session_id: SmolStr::new_static(session_id),
+            host_url: Uri::parse("https://pds.example.com").unwrap().to_owned(),
+            authserver_url: SmolStr::new_static("https://issuer.example.com"),
+            authserver_token_endpoint: SmolStr::new_static("https://issuer.example.com/token"),
+            authserver_revocation_endpoint: None,
+            scopes: Scopes::empty(),
+            dpop_data: DpopClientData {
+                dpop_key: crate::utils::generate_key(&[SmolStr::new_static("ES256")]).unwrap(),
+                dpop_authserver_nonce: SmolStr::default(),
+                dpop_host_nonce: SmolStr::default(),
+            },
+            token_set: TokenSet {
+                iss: SmolStr::new_static("https://issuer.example.com"),
+                sub: account_did,
+                aud: SmolStr::new_static("https://pds.example.com"),
+                scope: None,
+                refresh_token: None,
+                access_token: SmolStr::new_static("access"),
+                token_type: OAuthTokenType::DPoP,
+                expires_at: None,
+            },
+            #[cfg(feature = "scope-check")]
+            resolved_scopes: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn memory_auth_store_lists_session_keys() {
+        let store = MemoryAuthStore::new();
+        let session = client_session("did:plc:alice", "state");
+        store.upsert_session(session).await.unwrap();
+
+        assert_eq!(
+            store.list_session_keys().await.unwrap(),
+            vec![SessionKey::new(
+                Did::new_static("did:plc:alice").unwrap(),
+                "state"
+            )]
+        );
+    }
+
+    #[tokio::test]
+    async fn memory_auth_store_selects_sessions_without_identifier_fallback() {
+        let store = MemoryAuthStore::new();
+        let alice = client_session("did:plc:alice", "state-a");
+        let alice_key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "state-a");
+        store.upsert_session(alice.clone()).await.unwrap();
+        store
+            .upsert_session(client_session("did:plc:bob", "state-b"))
+            .await
+            .unwrap();
+
+        let matched = store
+            .select_session(&SessionHint::Any)
+            .await
+            .unwrap()
+            .expect("any match");
+        assert_eq!(matched.key, alice_key);
+        assert_eq!(matched.session, alice);
+
+        let matched = store
+            .select_session(&SessionHint::Did(Did::new_static("did:plc:alice").unwrap()))
+            .await
+            .unwrap()
+            .expect("did match");
+        assert_eq!(matched.key, alice_key);
+
+        let matched = store
+            .select_session(&SessionHint::Key(alice_key.clone()))
+            .await
+            .unwrap()
+            .expect("key match");
+        assert_eq!(matched.key, alice_key);
+
+        assert!(
+            store
+                .select_session(&SessionHint::Identifier("alice@example.com".into()))
+                .await
+                .unwrap()
+                .is_none(),
+            "identifier hints must not fall back to Any"
+        );
+    }
+
+    #[derive(Clone, Default)]
+    struct CountingResolver {
+        handle_calls: Arc<tokio::sync::RwLock<usize>>,
+    }
+
+    impl IdentityResolver for CountingResolver {
+        fn options(&self) -> &jacquard_identity::resolver::ResolverOptions {
+            use std::sync::LazyLock;
+            static OPTS: LazyLock<jacquard_identity::resolver::ResolverOptions> =
+                LazyLock::new(jacquard_identity::resolver::ResolverOptions::default);
+            &OPTS
+        }
+
+        async fn resolve_handle<S: BosStr + Sync>(
+            &self,
+            _handle: &jacquard_common::types::string::Handle<S>,
+        ) -> Result<Did, jacquard_identity::resolver::IdentityError> {
+            *self.handle_calls.write().await += 1;
+            Ok(Did::new_static("did:plc:alice").unwrap())
+        }
+
+        async fn resolve_did_doc<S: BosStr + Sync>(
+            &self,
+            _did: &Did<S>,
+        ) -> Result<
+            jacquard_identity::resolver::DidDocResponse,
+            jacquard_identity::resolver::IdentityError,
+        > {
+            unreachable!("OAuth selector tests do not resolve DID documents")
+        }
+    }
+
+    #[tokio::test]
+    async fn oauth_session_selector_uses_store_before_handle_resolution() {
+        let store = MemoryAuthStore::new();
+        let resolver = CountingResolver::default();
+        let alice = client_session("did:plc:alice", "state");
+        store.upsert_session(alice.clone()).await.unwrap();
+
+        assert!(
+            OAuthSessionSelector::new(&store, &resolver)
+                .select_session(&SessionHint::Identifier("alice@example.com".into()))
+                .await
+                .unwrap()
+                .is_none(),
+            "identifier hints should not trigger resolver fallback"
+        );
+        assert_eq!(*resolver.handle_calls.read().await, 0);
+
+        let matched = OAuthSessionSelector::new(&store, &resolver)
+            .select_session(&SessionHint::Handle(
+                jacquard_common::types::string::Handle::new_static("alice.bsky.social").unwrap(),
+            ))
+            .await
+            .unwrap()
+            .expect("resolver fallback DID match");
+        assert_eq!(matched.session, alice);
+        assert_eq!(*resolver.handle_calls.read().await, 1);
+    }
+
+    #[tokio::test]
+    async fn arc_memory_auth_store_is_session_store() {
+        let store = Arc::new(MemoryAuthStore::new());
+        let session = client_session("did:plc:alice", "state");
+        let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "state");
+
+        SessionStore::set(&store, key.clone(), session.clone())
+            .await
+            .unwrap();
+        assert_eq!(SessionStore::get(&store, &key).await, Some(session));
+        assert_eq!(
+            SessionStore::list_keys(&store).await.unwrap(),
+            vec![key.clone()]
+        );
+        SessionStore::del(&store, &key).await.unwrap();
+        assert_eq!(SessionStore::get(&store, &key).await, None);
     }
 }
diff --git a/crates/jacquard-oauth/src/client.rs b/crates/jacquard-oauth/src/client.rs
index 73eb72b67..f4e7496ab 100644
--- a/crates/jacquard-oauth/src/client.rs
+++ b/crates/jacquard-oauth/src/client.rs
@@ -1,8 +1,8 @@
 use crate::{
     atproto::atproto_client_metadata,
-    authstore::ClientAuthStore,
+    authstore::{ClientAuthStore, OAuthSessionMatch},
     dpop::DpopExt,
-    error::{CallbackError, Result},
+    error::{CallbackError, OAuthError, Result},
     request::{OAuthMetadata, exchange_code, par},
     resolver::OAuthResolver,
     scopes::Scopes,
@@ -25,6 +25,7 @@ use jacquard_common::{
     deps::fluent_uri::Uri,
     error::{AuthError, ClientError, XrpcResult},
     http_client::HttpClient,
+    session::{SessionHint, SessionSelector, SessionStoreError},
     types::{did::Did, string::Handle},
     xrpc::{
         CallOptions, Response, XrpcClient, XrpcExt, XrpcRequest, XrpcResp, XrpcResponse,
@@ -50,6 +51,18 @@ use smol_str::{SmolStr, ToSmolStr};
 use std::{str::FromStr, sync::Arc};
 use tokio::sync::RwLock;
 
+/// Result of resuming an OAuth session or starting a new authorization flow.
+pub enum OAuthResumeOrLogin<T, S>
+where
+    T: OAuthResolver,
+    S: ClientAuthStore,
+{
+    /// A stored session was found and restored/refreshed.
+    Resumed(OAuthSession<T, S>),
+    /// No stored session matched; redirect the user to this login URL.
+    LoginUrl(String),
+}
+
 /// The top-level OAuth client responsible for driving the authorization flow.
 pub struct OAuthClient<T, S>
 where
@@ -331,7 +344,7 @@ where
         {
             Ok(token_set) => {
                 let scopes = if let Some(scope) = &token_set.scope {
-                    Scopes::new(SmolStr::from(scope.as_str()))
+                    Scopes::new(scope.as_str().to_smolstr())
                         .expect("Failed to parse scopes from token response")
                 } else {
                     Scopes::empty()
@@ -380,6 +393,55 @@ where
             .await
     }
 
+    /// Resume a stored session for `input`, or begin OAuth authorization and return a login URL.
+    pub async fn resume_or_start_auth_for<Str: BosStr>(
+        &self,
+        input: impl AsRef<str>,
+        options: AuthorizeOptions<Str>,
+    ) -> Result<OAuthResumeOrLogin<T, S>>
+    where
+        S: SessionSelector<OAuthSessionMatch, Error = SessionStoreError>,
+        Str: FromStr + Ord + Clone + core::fmt::Debug,
+        <Str as FromStr>::Err: core::fmt::Debug,
+    {
+        let input = input.as_ref();
+        let hint = oauth_hint_from_input(input);
+        match self.registry.store.select_session(&hint).await? {
+            Some(matched) => Ok(OAuthResumeOrLogin::Resumed(
+                self.restore(&matched.key.did, matched.key.session_id.as_str())
+                    .await?,
+            )),
+            None => Ok(OAuthResumeOrLogin::LoginUrl(
+                self.start_auth(input, options).await?,
+            )),
+        }
+    }
+
+    /// Resume a stored session for `hint`, or begin OAuth authorization from the hint identity.
+    pub async fn resume_or_start_auth<Str: BosStr>(
+        &self,
+        hint: &SessionHint,
+        options: AuthorizeOptions<Str>,
+    ) -> Result<OAuthResumeOrLogin<T, S>>
+    where
+        S: SessionSelector<OAuthSessionMatch, Error = SessionStoreError>,
+        Str: FromStr + Ord + Clone + core::fmt::Debug,
+        <Str as FromStr>::Err: core::fmt::Debug,
+    {
+        match self.registry.store.select_session(hint).await? {
+            Some(matched) => Ok(OAuthResumeOrLogin::Resumed(
+                self.restore(&matched.key.did, matched.key.session_id.as_str())
+                    .await?,
+            )),
+            None => {
+                let input = oauth_start_auth_input_from_hint(hint)?;
+                Ok(OAuthResumeOrLogin::LoginUrl(
+                    self.start_auth(input, options).await?,
+                ))
+            }
+        }
+    }
+
     /// Revoke a session by deleting it from the backing store.
     ///
     /// Note: this removes the session from local storage but does **not** call the authorization
@@ -394,6 +456,28 @@ where
     }
 }
 
+fn oauth_hint_from_input(input: &str) -> SessionHint<SmolStr> {
+    if let Ok(did) = Did::new(input) {
+        SessionHint::Did(did.convert())
+    } else if let Ok(handle) = Handle::new(input) {
+        SessionHint::Handle(handle.convert())
+    } else {
+        SessionHint::Identifier(SmolStr::from(input))
+    }
+}
+
+fn oauth_start_auth_input_from_hint(hint: &SessionHint) -> Result<SmolStr> {
+    match hint {
+        SessionHint::Did(did) => Ok(did.as_ref().to_smolstr()),
+        SessionHint::Handle(handle) => Ok(handle.as_ref().to_smolstr()),
+        SessionHint::Key(key) => Ok(key.did.as_str().to_smolstr()),
+        SessionHint::Identifier(identifier) => Ok(identifier.clone()),
+        SessionHint::Any => Err(OAuthError::InvalidRequest(
+            "cannot start OAuth authorization from SessionHint::Any without an identity".into(),
+        )),
+    }
+}
+
 /// Decode a percent-encoded audience string.
 ///
 /// The audience may contain percent-encoded characters like `%23` for `#`.
diff --git a/crates/jacquard-oauth/src/error.rs b/crates/jacquard-oauth/src/error.rs
index c55e1582c..9fc0bc81f 100644
--- a/crates/jacquard-oauth/src/error.rs
+++ b/crates/jacquard-oauth/src/error.rs
@@ -60,6 +60,11 @@ pub enum OAuthError {
     #[diagnostic(code(jacquard_oauth::form))]
     Form(#[from] serde_html_form::ser::Error),
 
+    /// Invalid OAuth helper input.
+    #[error("invalid OAuth request: {0}")]
+    #[diagnostic(code(jacquard_oauth::invalid_request))]
+    InvalidRequest(String),
+
     /// An error validating an authorization callback.
     #[error(transparent)]
     #[diagnostic(code(jacquard_oauth::callback))]
diff --git a/crates/jacquard-oauth/src/loopback.rs b/crates/jacquard-oauth/src/loopback.rs
index 1ddf41167..f7e31ab81 100644
--- a/crates/jacquard-oauth/src/loopback.rs
+++ b/crates/jacquard-oauth/src/loopback.rs
@@ -46,7 +46,7 @@
 #![cfg(feature = "loopback")]
 use crate::{
     atproto::AtprotoClientMetadata,
-    authstore::ClientAuthStore,
+    authstore::{ClientAuthStore, OAuthSessionMatch},
     client::OAuthClient,
     dpop::DpopExt,
     error::{CallbackError, OAuthError},
@@ -55,11 +55,23 @@ use crate::{
 };
 use jacquard_common::IntoStatic;
 use jacquard_common::deps::fluent_uri::Uri;
+use jacquard_common::session::{SessionHint, SessionSelector, SessionStoreError};
+use jacquard_common::types::{did::Did, string::Handle};
 use rouille::Server;
 use smol_str::{SmolStr, ToSmolStr};
 use std::net::SocketAddr;
 use tokio::sync::mpsc;
 
+fn oauth_hint_from_input(input: &str) -> SessionHint<SmolStr> {
+    if let Ok(did) = Did::new(input) {
+        SessionHint::Did(did.convert())
+    } else if let Ok(handle) = Handle::new(input) {
+        SessionHint::Handle(handle.convert())
+    } else {
+        SessionHint::Identifier(SmolStr::from(input))
+    }
+}
+
 /// Port selection strategy for the loopback OAuth callback server.
 #[derive(Clone, Debug)]
 pub enum LoopbackPort {
@@ -315,6 +327,26 @@ where
         }
         .into_static()
     }
+
+    /// Resume a stored session for the input identity, or drive the full OAuth flow using a local loopback server.
+    pub async fn resume_or_login_with_local_server(
+        &self,
+        input: impl AsRef<str>,
+        opts: AuthorizeOptions<SmolStr>,
+        cfg: LoopbackConfig,
+    ) -> crate::error::Result<super::client::OAuthSession<T, S>>
+    where
+        S: SessionSelector<OAuthSessionMatch, Error = SessionStoreError>,
+    {
+        let input_ref = input.as_ref();
+        let hint = oauth_hint_from_input(input_ref);
+        if let Some(matched) = self.registry.store.select_session(&hint).await? {
+            return self
+                .restore(&matched.key.did, matched.key.session_id.as_str())
+                .await;
+        }
+        self.login_with_local_server(input_ref, opts, cfg).await
+    }
 }
 
 #[cfg(feature = "scope-check")]
@@ -384,6 +416,26 @@ where
         handle_localhost_callback(handle, &flow_client, &cfg).await
     }
 
+    /// Resume a stored session for the input identity, or drive the full OAuth flow using a local loopback server.
+    pub async fn resume_or_login_with_local_server(
+        &self,
+        input: impl AsRef<str>,
+        opts: AuthorizeOptions<SmolStr>,
+        cfg: LoopbackConfig,
+    ) -> crate::error::Result<super::client::OAuthSession<T, S>>
+    where
+        S: SessionSelector<OAuthSessionMatch, Error = SessionStoreError>,
+    {
+        let input_ref = input.as_ref();
+        let hint = oauth_hint_from_input(input_ref);
+        if let Some(matched) = self.registry.store.select_session(&hint).await? {
+            return self
+                .restore(&matched.key.did, matched.key.session_id.as_str())
+                .await;
+        }
+        self.login_with_local_server(input_ref, opts, cfg).await
+    }
+
     /// Builds a [`crate::session::ClientData`] for use with the local loopback server method of OAuth.
     pub fn build_localhost_client_data(
         &self,
diff --git a/crates/jacquard-oauth/src/request.rs b/crates/jacquard-oauth/src/request.rs
index eca1a1bf7..ca697acaf 100644
--- a/crates/jacquard-oauth/src/request.rs
+++ b/crates/jacquard-oauth/src/request.rs
@@ -577,7 +577,7 @@ pub async fn par<
         .await?;
 
         let scopes = if let Some(scope) = &metadata.client_metadata.scope {
-            Scopes::new(SmolStr::from(scope.as_ref())).expect("Failed to parse scopes")
+            Scopes::new(scope.as_ref().to_smolstr()).expect("Failed to parse scopes")
         } else {
             Scopes::empty()
         };
@@ -655,7 +655,7 @@ where
     session_data.update_with_tokens(&TokenSet {
         iss,
         sub: session_data.token_set.sub.clone(),
-        aud: SmolStr::from(aud.as_str()),
+        aud: aud.as_str().to_smolstr(),
         scope: response.scope,
         access_token: response.access_token,
         refresh_token: response.refresh_token,
@@ -719,7 +719,7 @@ where
     Ok(TokenSet {
         iss,
         sub,
-        aud: SmolStr::from(aud.as_str()),
+        aud: aud.as_str().to_smolstr(),
         scope: token_response.scope,
         access_token: token_response.access_token,
         refresh_token: token_response.refresh_token,
diff --git a/crates/jacquard-oauth/src/scopes.rs b/crates/jacquard-oauth/src/scopes.rs
index b5a70bd1d..3ecd3eb9e 100644
--- a/crates/jacquard-oauth/src/scopes.rs
+++ b/crates/jacquard-oauth/src/scopes.rs
@@ -3035,7 +3035,7 @@ mod tests {
     fn test_scopes_buffer_size_limit() {
         // Test buffer exceeding u16 limit is rejected.
         let too_long = "a".repeat(u16::MAX as usize + 1);
-        let smol = SmolStr::from(too_long.as_str());
+        let smol = too_long.as_str().to_smolstr();
         let result = Scopes::new(smol);
         assert!(result.is_err());
     }
diff --git a/crates/jacquard-oauth/src/session.rs b/crates/jacquard-oauth/src/session.rs
index 405cb2712..a35d4742c 100644
--- a/crates/jacquard-oauth/src/session.rs
+++ b/crates/jacquard-oauth/src/session.rs
@@ -24,7 +24,7 @@ use jacquard_common::{
 };
 use jose_jwk::Key;
 use serde::{Deserialize, Serialize};
-use smol_str::{SmolStr, format_smolstr};
+use smol_str::{SmolStr, ToSmolStr, format_smolstr};
 use tokio::sync::Mutex;
 
 /// Provides DPoP key material and per-server nonces to the DPoP proof-building machinery.
@@ -139,7 +139,7 @@ impl<S: BosStr + Ord + AsRef<str>> ClientSessionData<S> {
     {
         if let Some(scope_str) = token_set.scope.as_ref() {
             // Parse scopes from the returned scope string, converting to the appropriate backing type
-            let scopes_smol = Scopes::new(SmolStr::from(scope_str.as_ref()))
+            let scopes_smol = Scopes::new(scope_str.as_ref().to_smolstr())
                 .expect("server returned invalid scopes in token refresh");
             self.scopes = scopes_smol.convert();
         }
diff --git a/crates/jacquard/src/client.rs b/crates/jacquard/src/client.rs
index dd671faf4..866b51fd8 100644
--- a/crates/jacquard/src/client.rs
+++ b/crates/jacquard/src/client.rs
@@ -54,6 +54,8 @@ pub use jacquard_common::session::{MemorySessionStore, SessionStore, SessionStor
 use jacquard_common::types::blob::{Blob, MimeType};
 use jacquard_common::types::collection::Collection;
 #[cfg(feature = "api")]
+use jacquard_common::types::did_doc::DidDocument;
+#[cfg(feature = "api")]
 use jacquard_common::types::ident::AtIdentifier;
 use jacquard_common::types::recordkey::{RecordKey, Rkey};
 use jacquard_common::types::string::AtUri;
@@ -479,7 +481,7 @@ impl Default for MemoryCredentialSession {
 /// App password session information from `com.atproto.server.createSession`
 ///
 /// Contains the access and refresh tokens along with user identity information.
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct AtpSession {
     /// Access token (JWT) used for authenticated requests
     pub access_jwt: SmolStr,
@@ -489,6 +491,28 @@ pub struct AtpSession {
     pub did: Did,
     /// User's handle (e.g., "alice.bsky.social")
     pub handle: Handle,
+    /// Account PDS endpoint, when known.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub pds: Option<Uri<String>>,
+}
+
+impl AtpSession {
+    /// Return the known account PDS endpoint, if present.
+    pub fn pds_endpoint(&self) -> Option<&Uri<String>> {
+        self.pds.as_ref()
+    }
+
+    /// Merge a refresh response into this session, preserving the existing PDS unless
+    /// the refresh response contains a parseable DID document PDS endpoint.
+    #[cfg(feature = "api")]
+    pub fn merge_refresh(&mut self, output: RefreshSessionOutput) {
+        let pds = pds_from_data(output.did_doc.as_ref()).or_else(|| self.pds.clone());
+        self.access_jwt = output.access_jwt;
+        self.refresh_jwt = output.refresh_jwt;
+        self.did = output.did;
+        self.handle = output.handle;
+        self.pds = pds;
+    }
 }
 
 impl IntoStatic for AtpSession {
@@ -499,14 +523,24 @@ impl IntoStatic for AtpSession {
     }
 }
 
+#[cfg(feature = "api")]
+pub(crate) fn pds_from_data<S: BosStr>(
+    data: Option<&jacquard_common::types::value::Data<S>>,
+) -> Option<Uri<String>> {
+    let doc: DidDocument = serde::Deserialize::deserialize(data?).ok()?;
+    doc.pds_endpoint().map(|uri| uri.to_owned())
+}
+
 #[cfg(feature = "api")]
 impl From<CreateSessionOutput> for AtpSession {
     fn from(output: CreateSessionOutput) -> Self {
+        let pds = pds_from_data(output.did_doc.as_ref());
         Self {
             access_jwt: output.access_jwt,
             refresh_jwt: output.refresh_jwt,
             did: output.did,
             handle: output.handle,
+            pds,
         }
     }
 }
@@ -514,11 +548,13 @@ impl From<CreateSessionOutput> for AtpSession {
 #[cfg(feature = "api")]
 impl From<RefreshSessionOutput> for AtpSession {
     fn from(output: RefreshSessionOutput) -> Self {
+        let pds = pds_from_data(output.did_doc.as_ref());
         Self {
             access_jwt: output.access_jwt,
             refresh_jwt: output.refresh_jwt,
             did: output.did,
             handle: output.handle,
+            pds,
         }
     }
 }
@@ -1239,7 +1275,7 @@ where
             CredentialSession::<S, T, W>::session_info(self)
                 .await
                 // Convert the SmolStr session id to CowStr<'static>.
-                .map(|key| (key.0, Some(key.1)))
+                .map(|key| (key.did, Some(key.session_id)))
         }
     }
     fn endpoint(&self) -> impl Future<Output = Uri<String>> {
diff --git a/crates/jacquard/src/client/bff_session.rs b/crates/jacquard/src/client/bff_session.rs
index 98eaee4be..4e2defcd9 100644
--- a/crates/jacquard/src/client/bff_session.rs
+++ b/crates/jacquard/src/client/bff_session.rs
@@ -151,7 +151,7 @@ impl ClientAuthStore for BrowserAuthStore {
 #[cfg(target_arch = "wasm32")]
 impl SessionStore<SessionKey, AtpSession> for BrowserAuthStore {
     fn get(&self, key: &SessionKey) -> impl Future<Output = Option<AtpSession>> + Send {
-        let key = Self::session_key(&key.0, &key.1);
+        let key = Self::session_key(&key.did, key.session_id.as_str());
         async move {
             match LocalStorage::get::<serde_json::Value>(&key) {
                 Ok(value) => {
@@ -170,7 +170,7 @@ impl SessionStore<SessionKey, AtpSession> for BrowserAuthStore {
         session: AtpSession,
     ) -> impl Future<Output = Result<(), SessionStoreError>> + Send {
         async move {
-            let key = Self::session_key(&key.0, &key.1);
+            let key = Self::session_key(&key.did, key.session_id.as_str());
 
             let value = serde_json::to_value(&session)
                 .map_err(|e| SessionStoreError::Other(format!("Serialize error: {}", e).into()))?;
@@ -184,7 +184,7 @@ impl SessionStore<SessionKey, AtpSession> for BrowserAuthStore {
     }
 
     fn del(&self, key: &SessionKey) -> impl Future<Output = Result<(), SessionStoreError>> + Send {
-        let key = Self::session_key(&key.0, &key.1);
+        let key = Self::session_key(&key.did, key.session_id.as_str());
         async move {
             LocalStorage::delete(&key);
             Ok(())
@@ -196,7 +196,7 @@ impl SessionStore<SessionKey, AtpSession> for BrowserAuthStore {
 #[cfg(target_arch = "wasm32")]
 impl SessionStore<SessionKey, SessionKey> for BrowserAuthStore {
     fn get(&self, key: &SessionKey) -> impl Future<Output = Option<SessionKey>> + Send {
-        let key = Self::session_key(&key.0, &key.1);
+        let key = Self::session_key(&key.did, key.session_id.as_str());
         async move {
             match LocalStorage::get::<serde_json::Value>(&key) {
                 Ok(value) => {
@@ -215,7 +215,7 @@ impl SessionStore<SessionKey, SessionKey> for BrowserAuthStore {
         session: SessionKey,
     ) -> impl Future<Output = Result<(), SessionStoreError>> + Send {
         async move {
-            let key = Self::session_key(&key.0, &key.1);
+            let key = Self::session_key(&key.did, key.session_id.as_str());
 
             let value = serde_json::to_value(&session)
                 .map_err(|e| SessionStoreError::Other(format!("Serialize error: {}", e).into()))?;
@@ -229,7 +229,7 @@ impl SessionStore<SessionKey, SessionKey> for BrowserAuthStore {
     }
 
     fn del(&self, key: &SessionKey) -> impl Future<Output = Result<(), SessionStoreError>> + Send {
-        let key = Self::session_key(&key.0, &key.1);
+        let key = Self::session_key(&key.did, key.session_id.as_str());
         async move {
             LocalStorage::delete(&key);
             Ok(())
diff --git a/crates/jacquard/src/client/credential_session.rs b/crates/jacquard/src/client/credential_session.rs
index 9f4649de9..20add88ff 100644
--- a/crates/jacquard/src/client/credential_session.rs
+++ b/crates/jacquard/src/client/credential_session.rs
@@ -9,29 +9,207 @@ use jacquard_common::{
     deps::fluent_uri::Uri,
     error::{AuthError, ClientError, XrpcResult},
     http_client::HttpClient,
-    session::SessionStore,
+    session::{MemorySessionStore, SessionHint, SessionSelector, SessionStore},
     types::{did::Did, string::Handle},
     xrpc::{CallOptions, Response, XrpcClient, XrpcExt, XrpcRequest, XrpcResp, XrpcResponse},
 };
 #[cfg(feature = "streaming")]
 use serde::Serialize;
-use smol_str::SmolStr;
+use smol_str::{SmolStr, ToSmolStr};
 use tokio::sync::RwLock;
 
 use crate::client::AtpSession;
-use jacquard_identity::resolver::{
-    DidDocResponse, IdentityError, IdentityResolver, ResolverOptions,
-};
-use std::any::Any;
-
 #[cfg(feature = "websocket")]
 use jacquard_common::websocket::{WebSocketClient, WebSocketConnection};
 #[cfg(feature = "websocket")]
 use jacquard_common::xrpc::XrpcSubscription;
+use jacquard_identity::resolver::{
+    DidDocResponse, IdentityError, IdentityResolver, ResolverOptions,
+};
+
+pub use jacquard_common::session::SessionKey;
+
+/// App-password session lookup result.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CredentialSessionMatch {
+    /// Matched session key.
+    pub key: SessionKey,
+    /// Stored app-password session for the matched key.
+    pub session: AtpSession,
+}
+
+/// Result of trying to resume an app-password session.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum CredentialResumeResult {
+    /// A stored session was found and activated.
+    Resumed(AtpSession),
+    /// No stored session matched; login credentials are required.
+    LoginRequired(CredentialLoginChallenge),
+}
+
+/// Login identity details derived from a failed resume hint.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CredentialLoginChallenge {
+    /// Login identifier known from the hint, if any.
+    pub identifier: Option<SmolStr>,
+    /// Session id known from the hint, if any.
+    pub session_id: Option<SmolStr>,
+}
+
+/// Options for hint/challenge-based app-password login helpers.
+#[derive(Debug, Clone)]
+pub struct CredentialLoginOptions<'a> {
+    /// App-password or account password.
+    pub password: CowStr<'a>,
+    /// Login identifier override, required when the challenge has no identifier.
+    pub identifier: Option<CowStr<'a>>,
+    /// Whether taken-down accounts are allowed.
+    pub allow_takendown: Option<bool>,
+    /// Optional auth factor token.
+    pub auth_factor_token: Option<CowStr<'a>>,
+    /// Explicit PDS/entryway endpoint to use for login.
+    pub pds: Option<Uri<String>>,
+}
+
+/// Resolver-backed app-password session selector.
+///
+/// This adapter returns richer credential-session match data while keeping selection pluggable:
+/// database-backed stores can provide their own [`SessionSelector`] implementation with more
+/// efficient indexed lookup.
+pub struct CredentialSessionSelector<'a, S, R> {
+    store: &'a S,
+    resolver: &'a R,
+}
+
+impl<'a, S, R> CredentialSessionSelector<'a, S, R> {
+    /// Create a selector over an app-password session store and identity resolver.
+    pub fn new(store: &'a S, resolver: &'a R) -> Self {
+        Self { store, resolver }
+    }
+}
+
+impl<S, R> SessionSelector<CredentialSessionMatch> for CredentialSessionSelector<'_, S, R>
+where
+    S: SessionStore<SessionKey, AtpSession>
+        + SessionSelector<CredentialSessionMatch, Error = ClientError>
+        + Sync,
+    R: IdentityResolver + Sync,
+{
+    type Error = ClientError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<CredentialSessionMatch>, Self::Error> {
+        if let Some(matched) = self.store.select_session(hint).await? {
+            return Ok(Some(matched));
+        }
+
+        let SessionHint::Handle(handle) = hint else {
+            return Ok(None);
+        };
 
-/// Storage key for app‑password sessions: `(account DID, session id)`.
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct SessionKey(pub Did, pub SmolStr);
+        let did = self.resolver.resolve_handle(handle).await?;
+        self.store.select_session(&SessionHint::Did(did)).await
+    }
+}
+
+/// Resolve a session hint against an app-password [`SessionStore`].
+///
+/// This is a convenience wrapper around [`CredentialSessionSelector`].
+pub async fn resolve_credential_session_hint<S, R>(
+    store: &S,
+    resolver: &R,
+    hint: &SessionHint,
+) -> Result<Option<CredentialSessionMatch>, ClientError>
+where
+    S: SessionStore<SessionKey, AtpSession>
+        + SessionSelector<CredentialSessionMatch, Error = ClientError>
+        + Sync,
+    R: IdentityResolver + Sync,
+{
+    CredentialSessionSelector::new(store, resolver)
+        .select_session(hint)
+        .await
+}
+
+async fn match_credential_session_key<S>(
+    store: &S,
+    key: SessionKey,
+) -> Result<Option<CredentialSessionMatch>, ClientError>
+where
+    S: SessionStore<SessionKey, AtpSession>,
+{
+    Ok(store
+        .get(&key)
+        .await
+        .map(|session| CredentialSessionMatch { key, session }))
+}
+
+impl SessionSelector<CredentialSessionMatch> for MemorySessionStore<SessionKey, AtpSession> {
+    type Error = ClientError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<CredentialSessionMatch>, Self::Error> {
+        match hint {
+            SessionHint::Any => {
+                let Some(key) = self.list_keys().await?.into_iter().next() else {
+                    return Ok(None);
+                };
+                match_credential_session_key(self, key).await
+            }
+            SessionHint::Did(did) => {
+                for key in self.list_keys().await? {
+                    if key.did.as_str() == did.as_ref() {
+                        if let Some(matched) = match_credential_session_key(self, key).await? {
+                            return Ok(Some(matched));
+                        }
+                    }
+                }
+                Ok(None)
+            }
+            SessionHint::Handle(handle) => {
+                for key in self.list_keys().await? {
+                    if let Some(session) = self.get(&key).await {
+                        if session.handle.as_str() == handle.as_ref() {
+                            return Ok(Some(CredentialSessionMatch { key, session }));
+                        }
+                    }
+                }
+                Ok(None)
+            }
+            SessionHint::Key(key) => match_credential_session_key(self, key.clone()).await,
+            SessionHint::Identifier(_) => Ok(None),
+        }
+    }
+}
+
+fn credential_challenge_from_hint(hint: &SessionHint) -> CredentialLoginChallenge {
+    match hint {
+        SessionHint::Any => CredentialLoginChallenge {
+            identifier: None,
+            session_id: None,
+        },
+        SessionHint::Did(did) => CredentialLoginChallenge {
+            identifier: Some(did.as_str().to_smolstr()),
+            session_id: None,
+        },
+        SessionHint::Handle(handle) => CredentialLoginChallenge {
+            identifier: Some(handle.as_str().to_smolstr()),
+            session_id: None,
+        },
+        SessionHint::Key(key) => CredentialLoginChallenge {
+            identifier: Some(key.did.as_str().to_smolstr()),
+            session_id: Some(key.session_id.clone()),
+        },
+        SessionHint::Identifier(identifier) => CredentialLoginChallenge {
+            identifier: Some(identifier.clone()),
+            session_id: None,
+        },
+    }
+}
 
 /// Stateful client for app‑password based sessions.
 ///
@@ -160,7 +338,9 @@ where
         let session = self.store.get(&key).await;
         let endpoint = self.endpoint().await;
         let mut opts = self.options.read().await.clone();
-        opts.auth = session.map(|s| AuthorizationToken::Bearer(s.refresh_jwt));
+        opts.auth = session
+            .as_ref()
+            .map(|s| AuthorizationToken::Bearer(s.refresh_jwt.clone()));
         let response = self
             .client
             .xrpc(endpoint.borrow())
@@ -173,7 +353,8 @@ where
                 .with_url("com.atproto.server.refreshSession")
         })?;
 
-        let new_session: AtpSession = refresh.into();
+        let mut new_session = session.unwrap_or_else(|| AtpSession::from(refresh.clone()));
+        new_session.merge_refresh(refresh);
         let token = AuthorizationToken::Bearer(new_session.access_jwt.clone());
         self.store.set(key, new_session).await.map_err(|e| {
             ClientError::from(e).with_context("failed to persist refreshed session to store")
@@ -201,10 +382,7 @@ where
         allow_takendown: Option<bool>,
         auth_factor_token: Option<CowStr<'_>>,
         pds: Option<Uri<String>>,
-    ) -> std::result::Result<AtpSession, ClientError>
-    where
-        S: Any + 'static,
-    {
+    ) -> std::result::Result<AtpSession, ClientError> {
         #[cfg(feature = "tracing")]
         let _span =
             tracing::info_span!("credential_session_login", identifier = %identifier).entered();
@@ -284,85 +462,142 @@ where
                 .with_help("check identifier and password are correct")
                 .with_url("com.atproto.server.createSession")
         })?;
-        let session = AtpSession::from(out);
+        let mut session = AtpSession::from(out);
+        if session.pds.is_none() {
+            session.pds = Some(jacquard_common::xrpc::normalize_base_uri(pds.clone()));
+        }
 
         let sid = session_id.unwrap_or_else(|| CowStr::new_static("session"));
-        let key = SessionKey(session.did.clone().convert::<SmolStr>(), SmolStr::from(sid));
+        let key = SessionKey::new(session.did.clone().convert::<SmolStr>(), SmolStr::from(sid));
         self.store
             .set(key.clone(), session.clone())
             .await
             .map_err(|e| ClientError::from(e).with_context("failed to persist session to store"))?;
-        // If using FileAuthStore, persist PDS for faster resume
-        if let Some(file_store) =
-            (&*self.store as &dyn Any).downcast_ref::<crate::client::token::FileAuthStore>()
-        {
-            let _ = file_store.set_atp_pds(&key, &pds);
-        }
         // Activate
         *self.key.write().await = Some(key);
-        let pds_uri = jacquard_common::xrpc::normalize_base_uri(pds);
+        let pds_uri = jacquard_common::xrpc::normalize_base_uri(session.pds.clone().unwrap_or(pds));
         *self.endpoint.write().await = Some(pds_uri);
 
         Ok(session)
     }
 
+    async fn activate_session(
+        &self,
+        key: SessionKey,
+        mut session: AtpSession,
+    ) -> std::result::Result<AtpSession, ClientError> {
+        let pds = if let Some(pds) = session.pds.clone() {
+            pds
+        } else {
+            let resp = self.client.resolve_did_doc(&session.did).await?;
+            let pds = resp
+                .into_owned()?
+                .pds_endpoint()
+                .map(|u| u.to_owned())
+                .ok_or_else(|| {
+                    ClientError::invalid_request("missing PDS endpoint")
+                        .with_help("DID document must include a PDS service endpoint")
+                })?;
+            session.pds = Some(jacquard_common::xrpc::normalize_base_uri(pds));
+            self.store
+                .set(key.clone(), session.clone())
+                .await
+                .map_err(|e| {
+                    ClientError::from(e).with_context("failed to persist session PDS to store")
+                })?;
+            session.pds.clone().expect("pds just set")
+        };
+
+        *self.key.write().await = Some(key);
+        *self.endpoint.write().await = Some(jacquard_common::xrpc::normalize_base_uri(pds));
+        Ok(session)
+    }
+
+    /// Try to resume a stored app-password session for the given hint.
+    pub async fn resume(&self, hint: &SessionHint) -> Result<CredentialResumeResult, ClientError>
+    where
+        S: SessionSelector<CredentialSessionMatch, Error = ClientError>,
+    {
+        match self.store.select_session(hint).await? {
+            Some(matched) => {
+                let session = self.activate_session(matched.key, matched.session).await?;
+                Ok(CredentialResumeResult::Resumed(session))
+            }
+            None => Ok(CredentialResumeResult::LoginRequired(
+                credential_challenge_from_hint(hint),
+            )),
+        }
+    }
+
+    /// Login using identity details from a resume challenge.
+    pub async fn login_from_challenge(
+        &self,
+        challenge: CredentialLoginChallenge,
+        options: CredentialLoginOptions<'_>,
+    ) -> Result<AtpSession, ClientError> {
+        let identifier = challenge
+            .identifier
+            .map(CowStr::from)
+            .or(options.identifier)
+            .ok_or_else(|| {
+                ClientError::invalid_request("missing login identifier").with_help(
+                    "provide CredentialLoginOptions::identifier for an Any resume challenge",
+                )
+            })?;
+        self.login(
+            identifier,
+            options.password,
+            challenge.session_id.map(CowStr::from),
+            options.allow_takendown,
+            options.auth_factor_token,
+            options.pds,
+        )
+        .await
+    }
+
+    /// Login using identity details derived from a session hint.
+    pub async fn login_with_hint(
+        &self,
+        hint: &SessionHint,
+        options: CredentialLoginOptions<'_>,
+    ) -> Result<AtpSession, ClientError> {
+        self.login_from_challenge(credential_challenge_from_hint(hint), options)
+            .await
+    }
+
+    /// Resume a stored session if available, otherwise login using the provided options.
+    pub async fn resume_or_login(
+        &self,
+        hint: &SessionHint,
+        options: CredentialLoginOptions<'_>,
+    ) -> Result<AtpSession, ClientError>
+    where
+        S: SessionSelector<CredentialSessionMatch, Error = ClientError>,
+    {
+        match self.resume(hint).await? {
+            CredentialResumeResult::Resumed(session) => Ok(session),
+            CredentialResumeResult::LoginRequired(challenge) => {
+                self.login_from_challenge(challenge, options).await
+            }
+        }
+    }
+
     /// Restore a previously persisted app-password session and set base endpoint.
     pub async fn restore(
         &self,
         did: Did,
         session_id: CowStr<'_>,
-    ) -> std::result::Result<(), ClientError>
-    where
-        S: Any + 'static,
-    {
+    ) -> std::result::Result<(), ClientError> {
         #[cfg(feature = "tracing")]
         let _span =
             tracing::info_span!("credential_session_restore", did = %did, session_id = %session_id)
                 .entered();
 
-        let key = SessionKey(did.clone(), SmolStr::from(session_id.clone()));
+        let key = SessionKey::new(did, SmolStr::from(session_id));
         let Some(sess) = self.store.get(&key).await else {
             return Err(ClientError::auth(AuthError::NotAuthenticated));
         };
-        // Try to read cached PDS; otherwise resolve via DID
-        let pds = if let Some(file_store) =
-            (&*self.store as &dyn Any).downcast_ref::<crate::client::token::FileAuthStore>()
-        {
-            file_store.get_atp_pds(&key).ok().flatten().or_else(|| None)
-        } else {
-            None
-        }
-        .unwrap_or({
-            let resp = self.client.resolve_did_doc(&did).await?;
-            resp.into_owned()?
-                .pds_endpoint()
-                .map(|u| u.to_owned())
-                .ok_or_else(|| {
-                    ClientError::invalid_request("missing PDS endpoint")
-                        .with_help("DID document must include a PDS service endpoint")
-                })?
-        });
-
-        // Activate
-        *self.key.write().await = Some(key.clone());
-        let pds_uri = jacquard_common::xrpc::normalize_base_uri(pds);
-        *self.endpoint.write().await = Some(pds_uri.clone());
-        // ensure store has the session (no-op if it existed)
-        self.store
-            .set(
-                SessionKey(
-                    sess.did.clone().convert::<SmolStr>(),
-                    SmolStr::from(session_id),
-                ),
-                sess,
-            )
-            .await?;
-        if let Some(file_store) =
-            (&*self.store as &dyn Any).downcast_ref::<crate::client::token::FileAuthStore>()
-        {
-            let _ = file_store.set_atp_pds(&key, &pds_uri);
-        }
-        Ok(())
+        self.activate_session(key, sess).await.map(|_| ())
     }
 
     /// Switch to a different stored session (and refresh endpoint/PDS).
@@ -370,41 +605,12 @@ where
         &self,
         did: Did,
         session_id: CowStr<'_>,
-    ) -> std::result::Result<(), ClientError>
-    where
-        S: Any + 'static,
-    {
-        let key = SessionKey(did.clone(), SmolStr::from(session_id));
-        if self.store.get(&key).await.is_none() {
+    ) -> std::result::Result<(), ClientError> {
+        let key = SessionKey::new(did, SmolStr::from(session_id));
+        let Some(sess) = self.store.get(&key).await else {
             return Err(ClientError::auth(AuthError::NotAuthenticated));
-        }
-        // Endpoint from store if cached, else resolve
-        let pds = if let Some(file_store) =
-            (&*self.store as &dyn Any).downcast_ref::<crate::client::token::FileAuthStore>()
-        {
-            file_store.get_atp_pds(&key).ok().flatten().or_else(|| None)
-        } else {
-            None
-        }
-        .unwrap_or({
-            let resp = self.client.resolve_did_doc(&did).await?;
-            resp.into_owned()?
-                .pds_endpoint()
-                .map(|u| u.to_owned())
-                .ok_or_else(|| {
-                    ClientError::invalid_request("missing PDS endpoint")
-                        .with_help("DID document must include a PDS service endpoint")
-                })?
-        });
-        *self.key.write().await = Some(key.clone());
-        let pds_uri = jacquard_common::xrpc::normalize_base_uri(pds);
-        *self.endpoint.write().await = Some(pds_uri.clone());
-        if let Some(file_store) =
-            (&*self.store as &dyn Any).downcast_ref::<crate::client::token::FileAuthStore>()
-        {
-            let _ = file_store.set_atp_pds(&key, &pds_uri);
-        }
-        Ok(())
+        };
+        self.activate_session(key, sess).await.map(|_| ())
     }
 
     /// Clear and delete the current session from the store.
diff --git a/crates/jacquard/src/client/token.rs b/crates/jacquard/src/client/token.rs
index 5a3a95a27..00aa54330 100644
--- a/crates/jacquard/src/client/token.rs
+++ b/crates/jacquard/src/client/token.rs
@@ -1,12 +1,14 @@
+use jacquard_common::IntoStatic;
 use jacquard_common::deps::fluent_uri::Uri;
-use jacquard_common::session::{FileTokenStore, SessionStore, SessionStoreError};
+use jacquard_common::session::{
+    FileTokenStore, SessionHint, SessionKey, SessionSelector, SessionStore, SessionStoreError,
+};
 use jacquard_common::types::string::{Datetime, Did};
 use jacquard_oauth::scopes::Scopes;
 use jacquard_oauth::session::{AuthRequestData, ClientSessionData, DpopClientData, DpopReqData};
 use jacquard_oauth::types::OAuthTokenType;
 use jose_jwk::Key;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 use smol_str::SmolStr;
 
 /// On-disk session records for app-password and OAuth flows, sharing a single JSON map.
@@ -20,22 +22,13 @@ pub enum StoredSession {
     OAuthState(OAuthState),
 }
 
-/// Minimal persisted representation of an app‑password session.
+/// Persisted representation of an app-password session plus its store-local session id.
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct StoredAtSession {
-    /// Access token (JWT)
-    access_jwt: String,
-    /// Refresh token (JWT)
-    refresh_jwt: String,
-    /// Account DID
-    did: String,
-    /// Optional PDS endpoint for faster resume
-    #[serde(skip_serializing_if = "std::option::Option::is_none")]
-    pds: Option<String>,
     /// Session id label (e.g., "session")
-    session_id: String,
-    /// Last known handle
-    handle: String,
+    pub session_id: String,
+    /// Stored app-password session.
+    pub session: crate::client::AtpSession,
 }
 
 /// Persisted OAuth client session (on-disk format).
@@ -265,6 +258,18 @@ impl FileAuthStore {
     pub fn new(path: impl AsRef<std::path::Path>) -> Self {
         Self(FileTokenStore::new(path))
     }
+
+    fn atp_key(key: &SessionKey) -> String {
+        format!("atp:{}", key)
+    }
+
+    fn oauth_key(key: &SessionKey) -> String {
+        format!("oauth:{}", key)
+    }
+
+    fn oauth_state_key(state: &str) -> String {
+        format!("oauth-state:{}", state)
+    }
 }
 
 impl jacquard_oauth::authstore::ClientAuthStore for FileAuthStore {
@@ -273,13 +278,11 @@ impl jacquard_oauth::authstore::ClientAuthStore for FileAuthStore {
         did: &Did<D>,
         session_id: &str,
     ) -> Result<Option<ClientSessionData>, SessionStoreError> {
-        let key = format!("{}_{}", did, session_id);
-        if let StoredSession::OAuth(session) = self
-            .0
-            .get(&key)
-            .await
-            .ok_or(SessionStoreError::Other("not found".into()))?
-        {
+        let key = SessionKey::new(did.borrow().into_static(), session_id);
+        let Some(value) = self.0.get_value(&Self::oauth_key(&key))? else {
+            return Ok(None);
+        };
+        if let StoredSession::OAuth(session) = serde_json::from_value(value)? {
             Ok(Some(session.into()))
         } else {
             Ok(None)
@@ -287,10 +290,11 @@ impl jacquard_oauth::authstore::ClientAuthStore for FileAuthStore {
     }
 
     async fn upsert_session(&self, session: ClientSessionData) -> Result<(), SessionStoreError> {
-        let key = format!("{}_{}", session.account_did, session.session_id);
-        self.0
-            .set(key, StoredSession::OAuth(session.into()))
-            .await?;
+        let key = SessionKey::new(session.account_did.clone(), session.session_id.clone());
+        self.0.set_value(
+            Self::oauth_key(&key),
+            serde_json::to_value(StoredSession::OAuth(session.into()))?,
+        )?;
         Ok(())
     }
 
@@ -299,31 +303,19 @@ impl jacquard_oauth::authstore::ClientAuthStore for FileAuthStore {
         did: &Did<D>,
         session_id: &str,
     ) -> Result<(), SessionStoreError> {
-        let key = format!("{}_{}", did, session_id);
-        let file = std::fs::read_to_string(&self.0.path)?;
-        let mut store: Value = serde_json::from_str(&file)?;
-        let key_string = key.to_string();
-        if let Some(store) = store.as_object_mut() {
-            store.remove(&key_string);
-
-            std::fs::write(&self.0.path, serde_json::to_string_pretty(&store)?)?;
-            Ok(())
-        } else {
-            Err(SessionStoreError::Other("invalid store".into()))
-        }
+        let key = SessionKey::new(did.borrow().into_static(), session_id);
+        self.0.remove_value(&Self::oauth_key(&key))
     }
 
     async fn get_auth_req_info(
         &self,
         state: &str,
     ) -> Result<Option<AuthRequestData>, SessionStoreError> {
-        let key = format!("authreq_{}", state);
-        if let StoredSession::OAuthState(auth_req) = self
-            .0
-            .get(&key)
-            .await
-            .ok_or(SessionStoreError::Other("not found".into()))?
-        {
+        let key = Self::oauth_state_key(state);
+        let Some(value) = self.0.get_value(&key)? else {
+            return Ok(None);
+        };
+        if let StoredSession::OAuthState(auth_req) = serde_json::from_value(value)? {
             Ok(Some(auth_req.into()))
         } else {
             Ok(None)
@@ -334,98 +326,43 @@ impl jacquard_oauth::authstore::ClientAuthStore for FileAuthStore {
         &self,
         auth_req_info: &AuthRequestData,
     ) -> Result<(), SessionStoreError> {
-        let key = format!("authreq_{}", auth_req_info.state);
+        let key = Self::oauth_state_key(&auth_req_info.state);
         let state = auth_req_info.clone().try_into().map_err(
             |e: jacquard_common::deps::fluent_uri::ParseError| {
                 SessionStoreError::Other(Box::new(e))
             },
         )?;
-        self.0.set(key, StoredSession::OAuthState(state)).await?;
+        self.0
+            .set_value(key, serde_json::to_value(StoredSession::OAuthState(state))?)?;
         Ok(())
     }
 
     async fn delete_auth_req_info(&self, state: &str) -> Result<(), SessionStoreError> {
-        let key = format!("authreq_{}", state);
-        let file = std::fs::read_to_string(&self.0.path)?;
-        let mut store: Value = serde_json::from_str(&file)?;
-        let key_string = key.to_string();
-        if let Some(store) = store.as_object_mut() {
-            store.remove(&key_string);
-
-            std::fs::write(&self.0.path, serde_json::to_string_pretty(&store)?)?;
-            Ok(())
-        } else {
-            Err(SessionStoreError::Other("invalid store".into()))
-        }
+        let key = Self::oauth_state_key(state);
+        self.0.remove_value(&key)
     }
-}
 
-impl FileAuthStore {
-    /// Update the persisted PDS endpoint for an app-password session (best-effort).
-    pub fn set_atp_pds(
-        &self,
-        key: &crate::client::credential_session::SessionKey,
-        pds: &Uri<String>,
-    ) -> Result<(), SessionStoreError> {
-        let key_str = format!("{}_{}", key.0, key.1);
-        let file = std::fs::read_to_string(&self.0.path)?;
-        let mut store: Value = serde_json::from_str(&file)?;
-        if let Some(map) = store.as_object_mut() {
-            if let Some(value) = map.get_mut(&key_str) {
-                if let Some(outer) = value.as_object_mut() {
-                    if let Some(inner) = outer.get_mut("Atp").and_then(|v| v.as_object_mut()) {
-                        inner.insert(
-                            "pds".to_string(),
-                            serde_json::Value::String(pds.as_str().to_string()),
-                        );
-                        std::fs::write(&self.0.path, serde_json::to_string_pretty(&store)?)?;
-                        return Ok(());
-                    }
-                }
+    async fn list_session_keys(&self) -> Result<Vec<SessionKey>, SessionStoreError> {
+        let mut keys = Vec::new();
+        for (_key, value) in self.0.entries()? {
+            if let Ok(StoredSession::OAuth(session)) =
+                serde_json::from_value::<StoredSession>(value)
+            {
+                keys.push(SessionKey::new(
+                    Did::new_owned(session.account_did).expect("stored DID should be valid"),
+                    session.session_id,
+                ));
             }
         }
-        Err(SessionStoreError::Other("invalid store".into()))
-    }
-
-    /// Read the persisted PDS endpoint for an app-password session, if present.
-    pub fn get_atp_pds(
-        &self,
-        key: &crate::client::credential_session::SessionKey,
-    ) -> Result<Option<Uri<String>>, SessionStoreError> {
-        let key_str = format!("{}_{}", key.0, key.1);
-        let file = std::fs::read_to_string(&self.0.path)?;
-        let store: Value = serde_json::from_str(&file)?;
-        if let Some(value) = store.get(&key_str) {
-            if let Some(obj) = value.as_object() {
-                if let Some(serde_json::Value::Object(inner)) = obj.get("Atp") {
-                    if let Some(serde_json::Value::String(pds)) = inner.get("pds") {
-                        return Ok(Uri::parse(pds.as_str()).ok().map(|u| u.to_owned()));
-                    }
-                }
-            }
-        }
-        Ok(None)
+        Ok(keys)
     }
 }
 
-impl
-    jacquard_common::session::SessionStore<
-        crate::client::credential_session::SessionKey,
-        crate::client::AtpSession,
-    > for FileAuthStore
-{
-    async fn get(
-        &self,
-        key: &crate::client::credential_session::SessionKey,
-    ) -> Option<crate::client::AtpSession> {
-        let key_str = format!("{}_{}", key.0, key.1);
-        if let Some(StoredSession::Atp(stored)) = self.0.get(&key_str).await {
-            Some(crate::client::AtpSession {
-                access_jwt: stored.access_jwt.into(),
-                refresh_jwt: stored.refresh_jwt.into(),
-                did: stored.did.into(),
-                handle: stored.handle.into(),
-            })
+impl SessionStore<SessionKey, crate::client::AtpSession> for FileAuthStore {
+    async fn get(&self, key: &SessionKey) -> Option<crate::client::AtpSession> {
+        let value = self.0.get_value(&Self::atp_key(key)).ok()??;
+        if let Ok(StoredSession::Atp(stored)) = serde_json::from_value::<StoredSession>(value) {
+            Some(stored.session)
         } else {
             None
         }
@@ -433,42 +370,148 @@ impl
 
     async fn set(
         &self,
-        key: crate::client::credential_session::SessionKey,
+        key: SessionKey,
         session: crate::client::AtpSession,
     ) -> Result<(), jacquard_common::session::SessionStoreError> {
-        let key_str = format!("{}_{}", key.0, key.1);
         let stored = StoredAtSession {
-            access_jwt: session.access_jwt.to_string(),
-            refresh_jwt: session.refresh_jwt.to_string(),
-            did: session.did.to_string(),
-            // pds endpoint is resolved on restore; do not persist
-            pds: None,
-            session_id: key.1.to_string(),
-            handle: session.handle.to_string(),
+            session_id: key.session_id.to_string(),
+            session,
         };
-        self.0.set(key_str, StoredSession::Atp(stored)).await
+        self.0.set_value(
+            Self::atp_key(&key),
+            serde_json::to_value(StoredSession::Atp(stored))?,
+        )
     }
 
     async fn del(
         &self,
-        key: &crate::client::credential_session::SessionKey,
+        key: &SessionKey,
     ) -> Result<(), jacquard_common::session::SessionStoreError> {
-        let key_str = format!("{}_{}", key.0, key.1);
-        // Manual removal to mirror existing pattern
-        let file = std::fs::read_to_string(&self.0.path)?;
-        let mut store: serde_json::Value = serde_json::from_str(&file)?;
-        if let Some(map) = store.as_object_mut() {
-            map.remove(&key_str);
-            std::fs::write(&self.0.path, serde_json::to_string_pretty(&store)?)?;
-            Ok(())
-        } else {
-            Err(jacquard_common::session::SessionStoreError::Other(
-                "invalid store".into(),
-            ))
+        self.0.remove_value(&Self::atp_key(key))
+    }
+
+    async fn list_keys(&self) -> Result<Vec<SessionKey>, SessionStoreError> {
+        let mut keys = Vec::new();
+        for (_key, value) in self.0.entries()? {
+            if let Ok(StoredSession::Atp(session)) = serde_json::from_value::<StoredSession>(value)
+            {
+                keys.push(SessionKey::new(
+                    session.session.did.clone(),
+                    session.session_id,
+                ));
+            }
+        }
+        Ok(keys)
+    }
+}
+
+impl SessionSelector<crate::client::credential_session::CredentialSessionMatch> for FileAuthStore {
+    type Error = jacquard_common::error::ClientError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<crate::client::credential_session::CredentialSessionMatch>, Self::Error>
+    {
+        match hint {
+            SessionHint::Any => {
+                let Some(key) = SessionStore::list_keys(self).await?.into_iter().next() else {
+                    return Ok(None);
+                };
+                Ok(SessionStore::get(self, &key).await.map(|session| {
+                    crate::client::credential_session::CredentialSessionMatch { key, session }
+                }))
+            }
+            SessionHint::Did(did) => {
+                for key in SessionStore::list_keys(self).await? {
+                    if key.did.as_str() == did.as_ref() {
+                        if let Some(session) = SessionStore::get(self, &key).await {
+                            return Ok(Some(
+                                crate::client::credential_session::CredentialSessionMatch {
+                                    key,
+                                    session,
+                                },
+                            ));
+                        }
+                    }
+                }
+                Ok(None)
+            }
+            SessionHint::Handle(handle) => {
+                for key in SessionStore::list_keys(self).await? {
+                    if let Some(session) = SessionStore::get(self, &key).await {
+                        if session.handle.as_str() == handle.as_ref() {
+                            return Ok(Some(
+                                crate::client::credential_session::CredentialSessionMatch {
+                                    key,
+                                    session,
+                                },
+                            ));
+                        }
+                    }
+                }
+                Ok(None)
+            }
+            SessionHint::Key(key) => Ok(SessionStore::get(self, key).await.map(|session| {
+                crate::client::credential_session::CredentialSessionMatch {
+                    key: key.clone(),
+                    session,
+                }
+            })),
+            SessionHint::Identifier(_) => Ok(None),
         }
     }
 }
 
+impl SessionSelector<jacquard_oauth::authstore::OAuthSessionMatch> for FileAuthStore {
+    type Error = SessionStoreError;
+
+    async fn select_session(
+        &self,
+        hint: &SessionHint,
+    ) -> Result<Option<jacquard_oauth::authstore::OAuthSessionMatch>, Self::Error> {
+        match hint {
+            SessionHint::Any => {
+                let Some(key) = jacquard_oauth::authstore::ClientAuthStore::list_session_keys(self)
+                    .await?
+                    .into_iter()
+                    .next()
+                else {
+                    return Ok(None);
+                };
+                oauth_match_for_key_file(self, key).await
+            }
+            SessionHint::Did(did) => {
+                for key in
+                    jacquard_oauth::authstore::ClientAuthStore::list_session_keys(self).await?
+                {
+                    if key.did.as_str() == did.as_ref() {
+                        if let Some(matched) = oauth_match_for_key_file(self, key).await? {
+                            return Ok(Some(matched));
+                        }
+                    }
+                }
+                Ok(None)
+            }
+            SessionHint::Handle(_) | SessionHint::Identifier(_) => Ok(None),
+            SessionHint::Key(key) => oauth_match_for_key_file(self, key.clone()).await,
+        }
+    }
+}
+
+async fn oauth_match_for_key_file(
+    store: &FileAuthStore,
+    key: SessionKey,
+) -> Result<Option<jacquard_oauth::authstore::OAuthSessionMatch>, SessionStoreError> {
+    Ok(jacquard_oauth::authstore::ClientAuthStore::get_session(
+        store,
+        &key.did,
+        key.session_id.as_str(),
+    )
+    .await?
+    .map(|session| jacquard_oauth::authstore::OAuthSessionMatch { key, session }))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -480,10 +523,45 @@ mod tests {
 
     fn temp_file() -> PathBuf {
         let mut p = std::env::temp_dir();
-        p.push(format!("jacquard-test-{}.json", std::process::id()));
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        p.push(format!("jacquard-test-{}-{nanos}.json", std::process::id()));
         p
     }
 
+    fn oauth_session(did: &'static str, session_id: &'static str) -> ClientSessionData {
+        let account_did = Did::new_static(did).unwrap();
+        ClientSessionData {
+            account_did: account_did.clone(),
+            session_id: SmolStr::new_static(session_id),
+            host_url: Uri::parse("https://pds.example.com").unwrap().to_owned(),
+            authserver_url: SmolStr::new_static("https://issuer.example.com"),
+            authserver_token_endpoint: SmolStr::new_static("https://issuer.example.com/token"),
+            authserver_revocation_endpoint: None,
+            scopes: Scopes::empty(),
+            dpop_data: DpopClientData {
+                dpop_key: jacquard_oauth::utils::generate_key(&[SmolStr::new_static("ES256")])
+                    .unwrap(),
+                dpop_authserver_nonce: SmolStr::default(),
+                dpop_host_nonce: SmolStr::default(),
+            },
+            token_set: jacquard_oauth::types::TokenSet {
+                iss: SmolStr::new_static("https://issuer.example.com"),
+                sub: account_did,
+                aud: SmolStr::new_static("https://pds.example.com"),
+                scope: None,
+                refresh_token: None,
+                access_token: SmolStr::new_static("access"),
+                token_type: OAuthTokenType::DPoP,
+                expires_at: None,
+            },
+            #[cfg(feature = "scope-check")]
+            resolved_scopes: None,
+        }
+    }
+
     #[tokio::test]
     async fn file_auth_store_roundtrip_atp() {
         let path = temp_file();
@@ -495,8 +573,9 @@ mod tests {
             refresh_jwt: "r".into(),
             did: Did::new_static("did:plc:alice").unwrap(),
             handle: Handle::new_static("alice.bsky.social").unwrap(),
+            pds: None,
         };
-        let key = SessionKey(session.did.clone(), "session".into());
+        let key = SessionKey::new(session.did.clone(), "session");
         jacquard_common::session::SessionStore::set(&store, key.clone(), session.clone())
             .await
             .unwrap();
@@ -507,4 +586,68 @@ mod tests {
         // clean up
         let _ = fs::remove_file(&path);
     }
+
+    #[tokio::test]
+    async fn file_auth_store_lists_only_atp_keys() {
+        let path = temp_file();
+        fs::write(&path, "{}").unwrap();
+        let store = FileAuthStore::new(&path);
+        let atp = AtpSession {
+            access_jwt: "a".into(),
+            refresh_jwt: "r".into(),
+            did: Did::new_static("did:plc:alice").unwrap(),
+            handle: Handle::new_static("alice.bsky.social").unwrap(),
+            pds: None,
+        };
+        let atp_key = SessionKey::new(atp.did.clone(), "session");
+        SessionStore::set(&store, atp_key.clone(), atp)
+            .await
+            .unwrap();
+        jacquard_oauth::authstore::ClientAuthStore::upsert_session(
+            &store,
+            oauth_session("did:plc:bob", "oauth-session"),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(
+            SessionStore::list_keys(&store).await.unwrap(),
+            vec![atp_key]
+        );
+        let _ = fs::remove_file(&path);
+    }
+
+    #[tokio::test]
+    async fn file_auth_store_lists_only_oauth_keys() {
+        let path = temp_file();
+        fs::write(&path, "{}").unwrap();
+        let store = FileAuthStore::new(&path);
+        let atp = AtpSession {
+            access_jwt: "a".into(),
+            refresh_jwt: "r".into(),
+            did: Did::new_static("did:plc:alice").unwrap(),
+            handle: Handle::new_static("alice.bsky.social").unwrap(),
+            pds: None,
+        };
+        SessionStore::set(&store, SessionKey::new(atp.did.clone(), "session"), atp)
+            .await
+            .unwrap();
+        jacquard_oauth::authstore::ClientAuthStore::upsert_session(
+            &store,
+            oauth_session("did:plc:bob", "oauth-session"),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(
+            jacquard_oauth::authstore::ClientAuthStore::list_session_keys(&store)
+                .await
+                .unwrap(),
+            vec![SessionKey::new(
+                Did::new_static("did:plc:bob").unwrap(),
+                "oauth-session",
+            )]
+        );
+        let _ = fs::remove_file(&path);
+    }
 }
diff --git a/crates/jacquard/tests/agent.rs b/crates/jacquard/tests/agent.rs
index f23e43cea..8c160eeb1 100644
--- a/crates/jacquard/tests/agent.rs
+++ b/crates/jacquard/tests/agent.rs
@@ -97,8 +97,9 @@ async fn agent_delegates_to_session_and_refreshes() {
         refresh_jwt: "ref1".into(),
         did: Did::new_static("did:plc:alice").unwrap(),
         handle: Handle::new_static("alice.bsky.social").unwrap(),
+        pds: None,
     };
-    let key = SessionKey(atp.did.clone(), "session".into());
+    let key = SessionKey::new(atp.did.clone(), "session");
     jacquard_common::session::SessionStore::set(store.as_ref(), key.clone(), atp)
         .await
         .unwrap();
diff --git a/crates/jacquard/tests/credential_session.rs b/crates/jacquard/tests/credential_session.rs
index a7293995d..bfd562b64 100644
--- a/crates/jacquard/tests/credential_session.rs
+++ b/crates/jacquard/tests/credential_session.rs
@@ -5,14 +5,16 @@ use bytes::Bytes;
 use http::{HeaderValue, Method, Response as HttpResponse, StatusCode};
 use jacquard::BosStr;
 use jacquard::client::AtpSession;
-use jacquard::client::credential_session::{CredentialSession, SessionKey};
+use jacquard::client::credential_session::{
+    CredentialResumeResult, CredentialSession, SessionKey, resolve_credential_session_hint,
+};
+use jacquard::deps::fluent_uri::Uri;
 use jacquard::identity::resolver::{DidDocResponse, IdentityResolver, ResolverOptions};
 use jacquard::types::did::Did;
 use jacquard::types::string::Handle;
 use jacquard::xrpc::XrpcClient;
 use jacquard_common::http_client::HttpClient;
-use jacquard_common::session::{MemorySessionStore, SessionStore};
-use smol_str::SmolStr;
+use jacquard_common::session::{MemorySessionStore, SessionHint, SessionStore};
 use tokio::sync::{Mutex, RwLock};
 
 #[derive(Clone, Default)]
@@ -23,6 +25,7 @@ struct MockClient {
     log: Arc<Mutex<Vec<http::Request<Vec<u8>>>>>,
     // Count calls to identity resolver helpers
     did_doc_calls: Arc<RwLock<usize>>,
+    handle_calls: Arc<RwLock<usize>>,
 }
 
 impl MockClient {
@@ -67,6 +70,7 @@ impl IdentityResolver for MockClient {
         handle: &Handle<S>,
     ) -> std::result::Result<Did, jacquard::identity::resolver::IdentityError> {
         // Return a fixed DID for any handle
+        *self.handle_calls.write().await += 1;
         assert!(handle.as_str().contains('.'));
         Ok(Did::new_static("did:plc:alice").unwrap())
     }
@@ -125,6 +129,186 @@ fn get_session_ok_body() -> Vec<u8> {
     .unwrap()
 }
 
+fn atp_session(did: &'static str, handle: &'static str) -> AtpSession {
+    AtpSession {
+        access_jwt: "acc".into(),
+        refresh_jwt: "ref".into(),
+        did: Did::new_static(did).unwrap(),
+        handle: Handle::new_static(handle).unwrap(),
+        pds: None,
+    }
+}
+
+#[tokio::test]
+async fn credential_session_hint_matcher_resolves_common_hints() {
+    let store = MemorySessionStore::<SessionKey, AtpSession>::default();
+    let client = MockClient::default();
+    let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "session");
+    store
+        .set(
+            key.clone(),
+            atp_session("did:plc:alice", "alice.bsky.social"),
+        )
+        .await
+        .unwrap();
+
+    let matched = resolve_credential_session_hint(&store, &client, &SessionHint::Any)
+        .await
+        .unwrap()
+        .expect("any match");
+    assert_eq!(matched.key, key);
+    assert_eq!(matched.session.handle.as_str(), "alice.bsky.social");
+    assert_eq!(matched.session.pds, None);
+
+    let matched = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Did(Did::new_static("did:plc:alice").unwrap()),
+    )
+    .await
+    .unwrap()
+    .expect("did match");
+    assert_eq!(matched.key, key);
+
+    let matched = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Handle(Handle::new_static("alice.bsky.social").unwrap()),
+    )
+    .await
+    .unwrap()
+    .expect("handle match");
+    assert_eq!(matched.key, key);
+
+    let matched = resolve_credential_session_hint(&store, &client, &SessionHint::Key(key.clone()))
+        .await
+        .unwrap()
+        .expect("key match");
+    assert_eq!(matched.key, key);
+
+    let missing = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Key(SessionKey::new(
+            Did::new_static("did:plc:bob").unwrap(),
+            "session",
+        )),
+    )
+    .await
+    .unwrap();
+    assert!(missing.is_none());
+
+    let identifier = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Identifier("alice@example.com".into()),
+    )
+    .await
+    .unwrap();
+    assert!(
+        identifier.is_none(),
+        "identifier hints must not fall back to Any"
+    );
+}
+
+#[tokio::test]
+async fn credential_resolver_selector_uses_store_before_handle_resolution() {
+    let store = MemorySessionStore::<SessionKey, AtpSession>::default();
+    let client = MockClient::default();
+    let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "session");
+    store
+        .set(
+            key.clone(),
+            atp_session("did:plc:alice", "alice.bsky.social"),
+        )
+        .await
+        .unwrap();
+
+    let matched = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Handle(Handle::new_static("alice.bsky.social").unwrap()),
+    )
+    .await
+    .unwrap()
+    .expect("store-level handle match");
+    assert_eq!(matched.key, key);
+    assert_eq!(
+        *client.handle_calls.read().await,
+        0,
+        "store selector should get first chance to satisfy handle hints"
+    );
+
+    let matched = resolve_credential_session_hint(
+        &store,
+        &client,
+        &SessionHint::Handle(Handle::new_static("alias.bsky.social").unwrap()),
+    )
+    .await
+    .unwrap()
+    .expect("resolver fallback did match");
+    assert_eq!(matched.key, key);
+    assert_eq!(
+        *client.handle_calls.read().await,
+        1,
+        "resolver should be used only after store selector misses the handle"
+    );
+}
+
+#[tokio::test]
+async fn credential_resume_returns_challenge_details_without_password() {
+    let store: Arc<MemorySessionStore<SessionKey, AtpSession>> = Arc::new(Default::default());
+    let client = Arc::new(MockClient::default());
+    let session = CredentialSession::new(store, client);
+
+    let result = session.resume(&SessionHint::Any).await.expect("resume any");
+    let CredentialResumeResult::LoginRequired(challenge) = result else {
+        panic!("expected login challenge for empty store");
+    };
+    assert_eq!(challenge.identifier, None);
+    assert_eq!(challenge.session_id, None);
+
+    let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "mobile");
+    let result = session
+        .resume(&SessionHint::Key(key.clone()))
+        .await
+        .expect("resume key");
+    let CredentialResumeResult::LoginRequired(challenge) = result else {
+        panic!("expected login challenge for missing key");
+    };
+    assert_eq!(challenge.identifier.as_deref(), Some(key.did.as_str()));
+    assert_eq!(challenge.session_id.as_deref(), Some("mobile"));
+
+    let result = session
+        .resume(&SessionHint::Identifier("alice@example.com".into()))
+        .await
+        .expect("resume identifier");
+    let CredentialResumeResult::LoginRequired(challenge) = result else {
+        panic!("expected login challenge for identifier");
+    };
+    assert_eq!(challenge.identifier.as_deref(), Some("alice@example.com"));
+    assert_eq!(challenge.session_id, None);
+}
+
+#[tokio::test]
+async fn credential_resume_uses_stored_pds_without_resolving_did_doc() {
+    let store: Arc<MemorySessionStore<SessionKey, AtpSession>> = Arc::new(Default::default());
+    let client = Arc::new(MockClient::default());
+    let session = CredentialSession::new(store.clone(), client.clone());
+    let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "session");
+    let mut stored = atp_session("did:plc:alice", "alice.bsky.social");
+    stored.pds = Some(Uri::parse("https://stored-pds").unwrap().to_owned());
+    store.set(key.clone(), stored.clone()).await.unwrap();
+
+    let result = session.resume(&SessionHint::Any).await.expect("resume any");
+    let CredentialResumeResult::Resumed(resumed) = result else {
+        panic!("expected resumed session");
+    };
+    assert_eq!(resumed.pds.as_ref().unwrap().as_str(), "https://stored-pds");
+    assert_eq!(session.endpoint().await.as_str(), "https://stored-pds");
+    assert_eq!(*client.did_doc_calls.read().await, 0);
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn credential_login_and_auto_refresh() {
     let client = Arc::new(MockClient::default());
@@ -255,10 +439,7 @@ async fn credential_login_and_auto_refresh() {
     );
 
     // Verify store updated with refreshed tokens
-    let key = SessionKey(
-        Did::new_static("did:plc:alice").unwrap(),
-        SmolStr::from("session"),
-    );
+    let key = SessionKey::new(Did::new_static("did:plc:alice").unwrap(), "session");
     let updated = store.get(&key).await.expect("session present");
     assert_eq!(updated.access_jwt.as_str(), "acc2");
     assert_eq!(updated.refresh_jwt.as_str(), "ref2");
diff --git a/crates/jacquard/tests/oauth_auto_refresh.rs b/crates/jacquard/tests/oauth_auto_refresh.rs
index b10646fc8..4c2dfb909 100644
--- a/crates/jacquard/tests/oauth_auto_refresh.rs
+++ b/crates/jacquard/tests/oauth_auto_refresh.rs
@@ -16,7 +16,7 @@ use jacquard_oauth::scopes::Scopes;
 use jacquard_oauth::session::SessionRegistry;
 use jacquard_oauth::session::{ClientData, ClientSessionData, DpopClientData};
 use jacquard_oauth::types::{OAuthAuthorizationServerMetadata, OAuthTokenType, TokenSet};
-use smol_str::SmolStr;
+use smol_str::{SmolStr, format_smolstr};
 use tokio::sync::Mutex;
 
 #[derive(Clone, Default)]
@@ -91,10 +91,10 @@ impl OAuthResolver for MockClient {
         // Return minimal metadata with supported auth method "none" and DPoP support
         let mut md = OAuthAuthorizationServerMetadata::default();
         md.issuer = SmolStr::from(issuer);
-        md.token_endpoint = SmolStr::from(format!("{}/token", issuer));
-        md.authorization_endpoint = SmolStr::from(format!("{}/authorize", issuer));
+        md.token_endpoint = format_smolstr!("{}/token", issuer);
+        md.authorization_endpoint = format_smolstr!("{}/authorize", issuer);
         md.require_pushed_authorization_requests = Some(true);
-        md.pushed_authorization_request_endpoint = Some(SmolStr::from(format!("{}/par", issuer)));
+        md.pushed_authorization_request_endpoint = Some(format_smolstr!("{}/par", issuer));
         md.token_endpoint_auth_methods_supported = Some(vec![SmolStr::from("none")]);
         md.dpop_signing_alg_values_supported = Some(vec![SmolStr::from("ES256")]);
         Ok(md)
diff --git a/crates/jacquard/tests/oauth_flow.rs b/crates/jacquard/tests/oauth_flow.rs
index 151eaeff6..f5a41a1ef 100644
--- a/crates/jacquard/tests/oauth_flow.rs
+++ b/crates/jacquard/tests/oauth_flow.rs
@@ -7,13 +7,14 @@ use jacquard::BosStr;
 use jacquard::client::Agent;
 use jacquard::xrpc::XrpcClient;
 use jacquard_common::http_client::HttpClient;
+use jacquard_common::session::SessionHint;
 use jacquard_oauth::atproto::AtprotoClientMetadata;
 use jacquard_oauth::authstore::ClientAuthStore;
 use jacquard_oauth::client::OAuthClient;
 use jacquard_oauth::resolver::OAuthResolver;
 use jacquard_oauth::scopes::Scopes;
 use jacquard_oauth::session::ClientData;
-use smol_str::SmolStr;
+use smol_str::{SmolStr, format_smolstr};
 
 #[derive(Clone, Default)]
 struct MockClient {
@@ -120,10 +121,10 @@ impl OAuthResolver for MockClient {
     > {
         let mut md = jacquard_oauth::types::OAuthAuthorizationServerMetadata::default();
         md.issuer = SmolStr::from(issuer);
-        md.authorization_endpoint = SmolStr::from(format!("{}/authorize", issuer));
-        md.token_endpoint = SmolStr::from(format!("{}/token", issuer));
+        md.authorization_endpoint = format_smolstr!("{}/authorize", issuer);
+        md.token_endpoint = format_smolstr!("{}/token", issuer);
         md.require_pushed_authorization_requests = Some(true);
-        md.pushed_authorization_request_endpoint = Some(SmolStr::from(format!("{}/par", issuer)));
+        md.pushed_authorization_request_endpoint = Some(format_smolstr!("{}/par", issuer));
         md.token_endpoint_auth_methods_supported = Some(vec![SmolStr::from("none")]);
         md.dpop_signing_alg_values_supported = Some(vec![SmolStr::from("ES256")]);
         Ok(md)
@@ -315,3 +316,40 @@ async fn oauth_end_to_end_mock_flow() {
 
     let _ = std::fs::remove_file(&path);
 }
+
+#[tokio::test]
+async fn oauth_resume_or_start_auth_rejects_any_without_identity() {
+    let client = MockClient::default();
+    let store = jacquard::client::FileAuthStore::new({
+        let mut path = std::env::temp_dir();
+        path.push(format!(
+            "jacquard-oauth-any-reject-{}.json",
+            std::process::id()
+        ));
+        std::fs::write(&path, "{}").unwrap();
+        path
+    });
+    let oauth = OAuthClient::new_from_resolver(
+        store,
+        client,
+        ClientData {
+            keyset: None,
+            config: AtprotoClientMetadata::new_localhost(
+                None,
+                Some(Scopes::new(SmolStr::new_static("atproto rpc:*")).unwrap()),
+            ),
+        },
+    );
+
+    let err = match oauth
+        .resume_or_start_auth(
+            &SessionHint::Any,
+            jacquard_oauth::types::AuthorizeOptions::<SmolStr>::default(),
+        )
+        .await
+    {
+        Ok(_) => panic!("Any cannot start auth without identity"),
+        Err(err) => err,
+    };
+    assert!(err.to_string().contains("cannot start OAuth authorization"));
+}
diff --git a/crates/jacquard/tests/restore_pds_cache.rs b/crates/jacquard/tests/restore_pds_cache.rs
index aa282eae2..aa1e14786 100644
--- a/crates/jacquard/tests/restore_pds_cache.rs
+++ b/crates/jacquard/tests/restore_pds_cache.rs
@@ -91,32 +91,18 @@ async fn restore_uses_cached_pds_when_present() {
         refresh_jwt: "ref".into(),
         did: Did::new_static("did:plc:alice").unwrap(),
         handle: Handle::new_static("alice.bsky.social").unwrap(),
+        pds: Some(
+            Uri::parse("https://pds-cached")
+                .expect("valid uri")
+                .to_owned(),
+        ),
     };
-    let key = SessionKey(session.did.clone(), "session".into());
+    let key = SessionKey::new(session.did.clone(), "session");
     jacquard_common::session::SessionStore::set(store.as_ref(), key.clone(), session)
         .await
         .unwrap();
     // Verify it is persisted
     assert!(SessionStore::get(store.as_ref(), &key).await.is_some());
-    // Persist PDS endpoint cache to avoid DID resolution on restore
-    store
-        .set_atp_pds(
-            &key,
-            &Uri::parse("https://pds-cached")
-                .expect("valid uri")
-                .to_owned(),
-        )
-        .unwrap();
-    assert_eq!(
-        store
-            .get_atp_pds(&key)
-            .ok()
-            .flatten()
-            .expect("pds cached")
-            .as_str(),
-        "https://pds-cached"
-    );
-
     let session = CredentialSession::new(store.clone(), resolver.clone());
     // Restore should pick cached PDS and NOT call resolve_did_doc
     session
diff --git a/crates/jacquard/tests/scope_check.rs b/crates/jacquard/tests/scope_check.rs
index c18b1873a..21d011331 100644
--- a/crates/jacquard/tests/scope_check.rs
+++ b/crates/jacquard/tests/scope_check.rs
@@ -21,7 +21,7 @@ use jacquard_oauth::scopes::{
 use jacquard_oauth::session::SessionRegistry;
 use jacquard_oauth::session::{ClientData, ClientSessionData, DpopClientData};
 use jacquard_oauth::types::{OAuthAuthorizationServerMetadata, OAuthTokenType, TokenSet};
-use smol_str::SmolStr;
+use smol_str::{SmolStr, format_smolstr};
 use std::collections::BTreeSet;
 use tokio::sync::Mutex;
 
@@ -97,10 +97,10 @@ impl OAuthResolver for MockClient {
     ) -> Result<OAuthAuthorizationServerMetadata, jacquard_oauth::resolver::ResolverError> {
         let mut md = OAuthAuthorizationServerMetadata::default();
         md.issuer = SmolStr::from(issuer);
-        md.token_endpoint = SmolStr::from(format!("{}/token", issuer));
-        md.authorization_endpoint = SmolStr::from(format!("{}/authorize", issuer));
+        md.token_endpoint = format_smolstr!("{}/token", issuer);
+        md.authorization_endpoint = format_smolstr!("{}/authorize", issuer);
         md.require_pushed_authorization_requests = Some(true);
-        md.pushed_authorization_request_endpoint = Some(SmolStr::from(format!("{}/par", issuer)));
+        md.pushed_authorization_request_endpoint = Some(format_smolstr!("{}/par", issuer));
         md.token_endpoint_auth_methods_supported = Some(vec![SmolStr::from("none")]);
         md.dpop_signing_alg_values_supported = Some(vec![SmolStr::from("ES256")]);
         Ok(md)