diff --git a/AGENTS.md b/AGENTS.md index 9d1ea358..b2d1bb30 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -67,9 +67,8 @@ Key store modules: - `backend/app/agent/memory_db.py` -- `MemoryStore` (per-user via `get_memory_store(id)`) - `backend/app/agent/stores.py` -- `HeartbeatStore`, `IdempotencyStore`, `LLMUsageStore`, `ToolConfigStore` - `backend/app/agent/dto.py` -- Pydantic DTOs: `UserData`, `StoredMessage`, `SessionState`, etc. -- `backend/app/agent/file_store.py` -- Compatibility shim (re-exports from above modules) - `backend/app/database.py` -- `Base`, `AsyncSessionLocal`, `db_session_async()`, `get_async_db()`, `get_async_engine()` -- `backend/app/models.py` -- All SQLAlchemy ORM model classes +- `backend/app/models/` -- SQLAlchemy ORM models, one module per product area. Importing the package is what registers every mapper on `Base.metadata`, so Alembic autogenerate and the test-suite TRUNCATE both depend on it. File storage is exposed through the Google Drive integration: each user grants the `drive.file` scope and uploads land in their own Drive under a top-level `Clawbolt` folder. The operator wires the OAuth client via `GOOGLE_DRIVE_CLIENT_ID` / `GOOGLE_DRIVE_CLIENT_SECRET`; without those, the file tools never load. @@ -85,11 +84,11 @@ Pool sizing is currently SQLAlchemy default (`pool_size=5`, `max_overflow=10`). ### Common SQLAlchemy 2.0 patterns -- Read: `(await db.execute(select(X).where(...))).scalar_one_or_none()`. For a scalar count use `await db.scalar(select(func.count(...)))`. +- Read: `(await db.execute(select(X).where(...))).scalar_one_or_none()`. For the shapes that repeat, use the helpers in `backend/app/query_helpers.py`: `get_or_404_async` (fetch one or 404), `count_rows(db, Model.id, *where)` (scalar count), `fetch_all(db, stmt)` (entities as a `list`), and `iso` / `iso_or_none` (timestamp serialization, `""` or `None` fallback). Hand-roll only where the query does not fit one, e.g. a count across a join. - DML rowcount: at runtime `(await db.execute(update/delete)).rowcount` returns `int`, but the stubs say `Result`. Cast to access cleanly: `cast("CursorResult[object]", await db.execute(...)).rowcount`. Reference `SessionStore.delete_message` in `backend/app/agent/session_db.py`. - Bulk DML `synchronize_session`: the kwarg moved off `update()`/`delete()` constructors. Use `.execution_options(synchronize_session="fetch")` on the executable. Reference `_append_history_update` in `backend/app/agent/memory_db.py`. - Row-level lock: `(await db.execute(select(M).filter_by(id=x).with_for_update())).scalar_one_or_none()`. -- `.scalars().all()` returns `Sequence[T]`, not `list[T]`. Wrap with `list(...)` only when the consumer is typed for `list`. +- `.scalars().all()` returns `Sequence[T]`, not `list[T]`. Wrap with `list(...)` only when the consumer is typed for `list`; `fetch_all` already does. - Do not use the SQLAlchemy 1.x `db.query()` API. ### Encrypted columns: do not concat on the SQL side @@ -177,7 +176,7 @@ When you need realistic-looking data, use clearly synthetic values: `jane.doe@ex ## Architecture -- **PostgreSQL storage**: all structured data in PostgreSQL via SQLAlchemy 2.0 ORM. See `backend/app/database.py` and `backend/app/models.py`. Store modules in `backend/app/agent/` provide CRUD APIs. +- **PostgreSQL storage**: all structured data in PostgreSQL via SQLAlchemy 2.0 ORM. See `backend/app/database.py` and `backend/app/models/`. Store modules in `backend/app/agent/` provide CRUD APIs. - **Auth plugin infrastructure**: base.py (ABC), loader.py (dynamic import), dependencies.py (get_current_user), scoping.py (row-level auth). `AUTH_MODE` selects the model: `single_user` (default) resolves every request to the one user in the database; `multi_user` delegates to a resolver registered via `set_current_user_resolver()` and never falls back to the single-user path, since that would serve one tenant's data to an unauthenticated caller. - **`user_id` scoping** on every data class and endpoint from day one - **Message bus**: async inbound/outbound queues in `bus.py`. Channels publish inbound messages; the agent publishes outbound replies. The ``ChannelManager`` dispatches outbound messages to the correct channel. @@ -194,7 +193,7 @@ What the mode switches on, and where it lives: | Surface | Modules | |---|---| | Google OAuth sign-in, JWT sessions, admin API keys | `auth/google_oauth` router, `auth/oauth_flow.py`, `auth/jwt_auth.py`, `auth/session_auth.py`, `services/admin_api_keys.py` | -| Admin console, audit log, consent-gated shared data | `routers/admin.py`, `routers/admin_shared_data.py`, `routers/admin_reported_conversations.py`, `services/admin_audit.py`, `services/pii_redaction.py` | +| Admin console, audit log, consent-gated shared data | `routers/admin/`, `routers/admin_shared_data/`, `routers/admin_reported_conversations.py`, `services/admin_audit.py`, `services/pii_redaction.py` | | Account page, data export, deletion | `routers/account.py`, `services/data_export.py`, `services/user_deletion.py`, `services/inactive_cleanup.py` | | Per-tenant quotas and plans | `billing/` | | Operator monitoring and email | `routers/monitoring.py`, `services/health_monitor.py`, `services/admin_alerts.py`, `services/email_service.py` | @@ -339,7 +338,7 @@ cd frontend && npm run test # vitest suite passes ### Frontend generated types -When backend schemas change (`backend/app/schemas.py`, route signatures, response models, or endpoint docstrings), you **must** regenerate the frontend OpenAPI types. Never hand-edit `frontend/src/generated/api.d.ts`. CI will fail if the committed file doesn't match what the generator produces. +When backend schemas change (`backend/app/schemas/`, route signatures, response models, or endpoint docstrings), you **must** regenerate the frontend OpenAPI types. Never hand-edit `frontend/src/generated/api.d.ts`. CI will fail if the committed file doesn't match what the generator produces. The spec is exported in `multi_user` mode, so it documents every route the product can serve. The frontend ships as one bundle for both modes and calls those routes through a typed client; a spec narrowed to `single_user` would leave the sign-in, account, and admin calls untyped. diff --git a/TODOS.md b/TODOS.md index 2cfb1796..3d3ee389 100644 --- a/TODOS.md +++ b/TODOS.md @@ -16,5 +16,5 @@ Plan: the current user's preference is off. Grouped receipts likewise skip the URL line. 5. Tests: preference toggling, render output shape, LLM tool. -Blast radius: `backend/app/models.py`, `alembic/`, `backend/app/agent/tool_summary.py`, +Blast radius: `backend/app/models/`, `alembic/`, `backend/app/agent/tool_summary.py`, new preferences tool, new tests. Estimate: one day CC. diff --git a/backend/app/agent/approval.py b/backend/app/agent/approval.py index 7eba2186..bef8b469 100644 --- a/backend/app/agent/approval.py +++ b/backend/app/agent/approval.py @@ -774,7 +774,7 @@ async def cleanup_orphaned_approvals( lock_acquired = True try: - rows = (await db.execute(select(PendingApprovalRow))).scalars().all() + rows = await fetch_all(db, select(PendingApprovalRow)) snapshot = [(r.user_id, r.tool_name, r.channel, r.chat_id, r.created_at) for r in rows] await db.commit() except Exception: diff --git a/backend/app/agent/heartbeat_frequency.py b/backend/app/agent/heartbeat_frequency.py index 57bdfd1f..489e2993 100644 --- a/backend/app/agent/heartbeat_frequency.py +++ b/backend/app/agent/heartbeat_frequency.py @@ -17,18 +17,6 @@ "weekly": 10080, } -# How far back to look when building heartbeat history context for the LLM. -_HISTORY_LOOKBACK_DAYS = 7 - -# Literal prefix that identifies the heartbeat-driven (scheduled) message -# path to the agent. Phase 2's ``task_context`` starts with this string. -# ``backend/app/agent/tools/heartbeat_tools.py`` matches on the same -# constant in ``update_heartbeat``'s ``usage_hint`` to scope proactive -# pruning to the heartbeat path; without that gate, the user-driven -# agent could start removing HEARTBEAT.md lines mid-conversation. Both -# sides import this constant so the strings cannot drift. -SCHEDULED_TASK_PREFIX = "Execute this scheduled task now" - def parse_frequency_to_minutes(freq: str) -> int | None: """Convert a frequency string like ``15m``, ``2h``, ``1d`` to minutes. diff --git a/backend/app/agent/heartbeat_routing.py b/backend/app/agent/heartbeat_routing.py index 527d31fc..c3733e3c 100644 --- a/backend/app/agent/heartbeat_routing.py +++ b/backend/app/agent/heartbeat_routing.py @@ -11,7 +11,6 @@ from sqlalchemy import Select, select from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import Session from backend.app.channels import get_channel from backend.app.models import ChannelRoute, User @@ -24,7 +23,7 @@ def _heartbeat_route_select(user: User) -> Select[tuple[ChannelRoute]]: - """Pure builder shared by sync and async ``resolve_heartbeat_route`` peers.""" + """Select the enabled, pushable routes for *user*, newest first.""" return select(ChannelRoute).where( ChannelRoute.user_id == user.id, ChannelRoute.enabled.is_(True), @@ -35,8 +34,8 @@ def _heartbeat_route_select(user: User) -> Select[tuple[ChannelRoute]]: def _check_route_is_registered(user: User, route: ChannelRoute) -> tuple[str, ChannelRoute] | None: """Verify the channel is registered in this process and return the result tuple. - Pure helper shared between the sync and async ``resolve_heartbeat_route`` - peers so the post-query branching cannot drift. + Kept apart from the query so the post-query branching is testable + without a database. """ try: get_channel(route.channel) @@ -50,40 +49,11 @@ def _check_route_is_registered(user: User, route: ChannelRoute) -> tuple[str, Ch return route.channel, route -def resolve_heartbeat_route( - user: User, - db: Session, -) -> tuple[str, ChannelRoute] | None: - """Pick the user's single active messaging channel for heartbeat delivery. - - Returns ``(channel_name, route)`` on success, or ``None`` when no - pushable route can be found. - - Pure lookup: never mutates the database. Under single-channel enforcement - each user has at most one enabled non-webchat route, and the write paths - that flip ``enabled`` are responsible for keeping ``User.preferred_channel`` - in sync. - """ - route = db.execute(_heartbeat_route_select(user)).scalar_one_or_none() - - if route is None: - logger.debug( - "Heartbeat skipped for user %s: no pushable route configured", - user.id, - ) - return None - - return _check_route_is_registered(user, route) - - async def resolve_heartbeat_route_async( user: User, db: AsyncSession, ) -> tuple[str, ChannelRoute] | None: - """Async peer of :func:`resolve_heartbeat_route`. - - Same shape and semantics as the sync version; only the IO is async. - """ + """Resolve the channel a proactive message should go out on.""" route = (await db.execute(_heartbeat_route_select(user))).scalar_one_or_none() if route is None: diff --git a/backend/app/agent/inbound_recovery.py b/backend/app/agent/inbound_recovery.py index 1c311feb..550e67a5 100644 --- a/backend/app/agent/inbound_recovery.py +++ b/backend/app/agent/inbound_recovery.py @@ -183,7 +183,7 @@ def _orphan_to_stored_and_state( user_id=chat_session.user_id, messages=[stored], created_at=iso(chat_session.created_at), - last_message_at=(iso(chat_session.last_message_at)), + last_message_at=iso(chat_session.last_message_at), channel=chat_session.channel or "", ) return stored, state diff --git a/backend/app/agent/ingestion.py b/backend/app/agent/ingestion.py index 86ba77e6..6fe8fb68 100644 --- a/backend/app/agent/ingestion.py +++ b/backend/app/agent/ingestion.py @@ -40,6 +40,7 @@ from backend.app.logging_utils import mask_pii from backend.app.media.download import DownloadedMedia from backend.app.models import ChannelRoute, ChatSession, Message, ReportedConversation, User +from backend.app.query_helpers import fetch_all logger = logging.getLogger(__name__) @@ -235,7 +236,7 @@ async def _get_or_create_user(channel: str, sender_id: str) -> User: # messages to whichever account happens to be the only one, which # is a cross-tenant leak on a deployment early enough to have one # user. - all_users = (await db.execute(select(User))).scalars().all() + all_users = await fetch_all(db, select(User)) if len(all_users) == 1 and settings.auth_mode == "single_user": user = all_users[0] logger.debug("_get_or_create_user: single-tenant reuse -> user %s", user.id) diff --git a/backend/app/agent/media_staging.py b/backend/app/agent/media_staging.py index 97083e74..c27e7018 100644 --- a/backend/app/agent/media_staging.py +++ b/backend/app/agent/media_staging.py @@ -88,13 +88,6 @@ def _staging_root() -> Path: return root -def _user_dir(user_id: str) -> Path: - """Return ``/``, ensuring it exists.""" - d = _staging_root() / user_id - d.mkdir(parents=True, exist_ok=True) - return d - - def _store_disk_path(user_id: str, handle: str) -> str: """Suffix stored in ``staged_media.disk_path``, relative to the staging root. diff --git a/backend/app/agent/session_db.py b/backend/app/agent/session_db.py index 6fef67a1..406e9477 100644 --- a/backend/app/agent/session_db.py +++ b/backend/app/agent/session_db.py @@ -28,7 +28,7 @@ db_session_async, ) from backend.app.models import ChatSession, Message -from backend.app.query_helpers import iso +from backend.app.query_helpers import fetch_all, iso logger = logging.getLogger(__name__) @@ -258,7 +258,7 @@ async def load_session_async(self, session_id: str) -> SessionState | None: ).scalar_one_or_none() if cs is None: return None - messages = list((await db.execute(_select_messages_for_session(cs.id))).scalars().all()) + messages = await fetch_all(db, _select_messages_for_session(cs.id)) return _session_to_state(cs, messages) finally: await db.close() @@ -271,9 +271,7 @@ async def list_sessions_async(self) -> list[SessionState]: """Return all sessions with their messages for this user.""" db = AsyncSessionLocal() try: - sessions = ( - (await db.execute(_select_all_sessions_for_user(self.user_id))).scalars().all() - ) + sessions = await fetch_all(db, _select_all_sessions_for_user(self.user_id)) result = [] for cs in sessions: messages = list( diff --git a/backend/app/agent/stores.py b/backend/app/agent/stores.py index 2f450e5c..d23f086e 100644 --- a/backend/app/agent/stores.py +++ b/backend/app/agent/stores.py @@ -1,7 +1,7 @@ """Database-backed replacements for file-based stores. Replaces HeartbeatStore, IdempotencyStore, LLMUsageStore, and -ToolConfigStore from file_store.py. Uses the corresponding ORM models for +ToolConfigStore. Uses the corresponding ORM models for persistence, while keeping Pydantic DTOs as the public API surface. Follows the same AsyncSessionLocal() / try-finally pattern used in session_db.py. diff --git a/backend/app/agent/tool_assembly.py b/backend/app/agent/tool_assembly.py index fbf77779..0529da57 100644 --- a/backend/app/agent/tool_assembly.py +++ b/backend/app/agent/tool_assembly.py @@ -2,12 +2,11 @@ Three callers need the same list: the runtime turn in :mod:`backend.app.agent.router`, the heartbeat's Phase 2 agent, and the -system-prompt preview. They used to build it inline, three times, which -meant the preview could drift from what the agent was actually handed. +system-prompt preview. Assembling it here is what keeps the preview +showing what the agent was actually handed. -:func:`assemble_turn_tools` is that shared build. -:func:`build_initial_turn_tools` wraps it for preview callers, which have -no storage backend or outbound hook to supply. +:func:`build_initial_turn_tools` wraps :func:`assemble_turn_tools` for +preview callers, which have no storage backend or outbound hook to supply. """ from __future__ import annotations @@ -38,7 +37,14 @@ async def assemble_turn_tools( something to discover. Returns the tools and the specialist summaries, which callers log. + + Nothing populates the registry at startup, and an unpopulated registry + yields an empty tool list rather than an error. Priming it here means a + caller cannot get silently toolless by forgetting to. It is guarded and + cached, so the repeat calls cost nothing. """ + ensure_tool_modules_imported() + tools = await default_registry.create_core_tools( tool_context, excluded_factories=disabled_factories or None, @@ -89,11 +95,6 @@ async def build_initial_turn_tools( and usage hints (for system-prompt rendering or debugging) -- not their executors. """ - # The registry is auto-discovery-driven; ensure all *_tools modules - # have run their _register() side effects before we ask it for the - # current set of factories. Idempotent / cached after the first call. - ensure_tool_modules_imported() - tool_context = ToolContext( user=user, storage=None, diff --git a/backend/app/main.py b/backend/app/main.py index 27544aa9..1413a040 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -49,6 +49,7 @@ from backend.app.middleware.seo_meta import SeoMetaMiddleware from backend.app.models import ChannelRoute, User from backend.app.observability import setup_logging +from backend.app.query_helpers import fetch_all from backend.app.routers import ( account, admin, @@ -167,12 +168,10 @@ async def _enforce_single_channel() -> None: read-time drift-sync. """ async with db_session_async() as db: - users = (await db.execute(select(User))).scalars().all() + users = await fetch_all(db, select(User)) fixed = 0 for user in users: - routes = ( - (await db.execute(select(ChannelRoute).filter_by(user_id=user.id))).scalars().all() - ) + routes = await fetch_all(db, select(ChannelRoute).filter_by(user_id=user.id)) enabled_messaging = [r for r in routes if r.enabled and r.channel != "webchat"] if len(enabled_messaging) > 1: preferred_match = next( diff --git a/backend/app/query_helpers.py b/backend/app/query_helpers.py index 1c37d50b..8ea9a7e7 100644 --- a/backend/app/query_helpers.py +++ b/backend/app/query_helpers.py @@ -15,7 +15,6 @@ from fastapi import HTTPException from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession -from sqlalchemy.orm import Session if TYPE_CHECKING: from sqlalchemy.sql import ColumnElement @@ -25,24 +24,6 @@ T = TypeVar("T") -def get_or_404( - db: Session, - model: type[T], - detail: str = "Not found", - **filters: object, -) -> T: - """Query for a single row by filter or raise HTTP 404. - - Usage:: - - user = get_or_404(db, User, detail="User not found", id=user_id) - """ - row = db.execute(select(model).filter_by(**filters)).scalar_one_or_none() - if row is None: - raise HTTPException(status_code=404, detail=detail) - return row - - async def get_or_404_async( db: AsyncSession, model: type[T], diff --git a/backend/app/routers/admin/access.py b/backend/app/routers/admin/access.py index 4ba4ba30..6861c00a 100644 --- a/backend/app/routers/admin/access.py +++ b/backend/app/routers/admin/access.py @@ -3,7 +3,6 @@ from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException, Query -from sqlalchemy import func as sa_func from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -12,7 +11,7 @@ AllowedEmail, WaitlistEntry, ) -from backend.app.query_helpers import fetch_all, iso +from backend.app.query_helpers import count_rows, fetch_all, iso from backend.app.schemas.admin import ( AllowedEmailCreate, AllowedEmailListResponse, @@ -37,7 +36,7 @@ async def list_allowed_emails( db: AsyncSession = Depends(get_async_db), ) -> AllowedEmailListResponse: """List all pre-approved email addresses.""" - rows = (await db.execute(select(AllowedEmail).order_by(AllowedEmail.email))).scalars().all() + rows = await fetch_all(db, select(AllowedEmail).order_by(AllowedEmail.email)) ctx.detail = {"count": len(rows)} return AllowedEmailListResponse( total=len(rows), @@ -110,7 +109,7 @@ async def list_waitlist_entries( ) -> WaitlistListResponse: """List waitlist entries, newest first.""" ctx.detail = {"offset": offset, "limit": limit} - total = (await db.execute(select(sa_func.count(WaitlistEntry.id)))).scalar_one() or 0 + total = await count_rows(db, WaitlistEntry.id) rows = await fetch_all( db, select(WaitlistEntry).order_by(WaitlistEntry.created_at.desc()).offset(offset).limit(limit), diff --git a/backend/app/routers/admin/users.py b/backend/app/routers/admin/users.py index 481f389f..112880bd 100644 --- a/backend/app/routers/admin/users.py +++ b/backend/app/routers/admin/users.py @@ -25,7 +25,7 @@ Subscription, User, ) -from backend.app.query_helpers import iso_or_none +from backend.app.query_helpers import fetch_all, iso_or_none from backend.app.schemas.admin import ( AdminChannelRouteEntry, AdminToolConfigEntry, @@ -110,7 +110,7 @@ async def list_users( } # Get subscription data - subs = (await db.execute(select(Subscription))).scalars().all() + subs = await fetch_all(db, select(Subscription)) sub_map: dict[str, Subscription] = {s.user_id: s for s in subs} # Messages-this-month per user, aggregated from Message->Session->user_id. diff --git a/backend/app/routers/admin_llm_eval.py b/backend/app/routers/admin_llm_eval.py index 999767ec..53396daf 100644 --- a/backend/app/routers/admin_llm_eval.py +++ b/backend/app/routers/admin_llm_eval.py @@ -30,7 +30,6 @@ import json import logging -from datetime import datetime from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query @@ -42,7 +41,7 @@ from backend.app.config import settings from backend.app.database import get_async_db from backend.app.models import LLMEvalRun, LLMEvalTurnResult, Subscription, User -from backend.app.query_helpers import count_rows +from backend.app.query_helpers import count_rows, fetch_all, iso_or_none from backend.app.schemas.llm_eval import ( AdminLLMEvalDecision, AdminLLMEvalReportResponse, @@ -109,10 +108,6 @@ async def _effective_models(user_id: str, db: AsyncSession) -> tuple[str, str, s return settings.llm_endpoint, settings.llm_provider, model -def _iso(value: datetime | None) -> str | None: - return value.isoformat() if value is not None else None - - def _summary_of(run: LLMEvalRun) -> AdminLLMEvalSummary | None: if not run.summary_json: return None @@ -143,8 +138,8 @@ def _run_item( recommendation=run.recommendation, error=run.error, created_at=run.created_at.isoformat(), - started_at=_iso(run.started_at), - completed_at=_iso(run.completed_at), + started_at=iso_or_none(run.started_at), + completed_at=iso_or_none(run.completed_at), summary=_summary_of(run), ) @@ -463,7 +458,7 @@ async def list_runs( query = query.where(LLMEvalRun.user_id == user_id) total_query = total_query.where(LLMEvalRun.user_id == user_id) - runs = (await db.execute(query.limit(limit).offset(offset))).scalars().all() + runs = await fetch_all(db, query.limit(limit).offset(offset)) total = await db.scalar(total_query) or 0 # One query for the identities on this page rather than one per row. diff --git a/backend/app/routers/admin_shared_data/approvals.py b/backend/app/routers/admin_shared_data/approvals.py index a801c89c..2bfeeb26 100644 --- a/backend/app/routers/admin_shared_data/approvals.py +++ b/backend/app/routers/admin_shared_data/approvals.py @@ -94,7 +94,7 @@ async def list_shared_data_approval_events( ] return SharedDataApprovalEventListResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), items=items, total=len(items), ) diff --git a/backend/app/routers/admin_shared_data/compaction.py b/backend/app/routers/admin_shared_data/compaction.py index b884309c..c3f013bc 100644 --- a/backend/app/routers/admin_shared_data/compaction.py +++ b/backend/app/routers/admin_shared_data/compaction.py @@ -177,7 +177,7 @@ async def list_shared_data_compaction_events( total = (await db.execute(count_stmt)).scalar_one() or 0 return SharedDataCompactionEventListResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), items=items, total=int(total), ) diff --git a/backend/app/routers/admin_shared_data/consent.py b/backend/app/routers/admin_shared_data/consent.py index 402732c1..3c3c4d07 100644 --- a/backend/app/routers/admin_shared_data/consent.py +++ b/backend/app/routers/admin_shared_data/consent.py @@ -4,7 +4,7 @@ from datetime import UTC, datetime -from fastapi import APIRouter, HTTPException +from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession @@ -12,8 +12,6 @@ User, ) -router = APIRouter() - def _parse_date_range( start_date: str | None, end_date: str | None diff --git a/backend/app/routers/admin_shared_data/conversations.py b/backend/app/routers/admin_shared_data/conversations.py index b370a98b..bf1cfde4 100644 --- a/backend/app/routers/admin_shared_data/conversations.py +++ b/backend/app/routers/admin_shared_data/conversations.py @@ -275,7 +275,7 @@ async def list_shared_data_conversation_turns( return SharedDataConversationTurnsResponse( session_id=session.session_id, user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), turns=turns, total=len(turns), last_trim_seq=session.last_trim_seq, diff --git a/backend/app/routers/admin_shared_data/export.py b/backend/app/routers/admin_shared_data/export.py index 7ab1d7fa..7e033f3f 100644 --- a/backend/app/routers/admin_shared_data/export.py +++ b/backend/app/routers/admin_shared_data/export.py @@ -372,7 +372,7 @@ async def export_shared_data_user( # ---- profile + memory -------------------------------------------- profile = SharedDataProfileResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), soul_text=redact_pii(user.soul_text or ""), user_text=redact_pii(user.user_text or ""), heartbeat_text=redact_pii(user.heartbeat_text or ""), @@ -386,7 +386,7 @@ async def export_shared_data_user( ).scalar_one_or_none() memory = SharedDataMemoryDocumentResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), memory_text=redact_pii(mem_row.memory_text if mem_row and mem_row.memory_text else ""), history_text=redact_pii(mem_row.history_text if mem_row and mem_row.history_text else ""), updated_at=(mem_row.updated_at.isoformat() if mem_row and mem_row.updated_at else None), @@ -415,7 +415,7 @@ async def export_shared_data_user( SharedDataConversationTurnsResponse( session_id=sess.session_id, user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), turns=grouped, total=len(grouped), ) diff --git a/backend/app/routers/admin_shared_data/profile.py b/backend/app/routers/admin_shared_data/profile.py index 5c3fb846..0198e60c 100644 --- a/backend/app/routers/admin_shared_data/profile.py +++ b/backend/app/routers/admin_shared_data/profile.py @@ -65,7 +65,7 @@ async def get_shared_data_profile( return SharedDataProfileResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), soul_text=redact_pii(user.soul_text or ""), user_text=redact_pii(user.user_text or ""), heartbeat_text=redact_pii(user.heartbeat_text or ""), @@ -133,7 +133,7 @@ async def list_shared_data_heartbeat_logs( total = (await db.execute(count_stmt)).scalar_one() or 0 return SharedDataHeartbeatLogListResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), items=items, total=int(total), ) @@ -171,7 +171,7 @@ async def get_shared_data_memory( ).scalar_one_or_none() return SharedDataMemoryDocumentResponse( user_id=user.id, - consent_at=(iso_or_none(user.data_sharing_consent_at)), + consent_at=iso_or_none(user.data_sharing_consent_at), memory_text=redact_pii(doc.memory_text if doc else ""), history_text=redact_pii(doc.history_text if doc else ""), updated_at=(doc.updated_at.isoformat() if doc and doc.updated_at else None), diff --git a/backend/app/routers/admin_shared_data/summary.py b/backend/app/routers/admin_shared_data/summary.py index 64ec15dd..ce0f3ea7 100644 --- a/backend/app/routers/admin_shared_data/summary.py +++ b/backend/app/routers/admin_shared_data/summary.py @@ -255,7 +255,7 @@ async def list_shared_data_users( id=u.id, user_id=u.user_id, email=sub_emails.get(u.id, ""), - consent_at=(iso_or_none(u.data_sharing_consent_at)), + consent_at=iso_or_none(u.data_sharing_consent_at), conversation_count=activity.get(u.id, (0, None))[0], last_message_at=( activity[u.id][1].isoformat() diff --git a/backend/app/routers/user_profile.py b/backend/app/routers/user_profile.py index ae4996be..9497ae38 100644 --- a/backend/app/routers/user_profile.py +++ b/backend/app/routers/user_profile.py @@ -105,7 +105,7 @@ def _profile_response(c: User) -> UserProfileResponse: onboarding_complete=c.onboarding_complete, is_active=c.is_active, data_sharing_consent=c.data_sharing_consent, - data_sharing_consent_at=(iso_or_none(c.data_sharing_consent_at)), + data_sharing_consent_at=iso_or_none(c.data_sharing_consent_at), created_at=c.created_at.isoformat(), updated_at=c.updated_at.isoformat(), ) @@ -191,7 +191,7 @@ async def get_data_sharing_consent( """Return the current user's data sharing consent state.""" return DataSharingConsentResponse( data_sharing_consent=current_user.data_sharing_consent, - data_sharing_consent_at=(iso_or_none(current_user.data_sharing_consent_at)), + data_sharing_consent_at=iso_or_none(current_user.data_sharing_consent_at), ) @@ -216,7 +216,7 @@ async def update_data_sharing_consent( await db.refresh(user) return DataSharingConsentResponse( data_sharing_consent=user.data_sharing_consent, - data_sharing_consent_at=(iso_or_none(user.data_sharing_consent_at)), + data_sharing_consent_at=iso_or_none(user.data_sharing_consent_at), ) @@ -856,11 +856,7 @@ async def get_heartbeat_logs( db: AsyncSession = Depends(get_async_db), ) -> HeartbeatLogListResponse: """List heartbeat logs for the current user, most recent first.""" - total: int = ( - await db.scalar( - select(sa_func.count(HeartbeatLog.id)).where(HeartbeatLog.user_id == current_user.id) - ) - ) or 0 + total = await count_rows(db, HeartbeatLog.id, HeartbeatLog.user_id == current_user.id) logs = await fetch_all( db, diff --git a/backend/app/routers/user_sessions.py b/backend/app/routers/user_sessions.py index 0dd3e000..1589780c 100644 --- a/backend/app/routers/user_sessions.py +++ b/backend/app/routers/user_sessions.py @@ -21,12 +21,10 @@ from backend.app.auth.dependencies import get_current_user from backend.app.database import AsyncSessionLocal from backend.app.models import ChatSession, User -from backend.app.schemas.heartbeat import ( +from backend.app.schemas.user import ( BatchDeleteRequest, DeleteMessageResponse, DeleteMessagesResponse, -) -from backend.app.schemas.user import ( SessionDetailResponse, SessionMessage, SessionSystemPromptResponse, diff --git a/backend/app/schemas/heartbeat.py b/backend/app/schemas/heartbeat.py index 7f12a6d7..259206c7 100644 --- a/backend/app/schemas/heartbeat.py +++ b/backend/app/schemas/heartbeat.py @@ -2,7 +2,7 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel class HeartbeatLogItemResponse(BaseModel): @@ -26,20 +26,6 @@ class DeleteHeartbeatLogsResponse(BaseModel): deleted: int -class DeleteMessagesResponse(BaseModel): - status: str - messages_deleted: int - - -class BatchDeleteRequest(BaseModel): - seqs: list[int] = Field(..., min_length=1, max_length=1000) - - -class DeleteMessageResponse(BaseModel): - status: str - seq: int - - class AdminHeartbeatLogItem(BaseModel): """Heartbeat log metadata only. diff --git a/backend/app/schemas/user.py b/backend/app/schemas/user.py index 6f95af48..cc0c69c2 100644 --- a/backend/app/schemas/user.py +++ b/backend/app/schemas/user.py @@ -129,3 +129,17 @@ class ProfileResponse(BaseModel): id: str plan: str role: str + + +class DeleteMessagesResponse(BaseModel): + status: str + messages_deleted: int + + +class BatchDeleteRequest(BaseModel): + seqs: list[int] = Field(..., min_length=1, max_length=1000) + + +class DeleteMessageResponse(BaseModel): + status: str + seq: int diff --git a/backend/app/services/llm_endpoints.py b/backend/app/services/llm_endpoints.py index d8dbc179..8a322fc7 100644 --- a/backend/app/services/llm_endpoints.py +++ b/backend/app/services/llm_endpoints.py @@ -35,6 +35,7 @@ import asyncio import logging from collections.abc import Iterable +from typing import TypeVar, cast, get_args from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession @@ -44,6 +45,7 @@ from backend.app.database import AsyncSessionLocal from backend.app.enums import CacheControlMode, PricingMode, ReasoningMode from backend.app.models import LLMEndpoint, LLMEvalRun, Subscription +from backend.app.query_helpers import fetch_all from backend.app.schemas.llm import LLMEndpointItem from backend.app.services.llm_service import ( LLMTarget, @@ -104,7 +106,7 @@ async def _load_endpoints() -> dict[str, LLMEndpoint]: generation = _cache_generation db = AsyncSessionLocal() try: - rows = (await db.execute(select(LLMEndpoint))).scalars().all() + rows = await fetch_all(db, select(LLMEndpoint)) finally: await db.close() loaded = {row.name: row for row in rows} @@ -170,6 +172,25 @@ async def upsert_endpoint( return row +_ModeT = TypeVar("_ModeT", bound=str) + + +def _mode_or_auto(endpoint: str, field: str, value: str, allowed: tuple[_ModeT, ...]) -> _ModeT: + """Return *value* if it is one of *allowed*, else ``auto``. + + These columns are plain ``String(16)``, so nothing at the database level + stops a hand-written UPDATE from putting something outside the set in one. + Reporting it verbatim would fail ``LLMEndpointItem`` validation and 500 the + entire endpoint list over one bad row, which is a worse failure than + showing the default. The agent path already degrades this way; see + ``_reasoning_style_for``. + """ + if value in allowed: + return cast("_ModeT", value) + logger.warning("Endpoint %r has unknown %s %r; reporting %s", endpoint, field, value, _AUTO) + return cast("_ModeT", _AUTO) + + def endpoint_item(row: LLMEndpoint) -> LLMEndpointItem: """Serialize an endpoint. The stored key is reported as a boolean only.""" return LLMEndpointItem( @@ -177,9 +198,11 @@ def endpoint_item(row: LLMEndpoint) -> LLMEndpointItem: dialect=row.dialect, base_url=row.base_url, api_key_set=bool(row.api_key), - cache_control=row.cache_control, - reasoning=row.reasoning, - pricing=row.pricing, + cache_control=_mode_or_auto( + row.name, "cache_control", row.cache_control, get_args(CacheControlMode) + ), + reasoning=_mode_or_auto(row.name, "reasoning", row.reasoning, get_args(ReasoningMode)), + pricing=_mode_or_auto(row.name, "pricing", row.pricing, get_args(PricingMode)), notes=row.notes, ) diff --git a/backend/app/services/user_deletion.py b/backend/app/services/user_deletion.py index 98687f1c..3529b725 100644 --- a/backend/app/services/user_deletion.py +++ b/backend/app/services/user_deletion.py @@ -21,15 +21,14 @@ User, UserPermissionSet, ) +from backend.app.query_helpers import fetch_all logger = logging.getLogger(__name__) async def archive_usage(db: AsyncSession, user: User) -> None: """Archive aggregate usage before deletion to prevent quota-reset abuse.""" - quotas = ( - (await db.execute(select(UsageQuota).where(UsageQuota.user_id == user.id))).scalars().all() - ) + quotas = await fetch_all(db, select(UsageQuota).where(UsageQuota.user_id == user.id)) total_messages = sum(q.messages_used for q in quotas) total_tokens = sum(q.tokens_used for q in quotas) diff --git a/fiximports.py b/fiximports.py new file mode 100644 index 00000000..0c2fa4f3 --- /dev/null +++ b/fiximports.py @@ -0,0 +1,28 @@ +import ast, pathlib, re, subprocess, collections + +HELPERS = {'fetch_all','count_rows','iso','iso_or_none','get_or_404_async','get_or_404'} +out = subprocess.run(['uv','run','ruff','check','backend/','--output-format','concise'], + capture_output=True, text=True).stdout +need = collections.defaultdict(set) +for line in out.splitlines(): + m = re.match(r'(\S+?):\d+:\d+: F821 Undefined name `(\w+)`', line) + if m and m.group(2) in HELPERS: + need[m.group(1)].add(m.group(2)) + +def add_import(path, names): + p = pathlib.Path(path); src = p.read_text(); tree = ast.parse(src) + for node in tree.body: + if isinstance(node, ast.ImportFrom) and node.module == 'backend.app.query_helpers': + want = sorted({a.name for a in node.names} | set(names)) + lines = src.splitlines(keepends=True) + del lines[node.lineno-1:node.end_lineno] + lines.insert(node.lineno-1, f"from backend.app.query_helpers import {', '.join(want)}\n") + p.write_text(''.join(lines)); return 'merged' + last = [n for n in tree.body if isinstance(n,(ast.Import,ast.ImportFrom))] + if not last: return 'skip' + lines = src.splitlines(keepends=True) + lines.insert(last[-1].end_lineno, f"from backend.app.query_helpers import {', '.join(sorted(names))}\n") + p.write_text(''.join(lines)); return 'inserted' + +for f, n in sorted(need.items()): + print(f"{add_import(f, n):9} {f} {sorted(n)}") diff --git a/frontend/src/extensions/admin/admin-api.ts b/frontend/src/extensions/admin/admin-api.ts index 96d56722..484f5248 100644 --- a/frontend/src/extensions/admin/admin-api.ts +++ b/frontend/src/extensions/admin/admin-api.ts @@ -1,7 +1,13 @@ import type { components } from '@/generated/api'; import client from '@/lib/api-client'; -// --- Types (match backend schemas exactly) --- +// --- Types --- +// +// Anything whose name matches a schema in the generated spec is aliased to it, +// so those cannot drift. The rest are hand-written because the backend name +// differs (SharedDataMessage vs SharedDataMessageItem, EvalRun vs +// AdminLLMEvalRunItem, ...); those are only as accurate as the last person to +// touch them, so prefer aliasing a renamed schema over editing one by hand. export interface AdminUser { id: string; @@ -231,12 +237,7 @@ export async function getUserUsage(userId: string): Promise { // Admin-triggered compaction of a user's currently-visible context. // ``keepRecent`` preserves the tail; ``hint`` is prepended to the // compaction LLM's block as ``[admin note: ...]``. -export interface CompactUserContextResult { - compacted_message_count: number; - new_watermark: number | null; - memory_updated: boolean; - event_id: number | null; -} +export type CompactUserContextResult = components['schemas']['CompactUserContextResponse']; export async function compactUserContext( id: string, @@ -553,7 +554,8 @@ export async function updateUserPlan( // --- LLM provider/model enumeration (uses OSS endpoints; available to admins) --- -export type ProviderInfo = components['schemas']['ProviderInfo']; +export type { ProviderInfo } from '@/types'; +import type { ProviderInfo } from '@/types'; /** * Module-scoped cache for ``listProviders`` and ``listProviderModels``. @@ -1283,28 +1285,7 @@ export type EvalRecommendation = | 'do_not_switch' | 'inconclusive'; -export interface EvalModelTotals { - provider: string; - model: string; - input_tokens: number; - output_tokens: number; - cache_read_tokens: number; - cache_creation_tokens: number; - cache_read_ratio: number; - /** - * Cached share of prompt tokens, read plus written. Prefer this over - * cache_read_ratio for any comparison: the read ratio depends on whether an - * earlier run left warm cache entries behind, so it swings between runs of - * the same models. - */ - cache_participation_ratio: number; - total_cost_usd: string; - // False when the pricing library has no entry for this model. The cost is - // then zero and must be rendered as "unknown", never as "free". - pricing_available: boolean; - latency_p50_ms: number; - latency_p95_ms: number; -} +export type EvalModelTotals = components['schemas']['AdminLLMEvalModelTotals']; export interface EvalSummary { turns_total: number; diff --git a/frontend/src/extensions/admin/tabs/config.tsx b/frontend/src/extensions/admin/tabs/config.tsx index 2126b38c..3747d53d 100644 --- a/frontend/src/extensions/admin/tabs/config.tsx +++ b/frontend/src/extensions/admin/tabs/config.tsx @@ -23,6 +23,24 @@ import { ReasoningEffortSelect, } from '../llm-picker'; + +// Typed against the generated schema rather than loose strings: if a value is +// ever removed backend-side, these stop compiling instead of silently offering +// an option every PUT would 422 on. +const CACHE_CONTROL_MODES: readonly LLMEndpointUpsert['cache_control'][] = [ + 'auto', + 'always', + 'never', +] as const; +const REASONING_MODES: readonly LLMEndpointUpsert['reasoning'][] = [ + 'auto', + 'thinking', + 'effort', + 'none', +] as const; +const PRICING_MODES: readonly LLMEndpointUpsert['pricing'][] = ['auto', 'unpriced'] as const; + + const inputClass = 'w-full px-3 py-2 text-sm bg-card border border-border rounded-[--radius-md] text-foreground placeholder:text-muted-foreground focus:outline-none focus:ring-2 focus:ring-primary/30'; @@ -358,10 +376,10 @@ function LLMEndpointsSection({ onChanged }: { onChanged: () => void }) { onChange={e => set({ [field]: e.target.value } as Partial)} > {(field === 'cache_control' - ? (['auto', 'always', 'never'] as const) + ? CACHE_CONTROL_MODES : field === 'reasoning' - ? (['auto', 'thinking', 'effort', 'none'] as const) - : (['auto', 'unpriced'] as const) + ? REASONING_MODES + : PRICING_MODES ).map(option => (