diff --git a/backend/app/alembic/versions/076_reencrypt_credentials_envelope.py b/backend/app/alembic/versions/076_reencrypt_credentials_envelope.py
new file mode 100644
index 000000000..d8bde71e2
--- /dev/null
+++ b/backend/app/alembic/versions/076_reencrypt_credentials_envelope.py
@@ -0,0 +1,37 @@
+"""re-encrypt stored credentials to the kms.v2 envelope format
+
+Revision ID: 076
+Revises: 075
+Create Date: 2026-08-07
+
+One-shot data backfill onto envelope encryption (KMS-wrapped data key + AES-GCM).
+Empty table (fresh DB) and Fernet-only (KMS inactive) environments are no-ops, so
+replaying is safe. Requires KMS/IAM access in the migrate runtime — the migrate
+container needs the same role/creds the app uses, else the upgrade aborts.
+"""
+from alembic import op
+from sqlmodel import Session
+
+from app.services.credentials.reencrypt import (
+    execute_credential_reencrypt,
+    execute_credential_reencrypt_fernet,
+)
+
+# revision identifiers, used by Alembic.
+revision = "076"
+down_revision = "075"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    bind = op.get_bind()
+    with Session(bind=bind) as session:
+        execute_credential_reencrypt(session=session)
+
+
+def downgrade():
+    # Reverse the envelope backfill: convert kms.v1/kms.v2 rows back to legacy Fernet.
+    bind = op.get_bind()
+    with Session(bind=bind) as session:
+        execute_credential_reencrypt_fernet(session=session)
diff --git a/backend/app/core/security.py b/backend/app/core/security.py
index 406f46ed8..96f7251ac 100644
--- a/backend/app/core/security.py
+++ b/backend/app/core/security.py
@@ -20,6 +20,7 @@
 from botocore.client import BaseClient
 from cryptography.fernet import Fernet
 from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
 from jwt.exceptions import InvalidTokenError
 from passlib.context import CryptContext
@@ -41,6 +42,8 @@
 
 # Marks KMS-encrypted credentials; rows without it are legacy Fernet.
 KMS_CIPHERTEXT_PREFIX = "kms.v1:"
+# Envelope-encrypted credentials: KMS-wrapped data key + AES-256-GCM payload.
+KMS_ENVELOPE_PREFIX = "kms.v2:"
 
 _kms_client: BaseClient | None = None
 
@@ -196,19 +199,32 @@ def get_password_hash(password: str) -> str:
 
 
 def encrypt_credentials(credentials: dict[str, Any]) -> str:
-    """Encrypt credentials for storage. KMS outside dev, Fernet otherwise.
+    """Encrypt credentials for storage. KMS envelope outside dev, Fernet otherwise.
 
