diff --git a/app/database/crud/discount_offer.py b/app/database/crud/discount_offer.py
index c0bc9800..d1add91e 100644
--- a/app/database/crud/discount_offer.py
+++ b/app/database/crud/discount_offer.py
@@ -1,11 +1,17 @@
+from __future__ import annotations
+
+import logging
 from datetime import datetime, timedelta
 from typing import Optional
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.database.crud.promo_offer_log import log_promo_offer_action
 from app.database.models import DiscountOffer
 
+logger = logging.getLogger(__name__)
+
 
 async def upsert_discount_offer(
     db: AsyncSession,
@@ -67,11 +73,35 @@ async def get_offer_by_id(db: AsyncSession, offer_id: int) -> Optional[DiscountO
     return result.scalar_one_or_none()
 
 
-async def mark_offer_claimed(db: AsyncSession, offer: DiscountOffer) -> DiscountOffer:
+async def mark_offer_claimed(
+    db: AsyncSession,
+    offer: DiscountOffer,
+    *,
+    details: Optional[dict] = None,
+) -> DiscountOffer:
     offer.claimed_at = datetime.utcnow()
     offer.is_active = False
     await db.commit()
     await db.refresh(offer)
+
+    try:
+        await log_promo_offer_action(
+            db,
+            user_id=offer.user_id,
+            offer_id=offer.id,
+            action="claimed",
+            source=offer.notification_type,
+            percent=offer.discount_percent,
+            effect_type=offer.effect_type,
+            details=details,
+        )
+    except Exception as exc:  # pragma: no cover - defensive logging
+        logger.warning(
+            "Failed to record promo offer claim log for offer %s: %s",
+            offer.id,
+            exc,
+        )
+
     return offer
 
 
@@ -88,9 +118,62 @@ async def deactivate_expired_offers(db: AsyncSession) -> int:
         return 0
 
     count = 0
+    log_payloads = []
     for offer in offers:
         offer.is_active = False
         count += 1
+        log_payloads.append(
+            {
+                "user_id": offer.user_id,
+                "offer_id": offer.id,
+                "source": offer.notification_type,
+                "percent": offer.discount_percent,
+                "effect_type": offer.effect_type,
+            }
+        )
 
     await db.commit()
+
+    for payload in log_payloads:
+        if not payload.get("user_id"):
+            continue
+        try:
+            await log_promo_offer_action(
+                db,
+                user_id=payload["user_id"],
+                offer_id=payload["offer_id"],
+                action="disabled",
+                source=payload.get("source"),
+                percent=payload.get("percent"),
+                effect_type=payload.get("effect_type"),
+                details={"reason": "offer_expired"},
+            )
+        except Exception as exc:  # pragma: no cover - defensive logging
+            logger.warning(
+                "Failed to record promo offer disable log for offer %s: %s",
+                payload.get("offer_id"),
+                exc,
+            )
+
     return count
+
+
+async def get_latest_claimed_offer_for_user(
+    db: AsyncSession,
+    user_id: int,
+    source: Optional[str] = None,
+) -> Optional[DiscountOffer]:
+    stmt = (
+        select(DiscountOffer)
+        .where(
+            DiscountOffer.user_id == user_id,
+            DiscountOffer.claimed_at.isnot(None),
+        )
+        .order_by(DiscountOffer.claimed_at.desc())
+    )
+
+    if source:
+        stmt = stmt.where(DiscountOffer.notification_type == source)
+
+    result = await db.execute(stmt)
+    return result.scalars().first()
diff --git a/app/database/crud/promo_offer_log.py b/app/database/crud/promo_offer_log.py
new file mode 100644
index 00000000..42d17b01
--- /dev/null
+++ b/app/database/crud/promo_offer_log.py
@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+import logging
+from typing import Dict, List, Optional, Tuple
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from app.database.models import PromoOfferLog
+
+logger = logging.getLogger(__name__)
+
+
+async def log_promo_offer_action(
+    db: AsyncSession,
+    *,
+    user_id: Optional[int],
+    offer_id: Optional[int],
+    action: str,
+    source: Optional[str] = None,
+    percent: Optional[int] = None,
+    effect_type: Optional[str] = None,
+    details: Optional[Dict[str, object]] = None,
+    commit: bool = True,
+) -> PromoOfferLog:
+    """Persist a promo offer log entry."""
+
+    entry = PromoOfferLog(
+        user_id=user_id,
+        offer_id=offer_id,
+        action=action,
+        source=source,
+        percent=percent,
+        effect_type=effect_type,
+        details=(details or {}).copy(),
+    )
+    db.add(entry)
+
+    if commit:
+        try:
+            await db.commit()
+            await db.refresh(entry)
+        except Exception:
+            logger.exception("Failed to commit promo offer log entry")
+            raise
+
+    return entry
+
+
+async def list_promo_offer_logs(
+    db: AsyncSession,
+    offset: int = 0,
+    limit: int = 20,
+) -> Tuple[List[PromoOfferLog], int]:
+    stmt = (
+        select(PromoOfferLog)
+        .options(
+            selectinload(PromoOfferLog.user),
+            selectinload(PromoOfferLog.offer),
+        )
+        .order_by(PromoOfferLog.created_at.desc())
+        .offset(offset)
+        .limit(limit)
+    )
+    result = await db.execute(stmt)
+    logs = result.scalars().all()
+
+    count_stmt = select(func.count(PromoOfferLog.id))
+    total = (await db.execute(count_stmt)).scalar() or 0
+
+    return logs, total
diff --git a/app/database/crud/user.py b/app/database/crud/user.py
index 9dacb13b..3bd0b7d9 100644
--- a/app/database/crud/user.py
+++ b/app/database/crud/user.py
@@ -20,6 +20,8 @@ from app.database.models import (
 )
 from app.config import settings
 from app.database.crud.promo_group import get_default_promo_group
+from app.database.crud.discount_offer import get_latest_claimed_offer_for_user
+from app.database.crud.promo_offer_log import log_promo_offer_action
 from app.utils.validators import sanitize_telegram_name
 
 logger = logging.getLogger(__name__)
@@ -282,10 +284,46 @@ async def subtract_user_balance(
     logger.error(f"   💸 Сумма к списанию: {amount_kopeks} копеек")
     logger.error(f"   📝 Описание: {description}")
     
+    log_context: Optional[Dict[str, object]] = None
+    if consume_promo_offer:
+        try:
+            current_percent = int(getattr(user, "promo_offer_discount_percent", 0) or 0)
+        except (TypeError, ValueError):
+            current_percent = 0
+
+        if current_percent > 0:
+            source = getattr(user, "promo_offer_discount_source", None)
+            log_context = {
+                "offer_id": None,
+                "percent": current_percent,
+                "source": source,
+                "effect_type": None,
+                "details": {
+                    "reason": "manual_charge",
+                    "description": description,
+                    "amount_kopeks": amount_kopeks,
+                },
+            }
+            try:
+                offer = await get_latest_claimed_offer_for_user(db, user.id, source)
+            except Exception as lookup_error:  # pragma: no cover - defensive logging
+                logger.warning(
+                    "Failed to fetch latest claimed promo offer for user %s: %s",
+                    user.id,
+                    lookup_error,
+                )
+                offer = None
+
+            if offer:
+                log_context["offer_id"] = offer.id
+                log_context["effect_type"] = offer.effect_type
+                if not log_context["percent"] and offer.discount_percent:
+                    log_context["percent"] = offer.discount_percent
+
     if user.balance_kopeks < amount_kopeks:
         logger.error(f"   ❌ НЕДОСТАТОЧНО СРЕДСТВ!")
         return False
-    
+
     try:
         old_balance = user.balance_kopeks
         user.balance_kopeks -= amount_kopeks
@@ -295,7 +333,7 @@ async def subtract_user_balance(
             user.promo_offer_discount_source = None
 
         user.updated_at = datetime.utcnow()
-        
+
         await db.commit()
         await db.refresh(user)
 
@@ -313,6 +351,25 @@ async def subtract_user_balance(
                 payment_method=payment_method,
             )
 
+        if consume_promo_offer and log_context:
+            try:
+                await log_promo_offer_action(
+                    db,
+                    user_id=user.id,
+                    offer_id=log_context.get("offer_id"),
+                    action="consumed",
+                    source=log_context.get("source"),
+                    percent=log_context.get("percent"),
+                    effect_type=log_context.get("effect_type"),
+                    details=log_context.get("details"),
+                )
+            except Exception as log_error:  # pragma: no cover - defensive logging
+                logger.warning(
+                    "Failed to record promo offer consumption log for user %s: %s",
+                    user.id,
+                    log_error,
+                )
+
         logger.error(f"   ✅ Средства списаны: {old_balance} → {user.balance_kopeks}")
         return True
         
diff --git a/app/database/models.py b/app/database/models.py
index e6dd34c2..affb5dd2 100644
--- a/app/database/models.py
+++ b/app/database/models.py
@@ -386,6 +386,7 @@ class User(Base):
     transactions = relationship("Transaction", back_populates="user")
     referral_earnings = relationship("ReferralEarning", foreign_keys="ReferralEarning.user_id", back_populates="user")
     discount_offers = relationship("DiscountOffer", back_populates="user")
+    promo_offer_logs = relationship("PromoOfferLog", back_populates="user")
     lifetime_used_traffic_bytes = Column(BigInteger, default=0)
     auto_promo_group_assigned = Column(Boolean, nullable=False, default=False)
     auto_promo_group_threshold_kopeks = Column(BigInteger, nullable=False, default=0)
@@ -821,6 +822,7 @@ class DiscountOffer(Base):
 
     user = relationship("User", back_populates="discount_offers")
     subscription = relationship("Subscription", back_populates="discount_offers")
+    logs = relationship("PromoOfferLog", back_populates="offer")
 
 
 class PromoOfferTemplate(Base):
@@ -863,6 +865,23 @@ class SubscriptionTemporaryAccess(Base):
     subscription = relationship("Subscription", back_populates="temporary_accesses")
     offer = relationship("DiscountOffer")
 
+
+class PromoOfferLog(Base):
+    __tablename__ = "promo_offer_logs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True)
+    offer_id = Column(Integer, ForeignKey("discount_offers.id", ondelete="SET NULL"), nullable=True, index=True)
+    action = Column(String(50), nullable=False)
+    source = Column(String(100), nullable=True)
+    percent = Column(Integer, nullable=True)
+    effect_type = Column(String(50), nullable=True)
+    details = Column(JSON, nullable=True)
+    created_at = Column(DateTime, default=func.now())
+
+    user = relationship("User", back_populates="promo_offer_logs")
+    offer = relationship("DiscountOffer", back_populates="logs")
+
 class BroadcastHistory(Base):
     __tablename__ = "broadcast_history"
     
diff --git a/app/database/universal_migration.py b/app/database/universal_migration.py
index fd1db22e..ba2506a3 100644
--- a/app/database/universal_migration.py
+++ b/app/database/universal_migration.py
@@ -1009,6 +1009,79 @@ async def create_promo_offer_templates_table():
         return False
 
 
+async def create_promo_offer_logs_table() -> bool:
+    table_exists = await check_table_exists('promo_offer_logs')
+    if table_exists:
+        logger.info("Таблица promo_offer_logs уже существует")
+        return True
+
+    try:
+        db_type = await get_database_type()
+        async with engine.begin() as conn:
+            if db_type == 'sqlite':
+                await conn.execute(text("""
+                    CREATE TABLE IF NOT EXISTS promo_offer_logs (
+                        id INTEGER PRIMARY KEY AUTOINCREMENT,
+                        user_id INTEGER NULL REFERENCES users(id) ON DELETE SET NULL,
+                        offer_id INTEGER NULL REFERENCES discount_offers(id) ON DELETE SET NULL,
+                        action VARCHAR(50) NOT NULL,
+                        source VARCHAR(100) NULL,
+                        percent INTEGER NULL,
+                        effect_type VARCHAR(50) NULL,
+                        details JSON NULL,
+                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
+                    );
+
+                    CREATE INDEX IF NOT EXISTS ix_promo_offer_logs_created_at ON promo_offer_logs(created_at DESC);
+                    CREATE INDEX IF NOT EXISTS ix_promo_offer_logs_user_id ON promo_offer_logs(user_id);
+                """))
+            elif db_type == 'postgresql':
+                await conn.execute(text("""
+                    CREATE TABLE IF NOT EXISTS promo_offer_logs (
+                        id SERIAL PRIMARY KEY,
+                        user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
+                        offer_id INTEGER REFERENCES discount_offers(id) ON DELETE SET NULL,
+                        action VARCHAR(50) NOT NULL,
+                        source VARCHAR(100),
+                        percent INTEGER,
+                        effect_type VARCHAR(50),
+                        details JSONB,
+                        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+                    );
+
+                    CREATE INDEX IF NOT EXISTS ix_promo_offer_logs_created_at ON promo_offer_logs(created_at DESC);
+                    CREATE INDEX IF NOT EXISTS ix_promo_offer_logs_user_id ON promo_offer_logs(user_id);
+                """))
+            elif db_type == 'mysql':
+                await conn.execute(text("""
+                    CREATE TABLE IF NOT EXISTS promo_offer_logs (
+                        id INT AUTO_INCREMENT PRIMARY KEY,
+                        user_id INT NULL,
+                        offer_id INT NULL,
+                        action VARCHAR(50) NOT NULL,
+                        source VARCHAR(100) NULL,
+                        percent INT NULL,
+                        effect_type VARCHAR(50) NULL,
+                        details JSON NULL,
+                        created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                        CONSTRAINT fk_promo_offer_logs_users FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE SET NULL,
+                        CONSTRAINT fk_promo_offer_logs_offers FOREIGN KEY (offer_id) REFERENCES discount_offers(id) ON DELETE SET NULL
+                    );
+
+                    CREATE INDEX ix_promo_offer_logs_created_at ON promo_offer_logs(created_at DESC);
+                    CREATE INDEX ix_promo_offer_logs_user_id ON promo_offer_logs(user_id);
+                """))
+            else:
+                logger.warning("Неизвестный тип БД для создания promo_offer_logs: %s", db_type)
+                return False
+
+        logger.info("✅ Таблица promo_offer_logs успешно создана")
+        return True
+    except Exception as e:
+        logger.error(f"Ошибка создания таблицы promo_offer_logs: {e}")
+        return False
+
+
 async def create_subscription_temporary_access_table():
     table_exists = await check_table_exists('subscription_temporary_access')
     if table_exists:
@@ -2528,6 +2601,13 @@ async def run_universal_migration():
         else:
             logger.warning("⚠️ Проблемы с таблицей promo_offer_templates")
 
+        logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ PROMO_OFFER_LOGS ===")
+        promo_logs_created = await create_promo_offer_logs_table()
+        if promo_logs_created:
+            logger.info("✅ Таблица promo_offer_logs готова")
+        else:
+            logger.warning("⚠️ Проблемы с таблицей promo_offer_logs")
+
         logger.info("=== СОЗДАНИЕ ТАБЛИЦЫ SUBSCRIPTION_TEMPORARY_ACCESS ===")
         temp_access_created = await create_subscription_temporary_access_table()
         if temp_access_created:
@@ -2742,6 +2822,7 @@ async def check_migration_status():
             "discount_offers_effect_column": False,
             "discount_offers_extra_column": False,
             "promo_offer_templates_table": False,
+            "promo_offer_logs_table": False,
             "subscription_temporary_access_table": False,
         }
         
@@ -2758,6 +2839,7 @@ async def check_migration_status():
         status["discount_offers_effect_column"] = await check_column_exists('discount_offers', 'effect_type')
         status["discount_offers_extra_column"] = await check_column_exists('discount_offers', 'extra_data')
         status["promo_offer_templates_table"] = await check_table_exists('promo_offer_templates')
+        status["promo_offer_logs_table"] = await check_table_exists('promo_offer_logs')
         status["subscription_temporary_access_table"] = await check_table_exists('subscription_temporary_access')
 
         status["welcome_texts_is_enabled_column"] = await check_column_exists('welcome_texts', 'is_enabled')
@@ -2815,6 +2897,7 @@ async def check_migration_status():
             "discount_offers_effect_column": "Колонка effect_type в discount_offers",
             "discount_offers_extra_column": "Колонка extra_data в discount_offers",
             "promo_offer_templates_table": "Таблица promo_offer_templates",
+            "promo_offer_logs_table": "Таблица promo_offer_logs",
             "subscription_temporary_access_table": "Таблица subscription_temporary_access",
         }
         
diff --git a/app/handlers/admin/promo_offers.py b/app/handlers/admin/promo_offers.py
index 85b1be48..4e25152e 100644
--- a/app/handlers/admin/promo_offers.py
+++ b/app/handlers/admin/promo_offers.py
@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+import html
 import logging
 import re
-from typing import List, Optional, Sequence, Tuple
+from typing import Dict, List, Optional, Sequence, Tuple
 
 from aiogram import Dispatcher, F, types
 from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
@@ -23,8 +24,9 @@ from app.database.crud.promo_offer_template import (
     list_promo_offer_templates,
     update_promo_offer_template,
 )
+from app.database.crud.promo_offer_log import list_promo_offer_logs
 from app.database.crud.user import get_users_for_promo_segment
-from app.database.models import PromoOfferTemplate, User
+from app.database.models import PromoOfferLog, PromoOfferTemplate, User
 from app.keyboards.inline import get_happ_download_button_row
 from app.localization.texts import get_texts
 from app.states import AdminStates
@@ -35,6 +37,22 @@ logger = logging.getLogger(__name__)
 
 
 SQUADS_PAGE_LIMIT = 10
+PROMO_OFFER_LOGS_PAGE_LIMIT = 10
+
+
+ACTION_LABEL_KEYS = {
+    "claimed": "ADMIN_PROMO_OFFER_LOGS_ACTION_CLAIMED",
+    "consumed": "ADMIN_PROMO_OFFER_LOGS_ACTION_CONSUMED",
+    "disabled": "ADMIN_PROMO_OFFER_LOGS_ACTION_DISABLED",
+}
+
+
+REASON_LABEL_KEYS = {
+    "manual_charge": "ADMIN_PROMO_OFFER_LOGS_REASON_MANUAL",
+    "autopay_consumed": "ADMIN_PROMO_OFFER_LOGS_REASON_AUTOPAY",
+    "offer_expired": "ADMIN_PROMO_OFFER_LOGS_REASON_EXPIRED",
+    "test_access_expired": "ADMIN_PROMO_OFFER_LOGS_REASON_TEST_EXPIRED",
+}
 
 
 OFFER_TYPE_CONFIG = {
@@ -123,6 +141,12 @@ def _build_templates_keyboard(templates: Sequence[PromoOfferTemplate], language:
                 callback_data=f"promo_offer_{template.id}",
             )
         ])
+    rows.append([
+        InlineKeyboardButton(
+            text=texts.t("ADMIN_PROMO_OFFER_LOGS", "📜 Лог операций"),
+            callback_data="promo_offer_logs_page_1",
+        )
+    ])
     rows.append([InlineKeyboardButton(text=texts.BACK, callback_data="admin_submenu_communications")])
     return InlineKeyboardMarkup(inline_keyboard=rows)
 
@@ -157,6 +181,128 @@ def _build_offer_detail_keyboard(template: PromoOfferTemplate, language: str) ->
     return InlineKeyboardMarkup(inline_keyboard=rows)
 
 
+def _format_promo_offer_log_entry(
+    entry: PromoOfferLog,
+    index: int,
+    texts,
+) -> str:
+    timestamp = entry.created_at.strftime("%d.%m.%Y %H:%M") if entry.created_at else "-"
+    action_key = ACTION_LABEL_KEYS.get(entry.action, "")
+    action_label = texts.get(action_key, entry.action.title())
+    lines = [f"{index}. <b>{timestamp}</b> — {action_label}"]
+
+    user = entry.user
+    if user:
+        username = f"@{user.username}" if user.username else f"ID{user.telegram_id}"
+        label = f"{username} (#{user.id})"
+    elif entry.user_id:
+        label = f"ID{entry.user_id}"
+    else:
+        label = texts.get("ADMIN_PROMO_OFFER_LOGS_UNKNOWN_USER", "Неизвестный пользователь")
+
+    lines.append(texts.get("ADMIN_PROMO_OFFER_LOGS_USER", "👤 {user}").format(user=html.escape(label)))
+
+    if entry.percent:
+        lines.append(
+            texts.get("ADMIN_PROMO_OFFER_LOGS_PERCENT", "📉 Скидка: {percent}%").format(
+                percent=entry.percent
+            )
+        )
+
+    effect_type = (entry.effect_type or "").lower()
+    if effect_type:
+        if effect_type == "test_access":
+            effect_label = texts.get("ADMIN_PROMO_OFFER_LOGS_EFFECT_TEST", "🧪 Тестовый доступ")
+        else:
+            effect_label = texts.get("ADMIN_PROMO_OFFER_LOGS_EFFECT_DISCOUNT", "💸 Скидка")
+        lines.append(effect_label)
+
+    if entry.source:
+        lines.append(
+            texts.get("ADMIN_PROMO_OFFER_LOGS_SOURCE", "🏷 Источник: {source}").format(
+                source=html.escape(entry.source)
+            )
+        )
+
+    details: Dict[str, object] = entry.details if isinstance(entry.details, dict) else {}
+    reason_key = details.get("reason")
+    if reason_key:
+        reason_label = texts.get(REASON_LABEL_KEYS.get(reason_key, ""), "")
+        if not reason_label:
+            reason_label = texts.get(
+                "ADMIN_PROMO_OFFER_LOGS_REASON_GENERIC",
+                "ℹ️ Действие: {reason}",
+            ).format(reason=html.escape(str(reason_key)))
+        lines.append(reason_label)
+
+    description = details.get("description")
+    if description:
+        lines.append(
+            texts.get("ADMIN_PROMO_OFFER_LOGS_DESCRIPTION", "📝 {description}").format(
+                description=html.escape(str(description))
+            )
+        )
+
+    amount = details.get("amount_kopeks")
+    if isinstance(amount, int):
+        lines.append(
+            texts.get("ADMIN_PROMO_OFFER_LOGS_AMOUNT", "💰 Сумма: {amount}").format(
+                amount=texts.format_price(amount)
+            )
+        )
+
+    squad_uuid = details.get("squad_uuid")
+    if squad_uuid:
+        lines.append(
+            texts.get("ADMIN_PROMO_OFFER_LOGS_SQUAD", "🌍 Сквад: {squad}").format(
+                squad=html.escape(str(squad_uuid))
+            )
+        )
+
+    new_squads = details.get("new_squads")
+    if isinstance(new_squads, (list, tuple)):
+        filtered = [html.escape(str(item)) for item in new_squads if item]
+        if filtered:
+            lines.append(
+                texts.get("ADMIN_PROMO_OFFER_LOGS_NEW_SQUADS", "🌍 Новые сквады: {squads}").format(
+                    squads=", ".join(filtered)
+                )
+            )
+
+    return "\n".join(lines)
+
+
+def _build_logs_keyboard(page: int, total_pages: int, language: str) -> InlineKeyboardMarkup:
+    texts = get_texts(language)
+    rows: List[List[InlineKeyboardButton]] = []
+    if total_pages > 1:
+        nav_row: List[InlineKeyboardButton] = []
+        if page > 1:
+            nav_row.append(
+                InlineKeyboardButton(
+                    text="⬅️",
+                    callback_data=f"promo_offer_logs_page_{page - 1}",
+                )
+            )
+        nav_row.append(
+            InlineKeyboardButton(
+                text=f"{page}/{total_pages}",
+                callback_data=f"promo_offer_logs_page_{page}",
+            )
+        )
+        if page < total_pages:
+            nav_row.append(
+                InlineKeyboardButton(
+                    text="➡️",
+                    callback_data=f"promo_offer_logs_page_{page + 1}",
+                )
+            )
+        rows.append(nav_row)
+
+    rows.append([InlineKeyboardButton(text=texts.BACK, callback_data="admin_promo_offers")])
+    return InlineKeyboardMarkup(inline_keyboard=rows)
+
+
 def _build_send_keyboard(template: PromoOfferTemplate, language: str) -> InlineKeyboardMarkup:
     config = OFFER_TYPE_CONFIG.get(template.offer_type, {})
     segments = config.get("allowed_segments", [])
@@ -289,6 +435,70 @@ async def show_promo_offer_details(callback: CallbackQuery, db_user: User, db: A
     await callback.answer()
 
 
+@admin_required
+@error_handler
+async def show_promo_offer_logs(callback: CallbackQuery, db_user: User, db: AsyncSession):
+    try:
+        if "_page_" in callback.data:
+            page = int(callback.data.split("_page_")[-1])
+        else:
+            page = 1
+    except (ValueError, AttributeError):
+        page = 1
+
+    if page < 1:
+        page = 1
+
+    limit = PROMO_OFFER_LOGS_PAGE_LIMIT
+    offset = (page - 1) * limit
+    logs, total = await list_promo_offer_logs(db, offset=offset, limit=limit)
+    total_pages = max(1, (total + limit - 1) // limit)
+
+    if page > total_pages and total > 0:
+        page = total_pages
+        offset = (page - 1) * limit
+        logs, _ = await list_promo_offer_logs(db, offset=offset, limit=limit)
+
+    texts = get_texts(db_user.language)
+    header = texts.t(
+        "ADMIN_PROMO_OFFER_LOGS_TITLE",
+        "📜 <b>Лог операций промо-предложений</b>",
+    )
+
+    if logs:
+        message_lines = [
+            header,
+            texts.get(
+                "ADMIN_PROMO_OFFER_LOGS_PAGINATION",
+                "Страница {page}/{total}",
+            ).format(page=page, total=total_pages),
+            "",
+        ]
+        for index, entry in enumerate(logs, start=offset + 1):
+            message_lines.append(_format_promo_offer_log_entry(entry, index, texts))
+            message_lines.append("")
+        message_text = "\n".join(message_lines).rstrip()
+    else:
+        message_text = "\n".join(
+            [
+                header,
+                "",
+                texts.get(
+                    "ADMIN_PROMO_OFFER_LOGS_EMPTY_BODY",
+                    "Записей пока нет.",
+                ),
+            ]
+        )
+
+    keyboard = _build_logs_keyboard(page, total_pages, db_user.language)
+    await callback.message.edit_text(
+        message_text,
+        reply_markup=keyboard,
+        parse_mode="HTML",
+    )
+    await callback.answer()
+
+
 async def _prompt_edit(callback: CallbackQuery, state: FSMContext, template_id: int, prompt: str, new_state):
     await state.update_data(
         selected_promo_offer=template_id,
@@ -965,6 +1175,7 @@ def register_handlers(dp: Dispatcher):
     dp.callback_query.register(back_to_offer_from_squads, F.data.startswith("promo_offer_squad_back_"))
     dp.callback_query.register(show_send_segments, F.data.startswith("promo_offer_send_menu_"))
     dp.callback_query.register(send_offer_to_segment, F.data.startswith("promo_offer_send_"))
+    dp.callback_query.register(show_promo_offer_logs, F.data.regexp(r"^promo_offer_logs_page_\d+$"))
     dp.callback_query.register(show_promo_offer_details, F.data.startswith("promo_offer_"))
 
     dp.message.register(process_edit_message_text, AdminStates.editing_promo_offer_message)
diff --git a/app/handlers/subscription.py b/app/handlers/subscription.py
index d4bea16f..39a14be9 100644
--- a/app/handlers/subscription.py
+++ b/app/handlers/subscription.py
@@ -5202,7 +5202,15 @@ async def claim_discount_offer(
             await callback.answer(error_message, show_alert=True)
             return
 
-        await mark_offer_claimed(db, offer)
+        await mark_offer_claimed(
+            db,
+            offer,
+            details={
+                "context": "test_access_claim",
+                "new_squads": newly_added,
+                "expires_at": expires_at.isoformat() if expires_at else None,
+            },
+        )
 
         expires_text = expires_at.strftime("%d.%m.%Y %H:%M") if expires_at else ""
         success_message = texts.get(
@@ -5237,7 +5245,14 @@ async def claim_discount_offer(
     db_user.promo_offer_discount_source = offer.notification_type
     db_user.updated_at = now
 
-    await mark_offer_claimed(db, offer)
+    await mark_offer_claimed(
+        db,
+        offer,
+        details={
+            "context": "discount_claim",
+            "discount_percent": discount_percent,
+        },
+    )
     await db.refresh(db_user)
 
     success_message = texts.get(
diff --git a/app/services/monitoring_service.py b/app/services/monitoring_service.py
index e0cc2968..d7eb678b 100644
--- a/app/services/monitoring_service.py
+++ b/app/services/monitoring_service.py
@@ -15,8 +15,10 @@ from app.config import settings
 from app.database.database import get_db
 from app.database.crud.discount_offer import (
     deactivate_expired_offers,
+    get_latest_claimed_offer_for_user,
     upsert_discount_offer,
 )
+from app.database.crud.promo_offer_log import log_promo_offer_action
 from app.database.crud.notification import (
     clear_notification_by_type,
     notification_sent,
@@ -795,9 +797,34 @@ class MonitoringService:
 
     @staticmethod
     async def _consume_user_promo_offer_discount(db: AsyncSession, user: User) -> None:
-        if MonitoringService._get_user_promo_offer_discount_percent(user) <= 0:
+        percent = MonitoringService._get_user_promo_offer_discount_percent(user)
+        if percent <= 0:
             return
 
+        source = getattr(user, "promo_offer_discount_source", None)
+        log_payload = {
+            "offer_id": None,
+            "percent": percent,
+            "source": source,
+            "effect_type": None,
+        }
+
+        try:
+            offer = await get_latest_claimed_offer_for_user(db, user.id, source)
+        except Exception as lookup_error:  # pragma: no cover - defensive logging
+            logger.warning(
+                "Failed to resolve latest claimed promo offer for user %s: %s",
+                user.id,
+                lookup_error,
+            )
+            offer = None
+
+        if offer:
+            log_payload["offer_id"] = offer.id
+            log_payload["effect_type"] = offer.effect_type
+            if not log_payload["percent"] and offer.discount_percent:
+                log_payload["percent"] = offer.discount_percent
+
         user.promo_offer_discount_percent = 0
         user.promo_offer_discount_source = None
         user.updated_at = datetime.utcnow()
@@ -805,6 +832,24 @@ class MonitoringService:
         await db.commit()
         await db.refresh(user)
 
+        try:
+            await log_promo_offer_action(
+                db,
+                user_id=user.id,
+                offer_id=log_payload.get("offer_id"),
+                action="consumed",
+                source=log_payload.get("source"),
+                percent=log_payload.get("percent"),
+                effect_type=log_payload.get("effect_type"),
+                details={"reason": "autopay_consumed"},
+            )
+        except Exception as log_error:  # pragma: no cover - defensive logging
+            logger.warning(
+                "Failed to record promo offer autopay log for user %s: %s",
+                user.id,
+                log_error,
+            )
+
     async def _process_autopayments(self, db: AsyncSession):
         try:
             current_time = datetime.utcnow()
diff --git a/app/services/promo_offer_service.py b/app/services/promo_offer_service.py
index a5a24416..de893572 100644
--- a/app/services/promo_offer_service.py
+++ b/app/services/promo_offer_service.py
@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import logging
 from datetime import datetime, timedelta
-from typing import List, Optional, Sequence, Tuple
+from typing import Dict, List, Optional, Sequence, Tuple
 
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -15,6 +15,7 @@ from app.database.models import (
     User,
 )
 from app.services.subscription_service import SubscriptionService
+from app.database.crud.promo_offer_log import log_promo_offer_action
 
 logger = logging.getLogger(__name__)
 
@@ -134,7 +135,10 @@ class PromoOfferService:
         now = datetime.utcnow()
         result = await db.execute(
             select(SubscriptionTemporaryAccess)
-            .options(selectinload(SubscriptionTemporaryAccess.subscription))
+            .options(
+                selectinload(SubscriptionTemporaryAccess.subscription),
+                selectinload(SubscriptionTemporaryAccess.offer),
+            )
             .where(
                 SubscriptionTemporaryAccess.is_active == True,  # noqa: E712
                 SubscriptionTemporaryAccess.expires_at <= now,
@@ -145,6 +149,7 @@ class PromoOfferService:
             return 0
 
         subscriptions_updates: dict[int, Tuple[Subscription, set[str]]] = {}
+        log_payloads: List[Dict[str, object]] = []
 
         for entry in entries:
             entry.is_active = False
@@ -157,6 +162,23 @@ class PromoOfferService:
             if not entry.was_already_connected:
                 bucket[1].add(entry.squad_uuid)
 
+            user_id = subscription.user_id
+            if user_id:
+                offer = entry.offer
+                log_payloads.append(
+                    {
+                        "user_id": user_id,
+                        "offer_id": entry.offer_id,
+                        "source": getattr(offer, "notification_type", None),
+                        "percent": getattr(offer, "discount_percent", None),
+                        "effect_type": getattr(offer, "effect_type", "test_access"),
+                        "details": {
+                            "reason": "test_access_expired",
+                            "squad_uuid": entry.squad_uuid,
+                        },
+                    }
+                )
+
         for subscription, squads_to_remove in subscriptions_updates.values():
             if not squads_to_remove:
                 continue
@@ -175,6 +197,24 @@ class PromoOfferService:
                     )
 
         await db.commit()
+        for payload in log_payloads:
+            try:
+                await log_promo_offer_action(
+                    db,
+                    user_id=payload["user_id"],
+                    offer_id=payload.get("offer_id"),
+                    action="disabled",
+                    source=payload.get("source"),
+                    percent=payload.get("percent"),
+                    effect_type=payload.get("effect_type"),
+                    details=payload.get("details"),
+                )
+            except Exception as exc:  # pragma: no cover - defensive logging
+                logger.warning(
+                    "Failed to record promo offer test access disable log for user %s: %s",
+                    payload.get("user_id"),
+                    exc,
+                )
         return len(entries)
 
 
diff --git a/locales/en.json b/locales/en.json
index e55f314e..aa465cdf 100644
--- a/locales/en.json
+++ b/locales/en.json
@@ -607,6 +607,28 @@
   "ADMIN_PROMO_OFFER_CTA_EXTEND": "Extend subscription",
   "ADMIN_PROMO_OFFER_BACK_TO_TEMPLATE": "↩️ Back to offer",
   "ADMIN_PROMO_OFFER_BACK_TO_LIST": "⬅️ Back to promo offers",
+  "ADMIN_PROMO_OFFER_LOGS": "📜 Activity log",
+  "ADMIN_PROMO_OFFER_LOGS_TITLE": "📜 <b>Promo offer activity log</b>",
+  "ADMIN_PROMO_OFFER_LOGS_PAGINATION": "Page {page}/{total}",
+  "ADMIN_PROMO_OFFER_LOGS_EMPTY_BODY": "No activity yet.",
+  "ADMIN_PROMO_OFFER_LOGS_UNKNOWN_USER": "Unknown user",
+  "ADMIN_PROMO_OFFER_LOGS_USER": "👤 {user}",
+  "ADMIN_PROMO_OFFER_LOGS_PERCENT": "📉 Discount: {percent}%",
+  "ADMIN_PROMO_OFFER_LOGS_EFFECT_TEST": "🧪 Test access",
+  "ADMIN_PROMO_OFFER_LOGS_EFFECT_DISCOUNT": "💸 Discount",
+  "ADMIN_PROMO_OFFER_LOGS_SOURCE": "🏷 Source: {source}",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_GENERIC": "ℹ️ Action: {reason}",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_MANUAL": "💳 Applied during manual payment",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_AUTOPAY": "🤖 Used in autopay renewal",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_EXPIRED": "⏳ Offer expired",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_TEST_EXPIRED": "⏳ Test access removed",
+  "ADMIN_PROMO_OFFER_LOGS_DESCRIPTION": "📝 {description}",
+  "ADMIN_PROMO_OFFER_LOGS_AMOUNT": "💰 Amount: {amount}",
+  "ADMIN_PROMO_OFFER_LOGS_SQUAD": "🌍 Squad: {squad}",
+  "ADMIN_PROMO_OFFER_LOGS_NEW_SQUADS": "🌍 New squads: {squads}",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_CLAIMED": "Claimed",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_CONSUMED": "Used",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_DISABLED": "Disabled",
   "ADMIN_SUPPORT_TICKETS": "🎫 Support tickets",
   "ADMIN_SUPPORT_AUDIT": "🧾 Moderator audit",
   "ADMIN_SUPPORT_SETTINGS": "🛟 Support settings",
diff --git a/locales/ru.json b/locales/ru.json
index 4a91528e..ba73f5f4 100644
--- a/locales/ru.json
+++ b/locales/ru.json
@@ -607,6 +607,28 @@
   "ADMIN_PROMO_OFFER_CTA_EXTEND": "Продлить подписку",
   "ADMIN_PROMO_OFFER_BACK_TO_TEMPLATE": "↩️ К предложению",
   "ADMIN_PROMO_OFFER_BACK_TO_LIST": "⬅️ К промопредложениям",
+  "ADMIN_PROMO_OFFER_LOGS": "📜 Лог операций",
+  "ADMIN_PROMO_OFFER_LOGS_TITLE": "📜 <b>Лог операций промо-предложений</b>",
+  "ADMIN_PROMO_OFFER_LOGS_PAGINATION": "Страница {page}/{total}",
+  "ADMIN_PROMO_OFFER_LOGS_EMPTY_BODY": "Записей пока нет.",
+  "ADMIN_PROMO_OFFER_LOGS_UNKNOWN_USER": "Неизвестный пользователь",
+  "ADMIN_PROMO_OFFER_LOGS_USER": "👤 {user}",
+  "ADMIN_PROMO_OFFER_LOGS_PERCENT": "📉 Скидка: {percent}%",
+  "ADMIN_PROMO_OFFER_LOGS_EFFECT_TEST": "🧪 Тестовый доступ",
+  "ADMIN_PROMO_OFFER_LOGS_EFFECT_DISCOUNT": "💸 Скидка",
+  "ADMIN_PROMO_OFFER_LOGS_SOURCE": "🏷 Источник: {source}",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_GENERIC": "ℹ️ Действие: {reason}",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_MANUAL": "💳 Списано при ручной оплате",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_AUTOPAY": "🤖 Применено при автопродлении",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_EXPIRED": "⏳ Предложение истекло",
+  "ADMIN_PROMO_OFFER_LOGS_REASON_TEST_EXPIRED": "⏳ Тестовый доступ отключён",
+  "ADMIN_PROMO_OFFER_LOGS_DESCRIPTION": "📝 {description}",
+  "ADMIN_PROMO_OFFER_LOGS_AMOUNT": "💰 Сумма: {amount}",
+  "ADMIN_PROMO_OFFER_LOGS_SQUAD": "🌍 Сквад: {squad}",
+  "ADMIN_PROMO_OFFER_LOGS_NEW_SQUADS": "🌍 Новые сквады: {squads}",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_CLAIMED": "Принято",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_CONSUMED": "Использовано",
+  "ADMIN_PROMO_OFFER_LOGS_ACTION_DISABLED": "Отключено",
   "ADMIN_SUPPORT_TICKETS": "🎫 Тикеты поддержки",
   "ADMIN_SUPPORT_AUDIT": "🧾 Аудит модераторов",
   "ADMIN_SUPPORT_SETTINGS": "🛟 Настройки поддержки",