fix: address the findings from reviewing this stack

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.
 

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.

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:

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.

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:

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

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)

backend/app/agent/media_staging.py

@@ -88,13 +88,6 @@ def _staging_root() -> Path:
     return root
 
 
-def _user_dir(user_id: str) -> Path:
-    """Return ``<staging_root>/<user_id>``, 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.
 

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(

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.

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,

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(

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],

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),

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.