Fix 21 safety and reliability defects found by audit

Cortex_Preview.py

@@ -11,6 +11,7 @@
 ROOT = Path(__file__).resolve().parent
 sys.path.insert(0, str(ROOT / "backend"))
 
+import httpx  # noqa: E402
 import ollama  # noqa: E402
 import uvicorn  # noqa: E402
 
@@ -84,11 +85,17 @@ def build_preview_app(
     database.migrate_from_json_if_needed()
     permanent_memory = PermanentMemoryManager(app_paths=paths)
     settings_repository = SQLiteSettingsRepository(
-        paths.database,
+        paths.settings_database,
         legacy=LegacySettingsReader(),
+        # Settings used to live inside the chat database; adopt them once so
+        # an upgrade does not silently revert to defaults.
+        adopt_from=paths.database,
     )
     ollama_host = os.environ.get("CORTEX_OLLAMA_HOST", "http://127.0.0.1:11434")
-    client = ollama.Client(host=ollama_host)
+    client = ollama.Client(
+        host=ollama_host,
+        timeout=httpx.Timeout(connect=5.0, read=600.0, write=30.0, pool=5.0),
+    )
 
     def gguf_directory() -> Path:
         # Re-read settings each call (cheap SQLite read, same pattern the API

backend/cortex_backend/api/app.py

@@ -5,6 +5,7 @@
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
+import logging
 import os
 import tempfile
 from pathlib import Path
@@ -42,6 +43,8 @@
 from .routes import build_router
 from .security import SessionManager
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(slots=True)
 class BackendDependencies:
@@ -159,7 +162,16 @@ async def lifespan(app: FastAPI):
             yield
         finally:
             app.state.ready = False
-            await app.state.jobs.shutdown()
+            # Runtime teardown below must run even if job shutdown raises --
+            # it is what actually terminates the llama-server child process,
+            # and a worker that refuses to unwind must not leave that
+            # process (and the GPU/RAM it holds) orphaned.
+            try:
+                await app.state.jobs.shutdown()
+            except Exception:
+                logger.exception(
+                    "Cortex job registry shutdown raised; continuing with runtime teardown."
+                )
             if app.state.execution_lifecycle is not None:
                 app.state.execution_lifecycle.stop()
             elif app.state.execution_coordinator is not None:

backend/cortex_backend/api/jobs.py

@@ -197,13 +197,22 @@ class JobRegistry:
     registry itself remains the single authority for lifecycle transitions.
     """
 
-    def __init__(self, *, poll_seconds: float = 0.025, max_terminal_jobs: int = 100):
+    def __init__(
+        self,
+        *,
+        poll_seconds: float = 0.025,
+        max_terminal_jobs: int = 100,
+        shutdown_grace_seconds: float = 10.0,
+    ):
         if poll_seconds <= 0:
             raise ValueError("poll_seconds must be positive")
         if max_terminal_jobs <= 0:
             raise ValueError("max_terminal_jobs must be positive")
+        if shutdown_grace_seconds <= 0:
+            raise ValueError("shutdown_grace_seconds must be positive")
         self._poll_seconds = poll_seconds
         self._max_terminal_jobs = max_terminal_jobs
+        self._shutdown_grace_seconds = shutdown_grace_seconds
         self._records: dict[str, _JobRecord] = {}
         self._active: dict[JobKind, str] = {}
         self._request_index: dict[tuple[JobKind, str, str], str] = {}
@@ -472,7 +481,23 @@ async def events(
             await asyncio.sleep(self._poll_seconds)
 
     async def shutdown(self) -> None:
-        """Request cancellation and wait for owned workers to finish safely."""
+        """Request cancellation and wait for owned workers to finish safely.
+
+        A worker that has already begun committing its result (see
+        :meth:`JobProgressSink.begin_commit`) is awaited without a bound --
+        that commit must finish so persisted state and the retained event
+        stream stay consistent. A worker that has not committed is only
+        cooperative on a best-effort basis: it may be blocked inside a
+        synchronous call (a model HTTP request with no read deadline, for
+        example) that never polls ``cancel_event``. Waiting on it
+        indefinitely would hang app shutdown -- and the llama-server child
+        process it is talking to -- for as long as that call takes, so the
+        first wait below is capped at ``shutdown_grace_seconds``. Anything
+        still pending after that grace period is re-checked: a worker that
+        committed *during* the grace period still gets the unbounded wait
+        it is owed; a worker that never committed is abandoned so shutdown
+        can proceed with the rest of teardown.
+        """
         with self._lock:
             self._accepting = False
             records = [
@@ -498,12 +523,31 @@ async def shutdown(self) -> None:
                     if self._active.get(record.kind) == record.job_id:
                         self._active.pop(record.kind, None)
             tasks = [record.task for record in self._records.values() if record.task]
-        pending = [task for task in tasks if task is not asyncio.current_task()]
-        if pending:
+        pending = {task for task in tasks if task is not asyncio.current_task()}
+        if not pending:
+            return
+        _, still_pending = await asyncio.wait(pending, timeout=self._shutdown_grace_seconds)
+        if not still_pending:
+            return
+        with self._lock:
+            committed_still_pending = [
+                record.task
+                for record in self._records.values()
+                if record.task in still_pending and record.commit_started
+            ]
+        abandoned = len(still_pending) - len(committed_still_pending)
+        if abandoned:
+            logging.warning(
+                "Cortex shutdown: %d job worker(s) did not observe cancellation within "
+                "%.0fs and were abandoned so shutdown could proceed.",
+                abandoned,
+                self._shutdown_grace_seconds,
+            )
+        if committed_still_pending:
             # Cancelling an asyncio task does not stop its ``to_thread``
             # worker. Waiting for the task lets the worker observe the event,
             # complete its cleanup, and finalize the cancellation itself.
-            await asyncio.gather(*pending, return_exceptions=True)
+            await asyncio.gather(*committed_still_pending, return_exceptions=True)
 
     async def _run(
         self,

backend/cortex_backend/api/routes.py

@@ -200,7 +200,9 @@ def handoff(request: Request) -> HandoffResponse:
         manager.validate_request_context(request)
         supplied = request.headers.get("X-Cortex-Handoff", "")
         expected = request.app.state.handoff_secret
-        if not expected or not hmac.compare_digest(supplied, expected):
+        if not expected or not hmac.compare_digest(
+            supplied.encode("latin-1"), expected.encode("utf-8")
+        ):
             raise HTTPException(status_code=401, detail="Cortex handoff unavailable.")
         token, expires_at = manager.issue_bootstrap_token()
         return HandoffResponse(bootstrap_token=token, expires_at=expires_at)

backend/cortex_backend/api/security.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 import hashlib
 import hmac
@@ -34,6 +34,10 @@ class SessionPrincipal:
     session_id: str
     installation_principal_id: str
     expires_at: datetime
+    # When this session was first issued -- fixed for its whole life, unlike
+    # expires_at, so a sliding renewal has an absolute lifetime to cap
+    # against instead of extending forever under continuous use.
+    issued_at: datetime
 
 
 class SessionManager:
@@ -44,15 +48,19 @@ def __init__(
         *,
         bootstrap_token: str | None = None,
         ttl_seconds: int = 3600,
+        max_lifetime_seconds: int = 24 * 3600,
         allowed_hosts: Iterable[str] = ("127.0.0.1", "localhost", "::1"),
         installation_principal_id: str | None = None,
     ):
         if ttl_seconds < 60:
             raise ValueError("session TTL must be at least 60 seconds")
+        if max_lifetime_seconds < ttl_seconds:
+            raise ValueError("max_lifetime_seconds must be at least ttl_seconds")
         self._bootstrap_token = bootstrap_token or secrets.token_urlsafe(32)
         self._bootstrap_expires_at = datetime.now(timezone.utc) + timedelta(minutes=5)
         self._bootstrap_used = False
         self._ttl = timedelta(seconds=ttl_seconds)
+        self._max_lifetime = timedelta(seconds=max_lifetime_seconds)
         self._allowed_hosts = frozenset(host.lower() for host in allowed_hosts)
         self._installation_principal_id = validate_installation_principal_id(
             installation_principal_id or secrets.token_hex(32)
@@ -89,30 +97,43 @@ def installation_principal_id(self) -> str:
     def exchange(self, bootstrap_token: str) -> SessionExchange:
         with self._lock:
             if self._bootstrap_used or self._bootstrap_expires_at <= datetime.now(timezone.utc) or not hmac.compare_digest(
-                bootstrap_token,
-                self._bootstrap_token,
+                bootstrap_token.encode("utf-8"),
+                self._bootstrap_token.encode("utf-8"),
             ):
                 raise SessionSecurityError("invalid bootstrap token")
             self._bootstrap_used = True
             raw_token = secrets.token_urlsafe(32)
-            expires_at = datetime.now(timezone.utc) + self._ttl
+            issued_at = datetime.now(timezone.utc)
             session_id = secrets.token_urlsafe(16)
-            self._sessions[self._digest(raw_token)] = SessionPrincipal(
+            principal = SessionPrincipal(
                 session_id=session_id,
                 installation_principal_id=self._installation_principal_id,
-                expires_at=expires_at,
-            )
-            return SessionExchange(
-                token=raw_token,
-                principal=self._sessions[self._digest(raw_token)],
+                expires_at=issued_at + self._ttl,
+                issued_at=issued_at,
             )
+            self._sessions[self._digest(raw_token)] = principal
+            return SessionExchange(token=raw_token, principal=principal)
 
     def authenticate(self, token: str) -> SessionPrincipal:
+        """Validate a bearer token and slide its expiry forward.
+
+        A session in continuous use must never expire mid-session -- every
+        successful authenticate() extends expires_at by the full TTL again,
+        capped at issued_at + max_lifetime so a session cannot renew itself
+        forever. A session that goes genuinely idle for longer than the TTL
+        still expires normally, since nothing calls authenticate() to renew
+        it while idle.
+        """
         now = datetime.now(timezone.utc)
+        digest = self._digest(token)
         with self._lock:
-            principal = self._sessions.get(self._digest(token))
+            principal = self._sessions.get(digest)
             if principal is None or principal.expires_at <= now:
                 raise SessionSecurityError("invalid or expired session")
+            renewed_expiry = min(now + self._ttl, principal.issued_at + self._max_lifetime)
+            if renewed_expiry > principal.expires_at:
+                principal = replace(principal, expires_at=renewed_expiry)
+                self._sessions[digest] = principal
             return principal
 
     def validate_request_context(self, request: Request) -> None:

backend/cortex_backend/core/paths.py

@@ -79,6 +79,17 @@ def permanent_memory(self) -> Path:
     def permanent_memory_backup(self) -> Path:
         return self.data_dir / "memory_bank.json.bak"
 
+    @property
+    def settings_database(self) -> Path:
+        """Settings kept out of the chat database.
+
+        Settings writes take a full-file backup copy first. Colocating them
+        with chat history meant every settings save byte-copied the entire
+        transcript store -- slow, disk-doubling, and able to fail a theme
+        toggle outright once the chat database grew large.
+        """
+        return self.data_dir / "cortex_settings.sqlite"
+
     @property
     def vector_database(self) -> Path:
         """Retain the dormant legacy path without enabling vector memory."""