server.py

import asyncio
import logging
import os
from typing import Any
from urllib.parse import urlparse

# Increase file descriptor limit for macOS/Linux to prevent
# [Errno 24] Too many open files. The 'resource' module is POSIX-only,
# so the import is inside the try/except - on Windows the ImportError is
# swallowed and we skip this block (Windows uses a different I/O model
# and does not have RLIMIT_NOFILE).
try:
    import resource
    soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
    # Target 4096 or the system hard limit, whichever is smaller
    target = min(4096, hard) if hard > 0 else 4096
    if soft < target:
        resource.setrlimit(resource.RLIMIT_NOFILE, (target, hard))
except Exception:
    pass

# Fix SSL certificate verification for Python 3.13 + macOS Homebrew
import ssl as _ssl

_ca_file = _ssl.get_default_verify_paths().cafile
if _ca_file and os.path.isfile(_ca_file):
    os.environ.setdefault('SSL_CERT_FILE', _ca_file)
    os.environ.setdefault('REQUESTS_CA_BUNDLE', _ca_file)

from dotenv import load_dotenv
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

# Load Environment Variables from user_data/
_ENV_DEMO_MODE_REQUESTED = os.environ.get("DEMO_MODE")
_ENV_FORCE_DEMO_REQUESTED = os.environ.get("CAIRNIQ_FORCE_DEMO")
env_path = os.path.join(os.getcwd(), "user_data", ".env")
load_dotenv(env_path, override=True)

# Move any plaintext secrets out of .env into the OS keychain, then hydrate
# os.environ with whatever the keychain holds. After this point the rest of
# the codebase can keep reading os.environ.get("OPENAI_API_KEY") etc.
from tools.secrets_store import (  # noqa: E402
    clear_incompatible_aws_session_token,
    keyring_status,
    load_secrets_into_env,
    migrate_env_to_keyring,
)

try:
    _ks = keyring_status()
    _migration = migrate_env_to_keyring(env_path)
    _n_loaded = load_secrets_into_env()
    # Strip any stale AWS_SESSION_TOKEN that's incompatible with the long-term
    # IAM user key we just loaded — see secrets_store.clear_incompatible_aws_session_token.
    _aws_token_cleanup = clear_incompatible_aws_session_token()

    if not _ks["available"]:
        # Headless Linux / Docker / no-backend Mac — the app still works from .env
        # via the fallback path, but flag it so users know secrets aren't encrypted.
        from agent.logger import log_to_component
        log_to_component(
            "server",
            "secrets",
            f"keychain unavailable ({_ks['reason']}); using plaintext .env fallback",
            level=logging.WARNING,
        )
    elif _migration.get("migrated") or _n_loaded:
        # Only emit a startup line when something actually happened, to keep logs quiet.
        _summary = []
        if _migration.get("migrated"):
            _summary.append(f"migrated {len(_migration['migrated'])} from .env")
        if _n_loaded:
            _summary.append(f"hydrated {_n_loaded} from keychain")
        if _aws_token_cleanup["cleared"]:
            _summary.append("stripped stale AWS_SESSION_TOKEN")
        from agent.logger import log_to_component
        log_to_component("server", "secrets", f"{', '.join(_summary)}")
except Exception as _secrets_err:  # noqa: BLE001 — never block server startup on this
    from agent.logger import log_to_component
    log_to_component("server", "secrets", f"non-fatal init issue: {_secrets_err}", level=logging.ERROR)

if str(_ENV_DEMO_MODE_REQUESTED or "").strip().lower() in {"1", "true", "yes", "y", "on"}:
    os.environ["DEMO_MODE"] = "true"
if str(_ENV_FORCE_DEMO_REQUESTED or "").strip().lower() in {"1", "true", "yes", "y", "on"}:
    os.environ["CAIRNIQ_FORCE_DEMO"] = "true"
    os.environ["DEMO_MODE"] = "true"

from contextlib import asynccontextmanager

