Making scope filter by product id

.claude/settings.local.json

@@ -34,7 +34,12 @@
       "Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)",
       "Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")",
       "Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")",
-      "Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")"
+      "Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")",
+      "Bash(sqlite3 app.db \"PRAGMA wal_checkpoint\\(FULL\\);\")",
+      "Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/login)",
+      "Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/admin/imported-parts)",
+      "Bash(rg -n \"hx-delete|router.delete\" app/web/admin.py app/templates/admin/*.html)",
+      "Bash(rg -n \"is_superuser\" tests/conftest.py tests/*.py)"
     ]
   }
 }

.env.example

@@ -25,6 +25,9 @@ AUTO_CREATE_TABLES=true
 COOKIE_NAME=access_token
 # Set true when serving over HTTPS.
 COOKIE_SECURE=false
+# How long a browser login lasts before being bounced back to /login.
+# Independent of ACCESS_TOKEN_EXPIRE_MINUTES, which governs JSON API tokens.
+SESSION_TIMEOUT_HOURS=24
 
 # --- LDAP (optional; direct-bind) ---
 # When enabled, the login page and /api/auth/ldap-login accept a directory

Makefile

@@ -1,4 +1,4 @@
-VERSION := $(shell git describe --tags --always)
+VERSION := $(shell git tag | tail -1)
 IMAGE_NAME := fitguide
 
 sync:

README.md

@@ -149,8 +149,9 @@ Every setting is read from an environment variable first, then `.env`
 |----------|---------|-------|
 | `SECRET_KEY` | dev placeholder | **set a strong value in production** |
 | `DATABASE_URL` | `sqlite+aiosqlite:///./app.db` | use `postgresql+asyncpg://...` for Postgres |
-| `ACCESS_TOKEN_EXPIRE_MINUTES` | `15` | |
+| `ACCESS_TOKEN_EXPIRE_MINUTES` | `15` | JSON API bearer token lifetime |
 | `REFRESH_TOKEN_EXPIRE_DAYS` | `7` | |
+| `SESSION_TIMEOUT_HOURS` | `24` | browser login (HTMX UI cookie) lifetime |
 | `AUTO_CREATE_TABLES` | `true` | create tables on startup (dev); use Alembic in prod |
 | `COOKIE_SECURE` | `false` | set `true` behind HTTPS |
 | `FIRST_SUPERUSER_EMAIL` / `_PASSWORD` | admin@example.com / changethis | used by seed |

alembic/versions/04425ce3d4d5_add_jobs_table.py

@@ -0,0 +1,44 @@
+"""add jobs table
+
+Revision ID: 04425ce3d4d5
+Revises: a3f7c19b2d40
+Create Date: 2026-08-05 15:21:09.527442
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = "04425ce3d4d5"
+down_revision: Union[str, None] = "a3f7c19b2d40"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "jobs",
+        sa.Column("id", sa.String(length=32), nullable=False),
+        sa.Column("kind", sa.String(length=64), nullable=False),
+        sa.Column("status", sa.String(length=16), nullable=False),
+        sa.Column("message", sa.Text(), nullable=False),
+        sa.Column("result", sa.JSON(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("(CURRENT_TIMESTAMP)"),
+            nullable=False,
+        ),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("jobs")
+    # ### end Alembic commands ###

app/core/config.py

@@ -33,9 +33,13 @@ class Settings(BaseSettings):
     database_url: str = "sqlite+aiosqlite:///./app.db"
     auto_create_tables: bool = True
 
-    # Web / cookies (the HTMX UI stores the access token in an httpOnly cookie)
+    # Web / cookies (the HTMX UI stores the access token in an httpOnly cookie).
+    # session_timeout_hours governs how long a browser login lasts before the
+    # user is bounced back to /login; it's independent of the JSON API's
+    # short-lived access_token_expire_minutes.
     cookie_name: str = "access_token"
     cookie_secure: bool = False
+    session_timeout_hours: int = 24
 
     # LDAP (direct-bind). When enabled, the login page offers an LDAP option.
     # At login the user's DN is built from ``ldap_bind_dn_template`` (with

app/core/security.py

@@ -47,10 +47,12 @@ def _create_token(
     return token, payload
 
 
-def create_access_token(subject: str | int) -> str:
+def create_access_token(
+    subject: str | int, expires_delta: timedelta | None = None
+) -> str:
     token, _ = _create_token(
         subject,
-        timedelta(minutes=settings.access_token_expire_minutes),
+        expires_delta or timedelta(minutes=settings.access_token_expire_minutes),
         ACCESS_TOKEN_TYPE,
     )
     return token

app/crud/augmented_part.py

@@ -1,4 +1,4 @@
-from sqlalchemy import distinct, select
+from sqlalchemy import distinct, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.associations import augmented_part_imported_parts
@@ -89,10 +89,19 @@ async def list_filtered(
     direction: str = "asc",
     skip: int = 0,
     limit: int = 100,
+    match_sku: bool = False,
 ) -> list[AugmentedPart]:
     q = select(AugmentedPart)
     if description.strip():
-        q = q.where(AugmentedPart.description.ilike(f"%{description.strip()}%"))
+        term = f"%{description.strip()}%"
+        if match_sku:
+            q = q.where(
+                or_(
+                    AugmentedPart.description.ilike(term), AugmentedPart.sku.ilike(term)
+                )
+            )
+        else:
+            q = q.where(AugmentedPart.description.ilike(term))
     if colour_id is not None:
         q = q.where(AugmentedPart.colour_id == colour_id)
     if manufacturer_id is not None:

app/models/__init__.py

@@ -14,6 +14,7 @@ from app.models.fit import Fit, FitLabel, FitNote
 from app.models.imported_part import ImportedPart
 from app.models.imported_part_config import ImportedPartConfig
 from app.models.iso_thread import IsoThread
+from app.models.job import Job
 from app.models.manufacturer import Manufacturer
 from app.models.part_colour import PartColour
 from app.models.part_type import PartType
@@ -40,6 +41,7 @@ __all__ = [
     "ImportedPart",
     "ImportedPartConfig",
     "IsoThread",
+    "Job",
     "Manufacturer",
     "PartColour",
     "PartType",

app/models/job.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+
+from sqlalchemy import JSON, DateTime, String, Text, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.core.database import Base
+
+
+class Job(Base):
+    """Tracks a long-running admin operation (sync, import, fit recompute).
+
+    Persisted (rather than kept in a module-level dict) so that any worker
+    process/replica handling a status-poll request can see progress recorded
+    by whichever process is actually running the job.
+    """
+
+    __tablename__ = "jobs"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+    kind: Mapped[str] = mapped_column(String(64), nullable=False)
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="running"
+    )  # running | done | error
+    message: Mapped[str] = mapped_column(Text, nullable=False, default="Working…")
+    result: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=None)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    finished_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True, default=None
+    )

app/services/auth.py

@@ -1,5 +1,7 @@
 """Token issuance / rotation shared by the JSON API and the HTMX UI."""
 
+from datetime import timedelta
+
 import jwt
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -13,9 +15,16 @@ from app.crud import token as token_crud
 from app.models.user import User
 
 
-async def issue_token_pair(db: AsyncSession, user: User) -> tuple[str, str]:
+async def issue_token_pair(
-    """Create an access + refresh token and persist the refresh token for revocation."""
+    db: AsyncSession, user: User, access_expires_delta: timedelta | None = None
-    access = create_access_token(user.id)
+) -> tuple[str, str]:
+    """Create an access + refresh token and persist the refresh token for revocation.
+
+    ``access_expires_delta`` overrides the access token's default lifetime
+    (``settings.access_token_expire_minutes``) — used by the web login flow to
+    issue a longer-lived session cookie without affecting API bearer tokens.
+    """
+    access = create_access_token(user.id, access_expires_delta)
     refresh, payload = create_refresh_token(user.id)
     await token_crud.store(
         db,

app/services/jobs.py

@@ -1,11 +1,12 @@
-"""Lightweight in-process tracking for long-running admin operations.
+"""DB-backed tracking for long-running admin operations.
 
-Jobs live in a module-level dict, so this only works correctly for a single
+Jobs are persisted to the ``jobs`` table rather than kept in a module-level
-OS process (the app's current deployment: one uvicorn worker, no multi-process
+dict, so that a status-poll request landing on a *different* worker
-or multi-replica setup — see ``entrypoint.sh`` / ``docker-compose.yml``). If
+process/replica than the one running the job can still see its progress —
-the app is ever scaled to multiple workers/replicas, this needs to move to
+required once the app runs with more than one uvicorn worker or replica
-shared storage (a DB table, Redis) so a poll request landing on a different
+(a module-level dict is process-local and would report "not found" for a job
-process can still see the job.
+started elsewhere, even while that job is still running and writing to the
+shared database).
 """
 
 from __future__ import annotations
