Upload 2 files

src/config.py

@@ -1,127 +1,153 @@
-"""Environment-backed configuration for LibBee."""
-
-from functools import lru_cache
-from pathlib import Path
-from typing import Literal
-
-from pydantic import field_validator
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-LIBBEE_VERSION = "4.1.0"
-
-
-class Settings(BaseSettings):
-    model_config = SettingsConfigDict(case_sensitive=False, env_file=".env", env_file_encoding="utf-8", extra="ignore")
-
-    openai_api_key: str = ""
-    anthropic_api_key: str = ""
+"""Environment-backed configuration for LibBee."""
+
+from functools import lru_cache
+from pathlib import Path
+from typing import Literal
+
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+LIBBEE_VERSION = "4.1.1"
+
+
+class Settings(BaseSettings):
+    model_config = SettingsConfigDict(case_sensitive=False, env_file=".env", env_file_encoding="utf-8", extra="ignore")
+
+    openai_api_key: str = ""
+    anthropic_api_key: str = ""
     primo_api_key: str = ""
-    contact_email: str = "libse@ku.ac.ae"
-
-    # The Worker sends this secret in X-LibBee-Origin-Token. Production
-    # protected routes fail closed when the value is missing.
-    origin_token: str = ""
-    environment: Literal["development", "production", "test"] = "production"
-    allowed_cors_origins: str = "http://localhost:8080,http://127.0.0.1:5500"
-
-    admin_password: str = ""
-    session_secret: str = "change-me"
-    maintenance_mode: bool = False
-    maintenance_message: str = "LibBee is receiving an update. Please try again shortly."
-    max_results: int = 5
-    max_question_chars: int = 2000
-    max_history_messages: int = 12
-    max_message_chars: int = 4000
+    openalex_api_key: str = ""
+    contact_email: str = "libse@ku.ac.ae"
+
+    # The Worker sends this secret in X-LibBee-Origin-Token. Production
+    # protected routes fail closed when the value is missing.
+    origin_token: str = ""
+    environment: Literal["development", "production", "test"] = "production"
+    allowed_cors_origins: str = "http://localhost:8080,http://127.0.0.1:5500"
+
+    admin_password: str = ""
+    session_secret: str = "change-me"
+    maintenance_mode: bool = False
+    maintenance_message: str = "LibBee is receiving an update. Please try again shortly."
+    max_results: int = 5
+    max_question_chars: int = 2000
+    max_history_messages: int = 12
+    max_message_chars: int = 4000
     max_concurrent_agent_requests: int = 20
-
-    # /data survives HF Space restarts when persistent storage is attached.
-    runtime_dir: str = "/data/libbee_runtime"
-    feedback_store_path: str = ""
-    metrics_store_path: str = ""
-    config_store_path: str = ""
-    rag_index_dir: str = ""
-    knowledge_dir: str = ""
-
-    # The authenticated admin page receives only a short-lived signed token;
-    # the permanent shared secret is never serialized into browser HTML.
-    cloudflare_worker_url: str = ""
-    cloudflare_admin_token: str = ""
-
-    # HF talks to LibCal directly because some hosted environments cannot open
-    # outbound connections to shared workers.dev hostnames.
-    libcal_client_id: str = ""
-    libcal_client_secret: str = ""
-    libcal_events_client_id: str = ""
-    libcal_events_client_secret: str = ""
-
-    @field_validator("max_results")
-    @classmethod
-    def reasonable_max_results(cls, value: int) -> int:
-        if not 1 <= value <= 20:
-            raise ValueError("max_results must be between 1 and 20")
-        return value
-
-    @field_validator("max_question_chars", "max_message_chars")
-    @classmethod
-    def reasonable_text_limit(cls, value: int) -> int:
-        if not 256 <= value <= 20_000:
-            raise ValueError("text limits must be between 256 and 20,000")
-        return value
-
-    @field_validator("max_history_messages")
-    @classmethod
-    def reasonable_history_limit(cls, value: int) -> int:
-        if not 0 <= value <= 50:
-            raise ValueError("max_history_messages must be between 0 and 50")
+    # Includes the entire JSON envelope and UTF-8 history. The original 32 KiB
+    # ceiling rejected otherwise valid conversations under the model limits.
+    max_request_body_bytes: int = 524_288
+
+    # /data survives HF Space restarts when persistent storage is attached.
+    runtime_dir: str = "/data/libbee_runtime"
+    feedback_store_path: str = ""
+    metrics_store_path: str = ""
+    config_store_path: str = ""
+    rag_index_dir: str = ""
+    knowledge_dir: str = ""
+
+    # The authenticated admin page receives only a short-lived signed token;
+    # the permanent shared secret is never serialized into browser HTML.
+    cloudflare_worker_url: str = ""
+    cloudflare_admin_token: str = ""
+
+    # HF talks to LibCal directly because some hosted environments cannot open
+    # outbound connections to shared workers.dev hostnames.
+    libcal_client_id: str = ""
+    libcal_client_secret: str = ""
+    libcal_events_client_id: str = ""
+    libcal_events_client_secret: str = ""
+
+    @field_validator("max_results")
+    @classmethod
+    def reasonable_max_results(cls, value: int) -> int:
+        if not 1 <= value <= 20:
+            raise ValueError("max_results must be between 1 and 20")
+        return value
+
+    @field_validator("max_question_chars", "max_message_chars")
+    @classmethod
+    def reasonable_text_limit(cls, value: int, info) -> int:
+        ceiling = 2000 if info.field_name == "max_question_chars" else 4000
+        if not 256 <= value <= ceiling:
+            raise ValueError(f"{info.field_name} must be between 256 and {ceiling}")
+        return value
+
+    @field_validator("max_history_messages")
+    @classmethod
+    def reasonable_history_limit(cls, value: int) -> int:
+        if not 0 <= value <= 12:
+            raise ValueError("max_history_messages must be between 0 and 12")
+        return value
+
+    @field_validator("max_concurrent_agent_requests")
+    @classmethod
+    def reasonable_concurrency(cls, value: int) -> int:
+        if not 1 <= value <= 100:
+            raise ValueError("max_concurrent_agent_requests must be between 1 and 100")
         return value
 
