fix: Add scrollable container to ToolSelectorModal for small screens

- Add max-h-[90vh] and flex-col to modal content container
- Wrap tools grid in max-h-[50vh] overflow-y-auto container
- Add overscroll-contain for smooth scroll behavior on mobile
- Fixes issue where 21 PDF tools overflow viewport on small screens

backend/app/extensions.py

@@ -70,7 +70,7 @@ def init_celery(app):
         "app.tasks.pdf_to_excel_tasks.*": {"queue": "pdf_tools"},
         "app.tasks.qrcode_tasks.*": {"queue": "default"},
         "app.tasks.html_to_pdf_tasks.*": {"queue": "convert"},
-        "app.tasks.pdf_ai_tasks.*": {"queue": "default"},
+        "app.tasks.pdf_ai_tasks.*": {"queue": "ai_heavy"},
         "app.tasks.pdf_convert_tasks.*": {"queue": "convert"},
         "app.tasks.pdf_extra_tasks.*": {"queue": "pdf_tools"},
         "app.tasks.image_extra_tasks.*": {"queue": "image"},

backend/app/routes/account.py

@@ -6,15 +6,24 @@ from app.extensions import limiter
 from app.services.account_service import (
     create_api_key,
     get_user_by_id,
+    has_task_access,
     list_api_keys,
+    record_usage_event,
     revoke_api_key,
 )
 from app.services.policy_service import get_usage_summary_for_user
+from app.services.credit_config import (
+    get_all_tool_costs,
+    get_credits_for_plan,
+    get_tool_credit_cost,
+    CREDIT_WINDOW_DAYS,
+)
+from app.services.credit_service import deduct_credits, get_credit_summary
 from app.services.stripe_service import (
     is_stripe_configured,
     get_stripe_price_id,
 )
-from app.utils.auth import get_current_user_id
+from app.utils.auth import get_current_user_id, has_session_task_access
 import stripe
 import logging
 
@@ -38,6 +47,19 @@ def get_usage_route():
     return jsonify(get_usage_summary_for_user(user_id, user["plan"])), 200
 
 
+@account_bp.route("/credit-info", methods=["GET"])
+@limiter.limit("60/hour")
+def get_credit_info_route():
+    """Return public credit/pricing info (no auth required)."""
+    return jsonify({
+        "plans": {
+            "free": {"credits": get_credits_for_plan("free"), "window_days": CREDIT_WINDOW_DAYS},
+            "pro": {"credits": get_credits_for_plan("pro"), "window_days": CREDIT_WINDOW_DAYS},
+        },
+        "tool_costs": get_all_tool_costs(),
+    }), 200
+
+
 @account_bp.route("/subscription", methods=["GET"])
 @limiter.limit("60/hour")
 def get_subscription_status():
@@ -159,3 +181,62 @@ def revoke_api_key_route(key_id: int):
         return jsonify({"error": "API key not found or already revoked."}), 404
 
     return jsonify({"message": "API key revoked."}), 200
+
+
+@account_bp.route("/claim-task", methods=["POST"])
+@limiter.limit("60/hour")
+def claim_task_route():
+    """Adopt an anonymous task into the authenticated user's history.
+
+    Called after a guest signs up or logs in to record the previously
+    processed task in their account and deduct credits.
+    """
+    user_id = get_current_user_id()
+    if user_id is None:
+        return jsonify({"error": "Authentication required."}), 401
+
+    data = request.get_json(silent=True) or {}
+    task_id = str(data.get("task_id", "")).strip()
+    tool = str(data.get("tool", "")).strip()
+
+    if not task_id or not tool:
+        return jsonify({"error": "task_id and tool are required."}), 400
+
+    # Verify this task belongs to the caller's session
+    if not has_session_task_access(task_id):
+        return jsonify({"error": "Task not found in your session."}), 403
+
+    # Skip if already claimed (idempotent)
+    if has_task_access(user_id, "web", task_id):
+        summary = get_credit_summary(user_id, "free")
+        return jsonify({"claimed": True, "credits": summary}), 200
+
+    user = get_user_by_id(user_id)
+    if user is None:
+        return jsonify({"error": "User not found."}), 404
+
+    plan = user.get("plan", "free")
+    cost = get_tool_credit_cost(tool)
+
+    # Deduct credits
+    try:
+        deduct_credits(user_id, plan, tool)
+    except ValueError:
+        return jsonify({
+            "error": "Insufficient credits to claim this file.",
+            "credits_required": cost,
+        }), 429
+
+    # Record usage event so the task appears in history
+    record_usage_event(
+        user_id=user_id,
+        source="web",
+        tool=tool,
+        task_id=task_id,
+        event_type="accepted",
+        api_key_id=None,
+        cost_points=cost,
+    )
+
+    summary = get_credit_summary(user_id, plan)
+    return jsonify({"claimed": True, "credits": summary}), 200

backend/app/routes/barcode.py

@@ -52,7 +52,7 @@ def generate_barcode_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="barcode")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/compress.py

@@ -38,7 +38,7 @@ def compress_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="compress-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/compress_image.py

@@ -43,7 +43,7 @@ def compress_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="compress-image")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/convert.py

@@ -33,7 +33,7 @@ def pdf_to_word_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-to-word")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -75,7 +75,7 @@ def word_to_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="word-to-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/download.py

@@ -34,6 +34,13 @@ def download_file(task_id: str, filename: str):
             assert_api_task_access(actor, task_id)
         else:
             actor = resolve_web_actor()
+            # Download gate: anonymous users must register before downloading
+            if actor.actor_type == "anonymous":
+                return (
+                    {"error": "signup_required",
+                     "message": "Create a free account to download your file."},
+                    401,
+                )
             assert_web_task_access(actor, task_id)
     except PolicyError as exc:
         abort(exc.status_code, exc.message)

backend/app/routes/flowchart.py