@@ -13,62 +14,83 @@ from __future__ import annotations
 import asyncio
 import uuid
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass, field
+from datetime import UTC, datetime, timedelta
-from datetime import UTC, datetime
 
-_MAX_FINISHED_AGE_SECONDS = 3600
+from sqlalchemy import delete
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import AsyncSessionLocal
+from app.models.job import Job
+
+_MAX_FINISHED_AGE = timedelta(hours=1)
 
 
-@dataclass
+async def get(db: AsyncSession, job_id: str) -> Job | None:
-class Job:
+    return await db.get(Job, job_id)
-    id: str
-    kind: str
-    status: str = "running"  # running | done | error
-    message: str = "Working…"
-    result: object | None = None
-    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
-    finished_at: datetime | None = None
 
 
-_jobs: dict[str, Job] = {}
+async def start(kind: str, run: Callable[[str], Awaitable[None]]) -> Job:
+    """Create a job row, launch ``run(job_id)`` in the background, and return
+    the job immediately.
 
-
+    ``run`` is responsible for calling ``mark_done``/``mark_error`` to record
-def get(job_id: str) -> Job | None:
+    the outcome, including on its own errors. Any exception that escapes
-    return _jobs.get(job_id)
+    anyway is caught here as a last resort so the job doesn't hang forever in
-
+    "running" state.
-
-def _evict_stale() -> None:
-    now = datetime.now(UTC)
-    stale = [
-        job_id
-        for job_id, job in _jobs.items()
-        if job.finished_at
-        and (now - job.finished_at).total_seconds() > _MAX_FINISHED_AGE_SECONDS
-    ]
-    for job_id in stale:
-        del _jobs[job_id]
-
-
-def start(kind: str, run: Callable[[Job], Awaitable[None]]) -> Job:
-    """Create a job, launch ``run(job)`` in the background, and return it immediately.
-
-    ``run`` is responsible for updating ``job.status``/``job.message``/``job.result``
-    to reflect the outcome, including catching its own errors. Any exception that
-    escapes anyway is caught here as a last resort so the job doesn't hang forever
-    in "running" state.
     """
-    _evict_stale()
+    async with AsyncSessionLocal() as db:
-    job = Job(id=uuid.uuid4().hex, kind=kind)
+        await _evict_stale(db)
-    _jobs[job.id] = job
+        job = Job(id=uuid.uuid4().hex, kind=kind, status="running", message="Working…")
+        db.add(job)
+        await db.commit()
 
     async def _runner() -> None:
         try:
-            await run(job)
+            await run(job.id)
         except Exception as exc:
-            job.status = "error"
+            await mark_error(job.id, f"{kind} failed: {exc}")
-            job.message = f"{kind} failed: {exc}"
-        finally:
-            job.finished_at = datetime.now(UTC)
 
     asyncio.create_task(_runner())
     return job
+
+
+async def mark_done(job_id: str, message: str, result: dict | None = None) -> None:
+    await _update(job_id, status="done", message=message, result=result)
+
+
+async def mark_error(job_id: str, message: str) -> None:
+    await _update(job_id, status="error", message=message)
+
+
+async def set_message(job_id: str, message: str) -> None:
+    await _update(job_id, message=message)
+
+
+async def _update(
+    job_id: str,
+    *,
+    status: str | None = None,
+    message: str | None = None,
+    result: dict | None = None,
+) -> None:
+    async with AsyncSessionLocal() as db:
+        job = await db.get(Job, job_id)
+        if job is None:
+            return
+        if status is not None:
+            job.status = status
+            if status in ("done", "error"):
+                job.finished_at = datetime.now(UTC)
+        if message is not None:
+            job.message = message
+        if result is not None:
+            job.result = result
+        await db.commit()
+
+
+async def _evict_stale(db: AsyncSession) -> None:
+    cutoff = datetime.now(UTC) - _MAX_FINISHED_AGE
+    await db.execute(
+        delete(Job).where(Job.finished_at.is_not(None), Job.finished_at < cutoff)
+    )
+    await db.commit()

app/services/scope_guide.py

@@ -33,9 +33,12 @@ async def build(
     """Return the context for ``scopes.html``.
 
     The scope rows are narrowed by the scope-attribute filters (``description``
-    matches the scope's description; the ``*_id`` filters match exactly).
+    matches either the scope's description or its SKU/product ID; the ``*_id``
-    ``fits_part_type_id`` keeps only scopes that have at least one (non-``none``)
+    filters match exactly). ``fits_part_type_id`` keeps only scopes that have
-    fit for that part type.
+    at least one (non-``none``)
+    fit for that part type. Callers that paginate should slice ``rows``
+    themselves — this always returns the full filtered set, since
+    ``fits_part_type_id`` is a post-filter that can't be pushed into the DB query.
 
     Shape::
 
