From a88f1aab3410cc09a097b37d7d17c644a9b09481 Mon Sep 17 00:00:00 2001
From: Myst <1592048+LeMyst@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:11:00 +0200
Subject: [PATCH 1/4] Add check_constraints() helper to query Wikibase
 constraint violations

Uses the target instance's own wbcheckconstraints API action (from the
WikibaseQualityConstraints extension) instead of a hardcoded, Wikidata-specific
rule set, so it works against any Wikibase instance with the extension installed.

Closes #154

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 test/conftest.py                  |  6 +++++
 test/test_wbi_helpers.py          | 35 +++++++++++++++++++++++++++-
 wikibaseintegrator/wbi_helpers.py | 38 +++++++++++++++++++++++++++++++
 3 files changed, 78 insertions(+), 1 deletion(-)

diff --git a/test/conftest.py b/test/conftest.py
index 4fc8797e..da813660 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -79,6 +79,7 @@ def __init__(self, mocker: requests_mock_lib.Mocker, base_url: str = 'https://wi
         self.search_results: list[dict] = []  # wbsearchentities results
         self.fulltext_results: list[dict] = []  # list=search results
         self.sparql_bindings: list[dict] = []  # bindings returned by the SPARQL endpoint
+        self.constraint_results: dict[str, list[dict]] = {}  # wbcheckconstraints results, keyed by entity id
         self.valid_credentials: dict[str, str] = {}  # user -> password accepted by (client)login
         self.login_token = 'aabbccddeeff+\\'
         self.csrf_token = '0123456789abcdef+\\'
@@ -368,6 +369,11 @@ def _action_wbremoveclaims(self, params: dict[str, str]) -> dict:
     def _action_delete(self, params: dict[str, str]) -> dict:
         return {'delete': {'title': params.get('title', ''), 'reason': params.get('reason', 'mock deletion'), 'logid': 1}}
 
+    def _action_wbcheckconstraints(self, params: dict[str, str]) -> dict:
+        requested_ids = params['id'].split('|') if 'id' in params else list(self.constraint_results.keys())
+        claims = {entity_id: deepcopy(self.constraint_results[entity_id]) for entity_id in requested_ids if entity_id in self.constraint_results}
+        return {'claims': claims, 'success': 1}
+
 
 # ---------------------------------------------------------------------- #
 # SPARQL helpers usable from the tests
diff --git a/test/test_wbi_helpers.py b/test/test_wbi_helpers.py
index b23db7a9..e4ca3b43 100644
--- a/test/test_wbi_helpers.py
+++ b/test/test_wbi_helpers.py
@@ -11,7 +11,7 @@
 from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed
 from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent,
                                             lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims,
-                                            search_entities)
+                                            search_entities, check_constraints)
 
 
 class FakeLogin:
@@ -254,6 +254,39 @@ def test_search_strict_language_and_limit_parameters(self, wikibase):
         assert len(results) == 10
 
 
+class TestCheckConstraints:
+    def test_requires_entity_id_or_claim_id(self):
+        with pytest.raises(ValueError):
+            check_constraints()
+
+    def test_check_by_entity_id(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': [{'status': 'violation', 'property': 'P17'}]}]
+        }
+
+        result = check_constraints(entity_id='Q582')
+        assert wikibase.last_request['action'] == 'wbcheckconstraints'
+        assert wikibase.last_request['id'] == 'Q582'
+        assert result['claims']['Q582'][0]['results'][0]['status'] == 'violation'
+
+    def test_check_by_multiple_entity_ids(self, wikibase):
+        wikibase.constraint_results = {
+            'Q582': [{'id': 'Q582$1', 'results': []}],
+            'Q1': [{'id': 'Q1$1', 'results': []}],
+        }
+
+        check_constraints(entity_id=['Q582', 'Q1'])
+        assert wikibase.last_request['id'] == 'Q582|Q1'
+
+    def test_check_by_claim_id(self, wikibase):
+        check_constraints(claim_id='Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a')
+        assert wikibase.last_request['claimid'] == 'Q582$1d2e3f4a-5b6c-7d8e-9f0a-1b2c3d4e5f6a'
+
+    def test_check_with_status_filter(self, wikibase):
+        check_constraints(entity_id='Q582', status=['violation', 'warning'])
+        assert wikibase.last_request['status'] == 'violation|warning'
+
+
 class TestFulltextSearch:
     def test_fulltext_search(self, wikibase):
         wikibase.fulltext_results = [{'ns': 0, 'title': 'Q582', 'pageid': 892, 'snippet': 'Villeurbanne'}]
diff --git a/wikibaseintegrator/wbi_helpers.py b/wikibaseintegrator/wbi_helpers.py
index 52f6a31b..22131f2f 100644
--- a/wikibaseintegrator/wbi_helpers.py
+++ b/wikibaseintegrator/wbi_helpers.py
@@ -433,6 +433,44 @@ def remove_claims(claim_id: str, summary: str | None = None, baserevid: int | No
     return mediawiki_api_call_helper(data=params, is_bot=is_bot, **kwargs)
 
 
+def check_constraints(entity_id: str | list[str] | None = None, claim_id: str | list[str] | None = None, status: list[str] | None = None, allow_anonymous: bool = True,
+                      **kwargs: Any) -> dict:
+    """
+    Check the constraint violations of one or more entities or claims, using the target Wikibase instance's own constraint checker
+    (the ``wbcheckconstraints`` API module provided by the `WikibaseQualityConstraints
+    <https://www.mediawiki.org/wiki/Extension:WikibaseQualityConstraints>`_ extension).
+
+    This relies entirely on the constraints configured on the Wikibase instance being queried (e.g. on Wikidata itself), rather than
+    on a hardcoded or Wikidata-specific set of rules, so it works the same way against any Wikibase instance that has the extension
+    installed. If the extension isn't installed, the underlying API call will fail with a :class:`~wikibaseintegrator.wbi_exceptions.MWApiError`.
+
+    :param entity_id: One or more entity IDs (item, property, lexeme, etc.) whose claims should be checked.
+    :param claim_id: One or more claim GUIDs to check, instead of checking whole entities.
+    :param status: Only return results with one of these statuses, e.g. ``['violation']``. By default, the API returns every status.
+    :param allow_anonymous: Allow anonymous interaction with the MediaWiki API. 'True' by default since this is a read-only action.
+    :param kwargs: Extra parameters for mediawiki_api_call_helper()
+    :return: The data returned by the API as a dictionary
+    """
+    if not entity_id and not claim_id:
+        raise ValueError("You must provide either 'entity_id' or 'claim_id'.")
+
+    params: dict[str, Any] = {
+        'action': 'wbcheckconstraints',
+        'format': 'json'
+    }
+
+    if entity_id:
+        params.update({'id': '|'.join(entity_id) if isinstance(entity_id, list) else entity_id})
+
+    if claim_id:
+        params.update({'claimid': '|'.join(claim_id) if isinstance(claim_id, list) else claim_id})
+
+    if status:
+        params.update({'status': '|'.join(status)})
+
+    return mediawiki_api_call_helper(data=params, allow_anonymous=allow_anonymous, **kwargs)
+
+
 def search_entities(search_string: str, language: str | None = None, strict_language: bool = False, search_type: str = 'item', max_results: int = 50, dict_result: bool = False,
                     allow_anonymous: bool = True, **kwargs: Any) -> list[dict[str, Any]]:
     """

From 6c766c5e8a2297161a790d3d89038eb99c9afb57 Mon Sep 17 00:00:00 2001
From: Myst <1592048+LeMyst@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:12:52 +0200
Subject: [PATCH 2/4] Fix isort ordering in test_wbi_helpers.py imports

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 test/test_wbi_helpers.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/test/test_wbi_helpers.py b/test/test_wbi_helpers.py
index e4ca3b43..42f373b1 100644
--- a/test/test_wbi_helpers.py
+++ b/test/test_wbi_helpers.py
@@ -9,9 +9,9 @@
 
 from wikibaseintegrator.wbi_config import config as wbi_config
 from wikibaseintegrator.wbi_exceptions import AnonymousEditNotAllowedError, MaxRetriesReachedException, ModificationFailed, MWApiError, NonExistentEntityError, SaveFailed
-from wikibaseintegrator.wbi_helpers import (download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances, get_user_agent,
-                                            lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items, remove_claims,
-                                            search_entities, check_constraints)
+from wikibaseintegrator.wbi_helpers import (check_constraints, download_entity_ttl, execute_sparql_query, format2wbi, format_amount, fulltext_search, generate_entity_instances,
+                                            get_user_agent, lexeme_edit_sense, lexeme_remove_form, lexeme_remove_sense, mediawiki_api_call, mediawiki_api_call_helper, merge_items,
+                                            remove_claims, search_entities)
 
 
 class FakeLogin:

From 509b9d96bfd032dccc0a625f3977e9ef2db9dd28 Mon Sep 17 00:00:00 2001
From: Myst <1592048+LeMyst@users.noreply.github.com>
Date: Sat, 4 Jul 2026 10:35:16 +0200
Subject: [PATCH 3/4] Drop mwoauth dependency, use oauthlib/requests-oauthlib
 directly

The 3-legged OAuth1 handshake (Special:OAuth/initiate, authenticate,
token) is reimplemented with requests_oauthlib.OAuth1 and requests,
mirroring what mwoauth did internally, removing the extra dependency.
Also fixes continue_oauth(), which previously crashed with an
AttributeError because mediawiki_api_url/instantiation_time were never
initialized after completing the handshake.

Fixes #318

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
---
 poetry.lock                     | 45 ++----------------------
 pyproject.toml                  |  1 -
 test/test_wbi_login.py          | 21 +++++++++++
 wikibaseintegrator/wbi_login.py | 62 ++++++++++++++++++++++++---------
 4 files changed, 69 insertions(+), 60 deletions(-)

diff --git a/poetry.lock b/poetry.lock
index b3420926..78560e3b 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1897,27 +1897,6 @@ files = [
 [package.dependencies]
 typing-extensions = {version = "*", markers = "python_version < \"3.11\""}
 
-[[package]]
-name = "mwoauth"
-version = "0.4.0"
-description = "A generic MediaWiki OAuth handshake helper."
-optional = false
-python-versions = "*"
-groups = ["main"]
-files = [
-    {file = "mwoauth-0.4.0-py3-none-any.whl", hash = "sha256:fed9bc7d6bbabb5f691b918af0ac844e13c9b75d5fa51a898f36d54d798b5fe1"},
-    {file = "mwoauth-0.4.0.tar.gz", hash = "sha256:22e3403e748e70146f8eccc1430fe542c9f9c4ff677eff424a52e644f6d8f7c5"},
-]
-
-[package.dependencies]
-oauthlib = "*"
-PyJWT = ">=1.0.1"
-requests = "*"
-requests-oauthlib = "*"
-
-[package.extras]
-flask = ["flask"]
-
 [[package]]
 name = "mypy"
 version = "2.1.0"
@@ -2421,24 +2400,6 @@ files = [
 [package.extras]
 windows-terminal = ["colorama (>=0.4.6)"]
 
-[[package]]
-name = "pyjwt"
-version = "2.13.0"
-description = "JSON Web Token implementation in Python"
-optional = false
-python-versions = ">=3.9"
-groups = ["main"]
-files = [
-    {file = "pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728"},
-    {file = "pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423"},
-]
-
-[package.dependencies]
-typing_extensions = {version = ">=4.0", markers = "python_version < \"3.11\""}
-
-[package.extras]
-crypto = ["cryptography (>=3.4.0)"]
-
 [[package]]
 name = "pylint"
 version = "4.0.6"
@@ -3758,12 +3719,12 @@ version = "4.16.0"
 description = "Backported and Experimental Type Hints for Python 3.9+"
 optional = false
 python-versions = ">=3.9"
-groups = ["main", "coverage", "dev", "docs", "notebooks"]
+groups = ["coverage", "dev", "docs", "notebooks"]
 files = [
     {file = "typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8"},
     {file = "typing_extensions-4.16.0.tar.gz", hash = "sha256:dc983d19a509c94dba722ee6abd33940f7c05a89e243c47e907eb4db6f1a43e5"},
 ]
-markers = {main = "python_version == \"3.10\"", coverage = "python_version == \"3.10\"", docs = "python_version >= \"3.11\" and python_version < \"3.13\""}
+markers = {coverage = "python_version == \"3.10\"", docs = "python_version >= \"3.11\" and python_version < \"3.13\""}
 
 [[package]]
 name = "tzdata"
@@ -3981,4 +3942,4 @@ files = [
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.10"
-content-hash = "505ee90d158ed8020002f7f485cdc68fa0d5d0f7520267948cf8210d1c5983cb"
+content-hash = "3c3710f79630363e3c72c5cde954e2b7833f78db41fc4e10dbd8d9e9ddd90f57"
diff --git a/pyproject.toml b/pyproject.toml
index 2ac9275f..c53be093 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -41,7 +41,6 @@ Changelog = "https://github.com/LeMyst/WikibaseIntegrator/releases"
 [tool.poetry.dependencies]
 python = "^3.10"
 backoff = "^2.2.1"
-mwoauth = "^0.4.0"
 oauthlib = "^3.2.2"
 requests = "^2.32.3"
 requests-oauthlib = "^2.0.0"
diff --git a/test/test_wbi_login.py b/test/test_wbi_login.py
index 3e9b0c0f..40b0a866 100644
--- a/test/test_wbi_login.py
+++ b/test/test_wbi_login.py
@@ -75,6 +75,27 @@ def test_csrf_error_raises_login_error(self, credentials):
         with pytest.raises(LoginError):
             wbi_login.OAuth1(consumer_token='wrong', consumer_secret='wrong', access_token='wrong', access_secret='wrong')
 
+    def test_three_legged_flow(self, credentials, requests_mock):
+        # Step 1 (initiate) and step 3 (token) both post to Special:OAuth on the index endpoint.
+        requests_mock.post(credentials.mediawiki_index_url, [
+            {'text': 'oauth_token=request-key&oauth_token_secret=request-secret'},
+            {'text': 'oauth_token=access-key&oauth_token_secret=access-secret'},
+        ])
+
+        login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret')
+        assert 'oauth_token=request-key' in login.redirect
+        assert login.request_token_key == 'request-key'
+
+        login.continue_oauth('https://example.org/callback?oauth_token=request-key&oauth_verifier=some-verifier')
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_three_legged_flow_wrong_request_token(self, credentials, requests_mock):
+        requests_mock.post(credentials.mediawiki_index_url, text='oauth_token=request-key&oauth_token_secret=request-secret')
+
+        login = wbi_login.OAuth1(consumer_token='consumer-token', consumer_secret='consumer-secret')
+        with pytest.raises(LoginError):
+            login.continue_oauth('https://example.org/callback?oauth_token=unexpected-key&oauth_verifier=some-verifier')
+
 
 class TestOAuth2:
     def test_successful_flow(self, credentials, requests_mock):
diff --git a/wikibaseintegrator/wbi_login.py b/wikibaseintegrator/wbi_login.py
index 1a109f3f..0fbc6ffc 100644
--- a/wikibaseintegrator/wbi_login.py
+++ b/wikibaseintegrator/wbi_login.py
@@ -5,11 +5,13 @@
 import time
 import webbrowser
 from typing import Any, cast
+from urllib.parse import parse_qs, urlencode
 
-from mwoauth import ConsumerToken, Handshaker, OAuthException
+import requests
 from oauthlib.oauth2 import BackendApplicationClient, InvalidClientError
 from requests import Session
 from requests.cookies import RequestsCookieJar
+from requests_oauthlib import OAuth1 as OAuth1Auth
 from requests_oauthlib import OAuth1Session, OAuth2Session
 
 from wikibaseintegrator.wbi_backoff import wbi_backoff
@@ -176,19 +178,35 @@ def __init__(self, consumer_token: str | None = None, consumer_secret: str | Non
             super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
         else:
             # Oauth procedure, based on https://www.mediawiki.org/wiki/OAuth/For_Developers
-            # Construct a "consumer" from the key/secret provided by MediaWiki
-            self.oauth1_consumer_token = ConsumerToken(consumer_token, consumer_secret)
+            self.consumer_token = consumer_token
+            self.consumer_secret = consumer_secret
+            self.mediawiki_index_url = mediawiki_index_url
+            self.mediawiki_api_url = str(mediawiki_api_url or config['MEDIAWIKI_API_URL'])
+            self.token_renew_period = token_renew_period
+            self.user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)
 
-            # Construct handshaker with wiki URI and consumer
-            self.handshaker = Handshaker(mw_uri=mediawiki_index_url, consumer_token=self.oauth1_consumer_token, callback=callback_url,
-                                         user_agent=get_user_agent(user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)))
+            # Step 1: Initiate -- ask MediaWiki for a temporary key/secret for the user
+            auth = OAuth1Auth(consumer_token, client_secret=consumer_secret, callback_uri=callback_url)
+            response = requests.post(url=mediawiki_index_url, params={'title': "Special:OAuth/initiate"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT'])
+
+            request_token = self._parse_token_response(response.text)
+            self.request_token_key = request_token['oauth_token']
+            self.request_token_secret = request_token['oauth_token_secret']
 
-            # Step 1: Initialize -- ask MediaWiki for a temp key/secret for user
             # redirect -> authorization -> callback url
-            try:
-                self.redirect, self.request_token = self.handshaker.initiate(callback=callback_url)
-            except OAuthException as err:
-                raise LoginError(err) from err
+            params = {'title': "Special:OAuth/authenticate", 'oauth_token': self.request_token_key, 'oauth_consumer_key': consumer_token}
+            self.redirect = mediawiki_index_url + '?' + urlencode(params)
+
+    @staticmethod
+    def _parse_token_response(content: str) -> dict[str, str]:
+        if content.startswith("Error: "):
+            raise LoginError(content[len("Error: "):])
+
+        credentials = parse_qs(content)
+        if not credentials or 'oauth_token' not in credentials or 'oauth_token_secret' not in credentials:
+            raise LoginError(f"MediaWiki response lacks token information: {content!r}")
+
+        return {'oauth_token': credentials['oauth_token'][0], 'oauth_token_secret': credentials['oauth_token_secret'][0]}
 
     def continue_oauth(self, oauth_callback_data: str | None = None) -> None:
         """
@@ -205,17 +223,27 @@ def continue_oauth(self, oauth_callback_data: str | None = None) -> None:
 
         # input the url from redirect after authorization
         response_qs = oauth_callback_data.split('?')[-1]
+        callback_data = parse_qs(response_qs)
+
+        if not callback_data or 'oauth_token' not in callback_data or 'oauth_verifier' not in callback_data:
+            raise LoginError(f"Query string lacks token information: {callback_data!r}")
+
+        request_token_key = callback_data['oauth_token'][0]
+        verifier = callback_data['oauth_verifier'][0]
+
+        if self.request_token_key != request_token_key:
+            raise LoginError(f"Unexpected request token key {request_token_key!r}, expected {self.request_token_key!r}.")
 
         # Step 3: Complete -- obtain authorized key/secret for "resource owner"
-        access_token = self.handshaker.complete(self.request_token, response_qs)
+        auth = OAuth1Auth(self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=self.request_token_key, resource_owner_secret=self.request_token_secret, verifier=verifier)
+        response = requests.post(url=self.mediawiki_index_url, params={'title': "Special:OAuth/token"}, auth=auth, headers={'User-Agent': get_user_agent(self.user_agent)}, timeout=config['TIMEOUT'])
 
-        if self.oauth1_consumer_token is None:
-            raise ValueError("oauth1_consumer_token can't be None")
+        access_token = self._parse_token_response(response.text)
 
         # input the access token to return a csrf (edit) token
-        self.session = OAuth1Session(client_key=self.oauth1_consumer_token.key, client_secret=self.oauth1_consumer_token.secret, resource_owner_key=access_token.key,
-                                     resource_owner_secret=access_token.secret)
-        self.generate_edit_credentials()
+        session = OAuth1Session(client_key=self.consumer_token, client_secret=self.consumer_secret, resource_owner_key=access_token['oauth_token'],
+                                resource_owner_secret=access_token['oauth_token_secret'])
+        super().__init__(session=session, token_renew_period=self.token_renew_period, user_agent=self.user_agent, mediawiki_api_url=self.mediawiki_api_url)
 
 
 class Login(_Login):

From c16a406e5130987470bcde9e714281aa1648fc9b Mon Sep 17 00:00:00 2001
From: Myst <1592048+LeMyst@users.noreply.github.com>
Date: Sat, 4 Jul 2026 14:44:16 +0200
Subject: [PATCH 4/4] Recover from lost sessions instead of failing writes
 (#902)

Renewing the CSRF token only re-reads it off the current session
cookies, so it cannot resurrect a session the server has already
invalidated: writes then fail with assertuserfailed/assertbotfailed
("You are no longer logged in..."), no matter how often the token is
refreshed. Login and Clientlogin now keep their credentials so they
can redo a full login, and mediawiki_api_call transparently
re-authenticates and retries once when the API reports the session is
gone.
---
 test/test_wbi_helpers.py          | 31 ++++++++++++
 test/test_wbi_login.py            | 43 +++++++++++++++++
 wikibaseintegrator/wbi_helpers.py | 22 ++++++++-
 wikibaseintegrator/wbi_login.py   | 79 +++++++++++++++++++++++++------
 4 files changed, 159 insertions(+), 16 deletions(-)

diff --git a/test/test_wbi_helpers.py b/test/test_wbi_helpers.py
index 42f373b1..4686e40e 100644
--- a/test/test_wbi_helpers.py
+++ b/test/test_wbi_helpers.py
@@ -21,6 +21,7 @@ def __init__(self, mediawiki_api_url, edit_token='fakelogintoken+\\'):
         self.mediawiki_api_url = mediawiki_api_url
         self.edit_token = edit_token
         self.session = requests.Session()
+        self.reauthenticate_calls = 0
 
     def get_edit_token(self):
         return self.edit_token
@@ -28,6 +29,10 @@ def get_edit_token(self):
     def get_session(self):
         return self.session
 
+    def reauthenticate(self):
+        self.reauthenticate_calls += 1
+        self.edit_token = 'renewed-token+\\'
+
 
 class TestRetryBehaviour:
     """Behaviour of mediawiki_api_call when the instance is unhealthy."""
@@ -134,6 +139,32 @@ def test_generic_error(self, wikibase):
             mediawiki_api_call_helper(data={'action': 'wbeditentity', 'id': 'Q1', 'format': 'json'}, allow_anonymous=True)
 
 
+@pytest.mark.parametrize('error_code', ['assertuserfailed', 'assertbotfailed', 'notloggedin'])
+class TestSessionRecovery:
+    """
+    A CSRF token can look fresh yet the server has already dropped the underlying session (#902).
+    On assertuserfailed/assertbotfailed/notloggedin, the login object must be asked to fully
+    re-authenticate and the call retried, instead of failing outright.
+    """
+
+    def test_session_loss_triggers_reauthentication_and_retry(self, wikibase, error_code):
+        wikibase.fail_next(code=error_code, info='You are no longer logged in, so the action could not be completed.')
+        login = FakeLogin(mediawiki_api_url=wikibase.mediawiki_api_url)
+
+        result = mediawiki_api_call_helper(data={'action': 'query', 'format': 'json'}, login=login, mediawiki_api_url=wikibase.mediawiki_api_url)
+
+        assert result == {'batchcomplete': ''}
+        assert login.reauthenticate_calls == 1
+        # The retried request must carry the token obtained after re-authenticating, not the stale one.
+        assert wikibase.last_request['token'] == 'renewed-token+\\'
+
+    def test_session_loss_without_login_is_not_retried(self, wikibase, error_code):
+        # Without a login object there is nothing to re-authenticate with, so the error must surface as-is.
+        wikibase.fail_next(code=error_code, info='You are no longer logged in, so the action could not be completed.')
+        with pytest.raises(MWApiError):
+            mediawiki_api_call_helper(data={'action': 'query', 'format': 'json'}, allow_anonymous=True)
+
+
 class TestAuthenticationGuards:
     def test_anonymous_must_be_explicit(self, wikibase):
         # allow_anonymous=False without login object
diff --git a/test/test_wbi_login.py b/test/test_wbi_login.py
index 40b0a866..3c659bf5 100644
--- a/test/test_wbi_login.py
+++ b/test/test_wbi_login.py
@@ -6,6 +6,7 @@
 import pytest
 
 from wikibaseintegrator import wbi_login
+from wikibaseintegrator.wbi_helpers import edit_entity
 from wikibaseintegrator.wbi_login import LoginError
 
 
@@ -45,6 +46,29 @@ def test_get_edit_cookie(self, credentials):
         login = wbi_login.Login(user='TestUser@bot', password='botpassword')
         assert login.get_edit_cookie() is login.get_session().cookies
 
+    def test_reauthenticate_redoes_full_login(self, credentials):
+        # generate_edit_credentials() alone can't recover a session the server already dropped (#902):
+        # reauthenticate() must redo the login-token + login round trip, not just fetch a new csrf token.
+        login = wbi_login.Login(user='TestUser@bot', password='botpassword')
+        requests_before = len(credentials.requests)
+
+        login.reauthenticate()
+
+        actions = [request.get('action') or request.get('meta') for request in credentials.requests[requests_before:]]
+        assert actions == ['query', 'login', 'query']
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_session_loss_is_recovered_on_write(self, credentials):
+        # Simulate the server having invalidated the session mid-run: the very next write must
+        # transparently re-login and retry instead of raising MWApiError.
+        login = wbi_login.Login(user='TestUser@bot', password='botpassword')
+        credentials.fail_next(code='assertbotfailed', info='You do not have the "bot" right, so the action could not be completed.')
+
+        result = edit_entity(data={}, id='Q1', login=login, is_bot=True)
+
+        assert result['success'] == 1
+        assert login.get_edit_token() == credentials.csrf_token
+
 
 class TestClientLogin:
     def test_successful_login(self, credentials):
@@ -55,6 +79,25 @@ def test_wrong_credentials(self, credentials):
         with pytest.raises(LoginError):
             wbi_login.Clientlogin(user='wrong', password='wrong')
 
+    def test_reauthenticate_redoes_full_login(self, credentials):
+        login = wbi_login.Clientlogin(user='TestUser', password='password')
+        requests_before = len(credentials.requests)
+
+        login.reauthenticate()
+
+        actions = [request.get('action') or request.get('meta') for request in credentials.requests[requests_before:]]
+        assert actions == ['query', 'clientlogin', 'query']
+        assert login.get_edit_token() == credentials.csrf_token
+
+    def test_session_loss_is_recovered_on_write(self, credentials):
+        login = wbi_login.Clientlogin(user='TestUser', password='password')
+        credentials.fail_next(code='assertuserfailed', info='You are no longer logged in, so the action could not be completed.')
+
+        result = edit_entity(data={}, id='Q1', login=login)
+
+        assert result['success'] == 1
+        assert login.get_edit_token() == credentials.csrf_token
+
 
 class TestAnonymousToken:
     def test_anonymous_csrf_token_is_rejected(self, credentials):
diff --git a/wikibaseintegrator/wbi_helpers.py b/wikibaseintegrator/wbi_helpers.py
index 22131f2f..4729daac 100644
--- a/wikibaseintegrator/wbi_helpers.py
+++ b/wikibaseintegrator/wbi_helpers.py
@@ -53,13 +53,23 @@ class BColors:
 default_session = requests.Session()
 
 
-def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, session: Session | None = None, max_retries: int = 100, retry_after: int = 60, **kwargs: Any) -> dict:
+# MediaWiki error codes meaning the server no longer considers the current session authenticated
+# (e.g. its session store evicted/expired the session, see #902). A CSRF token fetched right before the
+# failing call can still look valid, since generate_edit_credentials() reads it off the very session
+# cookies that are now rejected; recovering requires a full re-login instead.
+SESSION_LOST_ERROR_CODES = {'assertuserfailed', 'assertbotfailed', 'notloggedin'}
+
+
+def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, session: Session | None = None, login: _Login | None = None, max_retries: int = 100, retry_after: int = 60,
+                       **kwargs: Any) -> dict:
     """
     A function to call the MediaWiki API.
 
     :param method: 'GET' or 'POST'
     :param mediawiki_api_url:
     :param session: If a session is passed, it will be used. Otherwise, a new requests session is created
+    :param login: If provided and the API reports that the session is no longer authenticated (see
+                  SESSION_LOST_ERROR_CODES), it is used to fully re-authenticate before retrying.
     :param max_retries: If api request fails due to rate limiting, maxlag, or readonly mode, retry up to `max_retries` times
     :param retry_after: Number of seconds to wait before retrying request (see max_retries)
     :param kwargs: Any additional keyword arguments to pass to requests.request
@@ -126,6 +136,14 @@ def mediawiki_api_call(method: str, mediawiki_api_url: str | None = None, sessio
                 sleep(retry_after)
                 continue
 
+            # session no longer valid: re-authenticate and retry instead of failing outright (#902)
+            if 'code' in json_data['error'] and json_data['error']['code'] in SESSION_LOST_ERROR_CODES and login is not None:
+                log.warning("%s: session no longer valid (%s). Re-authenticating and retrying.", datetime.datetime.now(datetime.timezone.utc), json_data['error']['code'])
+                login.reauthenticate()
+                if 'data' in kwargs and kwargs['data'] and 'token' in kwargs['data']:
+                    kwargs['data']['token'] = login.get_edit_token()
+                continue
+
             # non-existent error
             if 'code' in json_data['error'] and json_data['error']['code'] in ['no-such-entity', 'missingtitle']:
                 raise NonExistentEntityError(json_data['error'])
@@ -227,7 +245,7 @@ def mediawiki_api_call_helper(data: dict[str, Any], login: _Login | None = None,
 
     log.debug(data)
 
-    return mediawiki_api_call('POST', mediawiki_api_url=mediawiki_api_url, session=session, data=data, headers=headers, max_retries=max_retries, retry_after=retry_after, **kwargs)
+    return mediawiki_api_call('POST', mediawiki_api_url=mediawiki_api_url, session=session, login=login, data=data, headers=headers, max_retries=max_retries, retry_after=retry_after, **kwargs)
 
 
 @wbi_backoff()
diff --git a/wikibaseintegrator/wbi_login.py b/wikibaseintegrator/wbi_login.py
index 0fbc6ffc..d3715083 100644
--- a/wikibaseintegrator/wbi_login.py
+++ b/wikibaseintegrator/wbi_login.py
@@ -104,6 +104,21 @@ def get_session(self) -> Session:
         """
         return self.session
 
