refactor(auth): streamline supabase admin access in shares router

fix(input-limit): adjust max topic length and refine validation logic

Author: sanjeevafk

api/routers/shares.py

@@ -6,7 +6,8 @@
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel, Field, field_validator
 
-from auth import get_supabase_admin, verify_token, verify_token_optional
+import auth
+from auth import verify_token, verify_token_optional
 from config import get_settings
 from logging_config import anonymize_user_id, logger
 import services.share_manager as share_manager
@@ -195,7 +196,7 @@ async def create_share(
     user = auth_data["user"]
     user_id = str(user.id)
 
-    supabase = get_supabase_admin()
+    supabase = auth.get_supabase_admin()
     if not supabase:
         raise HTTPException(status_code=500, detail="Database connection error")
 
@@ -408,7 +409,7 @@ async def create_share(
 
 @router.get("/shares/{share_token}", response_model=ShareSnapshotResponse)
 async def get_share(share_token: str, request: Request, auth_data: Optional[dict] = Depends(verify_token_optional)):
-    supabase = get_supabase_admin()
+    supabase = auth.get_supabase_admin()
     if not supabase:
         raise HTTPException(status_code=500, detail="Database connection error")
 
@@ -461,7 +462,7 @@ async def revoke_share(share_id: str, auth_data: dict = Depends(verify_token)):
     user = auth_data["user"]
     user_id = str(user.id)
 
-    supabase = get_supabase_admin()
+    supabase = auth.get_supabase_admin()
     if not supabase:
         raise HTTPException(status_code=500, detail="Database connection error")
 
@@ -504,7 +505,7 @@ async def list_user_shares(
     user = auth_data["user"]
     user_id = str(user.id)
 
-    supabase = get_supabase_admin()
+    supabase = auth.get_supabase_admin()
     if not supabase:
         raise HTTPException(status_code=500, detail="Database connection error")
 

api/services/input_limit.py

@@ -21,6 +21,10 @@ def get_max_input_tokens_for_alias(alias: str, mode: str) -> int:
     mode_lower = (mode or "").strip().lower()
     alias_lower = (alias or "").strip().lower()
 
+    # Socratic mode: Always conservative (questions don't need massive context)
+    if mode_lower == SOCRATIC_MODE:
+        return 6000   # ~24K chars
+
     # Gemini Pro: 2M token window
     if "gemini-pro" in alias_lower or alias_lower == "technical-primary":
         return 18000  # ~72K chars; stay well under 2M
@@ -41,10 +45,6 @@ def get_max_input_tokens_for_alias(alias: str, mode: str) -> int:
     if "openrouter" in alias_lower:
         return 5000   # ~20K chars minimum safe
 
-    # Socratic mode: Always conservative (questions don't need massive context)
-    if mode_lower == SOCRATIC_MODE:
-        return 6000   # ~24K chars
-    
     # Default fallback
     return 8000  # ~32K chars
 
@@ -76,15 +76,16 @@ async def truncate_input_if_needed(
         "truncation_reason": None,
     }
 
-    if input_tokens <= max_tokens:
-        return user_input, metadata
-    
-    # Truncate to ~75% of max (buffer for system prompt + output)
-    safe_token_limit = int(max_tokens * 0.75)
+    # Truncate to ~80% of max (buffer for system prompt + output)
+    safe_token_limit = int(max_tokens * 0.8)
 
     # Estimate: 1 token ≈ 4 chars
     safe_char_limit = safe_token_limit * 4
 
+    # Allow both token and character thresholds to trigger truncation.
+    if input_tokens <= max_tokens and len(user_input) <= safe_char_limit:
+        return user_input, metadata
+
     # Truncate at sentence boundary (don't split mid-word)
     truncated = user_input[:safe_char_limit].rstrip()
 
@@ -102,9 +103,12 @@ async def truncate_input_if_needed(
 
         truncated_tokens = count_prompt_tokens(truncated)
         metadata["truncation_reason"] = (
-            f"Input ({input_tokens} tokens) exceeded {mode} mode limit ({max_tokens} tokens) "
-            f"for model alias '{alias}'. Truncated to {truncated_tokens} tokens."
+            f"Input ({input_tokens} tokens) exceeded safe limits for {mode} mode "
+            f"(max {max_tokens} tokens) on model alias '{alias}'. "
+            f"Truncated from {len(user_input)} to {len(truncated)} characters "
+            f"(~{truncated_tokens} tokens)."
         )
         truncated += f"\n\n_[Input truncated from {len(user_input)} to {len(truncated)} characters]_"
+        metadata["truncated_to"] = len(truncated)
 
     return truncated, metadata

api/utils.py

@@ -6,7 +6,7 @@
 import os
 import re
 
-MAX_TOPIC_LENGTH = 100000  # Increased to support large text input (100K chars)
+MAX_TOPIC_LENGTH = 200
 _logger = logging.getLogger(__name__)
 
 LEARNING_MODE = "learn"
@@ -39,8 +39,8 @@ def sanitize_topic(topic: str) -> str:
     topic = topic.strip()
     if len(topic) > MAX_TOPIC_LENGTH:
         raise ValueError(f"Topic exceeds {MAX_TOPIC_LENGTH} chars")
-    # Note: Removed character whitelist validation to support large text input
-    # (articles, code blocks, documentation). HTML escaping provides XSS protection.
+    if re.search(r"[<>]", topic):
+        raise ValueError("Topic contains invalid characters")
     return html.escape(topic)