from agent.graph import build_graph
from agent.logger import bind_log_context, log_to_component, reset_log_context
from api.dependencies import get_connection_manager, set_agent
from api.routers import chat, dashboard, memory, news, pages, portfolio, settings
from tools.user_profile import (
    ensure_demo_profile,
    get_demo_profile_name,
    is_demo_mode,
    is_known_profile,
    normalize_profile_name,
    reset_profile,
    set_active_profile,
)

if is_demo_mode():
    os.environ["DEMO_MODE"] = "true"
    os.environ["ACTIVE_PROFILE"] = get_demo_profile_name()
    os.environ["QUESTRADE_ENABLED"] = "false"
    os.environ["ALPACA_PAPER_MODE"] = "true"
    ensure_demo_profile()

def check_requirements():
    import importlib.metadata
    import re

    req_file = os.path.join(os.getcwd(), "requirements.txt")
    if not os.path.exists(req_file):
        return

    try:
        with open(req_file) as f:
            lines = f.readlines()

        mismatched = []
        for line in lines:
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            match = re.match(r"^([a-zA-Z0-9_\-]+)", line)
            if match:
                pkg_name = match.group(1)
                try:
                    importlib.metadata.version(pkg_name)
                except importlib.metadata.PackageNotFoundError:
                    mismatched.append(pkg_name)

        if mismatched:
            print(
                f"\n\n======================================================================\n"
                f"🚨 WARNING: The following dependencies in requirements.txt are missing:\n"
                f"   {', '.join(mismatched)}\n\n"
                f"   Please run './install.sh' to update your environment!\n"
                f"======================================================================\n"
            )
            log_to_component(
                "server",
                "Startup",
                f"Missing dependencies in virtual environment: {', '.join(mismatched)}",
                level=logging.WARNING
            )
    except Exception as e:
        log_to_component("server", "Startup", f"Failed to verify requirements: {e}", level=logging.WARNING)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Check for missing dependencies on startup
    check_requirements()

    # Register LiteLLM callback to track DSPy LLM costs
    try:
        import litellm
        from litellm.integrations.custom_logger import CustomLogger

        from agent.cost_tracker import accumulate_cost as _acc_cost

        class _CostCallback(CustomLogger):
            def log_success_event(self, kwargs, response_obj, start_time, end_time):
                try:
                    usage = getattr(response_obj, "usage", None)
                    if usage is None and isinstance(response_obj, dict):
                        usage = response_obj.get("usage")
                    if usage:
                        inp = getattr(usage, "prompt_tokens", 0) or (usage.get("prompt_tokens", 0) if isinstance(usage, dict) else 0)
                        out = getattr(usage, "completion_tokens", 0) or (usage.get("completion_tokens", 0) if isinstance(usage, dict) else 0)
                        cache_read = 0
                        cache_detail = getattr(usage, "prompt_tokens_details", None)
                        if cache_detail:
                            cache_read = getattr(cache_detail, "cached_tokens", 0) or 0
                        elif isinstance(usage, dict):
                            cd = usage.get("prompt_tokens_details") or {}
                            cache_read = cd.get("cached_tokens", 0) or 0
                        model = kwargs.get("model", "") or ""
                        if inp or out:
                            _acc_cost(inp, out, model, cache_read_tokens=cache_read)
                except Exception as e:
                    log_to_component("server", "Error", f"Error in callback parsing: {e}", level=logging.ERROR)

        litellm.callbacks.append(_CostCallback())
    except Exception as e:
        log_to_component("server", "Startup", f"Error registering callback: {e}", level=logging.WARNING)

    log_to_component("server", "Startup", "Initializing LangGraph agent...")
    try:
        new_agent = build_graph(use_memory=True)
        set_agent(new_agent)
        log_to_component("server", "Startup", "Agent initialized successfully.")
    except Exception as e:
        log_to_component("server", "Startup", f"FAILED to initialize agent: {e}", level=logging.ERROR)

    # Clean old cache files at startup
    try:
        from tools.daily_cache import cleanup_old
        removed = cleanup_old(7)
        if removed:
            log_to_component("server", "Startup", f"Cleaned {removed} old cache files")
    except Exception as e:
        log_to_component("server", "Startup", f"Error cleaning old cache files: {e}", level=logging.WARNING)

    # Seed user_data/funnel_config.json from the example on first run (cross-platform;
    # covers manual venv installs that don't run install.ps1). Fail-safe — the scanner
    # uses built-in defaults if anything goes wrong.
    try:
        from tools.opportunity_scanner import seed_funnel_config_if_missing
        if seed_funnel_config_if_missing():
            log_to_component("server", "Startup", "Seeded user_data/funnel_config.json from example")
    except Exception as e:
        log_to_component("server", "Startup", f"Error seeding funnel config: {e}", level=logging.WARNING)

    # Fetch live USD_TO_CAD exchange rate asynchronously to not block startup
    async def fetch_exchange_rate():
        try:
            import yfinance as yf

            from tools.daily_cache import set_cached
            def _fetch():
                ticker = yf.Ticker("USDCAD=X")
                hist = ticker.history(period="5d", timeout=40)
                if not hist.empty:
                    return float(hist["Close"].iloc[-1])
                return None

            rate = await asyncio.to_thread(_fetch)
            if rate:
                os.environ["USD_TO_CAD"] = str(rate)
                set_cached("usd_cad_rate", rate)
                log_to_component("server", "Startup", f"Injected live USD/CAD rate: {rate:.4f}")
        except Exception as e:
            log_to_component("server", "Startup", f"Failed to fetch live USD/CAD rate: {e}", level=logging.WARNING)

    asyncio.create_task(fetch_exchange_rate())

    yield
    # No teardown logic specified