-    KMS Encrypt caps plaintext at 4096 bytes, so payloads must stay under that.
+    Envelope format: `kms.v2:<wrapped_dek>:<nonce>:<ciphertext>`, each segment
+    base64. The DEK is generated by KMS, used locally for AES-256-GCM, and only
+    its KMS-wrapped form is persisted; the plaintext DEK never leaves this call.
+
+    NOTE: THIS FUNCTION SHOULD ONLY BE USED FOR ENCRYPTING TO KMS ENVELOPE.
+    AND SHOULD NOT BE USED FOR ANY OTHER ENCRYPTION OR ROTATION METHOD.
     """
     try:
         credentials_str = json.dumps(credentials)
         if _use_kms():
-            response = get_kms_client().encrypt(
+            resp = get_kms_client().generate_data_key(
                 KeyId=settings.AWS_KMS_KEY_ID,
-                Plaintext=credentials_str.encode(),
+                KeySpec="AES_256",
+            )
+            dek = resp["Plaintext"]
+            wrapped_dek = resp["CiphertextBlob"]
+            nonce = os.urandom(12)
+            ct = AESGCM(dek).encrypt(nonce, credentials_str.encode(), None)
+            return (
+                f"{KMS_ENVELOPE_PREFIX}"
+                f"{base64.b64encode(wrapped_dek).decode()}:"
+                f"{base64.b64encode(nonce).decode()}:"
+                f"{base64.b64encode(ct).decode()}"
             )
-            encoded = base64.b64encode(response["CiphertextBlob"]).decode()
-            return f"{KMS_CIPHERTEXT_PREFIX}{encoded}"
         return get_fernet().encrypt(credentials_str.encode()).decode()
     except Exception as e:
         # Log the real cause (may carry AWS ARNs); never surface it to callers.
@@ -216,12 +232,35 @@ def encrypt_credentials(credentials: dict[str, Any]) -> str:
         raise ValueError("Failed to encrypt credentials")
 
 
+def encrypt_fernet(credentials: dict[str, Any]) -> str:
+    """Encrypt credentials with Fernet regardless of the active mode.
+
+    Forces the local Fernet path even when _use_kms() is true; used to downgrade
+    KMS rows back to Fernet. Output decrypts only under the current SECRET_KEY.
+    """
+    try:
+        return get_fernet().encrypt(json.dumps(credentials).encode()).decode()
+    except Exception as e:
+        logger.error(f"[encrypt_fernet] Encryption failed | error: {e}")
+        raise ValueError("Failed to encrypt credentials")
+
+
 def decrypt_credentials(encrypted_credentials: str) -> dict[str, Any]:
     """Decrypt stored credentials. Routing is by ciphertext prefix, not the
     active mode, so legacy Fernet rows always decrypt even after KMS cutover.
     """
     try:
-        if encrypted_credentials.startswith(KMS_CIPHERTEXT_PREFIX):
+        if encrypted_credentials.startswith(KMS_ENVELOPE_PREFIX):
+            payload = encrypted_credentials[len(KMS_ENVELOPE_PREFIX) :]
+            # Exactly 3 segments; a wrong count raises ValueError caught below.
+            wrapped_b64, nonce_b64, ct_b64 = payload.split(":")
+            wrapped_dek = base64.b64decode(wrapped_b64)
+            nonce = base64.b64decode(nonce_b64)
+            ct = base64.b64decode(ct_b64)
+            # KMS infers the key from the wrapped blob; do not pass KeyId.
+            dek = get_kms_client().decrypt(CiphertextBlob=wrapped_dek)["Plaintext"]
+            decrypted_str = AESGCM(dek).decrypt(nonce, ct, None).decode()
+        elif encrypted_credentials.startswith(KMS_CIPHERTEXT_PREFIX):
             blob = base64.b64decode(encrypted_credentials[len(KMS_CIPHERTEXT_PREFIX) :])
             response = get_kms_client().decrypt(CiphertextBlob=blob)
             decrypted_str = response["Plaintext"].decode()
diff --git a/backend/app/services/credentials/reencrypt.py b/backend/app/services/credentials/reencrypt.py
index 5b8885139..b12abb375 100644
--- a/backend/app/services/credentials/reencrypt.py
+++ b/backend/app/services/credentials/reencrypt.py
@@ -10,7 +10,14 @@
 
 from app.core.config import settings
 from app.core.db import engine
-from app.core.security import _use_kms, decrypt_credentials, encrypt_credentials
+from app.core.security import (
+    KMS_CIPHERTEXT_PREFIX,
+    KMS_ENVELOPE_PREFIX,
+    _use_kms,
+    decrypt_credentials,
+    encrypt_credentials,
+    encrypt_fernet,
+)
 from app.core.util import now
 from app.crud.credentials import list_all_credentials
 
@@ -24,7 +31,21 @@ def execute_credential_reencrypt(*, session: Session | None = None) -> dict[str,
         return _reencrypt(owned_session)
 
 
+def execute_credential_reencrypt_fernet(
+    *, session: Session | None = None
+) -> dict[str, int]:
+    if session is not None:
+        return _reencrypt_fernet(session)
+    with Session(engine) as owned_session:
+        return _reencrypt_fernet(owned_session)
+
+
 def _reencrypt(session: Session) -> dict[str, int]:
+    """
+    DO NOT USE MIGRATION BEYOND 076
+    THIS IS ONLY MEANT TO IMPLEMENT KMS ENVELOPE MIGRATION,
+    AND SHOULD NOT BE USED FOR ROTATION OF KMS CMK.
+    """
     if not _use_kms():
         logger.info(
             f"[execute_credential_reencrypt] Skipped, KMS inactive | "
@@ -39,6 +60,9 @@ def _reencrypt(session: Session) -> dict[str, int]:
     converted = 0
     try:
         for row in rows:
+            if row.credential.startswith(KMS_ENVELOPE_PREFIX):
+                # Already in the new envelope format; skip.
+                continue
             plaintext = decrypt_credentials(row.credential)
             new_ciphertext = encrypt_credentials(plaintext)
             if decrypt_credentials(new_ciphertext) != plaintext:
@@ -63,3 +87,53 @@ def _reencrypt(session: Session) -> dict[str, int]:
         f"[execute_credential_reencrypt] Done | total: {total}, converted: {converted}"
     )
     return {"total": total, "converted": converted}
+
+
+def _reencrypt_fernet(session: Session) -> dict[str, int]:
+    """Downgrade KMS credentials (kms.v1/kms.v2) back to legacy Fernet — nothing else.
+
+    Only KMS rows are converted; Fernet rows are left untouched. Needs KMS access to
+    decrypt the existing rows; output decrypts only under the current SECRET_KEY.
+    One-shot, atomic — any failure rolls back and re-raises.
+    """
+    if not _use_kms():
+        logger.info(
+            f"[_reencrypt_fernet] Skipped, KMS inactive | "
+            f"environment: {settings.ENVIRONMENT}"
+        )
+        return {"total": 0, "converted": 0}
+
+    rows = list_all_credentials(session=session)
+    total = len(rows)
+    logger.info(f"[_reencrypt_fernet] Starting | total: {total}")
+
+    converted = 0
+    try:
+        for row in rows:
+            # Convert only KMS rows; anything without a KMS prefix is already Fernet.
+            if not row.credential.startswith(
+                (KMS_ENVELOPE_PREFIX, KMS_CIPHERTEXT_PREFIX)
+            ):
+                continue
+            plaintext = decrypt_credentials(row.credential)
+            new_ciphertext = encrypt_fernet(plaintext)
+            if decrypt_credentials(new_ciphertext) != plaintext:
+                raise ValueError(f"roundtrip mismatch for credential id {row.id}")
+
+            row.credential = new_ciphertext
+            row.updated_at = now()
+            session.add(row)
+            converted += 1
+
+        session.commit()
+    except Exception as e:
+        session.rollback()
+        logger.error(
+            f"[_reencrypt_fernet] Failed, rolled back all rows | "
+            f"converted-before-abort: {converted}/{total}, error: {e}",
+            exc_info=True,
+        )
+        raise
+
+    logger.info(f"[_reencrypt_fernet] Done | total: {total}, converted: {converted}")
+    return {"total": total, "converted": converted}
diff --git a/backend/app/tests/core/test_security.py b/backend/app/tests/core/test_security.py
index 412c48c25..681e281d9 100644
--- a/backend/app/tests/core/test_security.py
+++ b/backend/app/tests/core/test_security.py
@@ -1,3 +1,5 @@
+import base64
+import json
 from datetime import timedelta
 
 import boto3
@@ -11,6 +13,7 @@
 from app.core.security import (
     ALGORITHM,
     KMS_CIPHERTEXT_PREFIX,
+    KMS_ENVELOPE_PREFIX,
     APIKeyManager,
     create_access_token,
     create_refresh_token,
@@ -65,7 +68,7 @@ def test_kms_roundtrip(self, kms_key):
 
         encrypted = encrypt_credentials(creds)
 
-        assert encrypted.startswith(KMS_CIPHERTEXT_PREFIX)
+        assert encrypted.startswith(KMS_ENVELOPE_PREFIX)
         assert decrypt_credentials(encrypted) == creds
 
     def test_dual_read_fernet_row_with_kms_active(self, monkeypatch, kms_key):
@@ -78,6 +81,64 @@ def test_dual_read_fernet_row_with_kms_active(self, monkeypatch, kms_key):
         monkeypatch.setattr(settings, "ENVIRONMENT", "staging")
         assert decrypt_credentials(fernet_encrypted) == creds
 
+    def test_kms_v2_envelope_roundtrip(self, kms_key):
+        creds = {"openai": {"api_key": "sk-envelope-123"}}
+
+        encrypted = encrypt_credentials(creds)
+
+        assert encrypted.startswith(KMS_ENVELOPE_PREFIX)
+        segments = encrypted[len(KMS_ENVELOPE_PREFIX) :].split(":")
+        assert len(segments) == 3
+        for seg in segments:
+            base64.b64decode(seg)  # each segment must be valid base64
+        assert decrypt_credentials(encrypted) == creds
+
+    def test_kms_v2_large_payload_over_4096_bytes(self, kms_key):
+        # Direct KMS encrypt caps at 4096 bytes; envelope encryption has no such limit.
+        creds = {
+            "service_account": {"private_key": "k" * 5000, "client_email": "svc@x"}
+        }
+        assert len(json.dumps(creds).encode()) > 4096
+
+        encrypted = encrypt_credentials(creds)
+
+        assert encrypted.startswith(KMS_ENVELOPE_PREFIX)
+        assert decrypt_credentials(encrypted) == creds
+
+    def test_v1_row_still_decrypts_with_v2_active(self, kms_key):
+        creds = {"api_key": "sk-v1-legacy"}
+        blob = security._kms_client.encrypt(
+            KeyId=kms_key, Plaintext=json.dumps(creds).encode()
+        )["CiphertextBlob"]
+        v1_ciphertext = KMS_CIPHERTEXT_PREFIX + base64.b64encode(blob).decode()
+
+        assert decrypt_credentials(v1_ciphertext) == creds
+
+    def test_new_writes_produce_v2(self, kms_key):
+        encrypted = encrypt_credentials({"api_key": "sk-new"})
+
+        assert encrypted.startswith(KMS_ENVELOPE_PREFIX)
+        assert not encrypted.startswith(KMS_CIPHERTEXT_PREFIX)
+
+    def test_v2_tampered_ciphertext_raises(self, kms_key):
+        encrypted = encrypt_credentials({"api_key": "sk-tamper"})
+        wrapped_b64, nonce_b64, ct_b64 = encrypted[len(KMS_ENVELOPE_PREFIX) :].split(
+            ":"
+        )
+        ct = bytearray(base64.b64decode(ct_b64))
+        ct[0] ^= 0xFF
+        tampered = (
+            f"{KMS_ENVELOPE_PREFIX}{wrapped_b64}:{nonce_b64}:"
+            f"{base64.b64encode(bytes(ct)).decode()}"
+        )
+
+        with pytest.raises(ValueError, match="Failed to decrypt credentials"):
+            decrypt_credentials(tampered)
+
+    def test_v2_wrong_segment_count_raises(self, kms_key):
+        with pytest.raises(ValueError, match="Failed to decrypt credentials"):
+            decrypt_credentials(f"{KMS_ENVELOPE_PREFIX}onlyoneseg")
+
 
 class TestAPIKeyManager:
     """Test suite for APIKeyManager class."""
diff --git a/backend/app/tests/services/credentials/test_reencrypt.py b/backend/app/tests/services/credentials/test_reencrypt.py
index c77e2f9f2..3eb0df986 100644
--- a/backend/app/tests/services/credentials/test_reencrypt.py
+++ b/backend/app/tests/services/credentials/test_reencrypt.py
@@ -1,10 +1,17 @@
+import base64
+import json
+
 import boto3
 from moto import mock_aws
 from sqlmodel import Session
 
 import app.core.security as security
 from app.core.config import settings
-from app.core.security import KMS_CIPHERTEXT_PREFIX, decrypt_credentials
+from app.core.security import (
+    KMS_CIPHERTEXT_PREFIX,
+    KMS_ENVELOPE_PREFIX,
+    decrypt_credentials,
+)
 from app.models import Credential
 from app.services.credentials.reencrypt import execute_credential_reencrypt
 from app.tests.utils.test_data import create_test_credential
@@ -29,5 +36,35 @@ def test_reencrypt_converts_fernet_rows_to_kms(db: Session, monkeypatch):
         assert result["converted"] >= len(expected)
         for cid, plain in expected.items():
             row = db.get(Credential, cid)
-            assert row.credential.startswith(KMS_CIPHERTEXT_PREFIX)
+            assert row.credential.startswith(KMS_ENVELOPE_PREFIX)
+            assert decrypt_credentials(row.credential) == plain
+
+
+def test_reencrypt_converts_v1_row_to_v2(db: Session, monkeypatch):
+    creds, _ = create_test_credential(db)
+    expected = {c.id: decrypt_credentials(c.credential) for c in creds}
+
+    with mock_aws():
+        client = boto3.client("kms", region_name="ap-south-1")
+        key_id = client.create_key()["KeyMetadata"]["KeyId"]
+        monkeypatch.setattr(settings, "ENVIRONMENT", "staging")
+        monkeypatch.setattr(settings, "AWS_KMS_KEY_ID", key_id)
+        monkeypatch.setattr(security, "_kms_client", client)
+
+        # Rewrite each row as a legacy direct-KMS (v1) ciphertext.
+        for c in creds:
+            plain = expected[c.id]
+            blob = client.encrypt(KeyId=key_id, Plaintext=json.dumps(plain).encode())[
+                "CiphertextBlob"
+            ]
+            c.credential = KMS_CIPHERTEXT_PREFIX + base64.b64encode(blob).decode()
+            db.add(c)
+        db.commit()
+
+        result = execute_credential_reencrypt(session=db)
+
+        assert result["converted"] >= len(expected)
+        for cid, plain in expected.items():
+            row = db.get(Credential, cid)
+            assert row.credential.startswith(KMS_ENVELOPE_PREFIX)
             assert decrypt_credentials(row.credential) == plain
diff --git a/docs/wiki/modules/platform.md b/docs/wiki/modules/platform.md
index 0f34e5d87..07b5fa1aa 100644
--- a/docs/wiki/modules/platform.md
+++ b/docs/wiki/modules/platform.md
@@ -10,7 +10,7 @@ All paths relative to `backend/app/`.
 | Notifications | — | `notification` (`models/notification.py`) | `services/notifications/`, `crud/notification.py` |
 | Feature flags | `api/routes/features.py` | `feature_flag` (`models/feature_flag.py`) | `core/feature_flags/`, `crud/feature_flag.py` |
 | Languages | `api/routes/languages.py` | `global.languages` (`models/language.py`) | `crud/language.py` |
-| Credentials | `api/routes/credentials.py` | `credential` (`models/credentials.py`) | `crud/credentials.py`; provider keys per org/project |
+| Credentials | `api/routes/credentials.py` | `credential` (`models/credentials.py`) | `crud/credentials.py`; provider keys per org/project; envelope encryption (KMS-wrapped data key + AES-GCM), prefix-versioned ciphertexts |
 | Model config | `api/routes/model_config.py` | `model_config` (`models/model_config.py`) | `crud/model_config.py` |
 | Cron | `api/routes/cron.py` | — | triggers batch polling (`crud/evaluations/cron.py`) |
 | Jobs | — | `job` (`models/job.py`), `batch_job` (`models/batch_job.py`) | `crud/jobs.py`, `crud/job/`, `services/job_monitoring.py` |