@@ -69,6 +72,7 @@ async def build(
         sub_part_type_id=sub_part_type_id,
         part_type_id=scope_type.id,
         limit=1000,
+        match_sku=True,
     )
     scope_ids = {s.id for s in scopes}
     if not scope_ids:

app/static/css/style.css

@@ -567,6 +567,11 @@ textarea:focus {
   margin-bottom: 0.75rem;
 }
 
+/* Bulk actions (e.g. delete-selected above a checkbox table) */
+.bulk-actions {
+  margin-bottom: 0.75rem;
+}
+
 .filter-bar form {
   display: flex;
   flex-wrap: wrap;

app/templates/_scopes_table_card.html

@@ -18,9 +18,13 @@
   {%- endif -%}
 {%- endmacro -%}
 <div id="scopes-table-card" class="card" style="overflow-x:auto">
+  {% include "admin/_pagination.html" %}
   <table class="table sortable">
     <thead>
       <tr>
+        {% if can_write_scopes %}
+          <th data-sort-off><input type="checkbox" id="scopes-select-all" title="Select all"></th>
+        {% endif %}
         <th>Scope</th>
         <th>Manufacturer</th>
         <th>Model</th>
@@ -36,7 +40,10 @@
       {% for row in rows %}
         {% set scope = row.scope %}
         {% set info = row.info %}
-        <tr>
+        <tr id="scope-row-{{ scope.id }}">
+          {% if can_write_scopes %}
+            <td><input type="checkbox" name="scope_ids" value="{{ scope.id }}"></td>
+          {% endif %}
           <td>
             <strong><a href="/admin/augmented-parts/{{ scope.id }}">{{ scope.description }}</a></strong>
             {% if scope.sku %}<br><span class="muted">{{ scope.sku }}</span>{% endif %}
@@ -70,7 +77,7 @@
         </tr>
       {% else %}
         <tr>
-          <td colspan="{{ 6 + column_types | length }}" class="muted">
+          <td colspan="{{ 6 + column_types | length + (1 if can_write_scopes else 0) }}" class="muted">
             No scopes match your search.
           </td>
         </tr>

app/templates/admin/imported_parts.html

@@ -29,11 +29,9 @@
     <button class="btn btn-primary"
             hx-post="/admin/imported-parts/refresh"
             hx-target="#imported-parts-section"
-            hx-swap="outerHTML"
+            hx-swap="outerHTML">
-            hx-indicator="#refresh-indicator">
       Refresh from OData
     </button>
-    <span id="refresh-indicator" class="htmx-indicator muted"> Loading…</span>
   </div>
 
   {% include "admin/_imported_parts_table.html" %}

app/templates/scopes.html

@@ -15,7 +15,7 @@
           hx-target="#scopes-table-card"
           hx-trigger="input delay:300ms, change"
           hx-swap="outerHTML">
-      <input type="text" name="description" placeholder="Search scopes…"
+      <input type="text" name="description" placeholder="Search by description or product ID…"
              value="{{ filters.description or '' }}">
       <select name="manufacturer_id">
         <option value="">All manufacturers</option>
@@ -44,6 +44,50 @@
     </form>
   </div>
 
+  {% if can_write_scopes %}
+    <div class="bulk-actions">
+      <button id="delete-selected-scopes-btn" type="button" class="btn btn-danger" disabled
+              hx-post="/scopes/bulk-delete"
+              hx-include="[name='scope_ids']:checked, [name='description'], [name='manufacturer_id'], [name='scope_model_id'], [name='sub_part_type_id'], [name='fits_part_type_id']"
+              hx-target="#scopes-table-card"
+              hx-swap="outerHTML"
+              hx-confirm="Delete selected scopes?">Delete selected</button>
+    </div>
+  {% endif %}
+
   {% include "_scopes_table_card.html" %}
 </main>
+
+{% if can_write_scopes %}
+<script>
+(function () {
+  function updateDeleteButton() {
+    var count = document.querySelectorAll('input[name="scope_ids"]:checked').length;
+    var btn = document.getElementById('delete-selected-scopes-btn');
+    if (!btn) return;
+    btn.textContent = count > 0 ? 'Delete selected (' + count + ')' : 'Delete selected';
+    btn.disabled = count === 0;
+    btn.setAttribute('hx-confirm', 'Delete ' + count + ' selected scope' + (count === 1 ? '' : 's') + '?');
+  }
+
+  document.addEventListener('change', function (e) {
+    if (!e.target) return;
+
+    if (e.target.id === 'scopes-select-all') {
+      document.querySelectorAll('input[name="scope_ids"]').forEach(function (cb) {
+        cb.checked = e.target.checked;
+      });
+    }
+
+    if (e.target.id === 'scopes-select-all' || e.target.name === 'scope_ids') {
+      updateDeleteButton();
+    }
+  });
+
+  // Re-run after every HTMX swap (filter results, post-delete) so the button
+  // state reflects the freshly rendered checkboxes, which start unchecked.
+  document.addEventListener('htmx:afterSwap', updateDeleteButton);
+}());
+</script>
+{% endif %}
 {% endblock %}

app/web/admin.py

@@ -1,8 +1,5 @@
 """HTMX-driven admin dashboard for users, roles, and thread standards."""
 
-import math
-import urllib.parse
-
 from fastapi import (
     APIRouter,
     Depends,
@@ -17,9 +14,14 @@ from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import AsyncSessionLocal, get_db
+from app.crud import attachment_point as attachment_point_crud
+from app.crud import attachment_point_name as apn_crud
 from app.crud import audit as audit_crud
 from app.crud import augmented_part as augmented_part_crud
 from app.crud import augmented_part_note as augmented_part_note_crud
+from app.crud import clamp_connection as clamp_connection_crud
+from app.crud import fit as fit_crud
+from app.crud import fit_note as fit_note_crud
 from app.crud import imported_part as imported_part_crud
 from app.crud import imported_part_config as imported_part_config_crud
 from app.crud import iso_thread as iso_thread_crud
@@ -29,11 +31,6 @@ from app.crud import part_type as part_type_crud
 from app.crud import role as role_crud
 from app.crud import scope_info as scope_info_crud
 from app.crud import scope_model as scope_model_crud
-from app.crud import attachment_point as attachment_point_crud
-from app.crud import attachment_point_name as apn_crud
-from app.crud import clamp_connection as clamp_connection_crud
-from app.crud import fit as fit_crud
-from app.crud import fit_note as fit_note_crud
 from app.crud import static_connection as static_connection_crud
 from app.crud import sub_part_type as sub_part_type_crud
 from app.crud import unc_thread as unc_thread_crud
@@ -48,6 +45,8 @@ from app.services import fit_matching as fit_matching_service
 from app.services import jobs
 from app.services import odata as odata_service
 from app.services import scope_import as scope_import_service
+from app.web.pagination import paginate as _paginate
+from app.web.pagination import query_url as _query_url
 from app.web.templates import templates
 
 router = APIRouter(prefix="/admin", include_in_schema=False)
@@ -55,42 +54,6 @@ router = APIRouter(prefix="/admin", include_in_schema=False)
 _PAGE_SIZE = 50
 
 
-def _query_url(base_path: str, params: dict) -> str:
-    clean = {k: v for k, v in params.items() if v not in (None, "", 0)}
-    return f"{base_path}?{urllib.parse.urlencode(clean)}"
-
-
-def _paginate(
-    base_path: str,
-    filter_params: dict,
-    page: int,
-    page_size: int,
-    total: int,
-    target_id: str,
-) -> dict:
-    total_pages = max(1, math.ceil(total / page_size))
-    page = max(1, min(page, total_pages))
-    skip = (page - 1) * page_size
-
-    def _url(p: int) -> str:
-        return _query_url(
-            base_path, {**filter_params, "page": p, "page_size": page_size}
-        )
-
-    return {
-        "page": page,
-        "page_size": page_size,
-        "total": total,
-        "total_pages": total_pages,
-        "skip": skip,
-        "showing_from": skip + 1 if total > 0 else 0,
-        "showing_to": min(skip + page_size, total),
-        "prev_url": _url(page - 1) if page > 1 else None,
-        "next_url": _url(page + 1) if page < total_pages else None,
-        "target_id": target_id,
-    }
-
-
 def _sort_headers(
     base_path: str,
     filter_params: dict,
@@ -1366,6 +1329,19 @@ async def save_imported_part_config(
     return HTMLResponse("Configuration saved.", status_code=200)
 
 
+_IMPORTED_PARTS_REFRESH_TARGET_ID = "imported-parts-section"
+
+
+async def _imported_parts_table_ctx(
+    db: AsyncSession, *, sync_result: str | None = None, sync_error: str | None = None
+) -> dict:
+    return {
+        **await _default_imported_parts_table_ctx(db),
+        "sync_result": sync_result,
+        "sync_error": sync_error,
+    }
+
+
 @router.post("/imported-parts/refresh", response_class=HTMLResponse)
 async def refresh_imported_parts(
     request: Request,
@@ -1373,45 +1349,94 @@ async def refresh_imported_parts(
     actor: User = Depends(web_require("imported_parts:write")),
 ):
     config = await imported_part_config_crud.get(db)
-
-    async def _imported_table_ctx(sync_result=None, sync_error=None):
-        return {
-            **await _default_imported_parts_table_ctx(db),
-            "sync_result": sync_result,
-            "sync_error": sync_error,
-        }
-
     if not config or not config.odata_url:
         return templates.TemplateResponse(
             request,
             "admin/_imported_parts_table.html",
-            await _imported_table_ctx(sync_error="OData URL is not configured."),
+            await _imported_parts_table_ctx(
+                db, sync_error="OData URL is not configured."
+            ),
         )
+
+    actor_id = actor.id
+
+    async def _run(job_id: str) -> None:
+        async with AsyncSessionLocal() as db:
+            config = await imported_part_config_crud.get(db)
+            if not config or not config.odata_url:
+                await jobs.mark_error(job_id, "OData URL is not configured.")
+                return
             try:
                 records = await odata_service.fetch(
                     config.odata_url, config.username, config.password
                 )
                 created, updated = await imported_part_crud.upsert_bulk(db, records)
+            except Exception as exc:
+                await jobs.mark_error(job_id, f"Sync failed: {exc}")
+                return
+            actor_row = await user_crud.get(db, actor_id)
             await audit_crud.log(
                 db,
-            actor=actor,
+                actor=actor_row,
                 action="update",
                 resource_type="ImportedPart",
                 resource_id=None,
-            description=f"OData sync: {created} created, {updated} updated ({len(records)} total records)",
+                description=(
-        )
+                    f"OData sync: {created} created, {updated} updated "
-        return templates.TemplateResponse(
+                    f"({len(records)} total records)"
-            request,
-            "admin/_imported_parts_table.html",
-            await _imported_table_ctx(
-                sync_result=f"Sync complete: {created} created, {updated} updated."
                 ),
             )
-    except Exception as exc:
+            await jobs.mark_done(
+                job_id,
+                message=f"Sync complete: {created} created, {updated} updated.",
+            )
+
+    job = await jobs.start("imported_parts_refresh", _run)
+    return templates.TemplateResponse(
+        request,
+        "admin/_job_progress.html",
+        {
+            "container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
+            "poll_url": f"/admin/imported-parts/refresh/jobs/{job.id}",
+            "job": job,
+        },
+    )
+
+
+@router.get("/imported-parts/refresh/jobs/{job_id}", response_class=HTMLResponse)
+async def imported_parts_refresh_job_status(
+    request: Request,
+    job_id: str,
+    db: AsyncSession = Depends(get_db),
+    _: User = Depends(web_require("imported_parts:write")),
+):
+    job = await jobs.get(db, job_id)
+    if job is None:
         return templates.TemplateResponse(
             request,
             "admin/_imported_parts_table.html",
-            await _imported_table_ctx(sync_error=f"Sync failed: {exc}"),
+            await _imported_parts_table_ctx(
+                db, sync_error="Refresh job not found — it may have expired."
+            ),
+        )
+    if job.status == "running":
+        return templates.TemplateResponse(
+            request,
+            "admin/_job_progress.html",
+            {
+                "container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
+                "poll_url": f"/admin/imported-parts/refresh/jobs/{job_id}",
+                "job": job,
+            },
+        )
+    return templates.TemplateResponse(
+        request,
+        "admin/_imported_parts_table.html",
+        await _imported_parts_table_ctx(
+            db,
+            sync_result=job.message if job.status == "done" else None,
+            sync_error=job.message if job.status == "error" else None,
+        ),
     )
 
 
@@ -1511,13 +1536,12 @@ async def upload_scope_import(
     data = await file.read()
     actor_id = actor.id
 
-    async def _run(job: jobs.Job) -> None:
+    async def _run(job_id: str) -> None:
         async with AsyncSessionLocal() as db:
             try:
                 result = await scope_import_service.import_from_bytes(db, data)
             except Exception as exc:  # noqa: BLE001
-                job.status = "error"
+                await jobs.mark_error(job_id, f"Could not read spreadsheet: {exc}")
-                job.message = f"Could not read spreadsheet: {exc}"
                 return
             actor_row = await user_crud.get(db, actor_id)
             await audit_crud.log(
@@ -1532,10 +1556,16 @@ async def upload_scope_import(
                     f"{result.errors} errors"
                 ),
             )
-            job.status = "done"
+            await jobs.mark_done(
-            job.result = result
+                job_id,
+                message=(
+                    f"Scope import complete: {result.created} created, "
+                    f"{result.updated} updated."
+                ),
+                result=result.model_dump(mode="json"),
+            )
 
-    job = jobs.start("scope_import", _run)
+    job = await jobs.start("scope_import", _run)
     return templates.TemplateResponse(
         request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
     )
@@ -1545,9 +1575,10 @@ async def upload_scope_import(
 async def scope_import_job_status(
     request: Request,
     job_id: str,
+    db: AsyncSession = Depends(get_db),
     _: User = Depends(web_require("augmented_parts:write")),
 ):
-    job = jobs.get(job_id)
+    job = await jobs.get(db, job_id)
     if job is None:
         return templates.TemplateResponse(
             request,
@@ -1718,13 +1749,12 @@ async def sync_augmented_parts(
 ):
     actor_id = actor.id
 
-    async def _run(job: jobs.Job) -> None:
+    async def _run(job_id: str) -> None:
         async with AsyncSessionLocal() as db:
             try:
                 result = await augmented_part_sync_service.sync(db)
             except Exception as exc:
-                job.status = "error"
+                await jobs.mark_error(job_id, f"Sync failed: {exc}")
-                job.message = f"Sync failed: {exc}"
                 return
             actor_row = await user_crud.get(db, actor_id)
             await audit_crud.log(
@@ -1738,18 +1768,21 @@ async def sync_augmented_parts(
                     f"{result.skipped} skipped (no type match)"
                 ),
             )
-            job.status = "done"
+            await jobs.mark_done(
-            job.message = (
+                job_id,
-                f"Sync complete: {result.created} created, {result.deleted} deleted"
+                message=(
+                    f"Sync complete: {result.created} created, "
+                    f"{result.deleted} deleted"
                     + (
                         f", {result.skipped} skipped (type not recognised)"
                         if result.skipped
                         else ""
                     )
                     + "."
+                ),
             )
 
-    job = jobs.start("augmented_part_sync", _run)
+    job = await jobs.start("augmented_part_sync", _run)
     return templates.TemplateResponse(
         request,
         "admin/_job_progress.html",
@@ -1768,7 +1801,7 @@ async def augmented_parts_sync_job_status(
     db: AsyncSession = Depends(get_db),
     _: User = Depends(web_require("augmented_parts:write")),
 ):
-    job = jobs.get(job_id)
+    job = await jobs.get(db, job_id)
     if job is None:
         form_ctx = await _augmented_part_form_ctx(db)
         return templates.TemplateResponse(
@@ -2483,7 +2516,7 @@ async def recompute_augmented_part_fits(
 ):
     actor_id = actor.id
 
-    async def _run(job: jobs.Job) -> None:
+    async def _run(job_id: str) -> None:
         async with AsyncSessionLocal() as db:
             result = await fit_matching_service.recompute_for_part(
                 db, augmented_part_id
@@ -2500,10 +2533,11 @@ async def recompute_augmented_part_fits(
                     f"{result.created} created, {result.removed} removed"
                 ),
             )
-            job.status = "done"
+            await jobs.mark_done(
-            job.message = f"{result.created} created, {result.removed} removed."
+                job_id, f"{result.created} created, {result.removed} removed."
+            )
 
-    job = jobs.start("fit_recompute_part", _run)
+    job = await jobs.start("fit_recompute_part", _run)
     return templates.TemplateResponse(
         request,
         "admin/_job_progress.html",
@@ -2527,7 +2561,7 @@ async def augmented_part_fits_recompute_job_status(
     db: AsyncSession = Depends(get_db),
     actor: User = Depends(web_require("fits:write")),
 ):
-    job = jobs.get(job_id)
+    job = await jobs.get(db, job_id)
     if job is None:
         ctx = await _augmented_part_fits_ctx(
             db,
@@ -2774,7 +2808,7 @@ async def recompute_all_fits(
 ):
     actor_id = actor.id
 
-    async def _run(job: jobs.Job) -> None:
+    async def _run(job_id: str) -> None:
         async with AsyncSessionLocal() as db:
             result = await fit_matching_service.recompute_all(db)
             actor_row = await user_crud.get(db, actor_id)
@@ -2789,13 +2823,15 @@ async def recompute_all_fits(
                     f"{result.removed} removed"
                 ),
             )
-            job.status = "done"
+            await jobs.mark_done(
-            job.message = (
+                job_id,
+                message=(
                     f"Recompute complete: {result.created} created, "
                     f"{result.removed} removed."
+                ),
             )
 
-    job = jobs.start("fit_recompute_all", _run)
+    job = await jobs.start("fit_recompute_all", _run)
     return templates.TemplateResponse(
         request,
         "admin/_job_progress_inline.html",
@@ -2813,9 +2849,10 @@ async def recompute_all_fits(
 async def recompute_all_fits_job_status(
     request: Request,
     job_id: str,
+    db: AsyncSession = Depends(get_db),
     _: User = Depends(web_require("fits:write")),
 ):
-    job = jobs.get(job_id)
+    job = await jobs.get(db, job_id)
     if job is None:
         return HTMLResponse("Recompute job not found — it may have expired.")
     if job.status == "running":

app/web/pages.py

@@ -1,3 +1,5 @@
+from datetime import timedelta
+
 from fastapi import APIRouter, Depends, Form, HTTPException, Query, Request, status
 from fastapi.responses import HTMLResponse, RedirectResponse, Response
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -36,7 +38,7 @@ def _set_auth_cookie(response: Response, access_token: str) -> None:
         httponly=True,
         secure=settings.cookie_secure,
         samesite="lax",
-        max_age=settings.access_token_expire_minutes * 60,
+        max_age=settings.session_timeout_hours * 3600,
         path="/",
     )
 
@@ -79,7 +81,9 @@ async def login_submit(
             },
             status_code=status.HTTP_401_UNAUTHORIZED,
         )
-    access, _ = await issue_token_pair(db, user)
+    access, _ = await issue_token_pair(
+        db, user, timedelta(hours=settings.session_timeout_hours)
+    )
     response = RedirectResponse(
         url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
     )
@@ -110,7 +114,9 @@ async def login_ldap_submit(
             },
             status_code=status.HTTP_401_UNAUTHORIZED,
         )
-    access, _ = await issue_token_pair(db, user)
+    access, _ = await issue_token_pair(
+        db, user, timedelta(hours=settings.session_timeout_hours)
+    )
     response = RedirectResponse(
         url=_safe_next(next_url), status_code=status.HTTP_303_SEE_OTHER
     )

app/web/pagination.py

@@ -0,0 +1,40 @@
+"""Shared pagination helpers for HTMX table cards (admin screens, scopes page)."""
+
+import math
+import urllib.parse
+
+
+def query_url(base_path: str, params: dict) -> str:
+    clean = {k: v for k, v in params.items() if v not in (None, "", 0)}
+    return f"{base_path}?{urllib.parse.urlencode(clean)}"
+
+
+def paginate(
+    base_path: str,
+    filter_params: dict,
+    page: int,
+    page_size: int,
+    total: int,
+    target_id: str,
+) -> dict:
+    total_pages = max(1, math.ceil(total / page_size))
+    page = max(1, min(page, total_pages))
+    skip = (page - 1) * page_size
+
+    def _url(p: int) -> str:
+        return query_url(
+            base_path, {**filter_params, "page": p, "page_size": page_size}
+        )
+
+    return {
+        "page": page,
+        "page_size": page_size,
+        "total": total,
+        "total_pages": total_pages,
+        "skip": skip,
+        "showing_from": skip + 1 if total > 0 else 0,
+        "showing_to": min(skip + page_size, total),
+        "prev_url": _url(page - 1) if page > 1 else None,
+        "next_url": _url(page + 1) if page < total_pages else None,
+        "target_id": target_id,
+    }

app/web/scopes.py

@@ -5,11 +5,13 @@ non-Scope part type listing the accessories that fit each scope. Supports HTMX
 search/filter over scope attributes and required accessory fits.
 """
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, Form, Request
 from fastapi.responses import HTMLResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.database import get_db
+from app.crud import audit as audit_crud
+from app.crud import augmented_part as augmented_part_crud
 from app.crud import manufacturer as manufacturer_crud
 from app.crud import part_type as part_type_crud
 from app.crud import scope_model as scope_model_crud
@@ -17,17 +19,30 @@ from app.crud import sub_part_type as sub_part_type_crud
 from app.models.user import User
 from app.services import scope_guide as scope_guide_service
 from app.web.admin import web_require
+from app.web.pagination import paginate
 from app.web.templates import templates
 
 router = APIRouter(include_in_schema=False)
 
+_PAGE_SIZE = 50
+_TABLE_TARGET = "#scopes-table-card"
+_FILTER_PATH = "/scopes/filter"
+
 
 def _to_int(value: str) -> int | None:
     return int(value) if value.strip() else None
 
 
+def _can_write_scopes(user: User) -> bool:
+    return user.is_superuser or "augmented_parts:write" in user.permissions
+
+
 async def _build(db: AsyncSession, filters: dict) -> dict:
-    """Run the guide query for the given raw (string) filter values."""
+    """Run the guide query for the given raw (string) filter values.
+
+    Always returns the full matching row set — pagination is applied by the
+    caller by slicing ``rows`` (see ``_paginated_guide``).
+    """
     return await scope_guide_service.build(
         db,
         description=filters["description"],
@@ -38,6 +53,22 @@ async def _build(db: AsyncSession, filters: dict) -> dict:
     )
 
 
+async def _paginated_guide(
+    db: AsyncSession, filters: dict, page: int, page_size: int
+) -> tuple[dict, dict]:
+    """Run the guide query, then slice its rows to the requested page.
+
+    Returns ``(guide, pagination)`` where ``guide["rows"]`` is already sliced.
+    """
+    guide = await _build(db, filters)
+    all_rows = guide["rows"]
+    pagination = paginate(
+        _FILTER_PATH, filters, page, page_size, len(all_rows), _TABLE_TARGET
+    )
+    guide["rows"] = all_rows[pagination["skip"] : pagination["skip"] + page_size]
+    return guide, pagination
+
+
 async def _filter_options(db: AsyncSession) -> dict:
     """Dropdown option lists for the filter bar."""
     scope_type = await part_type_crud.get_by_name(
@@ -73,11 +104,18 @@ async def scopes_page(
         "sub_part_type_id": sub_part_type_id,
         "fits_part_type_id": fits_part_type_id,
     }
-    guide = await _build(db, filters)
+    guide, pagination = await _paginated_guide(db, filters, 1, _PAGE_SIZE)
     return templates.TemplateResponse(
         request,
         "scopes.html",
-        {"user": user, "filters": filters, **guide, **(await _filter_options(db))},
+        {
+            "user": user,
+            "filters": filters,
+            "can_write_scopes": _can_write_scopes(user),
+            "pagination": pagination,
+            **guide,
+            **(await _filter_options(db)),
+        },
     )
 
 
@@ -89,8 +127,10 @@ async def scopes_filter(
     scope_model_id: str = "",
     sub_part_type_id: str = "",
     fits_part_type_id: str = "",
+    page: int = 1,
+    page_size: int = _PAGE_SIZE,
     db: AsyncSession = Depends(get_db),
-    _: User = Depends(web_require("augmented_parts:read", "fits:read")),
+    user: User = Depends(web_require("augmented_parts:read", "fits:read")),
 ):
     filters = {
         "description": description,
@@ -99,5 +139,60 @@ async def scopes_filter(
         "sub_part_type_id": sub_part_type_id,
         "fits_part_type_id": fits_part_type_id,
     }
-    guide = await _build(db, filters)
+    guide, pagination = await _paginated_guide(db, filters, page, page_size)
-    return templates.TemplateResponse(request, "_scopes_table_card.html", guide)
+    return templates.TemplateResponse(
+        request,
+        "_scopes_table_card.html",
+        {
+            **guide,
+            "pagination": pagination,
+            "can_write_scopes": _can_write_scopes(user),
+        },
+    )
+
+
+@router.post("/scopes/bulk-delete", response_class=HTMLResponse)
+async def bulk_delete_scopes(
+    request: Request,
+    scope_ids: list[int] = Form([]),
+    description: str = Form(""),
+    manufacturer_id: str = Form(""),
+    scope_model_id: str = Form(""),
+    sub_part_type_id: str = Form(""),
+    fits_part_type_id: str = Form(""),
+    db: AsyncSession = Depends(get_db),
+    actor: User = Depends(web_require("augmented_parts:write")),
+):
+    for augmented_part_id in scope_ids:
+        ap = await augmented_part_crud.get(db, augmented_part_id)
+        if ap is None:
+            continue
+        label = ap.sku or ap.description
+        await augmented_part_crud.delete(db, ap)
+        await audit_crud.log(
+            db,
+            actor=actor,
+            action="delete",
+            resource_type="AugmentedPart",
+            resource_id=augmented_part_id,
+            description=audit_crud.describe_delete(
+                "AugmentedPart", augmented_part_id, label
+            ),
+        )
+    filters = {
+        "description": description,
+        "manufacturer_id": manufacturer_id,
+        "scope_model_id": scope_model_id,
+        "sub_part_type_id": sub_part_type_id,
+        "fits_part_type_id": fits_part_type_id,
+    }
+    guide, pagination = await _paginated_guide(db, filters, 1, _PAGE_SIZE)
+    return templates.TemplateResponse(
+        request,
+        "_scopes_table_card.html",
+        {
+            **guide,
+            "pagination": pagination,
+            "can_write_scopes": _can_write_scopes(actor),
+        },
+    )

tests/test_admin_background_jobs.py

@@ -2,7 +2,11 @@ import asyncio
 import re
 
 import pytest
+import pytest_asyncio
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
+from app.core.database import Base
+from app.crud import imported_part_config as imported_part_config_crud
 from app.crud import role as role_crud
 from app.models.attachment_point import AttachmentPoint
 from app.models.attachment_point_name import AttachmentPointName
@@ -13,6 +17,25 @@ from tests.test_scope_import import _scope_row, _workbook
 _JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"')
 
 
+@pytest_asyncio.fixture
+async def db_sessionmaker(tmp_path):
+    """Override conftest's in-memory+StaticPool fixture for this module.
+
+    These tests run a real background asyncio task (the job) concurrently
+    with polling requests, both hitting the DB at once. StaticPool pins every
+    session to one shared raw connection, which can't safely serve two
+    callers at the same time; a real file-backed DB with its own connection
+    pool (how SQLite/Postgres actually behave outside tests) can.
+    """
+    db_path = tmp_path / "jobs_test.db"
+    engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    maker = async_sessionmaker(engine, expire_on_commit=False)
+    yield maker
+    await engine.dispose()
+
+
 async def _web_login(client, email, password="secret123"):
     resp = await client.post("/login", data={"email": email, "password": password})
     assert resp.status_code == 303
@@ -39,6 +62,7 @@ async def test_scope_import_runs_in_background(
     client, seeded, make_user, db_sessionmaker, monkeypatch
 ):
     monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
     async with db_sessionmaker() as db:
         perms = seeded["perms"]
         editor = await role_crud.create(
@@ -69,6 +93,7 @@ async def test_augmented_parts_sync_runs_in_background(
     client, seeded, make_user, db_sessionmaker, monkeypatch
 ):
     monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
     async with db_sessionmaker() as db:
         perms = seeded["perms"]
         editor = await role_crud.create(
@@ -93,6 +118,7 @@ async def test_fit_recompute_for_part_runs_in_background(
     client, seeded, make_user, db_sessionmaker, monkeypatch
 ):
     monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
     async with db_sessionmaker() as db:
         apn = AttachmentPointName(name="Objective")
         part_t = PartType(name="Scope")
@@ -137,6 +163,7 @@ async def test_fit_recompute_all_runs_in_background(
     client, seeded, make_user, db_sessionmaker, monkeypatch
 ):
     monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
     async with db_sessionmaker() as db:
         perms = seeded["perms"]
         editor = await role_crud.create(
@@ -154,3 +181,64 @@ async def test_fit_recompute_all_runs_in_background(
 
     final_body = await _poll_until_done(client, match.group(1))
     assert "Recompute complete" in final_body
+
+
+@pytest.mark.asyncio
+async def test_imported_parts_refresh_runs_in_background(
+    client, seeded, make_user, db_sessionmaker, monkeypatch
+):
+    monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
+
+    async def _fake_fetch(url, username, password):
+        return [{"InventoryID": "INV-1", "Description": "Widget"}]
+
+    monkeypatch.setattr("app.web.admin.odata_service.fetch", _fake_fetch)
+
+    async with db_sessionmaker() as db:
+        await imported_part_config_crud.upsert(
+            db, odata_url="https://example.test/odata", username=None, password=None
+        )
+        perms = seeded["perms"]
+        editor = await role_crud.create(
+            db,
+            name="imported-parts-editor",
+            permission_ids=[perms["imported_parts:write"].id],
+        )
+        editor_id = editor.id
+    await make_user("refresher@example.com", role_ids=[editor_id])
+    await _web_login(client, "refresher@example.com")
+
+    resp = await client.post("/admin/imported-parts/refresh")
+    assert resp.status_code == 200
+    assert 'id="imported-parts-section"' in resp.text
+    match = _JOB_URL_RE.search(resp.text)
+    assert match, resp.text
+
+    final_body = await _poll_until_done(client, match.group(1))
+    assert "Sync complete: 1 created, 0 updated." in final_body
+    assert "INV-1" in final_body
+
+
+@pytest.mark.asyncio
+async def test_imported_parts_refresh_without_config_skips_job(
+    client, seeded, make_user, db_sessionmaker, monkeypatch
+):
+    monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
+    monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
+    async with db_sessionmaker() as db:
+        perms = seeded["perms"]
+        editor = await role_crud.create(
+            db,
+            name="imported-parts-editor-noconfig",
+            permission_ids=[perms["imported_parts:write"].id],
+        )
+        editor_id = editor.id
+    await make_user("refresher2@example.com", role_ids=[editor_id])
+    await _web_login(client, "refresher2@example.com")
+
+    resp = await client.post("/admin/imported-parts/refresh")
+    assert resp.status_code == 200
+    assert "OData URL is not configured." in resp.text
+    # No background job started for this synchronous validation failure.
+    assert _JOB_URL_RE.search(resp.text) is None

tests/test_scope_guide.py

@@ -147,6 +147,26 @@ async def test_build_filters_by_description(db_sessionmaker):
     assert _descriptions(guide) == ["Alpha 3-18x50"]
 
 
+@pytest.mark.asyncio
+async def test_build_filters_by_sku(db_sessionmaker):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+        ids["s1"].sku = "ARM-12345"
+        await db.commit()
+        guide = await scope_guide.build(db, description="ARM-12345")
+    assert _descriptions(guide) == ["Alpha 3-18x50"]
+
+
+@pytest.mark.asyncio
+async def test_build_filters_by_partial_sku(db_sessionmaker):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+        ids["s1"].sku = "ARM-12345"
+        await db.commit()
+        guide = await scope_guide.build(db, description="12345")
+    assert _descriptions(guide) == ["Alpha 3-18x50"]
+
+
 @pytest.mark.asyncio
 async def test_build_filters_by_manufacturer(db_sessionmaker):
     async with db_sessionmaker() as db:
@@ -330,3 +350,162 @@ async def test_scopes_page_redirects_when_anonymous(client):
     assert resp.status_code == 302
     # Deep link is preserved so the user lands back here after logging in.
     assert resp.headers["location"] == "/login?next=%2Fscopes"
+
+
+@pytest.mark.asyncio
+async def test_scopes_page_hides_checkboxes_for_read_only_user(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        await _seed_guide_data(db)
+    await make_user("reader6@example.com", role_ids=[seeded["roles"]["user"].id])
+    await _web_login(client, "reader6@example.com")
+    resp = await client.get("/scopes")
+    assert resp.status_code == 200
+    assert 'name="scope_ids"' not in resp.text
+    assert "delete-selected-scopes-btn" not in resp.text
+
+
+@pytest.mark.asyncio
+async def test_scopes_page_shows_checkboxes_for_write_user(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+    await make_user("writer@example.com", is_superuser=True)
+    await _web_login(client, "writer@example.com")
+    resp = await client.get("/scopes")
+    assert resp.status_code == 200
+    assert f'name="scope_ids" value="{ids["s1"].id}"' in resp.text
+    assert "delete-selected-scopes-btn" in resp.text
+    assert 'hx-post="/scopes/bulk-delete"' in resp.text
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_scopes_removes_selected_augmented_parts(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+    await make_user("writer2@example.com", is_superuser=True)
+    await _web_login(client, "writer2@example.com")
+
+    resp = await client.post(
+        "/scopes/bulk-delete",
+        data={"scope_ids": [str(ids["s1"].id), str(ids["s2"].id)]},
+    )
+    assert resp.status_code == 200
+
+    async with db_sessionmaker() as db:
+        from app.crud import augmented_part as augmented_part_crud
+
+        assert await augmented_part_crud.get(db, ids["s1"].id) is None
+        assert await augmented_part_crud.get(db, ids["s2"].id) is None
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_scopes_leaves_unselected_rows(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+    await make_user("writer3@example.com", is_superuser=True)
+    await _web_login(client, "writer3@example.com")
+
+    resp = await client.post(
+        "/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
+    )
+    assert resp.status_code == 200
+
+    async with db_sessionmaker() as db:
+        from app.crud import augmented_part as augmented_part_crud
+
+        assert await augmented_part_crud.get(db, ids["s1"].id) is None
+        assert await augmented_part_crud.get(db, ids["s2"].id) is not None
+
+
+@pytest.mark.asyncio
+async def test_scopes_page_shows_single_page_summary(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        await _seed_guide_data(db)  # 2 scopes, well under the default page size
+    await make_user("pager1@example.com", role_ids=[seeded["roles"]["user"].id])
+    await _web_login(client, "pager1@example.com")
+    resp = await client.get("/scopes")
+    assert resp.status_code == 200
+    assert "Showing 2 records." in resp.text
+    assert "Next" not in resp.text
+
+
+@pytest.mark.asyncio
+async def test_scopes_filter_paginates_results(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        scope_t = PartType(name="Scope")
+        db.add(scope_t)
+        await db.flush()
+        for i in range(5):
+            db.add(
+                AugmentedPart(
+                    description=f"Scope {i:02d}",
+                    colour_id=1,
+                    manufacturer_id=1,
+                    part_type_id=scope_t.id,
+                )
+            )
+        await db.commit()
+    await make_user("pager2@example.com", role_ids=[seeded["roles"]["user"].id])
+    await _web_login(client, "pager2@example.com")
+
+    resp = await client.get("/scopes/filter", params={"page_size": 2})
+    assert resp.status_code == 200
+    body = resp.text
+    assert "Scope 00" in body
+    assert "Scope 01" in body
+    assert "Scope 02" not in body
+    assert "Showing 1–2 of 5." in body
+
+    resp2 = await client.get("/scopes/filter", params={"page_size": 2, "page": 3})
+    assert resp2.status_code == 200
+    body2 = resp2.text
+    assert "Scope 04" in body2
+    assert "Scope 00" not in body2
+    assert "Showing 5–5 of 5." in body2
+
+
+@pytest.mark.asyncio
+async def test_scopes_filter_clamps_out_of_range_page(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        await _seed_guide_data(db)  # 2 scopes
+    await make_user("pager3@example.com", role_ids=[seeded["roles"]["user"].id])
+    await _web_login(client, "pager3@example.com")
+
+    resp = await client.get("/scopes/filter", params={"page_size": 1, "page": 99})
+    assert resp.status_code == 200
+    # Clamped to the last real page (2), not an empty page 99.
+    assert "Zeta 1-6x24" in resp.text
+    assert "Alpha 3-18x50" not in resp.text
+
+
+@pytest.mark.asyncio
+async def test_bulk_delete_scopes_forbidden_without_write_permission(
+    client, seeded, make_user, db_sessionmaker
+):
+    async with db_sessionmaker() as db:
+        ids = await _seed_guide_data(db)
+    await make_user("reader7@example.com", role_ids=[seeded["roles"]["user"].id])
+    await _web_login(client, "reader7@example.com")
+
+    resp = await client.post(
+        "/scopes/bulk-delete", data={"scope_ids": [str(ids["s1"].id)]}
+    )
+    assert resp.status_code == 403
+
+    async with db_sessionmaker() as db:
+        from app.crud import augmented_part as augmented_part_crud
+
+        assert await augmented_part_crud.get(db, ids["s1"].id) is not None