app = FastAPI(title="CairnIQ API", lifespan=lifespan)


def _configured_allowed_origins() -> list[str]:
    """Return browser origins allowed to send credentialed requests."""
    raw = os.environ.get("CAIRNIQ_ALLOWED_ORIGINS") or os.environ.get("CORS_ALLOW_ORIGINS")
    if raw:
        origins = [origin.strip().rstrip("/") for origin in raw.split(",") if origin.strip()]
    else:
        origins = [
            "http://localhost:8000",
            "http://127.0.0.1:8000",
        ]
    return [origin for origin in origins if origin != "*"]


ALLOWED_ORIGINS = _configured_allowed_origins()
_ALLOWED_ORIGIN_SET = set(ALLOWED_ORIGINS)
_MUTATING_METHODS = {"POST", "PUT", "PATCH", "DELETE"}


def _origin_from_url(value: str) -> str | None:
    parsed = urlparse(value)
    if not parsed.scheme or not parsed.netloc:
        return None
    return f"{parsed.scheme}://{parsed.netloc}".rstrip("/")


def _is_allowed_origin(origin: str | None) -> bool:
    if not origin:
        return True
    return origin.rstrip("/") in _ALLOWED_ORIGIN_SET


def _resolve_request_profile(cookie_profile: str | None, env_profile: str | None) -> str:
    if is_demo_mode():
        return get_demo_profile_name()

    for raw_profile in (cookie_profile, env_profile):
        raw_name = str(raw_profile or "").strip()
        if not raw_name:
            continue
        profile = normalize_profile_name(raw_profile)
        if profile == "default" and raw_name != "default":
            continue
        if is_known_profile(profile):
            return profile
    return "default"

# CORS configuration
app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def no_cache_html_middleware(request: Request, call_next):
    """Tell browsers never to cache HTML page responses.

    Without this, Edge (and Chrome on return visits) serve stale HTML from
    the disk cache, which means users run old embedded JavaScript even after
    the server has been updated.  API/JSON responses and static assets are
    unaffected — only text/html responses get the no-cache directive.
    """
    response = await call_next(request)
    ct = response.headers.get("content-type", "")
    if "text/html" in ct:
        response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
        response.headers["Pragma"] = "no-cache"
        response.headers["Expires"] = "0"
    return response

@app.middleware("http")
async def local_origin_middleware(request: Request, call_next):
    """Reject credentialed browser writes from untrusted origins."""
    if request.method.upper() in _MUTATING_METHODS:
        origin = request.headers.get("origin")
        referer_origin = _origin_from_url(request.headers.get("referer", ""))
        if not _is_allowed_origin(origin) or (referer_origin and not _is_allowed_origin(referer_origin)):
            return JSONResponse({"error": "Origin not allowed"}, status_code=403)
    return await call_next(request)

