WIP [draft]: Admin Layer PR A — the rewrite (create)

dashboard/backend/api/routers/admin_analytics.py

@@ -15,7 +15,6 @@
     AnalyticsActivityPage,
     AnalyticsOverview,
     AnalyticsQueryService,
-    AnalyticsUserFilters,
     get_analytics_query_service,
     get_value_analytics_query_service,
 )
@@ -51,8 +50,6 @@
 _MODEL_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/\-:]{0,255}$")
 _POSITIVE_INTEGER_PATTERN = re.compile(r"^[0-9]+$")
 _USER_STATES = {"blocked", "needs_attention", "dormant", "onboarding", "active"}
-_USER_SORTS = {"last_activity", "joined_at", "recent_runs", "recent_failures"}
-_SORT_ORDERS = {"asc", "desc"}
 _ACTIVITY_SECTIONS = {"timeline", "runs", "usage", "sessions"}
 _LIFECYCLE_SEGMENTS = {"new", "onboarding", "growing", "core", "at_risk", "dormant"}
 _OPERATIONAL_STATES = {"blocked", "needs_attention", "healthy"}
@@ -295,74 +292,6 @@ def _value_user_filters(request: Request) -> tuple[UserValueFilters, int, int]:
     return filters, limit, offset
 
 
-def _user_filters(request: Request) -> tuple[AnalyticsUserFilters, int, int]:
-    values = _query_values(
-        request,
-        {
-            "q",
-            "status",
-            "last_activity_from",
-            "last_activity_to",
-            "sort",
-            "order",
-            "limit",
-            "offset",
-            "include_internal",
-        },
-    )
-    query = values.get("q")
-    if query is not None and len(query) > 100:
-        _invalid_query()
-    status = values.get("status")
-    if status is not None and status not in _USER_STATES:
-        _invalid_query()
-    sort = values.get("sort", "last_activity")
-    if sort not in _USER_SORTS:
-        _invalid_query()
-    order = values.get("order", "desc")
-    if order not in _SORT_ORDERS:
-        _invalid_query()
-
-    from_date = (
-        _parse_date(values["last_activity_from"])
-        if "last_activity_from" in values
-        else None
-    )
-    to_date = (
-        _parse_date(values["last_activity_to"])
-        if "last_activity_to" in values
-        else None
-    )
-    if from_date is not None and to_date is not None and to_date < from_date:
-        _invalid_query()
-    activity_start = _utc_midnight(from_date) if from_date else None
-    activity_end = (
-        _exclusive_date_end(to_date) - timedelta(microseconds=1)
-        if to_date
-        else None
-    )
-
-    try:
-        filters = AnalyticsUserFilters(
-            q=query,
-            status=status,
-            last_activity_from=activity_start,
-            last_activity_to=activity_end,
-            sort=sort,
-            order=order,
-            include_internal=(
-                _parse_bool(values["include_internal"])
-                if "include_internal" in values
-                else False
-            ),
-        )
-    except (ValidationError, ValueError):
-        _invalid_query()
-    limit = _parse_integer(values.get("limit", "50"), minimum=1, maximum=100)
-    offset = _parse_integer(values.get("offset", "0"), minimum=0)
-    return filters, limit, offset
-
-
 def _activity_query(request: Request) -> tuple[str, int, str | None]:
     values = _query_values(request, {"section", "limit", "cursor"})
     section = values.get("section")
@@ -380,7 +309,12 @@ def _raise_service_error(exc: Exception) -> Never:
         raise HTTPException(status_code=404, detail=_NOT_FOUND_DETAIL) from None
     if isinstance(exc, (ValidationError, ValueError)):
         raise HTTPException(status_code=422, detail=_INVALID_QUERY_DETAIL) from None