+    def reauthenticate(self) -> None:
+        """
+        Recover from a session that the server has invalidated (e.g. MediaWiki returning
+        'assertuserfailed'/'assertbotfailed'/'notloggedin' on an otherwise well-formed request, see #902).
+
+        Simply asking for a new CSRF token via generate_edit_credentials() cannot resurrect a session the
+        server has already dropped, since that call is itself authenticated by the same (now invalid)
+        session cookies. This default implementation is only adequate for auth methods where
+        generate_edit_credentials() actually re-establishes the underlying credentials (OAuth2, which
+        refreshes its access token first) or where authentication is per-request rather than session-based
+        (OAuth1). Login and Clientlogin override this to redo the full username/password login.
+        """
+        self.generate_edit_credentials()
+        self.instantiation_time = time.time()
+
 
 class OAuth2(_Login):
     @wbi_backoff()
@@ -264,6 +279,19 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
         user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)
         session = Session()
 
+        # Kept so reauthenticate() can redo this same flow after the server invalidates the session (#902).
+        self._user = user
+        self._password = password
+        self._login_kwargs = kwargs
+
+        headers = {
+            'User-Agent': get_user_agent(user_agent)
+        }
+        self._perform_login(session=session, mediawiki_api_url=mediawiki_api_url, headers=headers, **kwargs)
+
+        super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
+
+    def _perform_login(self, session: Session, mediawiki_api_url: str, headers: dict[str, str], **kwargs: Any) -> None:
         params_login = {
             'action': 'query',
             'meta': 'tokens',
@@ -271,10 +299,6 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
             'format': 'json'
         }
 