@app.middleware("http")
async def profile_middleware(request: Request, call_next):
    if is_demo_mode():
        profile_to_use = get_demo_profile_name()
        token = set_active_profile(profile_to_use)
        try:
            response = await call_next(request)
            return response
        finally:
            reset_profile(token)

    # Try to get profile from cookie, fallback to env var ACTIVE_PROFILE, fallback to 'default'
    cookie_profile = request.cookies.get("profile")
    env_profile = os.environ.get("ACTIVE_PROFILE")

    # Priority: valid cookie > valid environment variable > 'default'
    profile_to_use = _resolve_request_profile(cookie_profile, env_profile)

    token = set_active_profile(profile_to_use)
    try:
        response = await call_next(request)
        return response
    finally:
        reset_profile(token)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    manager = get_connection_manager()
    await manager.connect(websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.disconnect(websocket)
    except Exception as e:
        log_to_component("server", "WebSocket", f"WebSocket error: {e}", level=logging.WARNING)
        manager.disconnect(websocket)

class LogMessage(BaseModel):
    level: str
    phase: str
    message: str
    data: dict[str, Any] | None = None

@app.post("/api/logs/frontend")
async def log_from_frontend(payload_obj: LogMessage, request: Request):
    log_context_token = None
    try:
        payload = payload_obj.model_dump()
        data = payload.get("data") or {}

        log_context_token = bind_log_context(
            user_ip=request.client.host if request.client else "unknown",
            user_agent=request.headers.get("user-agent", "unknown")
        )
        level_map = {
            "INFO": logging.INFO,
            "WARNING": logging.WARNING,
            "ERROR": logging.ERROR,
            "CRITICAL": logging.CRITICAL
        }
        level = level_map.get(payload.get("level", "INFO"), logging.INFO)

        log_to_component(
            component="frontend",
            phase=payload.get("phase", "UI"),
            message=payload.get("message", "Browser event"),
            data=data,
            level=level
        )
        return {"status": "ok"}
    except Exception as e:
        log_to_component("server", "Error", f"Error logging frontend event: {e}", level=logging.ERROR)
        return {"status": "error", "message": "Failed to log frontend event"}
    finally:
        if log_context_token is not None:
            reset_log_context(log_context_token)

# Custom StaticFiles to disable caching during development
class NoCacheStaticFiles(StaticFiles):
    def is_not_modified(self, response_headers, request_headers) -> bool:
        return False

    async def get_response(self, path: str, scope):
        response = await super().get_response(path, scope)
        response.headers["Cache-Control"] = "no-store, no-cache, must-revalidate, max-age=0"
        response.headers["Pragma"] = "no-cache"
        response.headers["Expires"] = "0"
        return response

# Mount static files (disable caching in dev mode for hot-reload)
_DEV_MODE = os.environ.get("LOG_LEVEL", "INFO").upper() == "DEBUG" or os.environ.get("DEV_MODE", "").lower() in ("1", "true")
_StaticFilesClass = NoCacheStaticFiles if _DEV_MODE else StaticFiles
app.mount("/static", _StaticFilesClass(directory="static"), name="static")

# Include Routers
app.include_router(pages.router)
app.include_router(chat.router)
app.include_router(memory.router)
app.include_router(portfolio.router)
app.include_router(dashboard.router)
app.include_router(settings.router)
app.include_router(news.router)

if __name__ == "__main__":
    import uvicorn

    class FilterFrontendLogs(logging.Filter):
        def filter(self, record):
            return "/api/logs/frontend" not in record.getMessage()

    logging.getLogger("uvicorn.access").addFilter(FilterFrontendLogs())

    uvicorn.run(
        "server:app",
        host=os.environ.get("CAIRNIQ_HOST", "127.0.0.1"),
        port=int(os.environ.get("PORT", "8000")),
        reload=True,
    )