diff --git a/.env.example b/.env.example
index e1aa778a4..8a6d5cd24 100644
--- a/.env.example
+++ b/.env.example
@@ -42,6 +42,9 @@ MINT_INFO_TOS_URL="https://mint.host/tos"
 
 # Increment derivation path to rotate to a new keyset
 # Example: m/0'/0'/0' -> m/0'/0'/1'
+# NOTE: With automatic keyset rotation enabled (default), the mint manages this
+# automatically in the database on startup and during execution. You do NOT
+# need to manually increment MINT_DERIVATION_PATH in your .env file after a rotation.
 MINT_DERIVATION_PATH="m/0'/0'/0'"
 
 # Multiple derivation paths and units. Unit is parsed from the derivation path.
@@ -56,6 +59,13 @@ MINT_DERIVATION_PATH="m/0'/0'/0'"
 # e.g. for 100 ppk: up to 10 inputs = 1 sat / 1 cent fee, for up to 20 inputs = 2 sat / 2 cent fee
 MINT_INPUT_FEE_PPK=100
 
+# Automatic keyset rotations
+# When enabled (default: TRUE), active keysets are automatically rotated after the configured
+# interval (default: 90 days / 7,776,000 seconds). The old keyset is deactivated but remains
+# usable for redeeming existing proofs, while a new active keyset is generated.
+# MINT_KEYSET_ROTATION_ENABLED=TRUE
+# MINT_KEYSET_ROTATION_INTERVAL_SECONDS=7776000
+
 # To use SQLite, choose a directory to store the database
 MINT_DATABASE=data/mint
 # To use PostgreSQL, set the connection string
diff --git a/README.md b/README.md
index 9e95f165f..1e572b5d7 100644
--- a/README.md
+++ b/README.md
@@ -214,6 +214,26 @@ poetry run mint
 
 For testing, you can use Nutshell without a Lightning backend by setting `MINT_BACKEND_BOLT11_SAT=FakeWallet` in the `.env` file.
 