-    raise HTTPException(status_code=503, detail=_UNAVAILABLE_DETAIL) from None
+    # Category only, never the message: a psycopg OperationalError carries the
+    # DSN and a ValidationError carries field values. The class name is what
+    # tells "the pool is exhausted" from "a bad SQL statement" in prod logs,
+    # which the bare `from None` 503 never could (design D24, SS4.4).
+    print(f"ERROR: admin_analytics.unhandled category={type(exc).__name__[:80]}")
+    raise HTTPException(status_code=503, detail=_UNAVAILABLE_DETAIL) from exc
 
 
 def _record_access(

dashboard/backend/api/routers/backtests.py

@@ -1576,6 +1576,7 @@ def run_backtest_background(
     # decoding it at timeout time would fail even if we had the key. Used only
     # to decide whether a timeout has a Credits cost worth reporting.
     billing_mode: Optional[str] = None,
+    owner_user_id: Optional[int] = None,
 ):
     """Run backtest in background thread.
 
@@ -1731,6 +1732,8 @@ def run_backtest_background(
             # how that gap becomes its measured `starting` phase.
             "--launched-at", f"{launched_at:.3f}",
         ]
+        if owner_user_id is not None:
+            cmd += ["--owner-user-id", str(int(owner_user_id))]
 
         # Simulation capital is independent of the agent's portfolio sleeve.
         cmd += ["--initial-capital", str(resolve_initial_capital(initial_capital))]
@@ -3522,6 +3525,10 @@ def run_backtest_endpoint(
                 if execution_handoff_payload is not None and billing_mode is not None
                 else None
             ),
+            # The caller's OWN account, the same user_id _backtest_owner_key
+            # bills the slot to -- never the session the results file under,
+            # which for a built-in agent is the agent's, not the caller's.
+            "owner_user_id": user_id,
             **({"universe_selection": universe_selection} if universe_selection is not None else {}),
         },
         daemon=True

dashboard/backend/app.py

@@ -314,18 +314,6 @@ def bar_cache_background():
     except Exception as e:
         print(f"⚠️ legacy session sweep registration error: {e}")
 
-    try:
-        from dashboard.backend.domain.analytics.retention import (
-            analytics_retention_coordinator,
-        )
-        from dashboard.backend.domain.runs.service import register_reaper_sweep
-        register_reaper_sweep(analytics_retention_coordinator.run_if_due)
-        print("🧹 Analytics retention sweep registered with the reaper")
-    except Exception as e:
-        print(
-            "WARNING: analytics.retention_registration_failed "
-            f"category={type(e).__name__}"
-        )
 
     try:
         from dashboard.backend.domain.analytics.maintenance import (
@@ -360,6 +348,25 @@ def bar_cache_background():
             f"category={type(e).__name__}"
         )
 
+    try:
+        # Admin layer redesign PR A (design D23, SS6.9): the daily-facts job
+        # runs on its own thread, not as a reaper sweep -- a whole-population
+        # batch across three databases on the heartbeat thread would let a
+        # slow analytics night mark live runs as orphaned. The worker also
+        # owns rollup_day and the retention coordinator now, and runs the
+        # idempotent user_activity seed + history copy once before ticking.
+        from dashboard.backend.domain.analytics.daily_job import (
+            start_daily_facts_worker,
+        )
+        from dashboard.backend.domain.analytics.facts_migration import (
+            run_startup_migrations,
+        )
+        run_startup_migrations()
+        start_daily_facts_worker()
+        print("📊 Analytics daily-facts worker started")
+    except Exception as e:
+        print(f"⚠️ Analytics daily-facts worker start error: {e}")
+
     try:
         from dashboard.backend.domain.runs.service import start_reaper
         start_reaper()

dashboard/backend/database.py

@@ -10,6 +10,7 @@
 import json
 import os
 from pathlib import Path
+from datetime import date, timedelta
 from typing import List, Dict, Optional, Any
 
 from dashboard.backend.paths import DEFAULT_DB_PATH
@@ -146,6 +147,7 @@ def _init_schema(self):
                 output_tokens INTEGER DEFAULT 0,
                 est_cost_usd REAL DEFAULT 0,
                 metadata TEXT,
+                owner_user_id INTEGER,
                 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                 updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
             )
@@ -369,6 +371,11 @@ def _migrate_schema(self):
                  "ALTER TABLE agent_runs ADD COLUMN est_cost_usd REAL DEFAULT 0"),
                 ("metadata",
                  "ALTER TABLE agent_runs ADD COLUMN metadata TEXT"),
+                # Analytics attribution (design §6.4): the authenticated caller
+                # who started a dashboard backtest. Nullable and never
+                # backfilled -- scheduled leaderboard deploys have no caller.
+                ("owner_user_id",
+                 "ALTER TABLE agent_runs ADD COLUMN owner_user_id INTEGER"),
             ]
             for col_name, add_column_sql in token_columns:
                 if col_name not in columns:
@@ -642,7 +649,8 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
                    input_tokens: int = 0,
                    output_tokens: int = 0,
                    est_cost_usd: float = 0.0,
-                   metadata: Optional[Dict[str, Any]] = None) -> None:
+                   metadata: Optional[Dict[str, Any]] = None,
+                   owner_user_id: Optional[int] = None) -> None:
         """Insert a new backtest run with session_id, LLM model and token-cost tracking.
 
         ``llm_calls`` and ``llm_decisions`` are not two spellings of one number:
@@ -662,14 +670,15 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
              initial_equity, final_equity, total_return, sharpe_ratio,
              max_drawdown, num_trades, llm_model,
              llm_calls, llm_decisions, input_tokens, output_tokens,
-             est_cost_usd, metadata)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+             est_cost_usd, metadata, owner_user_id)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """, (run_id, session_id, agent_name, mode, start_date, end_date,
               initial_equity, final_equity, total_return, sharpe_ratio,
               max_drawdown, num_trades, llm_model,
               llm_calls, llm_decisions, input_tokens, output_tokens,
               est_cost_usd,
-              json.dumps(metadata) if metadata is not None else None))
+              json.dumps(metadata) if metadata is not None else None,
+              owner_user_id))
 
         conn.commit()
         conn.close()
@@ -920,6 +929,42 @@ def get_runs_by_mode(self, mode: str) -> List[Dict]:
 
         return [self._parse_run_row(dict(row)) for row in rows]
 
+    def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
+        """Operator-funded model cost per owner for runs updated on ``day``, in micro-USD.
+
+        The daily job's run step (design SS6.9 step 3): one statement grouped
+        by ``owner_user_id``, parameterised by two day bounds and nothing else.
+        Rows with a NULL owner -- everything before Task 2's column, and every
+        scheduled leaderboard deploy -- are skipped rather than attributed to
+        anyone. ``updated_at`` is CURRENT_TIMESTAMP text
+        ("YYYY-MM-DD HH:MM:SS", UTC) on both twins, so the bounds are text of
+        the same shape and compare correctly.
+
+        ``est_cost_usd`` is a float; converted with ``round(value * 1_000_000)``
+        and clamped at zero so a negative stored value cannot violate
+        ``user_daily_facts.operator_cost_micro``'s CHECK.
+        """
+        start = f"{day.isoformat()} 00:00:00"
+        end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00"
+        conn = self._get_connection()
+        cursor = conn.cursor()
+        cursor.execute(
+            """
+            SELECT owner_user_id, COALESCE(SUM(est_cost_usd), 0) AS cost_usd
+            FROM agent_runs
+            WHERE owner_user_id IS NOT NULL
+              AND updated_at >= ? AND updated_at < ?
+            GROUP BY owner_user_id
+            """,
+            (start, end),
+        )
+        rows = cursor.fetchall()
+        conn.close()
+        return {
+            int(row["owner_user_id"]): max(0, round(float(row["cost_usd"] or 0) * 1_000_000))
+            for row in rows
+        }
+
     def insert_trades(self, run_id: str, trades: List[Dict[str, Any]]) -> None:
         """Batch insert trade records for a backtest run."""
         if not trades:

dashboard/backend/database_postgres.py

@@ -42,6 +42,7 @@
 from __future__ import annotations
 
 import json
+from datetime import date, timedelta
 from typing import Any, Dict, List, Optional
 
 from dashboard.backend.database import (
@@ -126,6 +127,7 @@ def _init_schema(self) -> None:
                         output_tokens INTEGER DEFAULT 0,
                         est_cost_usd DOUBLE PRECISION DEFAULT 0,
                         metadata TEXT,
+                        owner_user_id INTEGER,
                         created_at TEXT NOT NULL {created_at_default},
                         updated_at TEXT NOT NULL {created_at_default},
                         baseline_djia_run_id TEXT,
@@ -275,6 +277,9 @@ def _init_schema(self) -> None:
                 cur.execute(
                     "ALTER TABLE agent_runs ADD COLUMN IF NOT EXISTS metadata TEXT"
                 )
+                cur.execute(
+                    "ALTER TABLE agent_runs ADD COLUMN IF NOT EXISTS owner_user_id INTEGER"
+                )
 
                 cur.execute(
                     "ALTER TABLE backtest_decisions ADD COLUMN IF NOT EXISTS "
@@ -487,7 +492,8 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
                    input_tokens: int = 0,
                    output_tokens: int = 0,
                    est_cost_usd: float = 0.0,
-                   metadata: Optional[Dict[str, Any]] = None) -> None:
+                   metadata: Optional[Dict[str, Any]] = None,
+                   owner_user_id: Optional[int] = None) -> None:
         """Insert or refresh a backtest run.
 
         Carries divergences 1-3 from the module docstring, all of them