@@ -39,7 +39,7 @@ def extract_flowchart_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-flowchart")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -75,7 +75,7 @@ def extract_sample_flowchart_route():
     """
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-flowchart-sample")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/html_to_pdf.py

@@ -34,7 +34,7 @@ def html_to_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="html-to-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/image.py

@@ -54,7 +54,7 @@ def convert_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="image-convert")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -127,7 +127,7 @@ def resize_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="image-resize")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -180,7 +180,7 @@ def convert_image_to_svg_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="image-to-svg")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/image_extra.py

@@ -54,7 +54,7 @@ def crop_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="image-crop")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -120,7 +120,7 @@ def rotate_flip_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="image-rotate-flip")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/ocr.py

@@ -52,7 +52,7 @@ def ocr_image_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="ocr-image")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -102,7 +102,7 @@ def ocr_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="ocr-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_ai.py

@@ -11,6 +11,10 @@ from app.services.policy_service import (
     resolve_web_actor,
     validate_actor_file,
 )
+from app.services.translation_guardrails import (
+    check_page_admission,
+    TranslationAdmissionError,
+)
 from app.utils.file_validator import FileValidationError
 from app.utils.sanitizer import generate_safe_path
 from app.tasks.pdf_ai_tasks import (
@@ -48,7 +52,7 @@ def chat_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="chat-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -104,7 +108,7 @@ def summarize_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="summarize-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -161,7 +165,7 @@ def translate_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="translate-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -175,6 +179,12 @@ def translate_pdf_route():
     task_id, input_path = generate_safe_path(ext, folder_type="upload")
     file.save(input_path)
 
+    # ── Page-count admission guard ──
+    try:
+        page_count = check_page_admission(input_path, actor.plan)
+    except TranslationAdmissionError as e:
+        return jsonify({"error": e.message}), e.status_code
+
     task = translate_pdf_task.delay(
         input_path,
         task_id,
@@ -213,7 +223,7 @@ def extract_tables_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="extract-tables")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_convert.py

@@ -40,7 +40,7 @@ def pdf_to_pptx_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-to-pptx")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -77,7 +77,7 @@ def excel_to_pdf_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="excel-to-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -116,7 +116,7 @@ def pptx_to_pdf_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pptx-to-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -167,7 +167,7 @@ def sign_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="sign-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_editor.py

@@ -54,7 +54,7 @@ def edit_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-edit")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_extra.py

@@ -41,7 +41,7 @@ def crop_pdf_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="crop-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -89,7 +89,7 @@ def flatten_pdf_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="flatten-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -126,7 +126,7 @@ def repair_pdf_route():
     file = request.files["file"]
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="repair-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -184,7 +184,7 @@ def edit_metadata_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="edit-metadata")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_to_excel.py

@@ -34,7 +34,7 @@ def pdf_to_excel_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-to-excel")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/pdf_tools.py

@@ -56,7 +56,7 @@ def merge_pdfs_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="merge-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -123,7 +123,7 @@ def split_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="split-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -183,7 +183,7 @@ def rotate_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="rotate-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -246,7 +246,7 @@ def add_page_numbers_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="page-numbers")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -304,7 +304,7 @@ def pdf_to_images_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="pdf-to-images")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -353,7 +353,7 @@ def images_to_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="images-to-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -424,7 +424,7 @@ def watermark_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="watermark-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -480,7 +480,7 @@ def protect_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="protect-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -532,7 +532,7 @@ def unlock_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="unlock-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -579,7 +579,7 @@ def remove_watermark_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="remove-watermark")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -638,7 +638,7 @@ def reorder_pdf_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="reorder-pdf")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 
@@ -690,7 +690,7 @@ def extract_pages_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="extract-pages")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/qrcode.py

@@ -45,7 +45,7 @@ def generate_qr_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="qr-code")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/removebg.py

@@ -38,7 +38,7 @@ def remove_bg_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="remove-bg")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/routes/video.py

@@ -59,7 +59,7 @@ def video_to_gif_route():
 
     actor = resolve_web_actor()
     try:
-        assert_quota_available(actor)
+        assert_quota_available(actor, tool="video-frames")
     except PolicyError as e:
         return jsonify({"error": e.message}), e.status_code
 

backend/app/services/account_service.py

@@ -228,6 +228,30 @@ def _init_postgres_tables(conn):
         ON file_events(created_at DESC)
     """)
 