-        headers = {
-            'User-Agent': get_user_agent(user_agent)
-        }
-
         allowed_kwargs = {'headers', 'proxies', 'timeout', 'verify'}
         filtered_kwargs = {key: value for key, value in kwargs.items() if key in allowed_kwargs}
         if len(filtered_kwargs) < len(kwargs):
@@ -285,8 +309,8 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
         login_token = session.post(mediawiki_api_url, data=params_login, headers=headers, **filtered_kwargs).json()['query']['tokens']['logintoken']
         params = {
             'action': 'login',
-            'lgname': user,
-            'lgpassword': password,
+            'lgname': self._user,
+            'lgpassword': self._password,
             'lgtoken': login_token,
             'format': 'json'
         }
@@ -305,7 +329,16 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
             for message in login_result['warnings']:
                 log.warning(f"* {message}: {login_result['warnings'][message]['*']}")
 
-        super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
+    def reauthenticate(self) -> None:
+        """
+        Redo the full bot-password login flow on the existing session, then refresh the CSRF token.
+        See _Login.reauthenticate() for why this full re-login is needed instead of just fetching a new token.
+        """
+        log.warning("Session no longer valid, re-authenticating as %s", self._user)
+        headers = {'User-Agent': self.session.headers.get('User-Agent', get_user_agent())}
+        self._perform_login(session=self.session, mediawiki_api_url=self.mediawiki_api_url, headers=headers, **self._login_kwargs)
+        self.generate_edit_credentials()
+        self.instantiation_time = time.time()
 
 
 class Clientlogin(_Login):