+### Automatic Keyset Rotations
+
+Nutshell supports automatic keyset rotations to ensure active keysets are regularly rotated. This behavior is **enabled by default** with a default rotation interval of **90 days**.
+
+When a keyset rotation occurs:
+1. A new keyset is activated (by automatically incrementing the derivation path counter, e.g., `m/0'/0'/0'` -> `m/0'/0'/1'`).
+2. The old keyset is set to inactive but remains usable for redeeming existing ecash proofs.
+3. The mint automatically recovers the latest active keyset from the database on subsequent restarts. You do **not** need to manually update `MINT_DERIVATION_PATH` in your `.env` file.
+
+#### Configuration
+You can customize or disable automatic keyset rotations in your `.env`:
+
+```bash
+# Enable or disable automatic rotations (default: TRUE)
+MINT_KEYSET_ROTATION_ENABLED=TRUE
+
+# Set the rotation interval in seconds (default: 7776000 for 90 days)
+MINT_KEYSET_ROTATION_INTERVAL_SECONDS=7776000
+```
+
 ### NUT-19 Caching with Redis
 
 To cache HTTP responses ([NUT-19](https://github.com/cashubtc/nuts/blob/main/19.md)), you can either install Redis manually or use the docker compose file in `docker/redis/docker-compose.yaml` to start Redis in a container.
diff --git a/cashu/core/settings.py b/cashu/core/settings.py
index b7f4bdad8..5059bd7b0 100644
--- a/cashu/core/settings.py
+++ b/cashu/core/settings.py
@@ -86,6 +86,18 @@ class MintSettings(CashuSettings):
         description="Interval (in seconds) for running regular tasks like the invoice checker.",
     )
 
+    mint_keyset_rotation_enabled: bool = Field(
+        default=True,
+        title="Keyset rotation enabled",
+        description="Whether to automatically rotate keysets when they exceed the interval.",
+    )
+    mint_keyset_rotation_interval_seconds: int = Field(
+        default=7776000,
+        gt=0,
+        title="Keyset rotation interval",
+        description="The interval in seconds after which active keysets are automatically rotated.",
+    )
+
     mint_retry_exponential_backoff_base_delay: int = Field(default=1)
     mint_retry_exponential_backoff_max_delay: int = Field(default=10)
 
diff --git a/cashu/mint/crud.py b/cashu/mint/crud.py
index 4a6b05491..035e4d661 100644
--- a/cashu/mint/crud.py
+++ b/cashu/mint/crud.py
@@ -887,6 +887,15 @@ async def store_keyset(
         keyset: MintKeyset,
         conn: Optional[Connection] = None,
     ) -> None:
+        # NOTE: back-fills timestamps on the caller's keyset object (mutation is
+        # intentional) so in-memory copies stay consistent with the stored row.
+        if not keyset.valid_from:
+            keyset.valid_from = db.timestamp_now_str()
+        if not keyset.valid_to:
+            keyset.valid_to = db.timestamp_now_str()
+        if not keyset.first_seen:
+            keyset.first_seen = db.timestamp_now_str()
+
         await (conn or db).execute(
             f"""
             INSERT INTO {db.table_with_schema("keysets")}
@@ -899,13 +908,9 @@ async def store_keyset(
                 "encrypted_seed": keyset.encrypted_seed,
                 "seed_encryption_method": keyset.seed_encryption_method,
                 "derivation_path": keyset.derivation_path,
-                "valid_from": db.to_timestamp(
-                    keyset.valid_from or db.timestamp_now_str()
-                ),
-                "valid_to": db.to_timestamp(keyset.valid_to or db.timestamp_now_str()),
-                "first_seen": db.to_timestamp(
-                    keyset.first_seen or db.timestamp_now_str()
-                ),
+                "valid_from": db.to_timestamp(keyset.valid_from),
+                "valid_to": db.to_timestamp(keyset.valid_to),
+                "first_seen": db.to_timestamp(keyset.first_seen),
                 "active": True,
                 "version": keyset.version,
                 "unit": keyset.unit.name,
@@ -1045,7 +1050,6 @@ async def update_keyset(
                 "version": keyset.version,
                 "unit": keyset.unit.name,
                 "input_fee_ppk": keyset.input_fee_ppk,
-                "balance": keyset.balance,
                 "final_expiry": keyset.final_expiry,  # NEW: Update final expiry
             },
         )
diff --git a/cashu/mint/keysets.py b/cashu/mint/keysets.py
index 0d9bf5fa6..fa1e5fa6c 100644
--- a/cashu/mint/keysets.py
+++ b/cashu/mint/keysets.py
@@ -1,4 +1,7 @@
 import base64
+import copy
+import datetime
+import time
 from typing import Dict, List, Optional
 
 from loguru import logger
@@ -11,6 +14,9 @@
 
 
 class LedgerKeysets(SupportsKeysets, SupportsSeed, SupportsDb):
+    keyset: MintKeyset
+    derivation_path: str
+
     # ------- KEYS -------
 
     def maybe_update_derivation_path(self, derivation_path: str) -> str:
@@ -40,6 +46,7 @@ async def rotate_next_keyset(
         max_order: Optional[int] = None,
         input_fee_ppk: Optional[int] = None,
         final_expiry: Optional[int] = None,
+        active_keyset_id: Optional[str] = None,
     ) -> MintKeyset:
         """
         This function:
@@ -54,64 +61,117 @@ async def rotate_next_keyset(
             max_order (Optional[int], optional): The number of keys to generate, which correspond to powers of 2.
             input_fee_ppk (Optional[int], optional):  The new keyset's fee
             final_expiry (Optional[int], optional): The keyset's expiration date, after which it might be dropped from the database.
+            active_keyset_id (Optional[str], optional): The active keyset ID that triggered rotation check.
         Returns:
             MintKeyset: Resulting keyset of the rotation
         """
 
         logger.info(f"Attempting keyset rotation for unit {str(unit)}")
 
-        # Select keyset with the greatest counter
-        selected_keyset = None
-        selected_keyset_counter = -1
-        for keyset in self.keysets.values():
-            if keyset.active and keyset.unit == unit:
-                keyset_derivation_path = keyset.derivation_path.split("/")
-                keyset_derivation_counter = int(
-                    keyset_derivation_path[-1].replace("'", "")
+        async with self.db.connect(lock_table="keysets") as conn:
+            # Sync in-memory keysets from DB inside the lock preserving object identity
+            db_keysets = await self.crud.get_keyset(db=self.db, conn=conn)
+            for k in db_keysets:
+                if k.id not in self.keysets:
+                    self.keysets[k.id] = k
+                else:
+                    self.keysets[k.id].active = k.active
+                    self.keysets[k.id].valid_from = k.valid_from
+                    self.keysets[k.id].valid_to = k.valid_to
+                    self.keysets[k.id].first_seen = k.first_seen
+                    self.keysets[k.id].final_expiry = k.final_expiry
+
+            # Avoid concurrent rotations if another task/instance has already rotated
+            if active_keyset_id:
+                target_keyset = self.keysets.get(active_keyset_id)
+                if target_keyset and not target_keyset.active:
+                    logger.info(
+                        f"Keyset {active_keyset_id} was already deactivated (likely rotated by another process/task). Skipping redundant rotation."
+                    )
+                    active_keyset = max(
+                        (
+                            k
+                            for k in self.keysets.values()
+                            if k.active and k.unit == unit
+                        ),
+                        key=lambda k: int(
+                            k.derivation_path.split("/")[-1].replace("'", "")
+                        ),
+                        default=None,
+                    )
+                    if active_keyset:
+                        if self.keyset and self.keyset.id == active_keyset_id:
+                            self.keyset = active_keyset
+                            self.derivation_path = active_keyset.derivation_path
+                            logger.info(
+                                f"Updated default keyset to {active_keyset.id} with derivation path {active_keyset.derivation_path}"
+                            )
+                        return active_keyset
+
+            # Select keyset with the greatest counter
+            selected_keyset = None
+            selected_keyset_counter = -1
+            for keyset in self.keysets.values():
+                if keyset.active and keyset.unit == unit:
+                    keyset_derivation_path = keyset.derivation_path.split("/")
+                    keyset_derivation_counter = int(
+                        keyset_derivation_path[-1].replace("'", "")
+                    )
+                    if keyset_derivation_counter > selected_keyset_counter:
+                        selected_keyset = keyset
+                        selected_keyset_counter = keyset_derivation_counter
+
+            # If no selected keyset, then there is no keyset for this unit
+            if not selected_keyset:
+                logger.error(
+                    f"Couldn't find suitable keyset for rotation with unit {str(unit)}"
+                )
+                raise Exception(
+                    f"Couldn't find suitable keyset for rotation with unit {str(unit)}"
                 )
-                if keyset_derivation_counter > selected_keyset_counter:
-                    selected_keyset = keyset
 
-        # If no selected keyset, then there is no keyset for this unit
-        if not selected_keyset:
-            logger.error(
-                f"Couldn't find suitable keyset for rotation with unit {str(unit)}"
-            )
-            raise Exception(
-                f"Couldn't find suitable keyset for rotation with unit {str(unit)}"
-            )
+            logger.info(f"Rotating keyset {selected_keyset.id}")
 
-        logger.info(f"Rotating keyset {selected_keyset.id}")
+            # New derivation path is just old derivation path with increased counter
+            new_derivation_path = selected_keyset.derivation_path.split("/")
+            new_derivation_path[-1] = (
+                str(int(new_derivation_path[-1].replace("'", "")) + 1) + "'"
+            )
 
-        # New derivation path is just old derivation path with increased counter
-        new_derivation_path = selected_keyset.derivation_path.split("/")
-        new_derivation_path[-1] = (
-            str(int(new_derivation_path[-1].replace("'", "")) + 1) + "'"
-        )
+            # keys amounts for this keyset: if amounts is None we use `self.amounts`
+            amounts = [2**i for i in range(max_order)] if max_order else self.amounts
+
+            # Generate the keyset
+            new_keyset = MintKeyset(
+                derivation_path="/".join(new_derivation_path),
+                seed=self.seed,
+                amounts=amounts,
+                input_fee_ppk=input_fee_ppk,
+                active=True,
+                final_expiry=final_expiry,
+            )
 
-        # keys amounts for this keyset: if amounts is None we use `self.amounts`
-        amounts = [2**i for i in range(max_order)] if max_order else self.amounts
-
-        # Generate the keyset
-        new_keyset = MintKeyset(
-            derivation_path="/".join(new_derivation_path),
-            seed=self.seed,
-            amounts=amounts,
-            input_fee_ppk=input_fee_ppk,
-            active=True,
-            final_expiry=final_expiry
-        )
+            logger.debug(f"New keyset was generated with Id {new_keyset.id}. Saving...")
+            await self.crud.store_keyset(keyset=new_keyset, db=self.db, conn=conn)
 
-        logger.debug(f"New keyset was generated with Id {new_keyset.id}. Saving...")
-        await self.crud.store_keyset(keyset=new_keyset, db=self.db)
-        self.keysets[new_keyset.id] = new_keyset
+            logger.debug(f"De-activating keyset {selected_keyset.id}...")
+            inactive_keyset = copy.copy(selected_keyset)
+            inactive_keyset.active = False
+            await self.crud.update_keyset(keyset=inactive_keyset, db=self.db, conn=conn)
 
-        logger.debug(f"De-activating keyset {selected_keyset.id}...")
+        # Update live state only after the database transaction has committed.
         selected_keyset.active = False
-        await self.crud.update_keyset(keyset=selected_keyset, db=self.db)
+        self.keysets[new_keyset.id] = new_keyset
         self.keysets[selected_keyset.id] = selected_keyset
 
-        logger.debug(f"Keyset {keyset.id} was de-activated")
+        if self.keyset and self.keyset.id == selected_keyset.id:
+            self.keyset = new_keyset
+            self.derivation_path = new_keyset.derivation_path
+            logger.info(
+                f"Updated default keyset to {new_keyset.id} with derivation path {new_keyset.derivation_path}"
+            )
+
+        logger.debug(f"Keyset {selected_keyset.id} was de-activated")
         return new_keyset
 
     async def activate_keyset(
@@ -245,6 +305,82 @@ async def inactivate_base64_keysets(self) -> None:
                 self.keysets[keyset.id] = keyset
                 await self.crud.update_keyset(keyset=keyset, db=self.db)
 
+    def _parse_valid_from(self, keyset: MintKeyset) -> float:
+        # Handles multiple types for keyset.valid_from because database drivers return
+        # different types (PostgreSQL returns datetime.datetime, SQLite stores/returns
+        # stringified timestamp integers/floats), while test mocks or JSON payloads
+        # may supply formatted datetime strings.
+        # NOTE: naive datetimes and "%Y-%m-%d %H:%M:%S" strings are interpreted in
+        # server-local time, matching db.timestamp_from_seconds. This round-trips
+        # correctly only as long as writer and reader share a timezone.
+        if not keyset.valid_from:
+            raise ValueError("keyset.valid_from is None")
+        try:
+            if isinstance(keyset.valid_from, datetime.datetime):
+                return keyset.valid_from.timestamp()
+            else:
+                return float(keyset.valid_from)
+        except (ValueError, TypeError):
+            return datetime.datetime.strptime(
+                keyset.valid_from, "%Y-%m-%d %H:%M:%S"
+            ).timestamp()
+
+    def should_rotate_keyset(self, keyset: MintKeyset) -> bool:
+        if not keyset.active or not keyset.valid_from:
+            return False
+        try:
+            valid_from_ts = self._parse_valid_from(keyset)
+        except Exception:
+            # Fail closed: if we cannot determine the keyset's age we must not
+            # rotate. A systemic parsing failure would otherwise rotate every
+            # active keyset on every regular-tasks tick, growing the keyset
+            # table unboundedly.
+            logger.error(
+                f"Could not parse valid_from: {keyset.valid_from}. Skipping rotation."
+            )
+            return False
+
+        return (
+            time.time() - valid_from_ts
+        ) >= settings.mint_keyset_rotation_interval_seconds
+
+    async def rotate_keysets_if_needed(self) -> None:
+        if not settings.mint_keyset_rotation_enabled:
+            return
+
+        active_keysets = [k for k in self.keysets.values() if k.active]
+        for keyset in active_keysets:
+            if self.should_rotate_keyset(keyset):
+                logger.warning(
+                    f"Active keyset {keyset.id} for unit {keyset.unit.name} is older than "
+                    f"the configured rotation interval ({settings.mint_keyset_rotation_interval_seconds}s). "
+                    f"Rotating now."
+                )
+                try:
+                    new_final_expiry = None
+                    if keyset.final_expiry is not None:
+                        try:
+                            valid_from_ts = self._parse_valid_from(keyset)
+                        except Exception:
+                            valid_from_ts = time.time()
+                        active_duration = int(time.time() - valid_from_ts)
+                        new_final_expiry = keyset.final_expiry + active_duration
+
+                    new_keyset = await self.rotate_next_keyset(
+                        unit=keyset.unit,
+                        max_order=len(keyset.amounts),
+                        input_fee_ppk=keyset.input_fee_ppk,
+                        final_expiry=new_final_expiry,
+                        active_keyset_id=keyset.id,
+                    )
+                    logger.info(
+                        f"Successfully rotated keyset {keyset.id} -> {new_keyset.id} for unit {keyset.unit.name}"
+                    )
+                except Exception as e:
+                    logger.error(
+                        f"Failed to automatically rotate keyset {keyset.id}: {e}"
+                    )
+
     def get_keyset(self, keyset_id: Optional[str] = None) -> Dict[int, str]:
         """Returns a dictionary of hex public keys of a specific keyset for each supported amount"""
         if keyset_id and keyset_id not in self.keysets:
diff --git a/cashu/mint/ledger.py b/cashu/mint/ledger.py
index 497d236c8..3b9bfbaef 100644
--- a/cashu/mint/ledger.py
+++ b/cashu/mint/ledger.py
@@ -154,6 +154,7 @@ async def startup_ledger(self) -> None:
 
     async def _startup_keysets(self) -> None:
         await self.init_keysets()
+        await self.rotate_keysets_if_needed()
         for derivation_path in settings.mint_derivation_path_list:
             derivation_path = self.maybe_update_derivation_path(derivation_path)
             await self.activate_keyset(derivation_path=derivation_path)
@@ -167,6 +168,7 @@ async def _run_regular_tasks(self) -> None:
         while True:
             try:
                 await self._check_pending_proofs_and_melt_quotes()
+                await self.rotate_keysets_if_needed()
                 await asyncio.sleep(settings.mint_regular_tasks_interval_seconds)
             except Exception as e:
                 logger.error(f"Ledger regular task failed: {e}")
diff --git a/tests/mint/test_mint_automatic_rotations.py b/tests/mint/test_mint_automatic_rotations.py
new file mode 100644
index 000000000..32490af19
--- /dev/null
+++ b/tests/mint/test_mint_automatic_rotations.py
@@ -0,0 +1,548 @@
+import asyncio
+import datetime
+import time
+
+import pytest
+
+from cashu.core.base import MintKeyset, Unit
+from cashu.core.settings import settings
+from cashu.mint.ledger import Ledger
+from cashu.mint.startup import ledger as global_ledger
+
+
+@pytest.fixture(autouse=True)
+def disable_global_ledger_rotation(monkeypatch: pytest.MonkeyPatch):
+    async def noop_rotate(*args, **kwargs):
+        pass
+
+    monkeypatch.setattr(global_ledger, "rotate_keysets_if_needed", noop_rotate)
+    yield
+
+
+@pytest.mark.asyncio
+async def test_should_rotate_keyset_behavior(ledger: Ledger):
+    # Get any active keyset
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+
+    # By default, freshly created keyset should not rotate
+    assert not ledger.should_rotate_keyset(keyset)
+
+    # If keyset is inactive, it should never rotate
+    keyset.active = False
+    assert not ledger.should_rotate_keyset(keyset)
+    keyset.active = True
+
+    # If valid_from is mocked in the far past, it should rotate
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    settings.mint_keyset_rotation_interval_seconds = 2592000  # 30 days
+    try:
+        original_valid_from = keyset.valid_from
+        keyset.valid_from = (
+            datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=31)
+        ).strftime("%Y-%m-%d %H:%M:%S")
+        assert ledger.should_rotate_keyset(keyset)
+    finally:
+        # Restore
+        keyset.valid_from = original_valid_from
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+
+
+@pytest.mark.asyncio
+async def test_should_not_rotate_on_unparseable_valid_from(ledger: Ledger):
+    """Fail closed: an unparseable valid_from must not trigger rotation."""
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+    original_valid_from = keyset.valid_from
+    try:
+        keyset.valid_from = "not-a-timestamp"
+        assert not ledger.should_rotate_keyset(keyset)
+    finally:
+        keyset.valid_from = original_valid_from
+
+
+@pytest.mark.asyncio
+async def test_automatic_keyset_rotation_flow(ledger: Ledger):
+    # Cancel background tasks to avoid race conditions with manual triggering
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    # Set a very short rotation interval
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+
+    try:
+        settings.mint_keyset_rotation_enabled = True
+        settings.mint_keyset_rotation_interval_seconds = 1
+
+        # Keep track of active keysets before rotation
+        active_keysets_before = {k.unit: k for k in ledger.keysets.values() if k.active}
+        assert len(active_keysets_before) > 0
+
+        # Wait to exceed the 1 second interval
+        await asyncio.sleep(1.5)
+
+        # Trigger automatic rotation check
+        await ledger.rotate_keysets_if_needed()
+
+        # Get active keysets after rotation
+        active_keysets_after = {k.unit: k for k in ledger.keysets.values() if k.active}
+
+        for unit, old_keyset in active_keysets_before.items():
+            new_keyset = active_keysets_after[unit]
+            # Verify a new keyset has been created and it differs from the old one
+            assert old_keyset.id != new_keyset.id
+
+            # Verify the old keyset is now inactive in memory and DB
+            assert not old_keyset.active
+            db_old_keysets = await ledger.crud.get_keyset(
+                db=ledger.db, id=old_keyset.id
+            )
+            assert len(db_old_keysets) == 1
+            assert not db_old_keysets[0].active
+
+            # Verify new keyset is active in memory and DB
+            assert new_keyset.active
+            db_new_keysets = await ledger.crud.get_keyset(
+                db=ledger.db, id=new_keyset.id
+            )
+            assert len(db_new_keysets) == 1
+            assert db_new_keysets[0].active
+
+            # Verify key parameters are preserved
+            assert new_keyset.input_fee_ppk == old_keyset.input_fee_ppk
+            assert len(new_keyset.amounts) == len(old_keyset.amounts)
+
+            # Verify derivation path counter has incremented
+            old_path = old_keyset.derivation_path.split("/")
+            new_path = new_keyset.derivation_path.split("/")
+            assert old_path[:-1] == new_path[:-1]
+            assert (
+                int(new_path[-1].replace("'", "")) - int(old_path[-1].replace("'", ""))
+                == 1
+            )
+
+            # If the rotated unit matches the default keyset's unit, verify that
+            # the default keyset and derivation path are updated on the ledger
+            if unit == ledger.keyset.unit:
+                assert ledger.keyset.id == new_keyset.id
+                assert ledger.derivation_path == new_keyset.derivation_path
+
+    finally:
+        # Restore settings
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+
+
+@pytest.mark.asyncio
+async def test_automatic_keyset_rotation_disabled(ledger: Ledger):
+    # Cancel background tasks to avoid race conditions with manual triggering
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    # Keep track of active keyset
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+
+    try:
+        settings.mint_keyset_rotation_enabled = False
+        settings.mint_keyset_rotation_interval_seconds = 1
+
+        # Wait to exceed interval
+        await asyncio.sleep(1.5)
+
+        # Trigger check (should do nothing since disabled)
+        await ledger.rotate_keysets_if_needed()
+
+        # Get active keyset for the same unit
+        active_keysets = [
+            k for k in ledger.keysets.values() if k.active and k.unit == keyset.unit
+        ]
+        assert len(active_keysets) == 1
+        assert active_keysets[0].id == keyset.id
+        assert active_keysets[0].active
+
+    finally:
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+
+
+@pytest.mark.asyncio
+async def test_automatic_keyset_rotation_preserves_grace_period(ledger: Ledger):
+    # Cancel background tasks to avoid race conditions with manual triggering
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    # Get any active keyset
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+
+    try:
+        settings.mint_keyset_rotation_enabled = True
+        settings.mint_keyset_rotation_interval_seconds = 1
+
+        # Set a mock final_expiry on the old keyset
+        keyset.final_expiry = 2000000000
+
+        # Generate the timestamp 5 seconds ago using the database's format to avoid timezone shifts
+        past_ts = int(time.time() - 5)
+        keyset.valid_from = ledger.db.timestamp_from_seconds(past_ts)
+
+        # Trigger automatic rotation check
+        await ledger.rotate_keysets_if_needed()
+
+        # Retrieve the new active keyset for this unit
+        new_keyset = next(
+            k for k in ledger.keysets.values() if k.active and k.unit == keyset.unit
+        )
+
+        # Verify a rotation occurred
+        assert keyset.id != new_keyset.id
+
+        # Expected new final_expiry should be original final_expiry (2000000000) + active_duration (approx 5)
+        assert new_keyset.final_expiry is not None
+        assert 2000000004 <= new_keyset.final_expiry <= 2000000008
+
+    finally:
+        # Restore settings
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+
+
+@pytest.mark.asyncio
+async def test_automatic_keyset_rotation_background(ledger: Ledger):
+    # Keep track of original settings
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+    original_tasks_interval = settings.mint_regular_tasks_interval_seconds
+
+    try:
+        # Set configuration so that background tasks and keyset rotations run very frequently
+        settings.mint_keyset_rotation_enabled = True
+        settings.mint_keyset_rotation_interval_seconds = 1
+        settings.mint_regular_tasks_interval_seconds = 2  # Check every 2 seconds
+
+        # Cancel existing regular tasks so we can restart with the new interval
+        for task in ledger.regular_tasks:
+            task.cancel()
+        ledger.regular_tasks = []
+
+        # Start a new regular tasks loop with the updated 2-second interval
+        ledger.regular_tasks.append(asyncio.create_task(ledger._run_regular_tasks()))
+
+        # Keep track of active keysets before background rotation
+        active_keysets_before = {k.unit: k for k in ledger.keysets.values() if k.active}
+        assert len(active_keysets_before) > 0
+
+        # Wait to exceed the rotation interval and allow the background task to run
+        # Keyset rotation interval is 1s, and task runs every 2s, so 2.5s is plenty of time
+        # for exactly one background rotation to run and complete.
+        await asyncio.sleep(2.5)
+
+        # Get active keysets after background rotation
+        active_keysets_after = {k.unit: k for k in ledger.keysets.values() if k.active}
+
+        # Verify background rotation occurred successfully
+        for unit, old_keyset in active_keysets_before.items():
+            new_keyset = active_keysets_after[unit]
+            assert old_keyset.id != new_keyset.id
+            assert not old_keyset.active
+            assert new_keyset.active
+
+            # If the rotated unit matches the default keyset's unit, verify that
+            # the default keyset and derivation path are updated on the ledger
+            if unit == ledger.keyset.unit:
+                assert ledger.keyset.id == new_keyset.id
+                assert ledger.derivation_path == new_keyset.derivation_path
+
+    finally:
+        # Restore settings and restart original tasks loop
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+        settings.mint_regular_tasks_interval_seconds = original_tasks_interval
+
+        for task in ledger.regular_tasks:
+            task.cancel()
+        ledger.regular_tasks = []
+        ledger.regular_tasks.append(asyncio.create_task(ledger._run_regular_tasks()))
+
+
+@pytest.mark.asyncio
+async def test_regression_non_atomic_rotation(ledger: Ledger):
+    """
+    Regression test: rotation is not atomic.
+    Verifies that if update_keyset fails, the entire transaction is rolled back.
+    The new keyset is not saved, and only the old active keyset remains in DB.
+    """
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    # Get active keyset and set its valid_from to the past
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+    original_valid_from = keyset.valid_from
+    keyset.valid_from = (
+        datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=365)
+    ).strftime("%Y-%m-%d %H:%M:%S")
+    await ledger.crud.update_keyset(keyset=keyset, db=ledger.db)
+
+    original_update_keyset = ledger.crud.update_keyset
+
+    async def mock_update_keyset(*args, **kwargs):
+        raise Exception("Simulated DB crash/disconnection during update_keyset")
+
+    ledger.crud.update_keyset = mock_update_keyset
+
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+
+    try:
+        settings.mint_keyset_rotation_enabled = True
+        settings.mint_keyset_rotation_interval_seconds = 1
+
+        # Trigger rotation check. This will catch and suppress the Exception we raise in update_keyset
+        await ledger.rotate_keysets_if_needed()
+
+        # Due to transactional rollback, the new keyset should NOT have been stored in the DB,
+        # and the old keyset should still be active.
+        db_all_keysets = await ledger.crud.get_keyset(db=ledger.db)
+        active_db_keysets = [
+            k for k in db_all_keysets if k.active and k.unit == keyset.unit
+        ]
+
+        # There should be exactly ONE active keyset for this unit (the old one)
+        assert len(active_db_keysets) == 1, (
+            f"Expected exactly 1 active keyset in DB, found: {len(active_db_keysets)}"
+        )
+        assert active_db_keysets[0].id == keyset.id, (
+            "The active keyset should be the original one"
+        )
+
+        # The failed transaction must not leak into the live ledger state.
+        active_memory_keysets = [
+            k for k in ledger.keysets.values() if k.active and k.unit == keyset.unit
+        ]
+        assert len(active_memory_keysets) == 1
+        assert active_memory_keysets[0].id == keyset.id
+        assert ledger.should_rotate_keyset(keyset)
+
+        # The live keyset remains eligible for a later retry.
+        ledger.crud.update_keyset = original_update_keyset
+        await ledger.rotate_keysets_if_needed()
+        active_memory_keysets = [
+            k for k in ledger.keysets.values() if k.active and k.unit == keyset.unit
+        ]
+        assert len(active_memory_keysets) == 1
+        assert active_memory_keysets[0].id != keyset.id
+
+    finally:
+        ledger.crud.update_keyset = original_update_keyset
+        keyset.valid_from = original_valid_from
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+
+
+@pytest.mark.asyncio
+async def test_regression_concurrent_rotation_race(ledger: Ledger):
+    """
+    Regression test: concurrent instances/manual rotation can race.
+    Verifies that with locks and deactivation status checks, concurrent rotation requests
+    safely serialize. The second request detects that the keyset was already rotated and skips gracefully,
+    raising no exceptions.
+    """
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    keyset = next(k for k in ledger.keysets.values() if k.active)
+    original_valid_from = keyset.valid_from
+    keyset.valid_from = (
+        datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=365)
+    ).strftime("%Y-%m-%d %H:%M:%S")
+
+    original_interval = settings.mint_keyset_rotation_interval_seconds
+    original_enabled = settings.mint_keyset_rotation_enabled
+
+    try:
+        settings.mint_keyset_rotation_enabled = True
+        settings.mint_keyset_rotation_interval_seconds = 1
+
+        # Run two concurrent rotations passing active_keyset_id
+        results = await asyncio.gather(
+            ledger.rotate_next_keyset(unit=keyset.unit, active_keyset_id=keyset.id),
+            ledger.rotate_next_keyset(unit=keyset.unit, active_keyset_id=keyset.id),
+            return_exceptions=True,
+        )
+
+        # Neither of them should raise an exception (both return a valid MintKeyset)
+        exceptions = [res for res in results if isinstance(res, BaseException)]
+        assert len(exceptions) == 0, f"Expected no exceptions, but got: {exceptions}"
+
+        # Both results should be MintKeysets and they should be identical (the same rotated keyset)
+        assert isinstance(results[0], MintKeyset)
+        assert isinstance(results[1], MintKeyset)
+        assert results[0].id == results[1].id, (
+            "Expected both parallel tasks to return the same rotated keyset ID"
+        )
+
+    finally:
+        keyset.valid_from = original_valid_from
+        settings.mint_keyset_rotation_interval_seconds = original_interval
+        settings.mint_keyset_rotation_enabled = original_enabled
+
+
+@pytest.mark.asyncio
+async def test_concurrent_rotation_updates_stale_default_keyset(ledger: Ledger):
+    """A worker that loses the rotation race adopts the winner's keyset."""
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    old_keyset = ledger.keyset
+    path_parts = old_keyset.derivation_path.split("/")
+    path_parts[-1] = str(int(path_parts[-1].replace("'", "")) + 1) + "'"
+    new_keyset = MintKeyset(
+        derivation_path="/".join(path_parts),
+        seed=ledger.seed,
+        amounts=old_keyset.amounts,
+        input_fee_ppk=old_keyset.input_fee_ppk,
+        active=True,
+    )
+
+    await ledger.crud.store_keyset(keyset=new_keyset, db=ledger.db)
+    old_keyset.active = False
+    await ledger.crud.update_keyset(keyset=old_keyset, db=ledger.db)
+
+    # Simulate another worker that has not yet observed the committed rotation.
+    old_keyset.active = True
+    ledger.keysets.pop(new_keyset.id, None)
+    ledger.keyset = old_keyset
+    ledger.derivation_path = old_keyset.derivation_path
+
+    result = await ledger.rotate_next_keyset(
+        unit=old_keyset.unit,
+        active_keyset_id=old_keyset.id,
+    )
+
+    assert result.id == new_keyset.id
+    assert ledger.keyset.id == new_keyset.id
+    assert ledger.derivation_path == new_keyset.derivation_path
+
+
+@pytest.mark.asyncio
+async def test_concurrent_rotation_selects_newest_active_keyset(ledger: Ledger):
+    """A losing worker must not adopt an older active keyset."""
+    for task in ledger.regular_tasks:
+        task.cancel()
+    ledger.regular_tasks = []
+
+    old_keyset = ledger.keyset
+    original_counter = int(old_keyset.derivation_path.split("/")[-1].replace("'", ""))
+
+    def keyset_at_counter(counter: int, *, active: bool) -> MintKeyset:
+        path_parts = old_keyset.derivation_path.split("/")
+        path_parts[-1] = f"{counter}'"
+        return MintKeyset(
+            derivation_path="/".join(path_parts),
+            seed=ledger.seed,
+            amounts=old_keyset.amounts,
+            input_fee_ppk=old_keyset.input_fee_ppk,
+            active=active,
+        )
+
+    rotated_keyset = keyset_at_counter(original_counter + 1, active=False)
+    newest_keyset = keyset_at_counter(original_counter + 2, active=True)
+    await ledger.crud.store_keyset(keyset=rotated_keyset, db=ledger.db)
+    await ledger.crud.update_keyset(keyset=rotated_keyset, db=ledger.db)
+    await ledger.crud.store_keyset(keyset=newest_keyset, db=ledger.db)
+
+    # The old keyset remains active and precedes the winner in insertion order.
+    ledger.keysets[rotated_keyset.id] = rotated_keyset
+    ledger.keyset = rotated_keyset
+    ledger.derivation_path = rotated_keyset.derivation_path
+
+    result = await ledger.rotate_next_keyset(
+        unit=old_keyset.unit,
+        active_keyset_id=rotated_keyset.id,
+    )
+
+    assert result.id == newest_keyset.id
+    assert ledger.keyset.id == newest_keyset.id
+    assert ledger.derivation_path == newest_keyset.derivation_path
+
+
+@pytest.mark.asyncio
+async def test_regression_highest_counter_selection_incomplete(ledger: Ledger):
+    """
+    Regression test: highest-counter selection is incomplete.
+    Verifies that selected_keyset_counter is updated properly and the keyset with the
+    absolute highest counter is selected for rotation.
+    """
+    # Create dummy active keysets with different counters
+    keyset_high = MintKeyset(
+        derivation_path="m/0/0/0/5'",
+        seed=ledger.seed,
+        amounts=ledger.amounts,
+        active=True,
+        unit="sat",
+    )
+    keyset_low_after = MintKeyset(
+        derivation_path="m/0/0/0/2'",
+        seed=ledger.seed,
+        amounts=ledger.amounts,
+        active=True,
+        unit="sat",
+    )
+    keyset_medium_after = MintKeyset(
+        derivation_path="m/0/0/0/4'",
+        seed=ledger.seed,
+        amounts=ledger.amounts,
+        active=True,
+        unit="sat",
+    )
+
+    # Backup original active keysets for Unit.sat to restore later
+    original_keysets = dict(ledger.keysets)
+    try:
+        # Clear out other active keysets for Unit.sat
+        for k_id, k in list(ledger.keysets.items()):
+            if k.unit == Unit.sat:
+                ledger.keysets.pop(k_id)
+
+        # Add them in order: High counter first, then lower ones.
+        ledger.keysets[keyset_high.id] = keyset_high
+        ledger.keysets[keyset_low_after.id] = keyset_low_after
+        ledger.keysets[keyset_medium_after.id] = keyset_medium_after
+
+        # Mock database writes so we don't pollute the DB during unit test
+        original_store = ledger.crud.store_keyset
+        original_update = ledger.crud.update_keyset
+
+        async def mock_noop(*args, **kwargs):
+            pass
+
+        ledger.crud.store_keyset = mock_noop
+        ledger.crud.update_keyset = mock_noop
+
+        try:
+            rotated_keyset = await ledger.rotate_next_keyset(unit=Unit.sat)
+
+            # Since high counter (5) is correctly selected, the new counter should be 5 + 1 = 6.
+            rotated_counter = int(
+                rotated_keyset.derivation_path.split("/")[-1].replace("'", "")
+            )
+            assert rotated_counter == 6, (
+                f"Expected rotated counter to be 6, got {rotated_counter}"
+            )
+
+        finally:
+            ledger.crud.store_keyset = original_store
+            ledger.crud.update_keyset = original_update
+
+    finally:
+        ledger.keysets = original_keysets