"""Configuration, read once from the environment. Values come from ~/ai/api.env on lin001 (0600, not in Git). .env.example in the repo root lists every key with no values. Nothing here has a secret default, and nothing here is ever logged. """ from __future__ import annotations import os from functools import lru_cache from pydantic import BaseModel class Settings(BaseModel): # --- imh (pending) ----------------------------------------------------- use_fixtures: bool = True # --- local ------------------------------------------------------------- pghost: str = "pg-ai" pgport: int = 5432 pgdatabase: str = "plant" pguser: str = "agent_ro" pgpassword: str = "" # --- Azure OpenAI ------------------------------------------------------ azure_openai_endpoint: str = "" azure_openai_api_key: str = "" azure_openai_api_version: str = "" chat_deployment: str = "" # flagship - final prose only cheap_deployment: str = "" # classifier, entities, tool selection embed_deployment: str = "" # text-embedding-3-small # --- no-LLM stub mode -------------------------------------------------- # Off by default and must stay that way. See api/stub.py for what it does # and, more importantly, what it does not prove. no_llm_stub: bool = False # --- behaviour --------------------------------------------------------- classifier_confidence_threshold: float = 0.7 site_timezone: str = "Australia/Sydney" # Where the CONTROLLED copy of a procedure actually lives. A site fact, not # a document fact - it is the same for every document, so it is not in # doc_chunks. It is here rather than left to the model because sending an # operator to a controlled copy that does not exist is worse than telling # them nothing. The default says who to ask, which is always true. controlled_copy_location: str = ( "Ask the WRPS document controller - this assistant does not hold " "controlled copies." ) max_rows_returned: int = 5000 query_timeout_seconds: int = 30 max_output_tokens: int = 1200 # --- Documents (Phase 9) ----------------------------------------------- # Two extra roles, and the split is the safety boundary, not bookkeeping. # # uploads_rw the queue, and withdraw. db/005 carries a trigger that # refuses to let this role set superseded = FALSE, so it can # make a document LESS visible and never more. # ingest_rw writes doc_chunks: approve and restore. Anything that makes # a document citable uses this one. # # In the design these are two components - ai-api and ai-docs-worker, the # second with no HTTP surface. There is no worker here, so both connections # live in this process. That is a real deviation and it is recorded in # BUILD-AI-CONTAINERS.md S14: the trigger still blocks the web role, but the # process holding uploads_rw also holds ingest_rw, so the separation is now # a code boundary rather than a deployment one. Splitting the worker out # later is a config change and a compose file, not a redesign. uploads_db_user: str = "uploads_rw" uploads_db_password: str = "" ingest_db_user: str = "ingest_rw" ingest_db_password: str = "" # Where uploaded files land before review. Never inside /datadisk/ai-docs: # that folder means "the documents this plant runs on", and an unreviewed # upload is not one of those. docs_inbox: str = "/inbox" max_upload_mb: int = 25 # --- Document identity ------------------------------------------------- # "authelia" - the actor is Remote-User from the forward-auth headers, and # a missing header is a 401. This is the design. # "demo" - the actor is TYPED BY THE PERSON on the form. Self-asserted, # unverified, and exactly what the design forbids ("identity # comes from the headers, never from the request body"). # # Demo mode is off unless asked for, the screens carry a banner saying the # name is unverified, and every row it writes is stored as `demo:` # with actor_groups = 'DEMO-UNVERIFIED'. That marking is the point: when # real auth goes on, a self-asserted audit row must still be tellable from # an authenticated one. Without it they are indistinguishable forever. doc_identity_mode: str = "authelia" # Who may change anything. In the design this is the AD group # AI_DocPublishers; with no AD group available it is a name list, swapped # for the group later with one config change. Empty means nobody, and the # API fails closed - approve, withdraw and restore all 403. doc_publishers: tuple[str, ...] = () # --- Cube -------------------------------------------------------------- cubejs_api_url: str = "http://cube:4000/cubejs-api/v1" cubejs_api_secret: str = "" # --- Langfuse ---------------------------------------------------------- langfuse_host: str = "http://langfuse:3000" langfuse_public_key: str = "" langfuse_secret_key: str = "" def dsn(self) -> str: """Postgres DSN. Never log the result - it carries the password.""" return ( f"postgresql://{self.pguser}:{self.pgpassword}" f"@{self.pghost}:{self.pgport}/{self.pgdatabase}" ) def docs_dsn(self, role: str) -> str: """DSN for one of the document roles. Never log the result. `role` is "uploads" or "ingest" - spelled out at every call site rather than defaulted, because picking the wrong one is the difference between a web request that can withdraw a document and one that can publish it. """ user, password = { "uploads": (self.uploads_db_user, self.uploads_db_password), "ingest": (self.ingest_db_user, self.ingest_db_password), }[role] return ( f"postgresql://{user}:{password}" f"@{self.pghost}:{self.pgport}/{self.pgdatabase}" ) def redacted(self) -> dict[str, object]: """Safe to log and safe to return from /healthz.""" secret = {"pgpassword", "azure_openai_api_key", "cubejs_api_secret", "langfuse_secret_key", "uploads_db_password", "ingest_db_password"} return { k: ("set" if v else "unset") if k in secret else v for k, v in self.model_dump().items() } @lru_cache def settings() -> Settings: env = os.environ return Settings( use_fixtures=env.get("USE_FIXTURES", "true").lower() == "true", pghost=env.get("PGHOST", "pg-ai"), pgport=int(env.get("PGPORT", "5432")), pgdatabase=env.get("PGDATABASE", "plant"), pguser=env.get("PGUSER", "agent_ro"), pgpassword=env.get("PGPASSWORD", ""), no_llm_stub=env.get("NO_LLM_STUB", "false").lower() == "true", azure_openai_endpoint=env.get("AZURE_OPENAI_ENDPOINT", ""), azure_openai_api_key=env.get("AZURE_OPENAI_API_KEY", ""), azure_openai_api_version=env.get("AZURE_OPENAI_API_VERSION", ""), chat_deployment=env.get("CHAT_DEPLOYMENT", ""), cheap_deployment=env.get("CHEAP_DEPLOYMENT", ""), embed_deployment=env.get("EMBED_DEPLOYMENT", ""), classifier_confidence_threshold=float( env.get("CLASSIFIER_CONFIDENCE_THRESHOLD", "0.7") ), site_timezone=env.get("SITE_TIMEZONE", "Australia/Sydney"), controlled_copy_location=env.get( "CONTROLLED_COPY_LOCATION", "Ask the WRPS document controller - this assistant does not hold " "controlled copies.", ), max_rows_returned=int(env.get("MAX_ROWS_RETURNED", "5000")), query_timeout_seconds=int(env.get("QUERY_TIMEOUT_SECONDS", "30")), max_output_tokens=int(env.get("MAX_OUTPUT_TOKENS", "1200")), uploads_db_user=env.get("UPLOADS_DB_USER", "uploads_rw"), uploads_db_password=env.get("UPLOADS_DB_PASSWORD", ""), ingest_db_user=env.get("INGEST_DB_USER", "ingest_rw"), ingest_db_password=env.get("INGEST_DB_PASSWORD", ""), docs_inbox=env.get("DOCS_INBOX", "/inbox"), max_upload_mb=int(env.get("MAX_UPLOAD_MB", "25")), doc_identity_mode=env.get("DOC_IDENTITY_MODE", "authelia").lower(), # Comma-separated. Blank entries dropped so a trailing comma in an env # file cannot silently authorise "". doc_publishers=tuple( n.strip() for n in env.get("DOC_PUBLISHERS", "").split(",") if n.strip() ), cubejs_api_url=env.get("CUBEJS_API_URL", "http://cube:4000/cubejs-api/v1"), cubejs_api_secret=env.get("CUBEJS_API_SECRET", ""), langfuse_host=env.get("LANGFUSE_HOST", "http://langfuse:3000"), langfuse_public_key=env.get("LANGFUSE_PUBLIC_KEY", ""), langfuse_secret_key=env.get("LANGFUSE_SECRET_KEY", ""), )