@@ -549,8 +555,8 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
                      initial_equity, final_equity, total_return, sharpe_ratio,
                      max_drawdown, num_trades, llm_model,
                      llm_calls, llm_decisions, input_tokens, output_tokens,
-                     est_cost_usd, metadata)
-                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                     est_cost_usd, metadata, owner_user_id)
+                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s::integer)
                     ON CONFLICT (run_id) DO UPDATE SET
                         session_id = EXCLUDED.session_id,
                         agent_name = EXCLUDED.agent_name,
@@ -570,6 +576,7 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
                         output_tokens = EXCLUDED.output_tokens,
                         est_cost_usd = EXCLUDED.est_cost_usd,
                         metadata = EXCLUDED.metadata,
+                        owner_user_id = EXCLUDED.owner_user_id,
                         updated_at = to_char(now() AT TIME ZONE 'utc', 'YYYY-MM-DD HH24:MI:SS')
                     """,
                     (
@@ -579,6 +586,7 @@ def insert_run(self, run_id: str, session_id: str, agent_name: str, mode: str,
                         llm_calls, llm_decisions, input_tokens, output_tokens,
                         est_cost_usd,
                         json.dumps(metadata) if metadata is not None else None,
+                        owner_user_id,
                     ),
                 )
 
@@ -1019,6 +1027,28 @@ def get_runs_by_mode(self, mode: str) -> List[Dict]:
                 rows = cur.fetchall()
         return [BacktestDatabase._parse_run_row(row) for row in rows]
 
+    def aggregate_operator_cost_for_day(self, day: date) -> Dict[int, int]:
+        """See the SQLite twin."""
+        start = f"{day.isoformat()} 00:00:00"
+        end = f"{(day + timedelta(days=1)).isoformat()} 00:00:00"
+        with self._get_connection() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    SELECT owner_user_id, COALESCE(SUM(est_cost_usd), 0) AS cost_usd
+                    FROM agent_runs
+                    WHERE owner_user_id IS NOT NULL
+                      AND updated_at >= %s AND updated_at < %s
+                    GROUP BY owner_user_id
+                    """,
+                    (start, end),
+                )
+                rows = cur.fetchall()
+        return {
+            int(row["owner_user_id"]): max(0, round(float(row["cost_usd"] or 0) * 1_000_000))
+            for row in rows
+        }
+
     def get_trades(self, run_id: str) -> List[Dict]:
         """Get all trades for a run.