+    cursor.execute("""
+        CREATE TABLE IF NOT EXISTS user_credit_windows (
+            id SERIAL PRIMARY KEY,
+            user_id INTEGER NOT NULL UNIQUE,
+            window_start_at TEXT NOT NULL,
+            window_end_at TEXT NOT NULL,
+            credits_allocated INTEGER NOT NULL,
+            credits_used INTEGER NOT NULL DEFAULT 0,
+            plan TEXT NOT NULL DEFAULT 'free',
+            updated_at TEXT NOT NULL,
+            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+        )
+    """)
+    cursor.execute("""
+        CREATE INDEX IF NOT EXISTS idx_ucw_user
+        ON user_credit_windows(user_id)
+    """)
+
+    # Add cost_points column to usage_events if missing
+    if not _column_exists(conn, "usage_events", "cost_points"):
+        cursor.execute(
+            "ALTER TABLE usage_events ADD COLUMN cost_points INTEGER NOT NULL DEFAULT 1"
+        )
+
 
 def _init_sqlite_tables(conn):
     conn.executescript(
@@ -316,6 +340,21 @@ def _init_sqlite_tables(conn):
 
         CREATE INDEX IF NOT EXISTS idx_file_events_created
         ON file_events(created_at DESC);
+
+        CREATE TABLE IF NOT EXISTS user_credit_windows (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            user_id INTEGER NOT NULL UNIQUE,
+            window_start_at TEXT NOT NULL,
+            window_end_at TEXT NOT NULL,
+            credits_allocated INTEGER NOT NULL,
+            credits_used INTEGER NOT NULL DEFAULT 0,
+            plan TEXT NOT NULL DEFAULT 'free',
+            updated_at TEXT NOT NULL,
+            FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
+        );
+
+        CREATE INDEX IF NOT EXISTS idx_ucw_user
+        ON user_credit_windows(user_id);
         """
     )
 
@@ -325,6 +364,8 @@ def _init_sqlite_tables(conn):
         conn.execute("ALTER TABLE users ADD COLUMN updated_at TEXT NOT NULL DEFAULT ''")
     if not _column_exists(conn, "users", "role"):
         conn.execute("ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'user'")
+    if not _column_exists(conn, "usage_events", "cost_points"):
+        conn.execute("ALTER TABLE usage_events ADD COLUMN cost_points INTEGER NOT NULL DEFAULT 1")
 
 
 def create_user(email: str, password: str) -> dict:
@@ -842,6 +883,7 @@ def record_usage_event(
     task_id: str,
     event_type: str,
     api_key_id: int | None = None,
+    cost_points: int = 1,
 ):
     if user_id is None:
         return
@@ -851,17 +893,17 @@ def record_usage_event(
             """
             INSERT INTO usage_events (
                 user_id, api_key_id, source, tool, task_id,
-                event_type, created_at, period_month
+                event_type, created_at, period_month, cost_points
             )
-            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
+            VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
         """
             if is_postgres()
             else """
             INSERT INTO usage_events (
                 user_id, api_key_id, source, tool, task_id,
-                event_type, created_at, period_month
+                event_type, created_at, period_month, cost_points
             )
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
         """
         )
         execute_query(
@@ -876,6 +918,7 @@ def record_usage_event(
                 event_type,
                 _utc_now(),
                 get_current_period_month(),
+                cost_points,
             ),
         )
 

backend/app/services/credit_config.py

@@ -0,0 +1,122 @@
+"""Unified Credit System — tool cost registry and credit constants.
+
+Every tool has a credit cost. Lighter tools cost 1 credit, heavier
+server-side conversions cost 2, CPU/ML-intensive tools cost 3,
+and AI-powered tools cost 5+.
+
+This module is the single source of truth for all credit-related
+constants consumed by policy_service, credit_service, and the
+frontend config endpoint.
+"""
+
+import os
+
+# ── Credit allocations per rolling 30-day window ────────────────
+FREE_CREDITS_PER_WINDOW = int(os.getenv("FREE_CREDITS_PER_WINDOW", "50"))
+PRO_CREDITS_PER_WINDOW = int(os.getenv("PRO_CREDITS_PER_WINDOW", "500"))
+CREDIT_WINDOW_DAYS = int(os.getenv("CREDIT_WINDOW_DAYS", "30"))
+
+# ── Guest demo budget (anonymous, pre-registration) ────────────
+GUEST_DEMO_BUDGET = int(os.getenv("GUEST_DEMO_BUDGET", "3"))
+GUEST_DEMO_TTL_HOURS = int(os.getenv("GUEST_DEMO_TTL_HOURS", "24"))
+
+# ── API quota (Pro only, per rolling window) ────────────────────
+PRO_API_CREDITS_PER_WINDOW = int(os.getenv("PRO_API_CREDITS_PER_WINDOW", "1000"))
+
+# ── Cost tiers ──────────────────────────────────────────────────
+TIER_LIGHT = 1       # Fast, in-memory or trivial server ops
+TIER_MEDIUM = 2      # Server-side conversion (LibreOffice, Ghostscript, etc.)
+TIER_HEAVY = 3       # CPU/ML-intensive (OCR, background removal, compression)
+TIER_AI = 5          # AI-powered tools (LLM API calls)
+
+# ── Per-tool credit costs ───────────────────────────────────────
+# Keys match the `tool` parameter passed to record_usage_event / routes.
+TOOL_CREDIT_COSTS: dict[str, int] = {
+    # ─── PDF Core (light operations) ────────────────────────────
+    "merge-pdf": TIER_LIGHT,
+    "split-pdf": TIER_LIGHT,
+    "rotate-pdf": TIER_LIGHT,
+    "reorder-pdf": TIER_LIGHT,
+    "extract-pages": TIER_LIGHT,
+    "page-numbers": TIER_LIGHT,
+    "watermark-pdf": TIER_LIGHT,
+    "protect-pdf": TIER_LIGHT,
+    "unlock-pdf": TIER_LIGHT,
+    "flatten-pdf": TIER_LIGHT,
+    "repair-pdf": TIER_LIGHT,
+    "pdf-metadata": TIER_LIGHT,
+    "crop-pdf": TIER_LIGHT,
+    "sign-pdf": TIER_LIGHT,
+    "pdf-to-images": TIER_LIGHT,
+    "images-to-pdf": TIER_LIGHT,
+
+    # ─── Conversion (medium — server-side rendering) ────────────
+    "pdf-to-word": TIER_MEDIUM,
+    "word-to-pdf": TIER_MEDIUM,
+    "pdf-to-excel": TIER_MEDIUM,
+    "excel-to-pdf": TIER_MEDIUM,
+    "pdf-to-pptx": TIER_MEDIUM,
+    "pptx-to-pdf": TIER_MEDIUM,
+    "html-to-pdf": TIER_MEDIUM,
+    "pdf-editor": TIER_MEDIUM,
+
+    # ─── Image (light to medium) ────────────────────────────────
+    "image-converter": TIER_LIGHT,
+    "image-resize": TIER_LIGHT,
+    "image-crop": TIER_LIGHT,
+    "image-rotate-flip": TIER_LIGHT,
+    "image-to-svg": TIER_MEDIUM,
+
+    # ─── Image / PDF heavy (CPU/ML) ────────────────────────────
+    "compress-pdf": TIER_HEAVY,
+    "compress-image": TIER_HEAVY,
+    "ocr": TIER_HEAVY,
+    "remove-background": TIER_HEAVY,
+    "remove-watermark-pdf": TIER_HEAVY,
+
+    # ─── Utility ────────────────────────────────────────────────
+    "qr-code": TIER_LIGHT,
+    "barcode-generator": TIER_LIGHT,
+    "video-to-gif": TIER_MEDIUM,
+    "word-counter": TIER_LIGHT,
+    "text-cleaner": TIER_LIGHT,
+
+    # ─── AI-powered ─────────────────────────────────────────────
+    "chat-pdf": TIER_AI,
+    "summarize-pdf": TIER_AI,
+    "translate-pdf": TIER_AI,
+    "extract-tables": TIER_AI,
+    "pdf-flowchart": TIER_AI,
+    # ─── Route-specific aliases ─────────────────────────────────────
+    # Some routes record a tool name that differs from the manifest slug.
+    # Both names must map to the same cost.
+    "barcode": TIER_LIGHT,               # manifest: barcode-generator
+    "image-convert": TIER_LIGHT,          # manifest: image-converter
+    "ocr-image": TIER_HEAVY,             # manifest: ocr
+    "ocr-pdf": TIER_HEAVY,              # manifest: ocr
+    "pdf-flowchart-sample": TIER_AI,     # manifest: pdf-flowchart
+    "pdf-edit": TIER_MEDIUM,             # manifest: pdf-editor
+    "edit-metadata": TIER_LIGHT,         # manifest: pdf-metadata
+    "remove-watermark": TIER_HEAVY,      # manifest: remove-watermark-pdf
+    "remove-bg": TIER_HEAVY,            # manifest: remove-background
+    "video-frames": TIER_MEDIUM,         # route alias for video-to-gif
+    "edit-pdf-text": TIER_MEDIUM,        # route alias for pdf-editor
+}
+
+# Default cost for any tool not explicitly listed
+DEFAULT_CREDIT_COST = TIER_LIGHT
+
+
+def get_tool_credit_cost(tool: str) -> int:
+    """Return the credit cost for a given tool slug."""
+    return TOOL_CREDIT_COSTS.get(tool, DEFAULT_CREDIT_COST)
+
+
+def get_credits_for_plan(plan: str) -> int:
+    """Return the total credits per window for a plan."""
+    return PRO_CREDITS_PER_WINDOW if plan == "pro" else FREE_CREDITS_PER_WINDOW
+
+
+def get_all_tool_costs() -> dict[str, int]:
+    """Return the full cost registry — used by the config API endpoint."""
+    return dict(TOOL_CREDIT_COSTS)

backend/app/services/credit_service.py

@@ -0,0 +1,268 @@
+"""Credit window management — rolling 30-day balance for registered users.
+
+Handles lazy window creation on first use, automatic reset after expiry,
+balance queries, and atomic credit deduction.
+"""
+
+import logging
+import os
+from datetime import datetime, timedelta, timezone
+
+from app.services.credit_config import (
+    CREDIT_WINDOW_DAYS,
+    get_credits_for_plan,
+    get_tool_credit_cost,
+)
+from app.utils.database import (
+    db_connection,
+    execute_query,
+    is_postgres,
+    row_to_dict,
+)
+
+logger = logging.getLogger(__name__)
+
+# ── Redis caching (optional) ───────────────────────────────────
+_BALANCE_CACHE_TTL = int(os.getenv("CREDIT_BALANCE_CACHE_TTL", "300"))  # 5 min
+
+
+def _get_redis():
+    try:
+        import redis
+
+        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+        return redis.Redis.from_url(redis_url, decode_responses=True)
+    except Exception:
+        return None
+
+
+def _balance_cache_key(user_id: int) -> str:
+    return f"credit_balance:{user_id}"
+
+
+def _invalidate_balance_cache(user_id: int) -> None:
+    r = _get_redis()
+    if r:
+        try:
+            r.delete(_balance_cache_key(user_id))
+        except Exception:
+            pass
+
+
+def _cache_balance(user_id: int, balance: int) -> None:
+    r = _get_redis()
+    if r:
+        try:
+            r.setex(_balance_cache_key(user_id), _BALANCE_CACHE_TTL, str(balance))
+        except Exception:
+            pass
+
+
+def _get_cached_balance(user_id: int) -> int | None:
+    r = _get_redis()
+    if r is None:
+        return None
+    try:
+        val = r.get(_balance_cache_key(user_id))
+        return int(str(val)) if val is not None else None
+    except Exception:
+        return None
+
+
+# ── Window helpers ──────────────────────────────────────────────
+
+def _utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _utc_now_dt() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _make_window_end(start_iso: str) -> str:
+    start = datetime.fromisoformat(start_iso)
+    end = start + timedelta(days=CREDIT_WINDOW_DAYS)
+    return end.isoformat()
+
+
+def _is_window_expired(window_end_at: str) -> bool:
+    end = datetime.fromisoformat(window_end_at)
+    if end.tzinfo is None:
+        end = end.replace(tzinfo=timezone.utc)
+    return _utc_now_dt() >= end
+
+
+def _get_window(conn, user_id: int) -> dict | None:
+    sql = (
+        "SELECT * FROM user_credit_windows WHERE user_id = %s"
+        if is_postgres()
+        else "SELECT * FROM user_credit_windows WHERE user_id = ?"
+    )
+    cursor = execute_query(conn, sql, (user_id,))
+    row = cursor.fetchone()
+    return row_to_dict(row)
+
+
+def _create_window(conn, user_id: int, plan: str) -> dict:
+    now = _utc_now()
+    credits = get_credits_for_plan(plan)
+    end = _make_window_end(now)
+
+    sql = (
+        """
+        INSERT INTO user_credit_windows
+            (user_id, window_start_at, window_end_at, credits_allocated, credits_used, plan, updated_at)
+        VALUES (%s, %s, %s, %s, 0, %s, %s)
+        ON CONFLICT (user_id) DO UPDATE SET
+            window_start_at = EXCLUDED.window_start_at,
+            window_end_at = EXCLUDED.window_end_at,
+            credits_allocated = EXCLUDED.credits_allocated,
+            credits_used = 0,
+            plan = EXCLUDED.plan,
+            updated_at = EXCLUDED.updated_at
+        """
+        if is_postgres()
+        else """
+        INSERT OR REPLACE INTO user_credit_windows
+            (user_id, window_start_at, window_end_at, credits_allocated, credits_used, plan, updated_at)
+        VALUES (?, ?, ?, ?, 0, ?, ?)
+        """
+    )
+    execute_query(conn, sql, (user_id, now, end, credits, plan, now))
+
+    return {
+        "user_id": user_id,
+        "window_start_at": now,
+        "window_end_at": end,
+        "credits_allocated": credits,
+        "credits_used": 0,
+        "plan": plan,
+        "updated_at": now,
+    }
+
+
+def _reset_window(conn, user_id: int, plan: str) -> dict:
+    """Reset an expired window — starts a fresh 30-day period."""
+    return _create_window(conn, user_id, plan)
+
+
+# ── Public API ──────────────────────────────────────────────────
+
+def get_or_create_credit_window(user_id: int, plan: str) -> dict:
+    """Return the active credit window, creating or resetting as needed.
+
+    This is the lazy initialization entrypoint:
+    - First call after registration creates the window.
+    - First call after window expiry resets it with a fresh allocation.
+    - Plan upgrades (free→pro) are reflected on the next reset.
+    """
+    with db_connection() as conn:
+        window = _get_window(conn, user_id)
+
+        if window is None:
+            window = _create_window(conn, user_id, plan)
+            logger.info("Created credit window for user %d (plan=%s)", user_id, plan)
+            return window
+
+        if _is_window_expired(window["window_end_at"]):
+            window = _reset_window(conn, user_id, plan)
+            _invalidate_balance_cache(user_id)
+            logger.info("Reset expired credit window for user %d (plan=%s)", user_id, plan)
+            return window
+
+        # If plan changed mid-window, update allocation (pro upgrade benefit)
+        expected_credits = get_credits_for_plan(plan)
+        if window["plan"] != plan and expected_credits > window["credits_allocated"]:
+            additional = expected_credits - window["credits_allocated"]
+            sql = (
+                """
+                UPDATE user_credit_windows
+                SET credits_allocated = credits_allocated + %s, plan = %s, updated_at = %s
+                WHERE user_id = %s
+                """
+                if is_postgres()
+                else """
+                UPDATE user_credit_windows
+                SET credits_allocated = credits_allocated + ?, plan = ?, updated_at = ?
+                WHERE user_id = ?
+                """
+            )
+            execute_query(conn, sql, (additional, plan, _utc_now(), user_id))
+            window["credits_allocated"] += additional
+            window["plan"] = plan
+            _invalidate_balance_cache(user_id)
+            logger.info(
+                "Upgraded credit window for user %d: +%d credits (plan=%s)",
+                user_id,
+                additional,
+                plan,
+            )
+
+        return window
+
+
+def get_rolling_balance(user_id: int, plan: str) -> int:
+    """Return remaining credits for the current window."""
+    cached = _get_cached_balance(user_id)
+    if cached is not None:
+        return cached
+
+    window = get_or_create_credit_window(user_id, plan)
+    balance = max(0, window["credits_allocated"] - window["credits_used"])
+    _cache_balance(user_id, balance)
+    return balance
+
+
+def deduct_credits(user_id: int, plan: str, tool: str) -> int:
+    """Deduct tool credits from the user's window. Returns the cost deducted.
+
+    Raises ValueError if insufficient credits.
+    """
+    cost = get_tool_credit_cost(tool)
+
+    with db_connection() as conn:
+        # Ensure window is current
+        window = _get_window(conn, user_id)
+        if window is None or _is_window_expired(window.get("window_end_at", "")):
+            # get_or_create handles reset
+            pass
+        window = get_or_create_credit_window(user_id, plan)
+
+        balance = window["credits_allocated"] - window["credits_used"]
+        if balance < cost:
+            raise ValueError(
+                f"Insufficient credits: {balance} remaining, {cost} required for {tool}."
+            )
+
+        sql = (
+            """
+            UPDATE user_credit_windows
+            SET credits_used = credits_used + %s, updated_at = %s
+            WHERE user_id = %s
+            """
+            if is_postgres()
+            else """
+            UPDATE user_credit_windows
+            SET credits_used = credits_used + ?, updated_at = ?
+            WHERE user_id = ?
+            """
+        )
+        execute_query(conn, sql, (cost, _utc_now(), user_id))
+
+    _invalidate_balance_cache(user_id)
+    return cost
+
+
+def get_credit_summary(user_id: int, plan: str) -> dict:
+    """Return a full credit summary for the account page."""
+    window = get_or_create_credit_window(user_id, plan)
+    balance = max(0, window["credits_allocated"] - window["credits_used"])
+    return {
+        "credits_allocated": window["credits_allocated"],
+        "credits_used": window["credits_used"],
+        "credits_remaining": balance,
+        "window_start_at": window["window_start_at"],
+        "window_end_at": window["window_end_at"],
+        "plan": window["plan"],
+        "window_days": CREDIT_WINDOW_DAYS,
+    }

backend/app/services/guest_budget_service.py

@@ -0,0 +1,86 @@
+"""Guest demo budget enforcement.
+
+Anonymous visitors get a small usage budget tracked by IP address
+via Redis (with Flask session fallback).  The budget prevents abuse
+of expensive tools before the download-gate forces registration.
+"""
+
+import os
+from flask import request, session
+
+from app.services.credit_config import GUEST_DEMO_BUDGET, GUEST_DEMO_TTL_HOURS
+
+_TTL_SECONDS = GUEST_DEMO_TTL_HOURS * 3600
+
+
+# ── Redis helpers ──────────────────────────────────────────────
+def _get_redis():
+    try:
+        import redis
+        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+        return redis.Redis.from_url(redis_url, decode_responses=True)
+    except Exception:
+        return None
+
+
+def _guest_redis_key(ip: str) -> str:
+    return f"guest_demo:{ip}"
+
+
+def _get_client_ip() -> str:
+    """Return the best-effort client IP for rate tracking."""
+    forwarded = request.headers.get("X-Forwarded-For", "")
+    if forwarded:
+        return forwarded.split(",")[0].strip()
+    return request.remote_addr or "unknown"
+
+
+# ── Public API ─────────────────────────────────────────────────
+
+def get_guest_remaining() -> int:
+    """Return how many demo operations the current guest has left."""
+    ip = _get_client_ip()
+    r = _get_redis()
+
+    if r is not None:
+        try:
+            used = r.get(_guest_redis_key(ip))
+            if used is None:
+                return GUEST_DEMO_BUDGET
+            return max(0, GUEST_DEMO_BUDGET - int(str(used)))
+        except Exception:
+            pass
+
+    # Fallback: Flask session
+    used = session.get("guest_demo_used", 0)
+    return max(0, GUEST_DEMO_BUDGET - used)
+
+
+def record_guest_usage() -> None:
+    """Increment the guest demo counter for the current visitor."""
+    ip = _get_client_ip()
+    r = _get_redis()
+
+    if r is not None:
+        try:
+            key = _guest_redis_key(ip)
+            pipe = r.pipeline()
+            pipe.incr(key)
+            pipe.expire(key, _TTL_SECONDS)
+            pipe.execute()
+            return
+        except Exception:
+            pass
+
+    # Fallback: Flask session
+    session["guest_demo_used"] = session.get("guest_demo_used", 0) + 1
+
+
+def assert_guest_budget_available() -> None:
+    """Raise ValueError if the guest has exhausted their demo budget."""
+    remaining = get_guest_remaining()
+    if remaining <= 0:
+        raise ValueError(
+            "You have used all your free demo tries. "
+            "Create a free account to continue."
+        )

backend/app/services/policy_service.py

@@ -12,6 +12,23 @@ from app.services.account_service import (
     normalize_plan,
     record_usage_event,
 )
+from app.services.credit_config import (
+    get_tool_credit_cost,
+    get_credits_for_plan,
+    get_all_tool_costs,
+    GUEST_DEMO_BUDGET,
+    GUEST_DEMO_TTL_HOURS,
+    PRO_API_CREDITS_PER_WINDOW,
+)
+from app.services.credit_service import (
+    deduct_credits,
+    get_credit_summary,
+    get_rolling_balance,
+)
+from app.services.guest_budget_service import (
+    assert_guest_budget_available,
+    record_guest_usage,
+)
 from app.utils.auth import get_current_user_id, logout_user_session
 from app.utils.auth import has_session_task_access, remember_task_access
 from app.utils.file_validator import validate_file
@@ -19,10 +36,6 @@ from app.utils.file_validator import validate_file
 FREE_PLAN = "free"
 PRO_PLAN = "pro"
 
-FREE_WEB_MONTHLY_LIMIT = 50
-PRO_WEB_MONTHLY_LIMIT = 500
-PRO_API_MONTHLY_LIMIT = 1000
-
 FREE_HISTORY_LIMIT = 25
 PRO_HISTORY_LIMIT = 250
 
@@ -56,15 +69,15 @@ def get_history_limit(plan: str) -> int:
 
 
 def get_web_quota_limit(plan: str, actor_type: str) -> int | None:
-    """Return the monthly accepted-task cap for one web actor."""
+    """Return the credit allocation for one web actor's window."""
     if actor_type == "anonymous":
         return None
-    return PRO_WEB_MONTHLY_LIMIT if normalize_plan(plan) == PRO_PLAN else FREE_WEB_MONTHLY_LIMIT
+    return get_credits_for_plan(normalize_plan(plan))
 
 
 def get_api_quota_limit(plan: str) -> int | None:
-    """Return the monthly accepted-task cap for one API actor."""
-    return PRO_API_MONTHLY_LIMIT if normalize_plan(plan) == PRO_PLAN else None
+    """Return the credit allocation for one API actor's window."""
+    return PRO_API_CREDITS_PER_WINDOW if normalize_plan(plan) == PRO_PLAN else None
 
 
 def ads_enabled(plan: str, actor_type: str) -> bool:
@@ -97,27 +110,19 @@ def get_effective_file_size_limits_mb(plan: str) -> dict[str, int]:
 def get_usage_summary_for_user(user_id: int, plan: str) -> dict:
     """Return usage/quota summary for one authenticated user."""
     normalized_plan = normalize_plan(plan)
-    current_period = get_current_period_month()
-    web_used = count_usage_events(
-        user_id, "web", event_type="accepted", period_month=current_period
-    )
-    api_used = count_usage_events(
-        user_id, "api", event_type="accepted", period_month=current_period
-    )
+    credit_info = get_credit_summary(user_id, normalized_plan)
 
     return {
         "plan": normalized_plan,
-        "period_month": current_period,
         "ads_enabled": ads_enabled(normalized_plan, "session"),
         "history_limit": get_history_limit(normalized_plan),
         "file_limits_mb": get_effective_file_size_limits_mb(normalized_plan),
+        "credits": credit_info,
+        "tool_costs": get_all_tool_costs(),
+        # Legacy fields for backward compatibility
         "web_quota": {
-            "used": web_used,
-            "limit": get_web_quota_limit(normalized_plan, "session"),
-        },
-        "api_quota": {
-            "used": api_used,
-            "limit": get_api_quota_limit(normalized_plan),
+            "used": credit_info["credits_used"],
+            "limit": credit_info["credits_allocated"],
         },
     }
 
@@ -173,21 +178,38 @@ def validate_actor_file(file_storage, allowed_types: list[str], actor: ActorCont
     )
 
 
-def assert_quota_available(actor: ActorContext):
-    """Ensure an actor still has accepted-task quota for the current month."""
+def assert_quota_available(actor: ActorContext, tool: str | None = None):
+    """Ensure an actor still has credits for the requested tool.
+
+    For registered users: checks rolling credit window balance.
+    For anonymous users: checks guest demo budget.
+    """
     if actor.user_id is None:
+        # Guest demo budget enforcement
+        try:
+            assert_guest_budget_available()
+        except ValueError:
+            raise PolicyError(
+                "You have used all your free demo tries. "
+                "Create a free account to continue.",
+                429,
+            )
         return
 
     if actor.source == "web":
-        limit = get_web_quota_limit(actor.plan, actor.actor_type)
-        if limit is None:
-            return
-        used = count_usage_events(actor.user_id, "web", event_type="accepted")
-        if used >= limit:
+        # Credit-based check
+        cost = get_tool_credit_cost(tool) if tool else 1
+        balance = get_rolling_balance(actor.user_id, actor.plan)
+        if balance < cost:
             if normalize_plan(actor.plan) == PRO_PLAN:
-                raise PolicyError("Your monthly Pro web quota has been reached.", 429)
+                raise PolicyError(
+                    f"Your Pro credit balance is exhausted ({balance} remaining, "
+                    f"{cost} required). Credits reset at the end of your 30-day window.",
+                    429,
+                )
             raise PolicyError(
-                "Your monthly free plan limit has been reached. Upgrade to Pro for higher limits.",
+                f"Your free credit balance is exhausted ({balance} remaining, "
+                f"{cost} required). Upgrade to Pro for more credits.",
                 429,
             )
         return
@@ -202,10 +224,29 @@ def assert_quota_available(actor: ActorContext):
 
 
 def record_accepted_usage(actor: ActorContext, tool: str, celery_task_id: str):
-    """Record one accepted usage event after task dispatch succeeds."""
+    """Record one accepted usage event and deduct credits after task dispatch."""
     if actor.source == "web":
         remember_task_access(celery_task_id)
 
+    cost = get_tool_credit_cost(tool)
+
+    # Deduct credits from the rolling window (registered users only)
+    if actor.user_id is not None and actor.source == "web":
+        try:
+            deduct_credits(actor.user_id, actor.plan, tool)
+        except ValueError:
+            # Balance check should have caught this in assert_quota_available.
+            # Log but don't block — the usage event is the source of truth.
+            import logging
+            logging.getLogger(__name__).warning(
+                "Credit deduction failed for user %d tool %s (insufficient balance at record time)",
+                actor.user_id,
+                tool,
+            )
+    elif actor.user_id is None and actor.source == "web":
+        # Record guest demo usage
+        record_guest_usage()
+
     record_usage_event(
         user_id=actor.user_id,
         source=actor.source,
@@ -213,6 +254,7 @@ def record_accepted_usage(actor: ActorContext, tool: str, celery_task_id: str):
         task_id=celery_task_id,
         event_type="accepted",
         api_key_id=actor.api_key_id,
+        cost_points=cost,
     )
 
 

backend/app/services/translation_guardrails.py

@@ -0,0 +1,151 @@
+"""Translation guardrails — admission control, caching, and cost protection.
+
+This module implements the guardrail model described in
+docs/tool-portfolio/05-ai-cost-and-performance-plan.md.
+"""
+
+import hashlib
+import logging
+import os
+from typing import Optional
+
+from flask import current_app
+
+logger = logging.getLogger(__name__)
+
+# ── Page-count admission tiers ──────────────────────────────────────
+# These limits define the maximum number of pages allowed per plan.
+# Free/anonymous users get a lower cap; Pro users get a higher cap.
+FREE_TRANSLATE_MAX_PAGES = int(os.getenv("FREE_TRANSLATE_MAX_PAGES", "10"))
+PRO_TRANSLATE_MAX_PAGES = int(os.getenv("PRO_TRANSLATE_MAX_PAGES", "50"))
+
+
+class TranslationAdmissionError(Exception):
+    """Raised when a translation job is rejected at admission."""
+
+    def __init__(self, message: str, status_code: int = 400):
+        super().__init__(message)
+        self.message = message
+        self.status_code = status_code
+
+
+def get_page_limit(plan: str) -> int:
+    """Return the page cap for a given plan."""
+    from app.services.account_service import normalize_plan
+
+    if normalize_plan(plan) == "pro":
+        return PRO_TRANSLATE_MAX_PAGES
+    return FREE_TRANSLATE_MAX_PAGES
+
+
+def count_pdf_pages(file_path: str) -> int:
+    """Return the number of pages in a PDF file."""
+    try:
+        from PyPDF2 import PdfReader
+
+        reader = PdfReader(file_path)
+        return len(reader.pages)
+    except Exception as e:
+        logger.warning("Failed to count PDF pages for admission: %s", e)
+        # If we can't count pages, allow the job through but log it
+        return 0
+
+
+def check_page_admission(file_path: str, plan: str) -> int:
+    """Verify a PDF is within the page limit for the given plan.
+
+    Returns the page count on success.
+    Raises TranslationAdmissionError if the file exceeds the limit.
+    """
+    page_count = count_pdf_pages(file_path)
+    if page_count == 0:
+        # Can't determine — allow through (OCR fallback scenario)
+        return page_count
+
+    limit = get_page_limit(plan)
+    if page_count > limit:
+        raise TranslationAdmissionError(
+            f"This PDF has {page_count} pages. "
+            f"Your plan allows up to {limit} pages for translation. "
+            f"Please upgrade your plan or use a smaller file.",
+            status_code=413,
+        )
+    return page_count
+
+
+# ── Content-hash caching ────────────────────────────────────────────
+# Redis-based cache keyed by file-content hash + target language.
+# Avoids re-translating identical documents.
+
+TRANSLATION_CACHE_TTL = int(os.getenv("TRANSLATION_CACHE_TTL", str(7 * 24 * 3600)))  # 7 days
+
+
+def _get_redis():
+    """Get Redis connection from Flask app config."""
+    try:
+        import redis
+
+        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
+        return redis.Redis.from_url(redis_url, decode_responses=True)
+    except Exception as e:
+        logger.debug("Redis not available for translation cache: %s", e)
+        return None
+
+
+def _compute_content_hash(file_path: str) -> str:
+    """Compute SHA-256 hash of file contents."""
+    sha = hashlib.sha256()
+    with open(file_path, "rb") as f:
+        for chunk in iter(lambda: f.read(8192), b""):
+            sha.update(chunk)
+    return sha.hexdigest()
+
+
+def _cache_key(content_hash: str, target_language: str, source_language: str) -> str:
+    """Build a Redis key for the translation cache."""
+    return f"translate_cache:{content_hash}:{source_language}:{target_language}"
+
+
+def get_cached_translation(
+    file_path: str, target_language: str, source_language: str = "auto"
+) -> Optional[dict]:
+    """Look up a cached translation result. Returns None on miss."""
+    r = _get_redis()
+    if r is None:
+        return None
+
+    try:
+        content_hash = _compute_content_hash(file_path)
+        key = _cache_key(content_hash, target_language, source_language)
+        import json
+
+        cached = r.get(key)
+        if cached:
+            logger.info("Translation cache hit for %s", key)
+            return json.loads(cached)
+    except Exception as e:
+        logger.debug("Translation cache lookup failed: %s", e)
+
+    return None
+
+
+def store_cached_translation(
+    file_path: str,
+    target_language: str,
+    source_language: str,
+    result: dict,
+) -> None:
+    """Store a successful translation result in Redis."""
+    r = _get_redis()
+    if r is None:
+        return
+
+    try:
+        import json
+
+        content_hash = _compute_content_hash(file_path)
+        key = _cache_key(content_hash, target_language, source_language)
+        r.setex(key, TRANSLATION_CACHE_TTL, json.dumps(result, ensure_ascii=False))
+        logger.info("Translation cached: %s (TTL=%ds)", key, TRANSLATION_CACHE_TTL)
+    except Exception as e:
+        logger.debug("Translation cache store failed: %s", e)

backend/app/tasks/pdf_ai_tasks.py

@@ -15,6 +15,10 @@ from app.services.pdf_ai_service import (
     PdfAiError,
 )
 from app.services.task_tracking_service import finalize_task_tracking
+from app.services.translation_guardrails import (
+    get_cached_translation,
+    store_cached_translation,
+)
 from app.utils.sanitizer import cleanup_task_files
 
 logger = logging.getLogger(__name__)
@@ -214,9 +218,24 @@ def translate_pdf_task(
             meta={"step": "Translating document with provider fallback..."},
         )
 
-        data = translate_pdf(
-            input_path, target_language, source_language=source_language
+        # ── Cache lookup — skip AI call if identical translation exists ──
+        cached = get_cached_translation(
+            input_path, target_language, source_language or "auto"
         )
+        if cached is not None:
+            data = cached
+            data["provider"] = f"{data.get('provider', 'unknown')} (cached)"
+        else:
+            data = translate_pdf(
+                input_path, target_language, source_language=source_language
+            )
+            # Store successful result for future cache hits
+            store_cached_translation(
+                input_path,
+                target_language,
+                source_language or "auto",
+                data,
+            )
 
         result = {
             "status": "completed",

backend/tests/test_config.py

@@ -46,7 +46,8 @@ class TestConfigEndpoint:
         usage = data["usage"]
         assert usage["plan"] == "free"
         assert "web_quota" in usage
-        assert "api_quota" in usage
+        assert "credits" in usage
+        assert usage["credits"]["credits_allocated"] == 50
 
     def test_max_upload_mb_is_correct(self, client):
         """max_upload_mb should equal the largest single-type limit."""

backend/tests/test_download.py

@@ -1,12 +1,24 @@
 """Tests for file download route."""
 import os
 
+from app.services.account_service import create_user
 from app.utils.auth import TASK_ACCESS_SESSION_KEY
 
 
 class TestDownload:
-    def test_download_nonexistent_file(self, client):
-        """Should return 404 for missing file."""
+    def test_download_anonymous_returns_401(self, client):
+        """Anonymous users should be blocked by the download gate."""
+        response = client.get('/api/download/some-task-id/output.pdf')
+        assert response.status_code == 401
+        assert response.get_json()['error'] == 'signup_required'
+
+    def test_download_nonexistent_file(self, client, app):
+        """Should return 404 for missing file when authenticated."""
+        with app.app_context():
+            user = create_user('download-test@example.com', 'pass12345')
+        with client.session_transaction() as session:
+            session['user_id'] = user['id']
+            session[TASK_ACCESS_SESSION_KEY] = ['some-task-id']
         response = client.get('/api/download/some-task-id/output.pdf')
         assert response.status_code == 404
 
@@ -22,10 +34,13 @@ class TestDownload:
         assert response.status_code in (400, 404)
 
     def test_download_valid_file(self, client, app):
-        """Should serve file if it exists."""
+        """Should serve file if it exists and user is authenticated."""
         task_id = 'test-download-id'
         filename = 'output.pdf'
 
+        with app.app_context():
+            user = create_user('download-valid@example.com', 'pass12345')
+
         # Create the file in the output directory
         output_dir = os.path.join(app.config['OUTPUT_FOLDER'], task_id)
         os.makedirs(output_dir, exist_ok=True)
@@ -34,6 +49,7 @@ class TestDownload:
             f.write(b'%PDF-1.4 test content')
 
         with client.session_transaction() as session:
+            session['user_id'] = user['id']
             session[TASK_ACCESS_SESSION_KEY] = [task_id]
 
         response = client.get(f'/api/download/{task_id}/{filename}')
@@ -45,26 +61,37 @@ class TestDownload:
         task_id = 'test-name-id'
         filename = 'output.pdf'
 
+        with app.app_context():
+            user = create_user('download-name@example.com', 'pass12345')
+
         output_dir = os.path.join(app.config['OUTPUT_FOLDER'], task_id)
         os.makedirs(output_dir, exist_ok=True)
         with open(os.path.join(output_dir, filename), 'wb') as f:
             f.write(b'%PDF-1.4')
 
         with client.session_transaction() as session:
+            session['user_id'] = user['id']
             session[TASK_ACCESS_SESSION_KEY] = [task_id]
 
         response = client.get(f'/api/download/{task_id}/{filename}?name=my-document.pdf')
         assert response.status_code == 200
 
     def test_download_requires_task_access(self, client, app):
-        """Should not serve an existing file without session or API ownership."""
+        """Should not serve an existing file without task access, even if authenticated."""
         task_id = 'protected-download-id'
         filename = 'output.pdf'
 
+        with app.app_context():
+            user = create_user('download-noaccess@example.com', 'pass12345')
+
         output_dir = os.path.join(app.config['OUTPUT_FOLDER'], task_id)
         os.makedirs(output_dir, exist_ok=True)
         with open(os.path.join(output_dir, filename), 'wb') as f:
             f.write(b'%PDF-1.4 protected')
 
+        with client.session_transaction() as session:
+            session['user_id'] = user['id']
+            # No TASK_ACCESS_SESSION_KEY set — user can't access this task
+
         response = client.get(f'/api/download/{task_id}/{filename}')
         assert response.status_code == 404