fix(gateway): pin a re-encryption to the ciphertext it read

docs/public/openapi.json

@@ -9278,6 +9278,12 @@
             "title": "Reencrypted",
             "type": "integer"
           },
+          "skipped": {
+            "default": 0,
+            "description": "Number of rows whose stored key changed between the read and the write, so the re-encryption was not applied. They already hold whoever wrote them last.",
+            "title": "Skipped",
+            "type": "integer"
+          },
           "unreadable": {
             "description": "Number of encrypted keys left untouched because they could not be decrypted.",
             "title": "Unreadable",
@@ -9299,6 +9305,12 @@
             "title": "Reencrypted",
             "type": "integer"
           },
+          "skipped": {
+            "default": 0,
+            "description": "Number of rows whose stored key changed between the read and the write, so the re-encryption was not applied. They already hold whoever wrote them last.",
+            "title": "Skipped",
+            "type": "integer"
+          },
           "unreadable": {
             "description": "Number of encrypted keys left untouched because they could not be decrypted.",
             "title": "Unreadable",

src/gateway/api/routes/providers.py

@@ -370,6 +370,13 @@ class ReencryptProviderCredentialsResponse(BaseModel):
 
     reencrypted: int = Field(description="Number of stored provider keys re-encrypted.")
     unreadable: int = Field(description="Number of encrypted keys left untouched because they could not be decrypted.")
+    skipped: int = Field(
+        default=0,
+        description=(
+            "Number of rows whose stored key changed between the read and the write, so the "
+            "re-encryption was not applied. They already hold whoever wrote them last."
+        ),
+    )
 
 
 class TestProviderRequest(BaseModel):
@@ -506,7 +513,7 @@ async def reencrypt_stored_provider_keys(
     by replacing the affected provider keys.
     """
     try:
-        reencrypted, unreadable = await reencrypt_credentials(db)
+        reencrypted, unreadable, skipped = await reencrypt_credentials(db)
         await db.commit()
     except SecretBoxUnavailableError as exc:
         await db.rollback()
@@ -519,7 +526,7 @@ async def reencrypt_stored_provider_keys(
         await refresh_provider_cache(db, config)
     except SQLAlchemyError:
         logger.warning("Provider overlay refresh failed after re-encrypting credentials; converges within TTL")
-    return ReencryptProviderCredentialsResponse(reencrypted=reencrypted, unreadable=unreadable)
+    return ReencryptProviderCredentialsResponse(reencrypted=reencrypted, unreadable=unreadable, skipped=skipped)
 
 
 @router.get("/provider-credentials")

src/gateway/api/routes/search_tools.py

@@ -170,6 +170,13 @@ class ReencryptSearchToolsResponse(BaseModel):
 
     reencrypted: int = Field(description="Number of stored search-tool keys re-encrypted.")
     unreadable: int = Field(description="Number of encrypted keys left untouched because they could not be decrypted.")
+    skipped: int = Field(
+        default=0,
+        description=(
+            "Number of rows whose stored key changed between the read and the write, so the "
+            "re-encryption was not applied. They already hold whoever wrote them last."
+        ),
+    )
 
 
 def _is_decryptable(row: SearchToolCredential) -> bool:
@@ -302,7 +309,7 @@ async def reencrypt_stored_search_tool_keys(
     tool's key.
     """
     try:
-        reencrypted, unreadable = await reencrypt_search_tools(db)
+        reencrypted, unreadable, skipped = await reencrypt_search_tools(db)
         await db.commit()
     except SecretBoxUnavailableError as exc:
         await db.rollback()
@@ -314,7 +321,7 @@ async def reencrypt_stored_search_tool_keys(
         await refresh_search_tool_cache(db, config)
     except SQLAlchemyError:
         logger.warning("Search tool overlay refresh failed after re-encrypting keys; converges within TTL")
-    return ReencryptSearchToolsResponse(reencrypted=reencrypted, unreadable=unreadable)
+    return ReencryptSearchToolsResponse(reencrypted=reencrypted, unreadable=unreadable, skipped=skipped)
 
 
 @router.post("", status_code=status.HTTP_201_CREATED)

src/gateway/services/provider_store_service.py

@@ -21,9 +21,9 @@
 
 import asyncio
 import time
-from typing import Any, Final
+from typing import Any, Final, cast
 
-from sqlalchemy import select
+from sqlalchemy import CursorResult, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from gateway.core.config import GatewayConfig
@@ -125,7 +125,15 @@ async def refresh_provider_cache(db: AsyncSession, config: GatewayConfig) -> set
     """Reload the overlay from the database, apply it, and return shadowed names."""
     global _cached_at  # noqa: PLW0603
 
-    rows = (await db.execute(select(ProviderCredential))).scalars().all()
+    # `populate_existing`: the session factory sets `expire_on_commit=False`,
+    # so a row already in the identity map keeps the values it was loaded
+    # with and this SELECT would hand them straight back. The rotation
+    # endpoint refreshes on the same session it just re-encrypted on, where
+    # that means a credential a concurrent PATCH replaced can return to the
+    # cache. Today nothing holds those rows alive that long, which makes it
+    # a garbage-collection timing question rather than a guarantee
+    # (CodeRabbit).
+    rows = (await db.execute(select(ProviderCredential).execution_options(populate_existing=True))).scalars().all()
     overlay: dict[str, dict[str, Any]] = {}
     for row in rows:
         try:
@@ -259,13 +267,25 @@ async def save_credential(
     return row
 
 
-async def reencrypt_credentials(db: AsyncSession) -> tuple[int, int]:
+async def reencrypt_credentials(db: AsyncSession) -> tuple[int, int, int]:
     """Re-encrypt stored provider keys with the current primary OTARI_SECRET_KEY.
 
-    Returns ``(reencrypted, unreadable)``. Rows without a stored key are ignored.
-    If any encrypted key cannot be decrypted with the configured key set, it is
-    left untouched and counted as unreadable so the operator can recover it by
+    Returns ``(reencrypted, unreadable, skipped)``. Rows without a stored key are
+    ignored. A key that cannot be decrypted with the configured key set is left
+    untouched and counted as unreadable, so the operator can recover it by
     replacing that provider's key.
+
+    Each row is written with a conditional UPDATE matching the ciphertext that
+    was read. Rotation reads every row, decrypts and re-encrypts, and nothing
+    pinned the write to what it had seen: an edit committing in that window was
+    overwritten with a re-encryption of the value it replaced — a silent lost
+    update on a credential (otari#1127). Zero rows matched means someone else
+    got there first, and that row is counted as skipped rather than clobbered.
+
+    Skipped is reported rather than retried. A rotation is run by hand, the
+    operator is watching, and a row whose value changed under them is already
+    encrypted with the primary key by whoever wrote it — so the honest answer is
+    "these were not mine to rewrite", not a loop that races the same edit again.
     """
     rows = (
         (await db.execute(select(ProviderCredential).where(ProviderCredential.encrypted_api_key.is_not(None))))
@@ -274,17 +294,31 @@ async def reencrypt_credentials(db: AsyncSession) -> tuple[int, int]:
     )
     reencrypted = 0
     unreadable = 0
+    skipped = 0
     for row in rows:
-        if row.encrypted_api_key is None:
+        original = row.encrypted_api_key
+        if original is None:
             continue
         try:
-            plaintext = decrypt_secret(row.encrypted_api_key)
+            plaintext = decrypt_secret(original)
         except SecretDecryptionError:
             unreadable += 1
             continue
-        row.encrypted_api_key = encrypt_secret(plaintext)
-        reencrypted += 1
-    return reencrypted, unreadable
+        # Core UPDATE rather than a mutation on the loaded row: the whole point
+        # is the WHERE, and an ORM flush would carry no condition at all.
+        result = await db.execute(
+            update(ProviderCredential)
+            .where(ProviderCredential.instance == row.instance, ProviderCredential.encrypted_api_key == original)
+            .values(encrypted_api_key=encrypt_secret(plaintext))
+            .execution_options(synchronize_session=False)
+        )
+        # `execute` is typed as returning Result; an UPDATE always yields a
+        # CursorResult, which is where rowcount lives.
+        if cast(CursorResult[Any], result).rowcount == 1:
+            reencrypted += 1
+        else:
+            skipped += 1
+    return reencrypted, unreadable, skipped
 
 
 async def delete_credential(db: AsyncSession, instance: str) -> bool:

src/gateway/services/search_tool_store_service.py

@@ -22,9 +22,9 @@
 
 import asyncio
 import time
-from typing import Any, Final
+from typing import Any, Final, cast
 
-from sqlalchemy import select
+from sqlalchemy import CursorResult, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from gateway.core.config import GatewayConfig
@@ -129,7 +129,15 @@ async def refresh_search_tool_cache(db: AsyncSession, config: GatewayConfig) ->
     """Reload the overlay from the database, apply it, and return shadowed names."""
     global _cached_at  # noqa: PLW0603
 
-    rows = (await db.execute(select(SearchToolCredential))).scalars().all()
+    # `populate_existing`: the session factory sets `expire_on_commit=False`,
+    # so a row already in the identity map keeps the values it was loaded
+    # with and this SELECT would hand them straight back. The rotation
+    # endpoint refreshes on the same session it just re-encrypted on, where
+    # that means a credential a concurrent PATCH replaced can return to the
+    # cache. Today nothing holds those rows alive that long, which makes it
+    # a garbage-collection timing question rather than a guarantee
+    # (CodeRabbit).
+    rows = (await db.execute(select(SearchToolCredential).execution_options(populate_existing=True))).scalars().all()
     overlay: dict[str, dict[str, Any]] = {}
     for row in rows:
         try:
@@ -262,13 +270,25 @@ async def save_search_tool(
     return row
 
 
-async def reencrypt_search_tools(db: AsyncSession) -> tuple[int, int]:
+async def reencrypt_search_tools(db: AsyncSession) -> tuple[int, int, int]:
     """Re-encrypt stored search-tool keys with the current primary OTARI_SECRET_KEY.
 
-    Returns ``(reencrypted, unreadable)``. Rows without a stored key are ignored.
-    A key that cannot be decrypted with the configured key set is left untouched
-    and counted as unreadable, so the operator can recover it by replacing that
-    tool's key.
+    Returns ``(reencrypted, unreadable, skipped)``. Rows without a stored key are
+    ignored. A key that cannot be decrypted with the configured key set is left
+    untouched and counted as unreadable, so the operator can recover it by
+    replacing that tool's key.
+
+    Each row is written with a conditional UPDATE matching the ciphertext that
+    was read. Rotation reads every row, decrypts and re-encrypts, and nothing
+    pinned the write to what it had seen: an edit committing in that window was
+    overwritten with a re-encryption of the value it replaced — a silent lost
+    update on a credential (otari#1127). Zero rows matched means someone else
+    got there first, and that row is counted as skipped rather than clobbered.
+
+    Skipped is reported rather than retried. A rotation is run by hand, the
+    operator is watching, and a row whose value changed under them is already
+    encrypted with the primary key by whoever wrote it — so the honest answer is
+    "these were not mine to rewrite", not a loop that races the same edit again.
     """
     rows = (
         (await db.execute(select(SearchToolCredential).where(SearchToolCredential.encrypted_api_key.is_not(None))))
@@ -277,17 +297,31 @@ async def reencrypt_search_tools(db: AsyncSession) -> tuple[int, int]:
     )
     reencrypted = 0
     unreadable = 0
+    skipped = 0
     for row in rows:
-        if row.encrypted_api_key is None:
+        original = row.encrypted_api_key
+        if original is None:
             continue
         try:
-            plaintext = decrypt_secret(row.encrypted_api_key)
+            plaintext = decrypt_secret(original)
         except SecretDecryptionError:
             unreadable += 1
             continue
-        row.encrypted_api_key = encrypt_secret(plaintext)
-        reencrypted += 1
-    return reencrypted, unreadable
+        # Core UPDATE rather than a mutation on the loaded row: the whole point
+        # is the WHERE, and an ORM flush would carry no condition at all.
+        result = await db.execute(
+            update(SearchToolCredential)
+            .where(SearchToolCredential.name == row.name, SearchToolCredential.encrypted_api_key == original)
+            .values(encrypted_api_key=encrypt_secret(plaintext))
+            .execution_options(synchronize_session=False)
+        )
+        # `execute` is typed as returning Result; an UPDATE always yields a
+        # CursorResult, which is where rowcount lives.
+        if cast(CursorResult[Any], result).rowcount == 1:
+            reencrypted += 1
+        else:
+            skipped += 1
+    return reencrypted, unreadable, skipped
 
 
 async def delete_search_tool(db: AsyncSession, name: str) -> bool: