feat(auth): adopt sessions locally

README.md

@@ -34,6 +34,17 @@ client.locks.release("build", lease)
 
 `get_session()` reads immutable local state. It does not refresh or validate the token.
 
+Copy a complete native session into another client's memory:
+
+```python
+session = source.auth.get_session()
+if session is not None:
+    fresh.auth.set_session(session)
+```
+
+`set_session()` copies the session without making a request or persisting credentials. It raises
+`ValueError` when the session type or any credential field is incomplete.
+
 Realtime is async. Channels wrap `centrifuge-python`; the underlying client and
 subscription objects are not part of the public API.
 

features/contract/auth.feature

@@ -16,3 +16,12 @@ Feature: SDK authentication contract
     Then the SDK operation succeeds
     And the current session belongs to the contract user
     And the current session exposes access and refresh tokens
+
+  @auth @SDK-AUTH-003
+  Scenario: A client adopts a supplied session locally
+    Given the confirmed contract user
+    When the client signs in with the contract user's credentials
+    And a fresh client adopts the current session
+    Then the SDK operation succeeds
+    And the current session belongs to the contract user
+    And the current session exposes access and refresh tokens

features/steps/sdk_contract_steps.py

@@ -11,6 +11,8 @@
     classify_error,
 )
 
+from volcano_sdk import VolcanoClient
+
 
 def _world(context: Any) -> ContractWorld:
     return context.contract
@@ -38,6 +40,19 @@ def read_current_session(context: Any) -> None:
     world.record(world.client.auth.get_session)
 
 
+@when("a fresh client adopts the current session")
+def adopt_current_session(context: Any) -> None:
+    world = _world(context)
+    source = world.client.auth.get_session()
+    assert source is not None
+    target = VolcanoClient(
+        api_url=world.fixture["api_url"],
+        anon_key=world.fixture["anon_key"],
+    )
+    world.record(lambda: target.auth.set_session(source))
+    world.client = target
+
+
 @then("the SDK operation succeeds")
 def operation_succeeds(context: Any) -> None:
     outcome = _world(context).last_outcome

src/volcano_sdk/auth.py

@@ -7,6 +7,33 @@
 from ._transport import Transport, invoke, response_payload
 from .models import Session
 
+_INCOMPLETE_SESSION = "Expected a complete Session"
+
+
+def _is_non_empty_string(value: object) -> bool:
+    return isinstance(value, str) and bool(value.strip())
+
+
+def _has_complete_values(session: Session) -> bool:
+    return all(
+        _is_non_empty_string(value)
+        for value in (
+            session.access_token,
+            session.refresh_token,
+            session.user_id,
+        )
+    )
+
+
+def _copy_complete_session(session: object) -> Session:
+    if not isinstance(session, Session) or not _has_complete_values(session):
+        raise ValueError(_INCOMPLETE_SESSION)
+    return Session(
+        access_token=session.access_token,
+        refresh_token=session.refresh_token,
+        user_id=session.user_id,
+    )
+
 
 class AuthContext(Protocol):
     """Client capabilities required by the authentication facade."""
@@ -34,6 +61,12 @@ def get_session(self) -> Session | None:
         """Return the immutable locally held session without validating it."""
         return self._client.current_session
 
+    def set_session(self, session: Session) -> Session:
+        """Copy a complete session into local client state."""
+        owned = _copy_complete_session(session)
+        self._client._set_session(owned)
+        return owned
+
     def sign_in(self, *, email: str, password: str) -> Session:
         """Sign in a user and store the returned session."""
         response = invoke(

tests/unit/test_contract_bindings.py

@@ -15,7 +15,7 @@
 
 ROOT = Path(__file__).parents[2]
 FEATURE_SHA256 = {
-    "auth.feature": "70289856dcca9854464ae92659363f3395b2fe79d92ebb8374973f29f5c1994d",
+    "auth.feature": "4145af3f1120331d100d8548b2abfcb98c0aeef552fd2b3304dc7e074e526ce2",
     "database.feature": (
         "4685b29357a621068b25984ff0de29cd4c504eebe5cfb597f0b999e29878a668"
     ),
@@ -60,6 +60,7 @@ def test_every_contract_phrase_is_bound_verbatim() -> None:
     }
     assert bound == {
         "a service-role client",
+        "a fresh client adopts the current session",
         "an authenticated client",
         "exactly the fixture row is returned",
         "one client subscribes and the other publishes the contract message",

tests/unit/test_state.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from typing import Any
 
+import pytest
+
 from volcano_sdk import Session, VolcanoClient
 
 
@@ -132,3 +134,60 @@ def test_auth_facade_reads_established_immutable_session_without_transport() ->
         user_id="user-123",
     )
     assert transport.authorizations == calls_after_sign_in
+
+
+def test_auth_facade_adopts_an_owned_session_without_transport() -> None:
+    transport = StateTransport()
+    client = VolcanoClient(anon_key="anon", _transport=transport)
+    supplied = Session(
+        access_token="adopted-access",
+        refresh_token="adopted-refresh",
+        user_id="adopted-user",
+    )
+
+    adopted = client.auth.set_session(supplied)
+
+    assert adopted == supplied
+    assert adopted is not supplied
+    assert client.auth.get_session() is adopted
+    assert transport.authorizations == []
+
+
+def test_auth_facade_adoption_replaces_the_current_session() -> None:
+    transport = StateTransport()
+    client = VolcanoClient(anon_key="anon", _transport=transport)
+    client.auth.sign_in(email="user@example.com", password="secret")
+    replacement = Session(
+        access_token="replacement-access",
+        refresh_token="replacement-refresh",
+        user_id="replacement-user",
+    )
+    calls_after_sign_in = list(transport.authorizations)
+
+    adopted = client.auth.set_session(replacement)
+
+    assert client.auth.get_session() is adopted
+    assert adopted == replacement
+    assert transport.authorizations == calls_after_sign_in
+
+
+@pytest.mark.parametrize(
+    "invalid",
+    [
+        object(),
+        Session(access_token=" ", refresh_token="refresh", user_id="user"),
+        Session(access_token="access", refresh_token="\t", user_id="user"),
+        Session(access_token="access", refresh_token="refresh", user_id="\n"),
+    ],
+)
+def test_auth_facade_rejects_incomplete_adoption_without_mutation(invalid: Any) -> None:
+    transport = StateTransport()
+    client = VolcanoClient(anon_key="anon", _transport=transport)
+    previous = client.auth.sign_in(email="user@example.com", password="secret")
+    calls_after_sign_in = list(transport.authorizations)
+
+    with pytest.raises(ValueError, match="complete Session"):
+        client.auth.set_session(invalid)
+
+    assert client.auth.get_session() is previous
+    assert transport.authorizations == calls_after_sign_in