-    @field_validator("max_concurrent_agent_requests")
+    @field_validator("max_request_body_bytes")
     @classmethod
-    def reasonable_concurrency(cls, value: int) -> int:
-        if not 1 <= value <= 100:
-            raise ValueError("max_concurrent_agent_requests must be between 1 and 100")
+    def reasonable_body_limit(cls, value: int) -> int:
+        if not 16_384 <= value <= 2_097_152:
+            raise ValueError("max_request_body_bytes must be between 16 KiB and 2 MiB")
         return value
 
     @property
-    def cors_origins(self) -> list[str]:
-        return [value.strip().rstrip("/") for value in self.allowed_cors_origins.split(",") if value.strip()]
-
-    @property
+    def runtime_defaults(self) -> dict:
+        """One source of defaults for admin, public config, and agent replies."""
+        return {
+            "max_results": self.max_results,
+            "maintenance_mode": self.maintenance_mode,
+            "welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.",
+            "custom_instructions": "",
+            "announcement": "",
+            "maintenance_message": self.maintenance_message,
+        }
+
+    @property
+    def cors_origins(self) -> list[str]:
+        return [value.strip().rstrip("/") for value in self.allowed_cors_origins.split(",") if value.strip()]
+
+    @property
     def runtime_path(self) -> Path:
         requested = Path(self.runtime_dir)
-        if requested.parent.exists() and requested.parent.is_dir():
-            return requested
-        return Path("/tmp/libbee_runtime")
-
-    @property
-    def feedback_path(self) -> Path:
-        return Path(self.feedback_store_path) if self.feedback_store_path else self.runtime_path / "feedback.jsonl"
-
-    @property
-    def metrics_path(self) -> Path:
-        return Path(self.metrics_store_path) if self.metrics_store_path else self.runtime_path / "metrics.json"
-
-    @property
-    def config_path(self) -> Path:
-        return Path(self.config_store_path) if self.config_store_path else self.runtime_path / "runtime_config.json"
-
-    @property
-    def rag_cache_dir(self) -> Path:
-        return Path(self.rag_index_dir) if self.rag_index_dir else self.runtime_path / "rag_cache"
-
-    @property
-    def kb_dir(self) -> Path:
-        if self.knowledge_dir:
-            return Path(self.knowledge_dir)
-        if Path("/app/data/knowledge").exists():
-            return Path("/app/data/knowledge")
-        return Path("data/knowledge")
-
-
-@lru_cache
-def get_settings() -> Settings:
-    settings = Settings()
-    settings.runtime_path.mkdir(parents=True, exist_ok=True)
-    settings.rag_cache_dir.mkdir(parents=True, exist_ok=True)
-    return settings
+        # Honour explicit paths, including nested directories not created yet.
+        # Only the absent default HF mount falls back to ephemeral local disk.
+        if self.runtime_dir == "/data/libbee_runtime" and not Path("/data").is_dir():
+            return Path("/tmp/libbee_runtime")
+        return requested
+
+    @property
+    def feedback_path(self) -> Path:
+        return Path(self.feedback_store_path) if self.feedback_store_path else self.runtime_path / "feedback.jsonl"
+
+    @property
+    def metrics_path(self) -> Path:
+        return Path(self.metrics_store_path) if self.metrics_store_path else self.runtime_path / "metrics.json"
+
+    @property
+    def config_path(self) -> Path:
+        return Path(self.config_store_path) if self.config_store_path else self.runtime_path / "runtime_config.json"
+
+    @property
+    def rag_cache_dir(self) -> Path:
+        return Path(self.rag_index_dir) if self.rag_index_dir else self.runtime_path / "rag_cache"
+
+    @property
+    def kb_dir(self) -> Path:
+        if self.knowledge_dir:
+            return Path(self.knowledge_dir)
+        if Path("/app/data/knowledge").exists():
+            return Path("/app/data/knowledge")
+        return Path("data/knowledge")
+
+
+@lru_cache
+def get_settings() -> Settings:
+    settings = Settings()
+    settings.runtime_path.mkdir(parents=True, exist_ok=True)
+    settings.rag_cache_dir.mkdir(parents=True, exist_ok=True)
+    return settings