@@ -326,6 +359,19 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
         user_agent = user_agent or (str(config['USER_AGENT']) if config['USER_AGENT'] is not None else None)
         session = Session()
 
+        # Kept so reauthenticate() can redo this same flow after the server invalidates the session (#902).
+        self._user = user
+        self._password = password
+        self._login_kwargs = kwargs
+
+        headers = {
+            'User-Agent': get_user_agent(user_agent)
+        }
+        self._perform_login(session=session, mediawiki_api_url=mediawiki_api_url, headers=headers, **kwargs)
+
+        super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
+
+    def _perform_login(self, session: Session, mediawiki_api_url: str, headers: dict[str, str], **kwargs: Any) -> None:
         params_login = {
             'action': 'query',
             'meta': 'tokens',
@@ -333,10 +379,6 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
             'format': 'json'
         }
 
-        headers = {
-            'User-Agent': get_user_agent(user_agent)
-        }
-
         allowed_kwargs = {'headers', 'proxies', 'timeout', 'verify'}
         filtered_kwargs = {key: value for key, value in kwargs.items() if key in allowed_kwargs}
         if len(filtered_kwargs) < len(kwargs):
@@ -348,8 +390,8 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
 
         params = {
             'action': 'clientlogin',
-            'username': user,
-            'password': password,
+            'username': self._user,
+            'password': self._password,
             'logintoken': login_token,
             'loginreturnurl': 'https://example.org/',
             'format': 'json'
@@ -376,7 +418,16 @@ def __init__(self, user: str | None = None, password: str | None = None, mediawi
             for message in login_result['warnings']:
                 log.warning(f"* {message}: {login_result['warnings'][message]['*']}")
 
-        super().__init__(session=session, token_renew_period=token_renew_period, user_agent=user_agent, mediawiki_api_url=mediawiki_api_url)
+    def reauthenticate(self) -> None:
+        """
+        Redo the full clientlogin flow on the existing session, then refresh the CSRF token.
+        See _Login.reauthenticate() for why this full re-login is needed instead of just fetching a new token.
+        """
+        log.warning("Session no longer valid, re-authenticating as %s", self._user)
+        headers = {'User-Agent': self.session.headers.get('User-Agent', get_user_agent())}
+        self._perform_login(session=self.session, mediawiki_api_url=self.mediawiki_api_url, headers=headers, **self._login_kwargs)
+        self.generate_edit_credentials()
+        self.instantiation_time = time.time()
 
 
 class LoginError(Exception):