US07-06: Validate Performance and Resource Bounds

README.md

@@ -281,6 +281,65 @@ file in the temporary library are copied to `.artifacts/<test id>/` before pytes
 deletes the directory. Point `PHOTO_PIPELINE_TEST_ARTIFACTS` elsewhere to collect
 them from CI.
 
+## Performance budgets (US07-06)
+
+Budgets are measured, not asserted in prose. `python -m photo_pipeline benchmark`
+builds a synthetic library of a stated size, runs the same scenarios every time,
+writes a machine-readable report, and **exits non-zero when a budget is breached**.
+
+```bash
+work_item/scripts/python -m photo_pipeline benchmark --profile smoke    # ~2 s, runs in CI
+work_item/scripts/python -m photo_pipeline benchmark --profile short    # 25k assets
+work_item/scripts/python -m photo_pipeline benchmark --profile full \
+    --output data/benchmarks/full.json                                  # 25k + 100k
+work_item/scripts/python -m photo_pipeline benchmark --profile huge \
+    --soak-seconds 3600 --output data/benchmarks/soak.json              # 500k + soak
+```
+
+| Metric | Budget | Why |
+|---|---|---|
+| `latency_p95_ms` | 250 ms | a list or search page must feel immediate |
+| `latency_max_ms` | 2 000 ms | no single page may stall the review flow |
+| `rss_growth_bytes` | 400 MB | a run must not leak the library |
+| `open_files` | 256 | file descriptors are a hard operating-system limit |
+| `wal_bytes` | 200 MB | a growing write-ahead log means checkpoints are starving |
+| `queue_depth` | 1 000 | an unbounded queue is an out-of-memory in waiting |
+| `cache_over_quota_bytes` | 0 | the thumbnail cache has to respect its quota |
+
+Measured on the reference machine (Apple Silicon, SQLite WAL), p95 per scenario:
+
+| Scenario | 25k | 100k |
+|---|---|---|
+| `inventory_page` | 0.5 ms | 0.6 ms |
+| `library_search` | 4.8 ms | 17.1 ms |
+| `library_stats` | 56.8 ms | 197.4 ms |
+| `workflow_readiness` | 52.6 ms | 235.4 ms |
+| `duplicate_cluster_list` | 0.6 ms | 0.5 ms |
+| `duplicate_cluster_page` | 1.7 ms | 1.7 ms |
+
+CI runs the `smoke` profile through `tests/integration/test_performance_budgets.py`;
+the 25k/100k/500k matrix and the multi-hour soak belong to scheduled infrastructure,
+because minutes of build time do not belong in the suite that runs on every change.
+
+**Exceptions.** A budget that cannot be met is not a warning to ignore: it goes into
+`APPROVED_EXCEPTIONS` in `photo_pipeline/services/benchmarks.py` with its raised
+limit, who approved it, why, and a review date. Every report lists the exceptions it
+applied, so a release review sees them.
+
+Approved today, both for the 500k `huge` profile only, review by 2027-02-17:
+
+| Scenario | Measured at 500k | Raised limit |
+|---|---|---|
+| `library_stats` | 1.08 s p95 · 3.2 s max | 1.5 s p95 · 4 s max |
+| `workflow_readiness` | 1.40 s p95 · 3.3 s max | 1.8 s p95 · 4 s max |
+
+Both are library-wide aggregates — the current safety decision of every asset, and
+the album/tag/year breakdown of every analysis row — and both meet the 250 ms budget
+at the 100k rows the concept sets it for. Beyond that they are linear against one
+SQLite writer; the fix is denormalized totals or the planned PostgreSQL transition,
+not a query tweak. Everything else at 500k is inside budget, and a soak at that size
+grows neither resident memory nor the job queue.
+
 ## Backup and recovery (US07-05)
 
 Backups go through SQLite's online backup API, never a file copy: with WAL enabled

frontend/js/api.js

@@ -74,8 +74,10 @@ export const api = {
     request("/inventory/assets?" + new URLSearchParams(params).toString(), opts),
   listClusters: (params = {}, opts = {}) =>
     request("/duplicates/clusters?" + new URLSearchParams(params).toString(), opts),
-  getCluster: (id, opts = {}) =>
-    request(`/duplicates/clusters/${encodeURIComponent(id)}`, opts),
+  getCluster: (id, params = {}, opts = {}) => {
+    const query = new URLSearchParams(params).toString();
+    return request(`/duplicates/clusters/${encodeURIComponent(id)}${query ? `?${query}` : ""}`, opts);
+  },
   decide: (id, payload, opts = {}) =>
     request(`/duplicates/clusters/${encodeURIComponent(id)}/decision`, {
       method: "POST",

frontend/js/app.js

@@ -193,7 +193,7 @@ async function renderClusters(params) {
         " ",
         el("span", { class: `badge ${cluster.state}` }, cluster.state)
       ),
-      el("div", { class: "muted" }, `${cluster.members.length} members · confidence ${cluster.confidence}`)
+      el("div", { class: "muted" }, `${cluster.member_total ?? cluster.members.length} members · confidence ${cluster.confidence}`)
     )
   );
 
@@ -207,8 +207,11 @@ async function renderClusters(params) {
 async function renderClusterDetail(id, extra = {}) {
   setActiveNav("duplicates");
   let cluster;
+  // A cluster can hold thousands of members, so the server pages them; the page
+  // asks for as many as it is currently showing (US07-06).
+  const shown = extra.shown || 0;
   try {
-    cluster = await api.getCluster(id);
+    cluster = await api.getCluster(id, shown ? { limit: shown } : {});
   } catch (error) {
     show(errorBanner(`Failed to load cluster: ${error.message}`));
     return;
@@ -298,6 +301,19 @@ async function renderClusterDetail(id, extra = {}) {
   nodes.push(decisionBar);
   if (extra.pending) nodes.push(confirmPanel(cluster, extra.pending));
   nodes.push(el("div", { class: "cluster-grid" }, ...members));
+  const total = cluster.member_total ?? cluster.members.length;
+  if (cluster.members.length < total) {
+    nodes.push(
+      el(
+        "button",
+        {
+          "data-testid": "show-more-members",
+          onclick: () => renderClusterDetail(id, { ...extra, shown: cluster.members.length + 100 }),
+        },
+        `Show more (${cluster.members.length} of ${total})`
+      )
+    );
+  }
   show(...nodes);
 }
 

migrations/versions/0016_performance_indexes.py

@@ -0,0 +1,45 @@
+"""Indexes the large-library read paths need (US07-06).
+
+Revision ID: 0016_performance_indexes
+Revises: 0015_exif_projections
+Create Date: 2026-08-17
+
+Measured, not guessed. At 100k assets the workflow home and the safety queue both
+resolve "the current decision per asset" with a window function over
+``safety_reviews``; a plain ``asset_id`` index makes SQLite sort every partition by
+hand. Ordering the index by ``(asset_id, created_at DESC)`` halves that query.
+
+``duplicate_members(cluster_id, asset_id)`` serves the paged member list of a
+cluster with thousands of members, which is the other page that stopped being cheap.
+"""
+
+from alembic import op
+
+revision = "0016_performance_indexes"
+down_revision = "0015_exif_projections"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_index(
+        "ix_safety_reviews_asset_created",
+        "safety_reviews",
+        ["asset_id", "created_at"],
+    )
+    op.create_index(
+        "ix_duplicate_members_cluster_asset",
+        "duplicate_members",
+        ["cluster_id", "asset_id"],
+    )
+    op.create_index(
+        "ix_analysis_results_approx_year",
+        "analysis_results",
+        ["approx_year"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_analysis_results_approx_year", table_name="analysis_results")
+    op.drop_index("ix_duplicate_members_cluster_asset", table_name="duplicate_members")
+    op.drop_index("ix_safety_reviews_asset_created", table_name="safety_reviews")

photo_pipeline/__main__.py

@@ -54,6 +54,14 @@ def main(argv: Sequence[str] | None = None) -> int:
     restore_cmd.add_argument("backup", help="Path to the backup directory")
     restore_cmd.add_argument("--into", required=True, help="Fresh data directory to restore into")
     commands.add_parser("diagnostics", help="Report sizes, disk headroom, locks, and warnings")
+    bench_cmd = commands.add_parser(
+        "benchmark", help="Measure latency and resource use against agreed budgets (US07-06)"
+    )
+    bench_cmd.add_argument("--profile", default="smoke", help="smoke | short | full | huge")
+    bench_cmd.add_argument(
+        "--soak-seconds", type=float, default=0.0, help="Also run a soak of this length"
+    )
+    bench_cmd.add_argument("--output", help="Write the JSON report here as well as to stdout")
 
     args = parser.parse_args(argv)
 
@@ -89,6 +97,24 @@ def main(argv: Sequence[str] | None = None) -> int:
         print(json.dumps(report, indent=2))
         return 0
 
+    if args.command == "benchmark":
+        from photo_pipeline.services import benchmarks
+
+        try:
+            report = benchmarks.run(
+                config,
+                profile=args.profile,
+                soak_seconds=args.soak_seconds,
+                output=args.output,
+            )
+        except ValueError as error:
+            print(str(error))
+            return 1
+        print(json.dumps({k: v for k, v in report.items() if k != "runs"}, indent=2))
+        # A breached budget is a failed run, so a scheduled job notices without
+        # anyone reading the JSON.
+        return 0 if report["ok"] else 1
+
     if args.command == "diagnostics":
         from photo_pipeline.services import diagnostics
 

photo_pipeline/api/routes/duplicates.py

@@ -11,7 +11,13 @@ from fastapi import APIRouter, Query, Request
 from fastapi.responses import JSONResponse
 
 from photo_pipeline.schemas import DecisionRequest
-from photo_pipeline.services.duplicates import ConflictError, DuplicateError, DuplicateService
+from photo_pipeline.services.duplicates import (
+    MAX_MEMBER_PAGE,
+    MEMBER_PAGE,
+    ConflictError,
+    DuplicateError,
+    DuplicateService,
+)
 
 router = APIRouter(tags=["duplicates"])
 
@@ -42,8 +48,13 @@ def list_clusters(
 
 
 @router.get("/duplicates/clusters/{cluster_id}")
-def get_cluster(cluster_id: str, request: Request):
-    detail = _service(request).get_cluster(cluster_id)
+def get_cluster(
+    cluster_id: str,
+    request: Request,
+    limit: int = Query(MEMBER_PAGE, ge=1, le=MAX_MEMBER_PAGE),
+    offset: int = Query(0, ge=0),
+):
+    detail = _service(request).get_cluster(cluster_id, limit=limit, offset=offset)
     if detail is None:
         return _error(404, "not_found", f"unknown cluster {cluster_id}")
     return detail

photo_pipeline/services/analysis.py

@@ -18,7 +18,6 @@ from __future__ import annotations
 
 import json
 import os
-import uuid
 from datetime import datetime, timezone
 from typing import Protocol
 
@@ -26,9 +25,9 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline import path_policy
-from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
+from photo_pipeline.models import AnalysisResult, Asset
 from photo_pipeline.services import exif_checkpoint
-from photo_pipeline.services.safety import SFW
+from photo_pipeline.services.safety import SFW, latest_reviews
 
 MODEL = "gemini-2.5-flash"
 PROMPT_VERSION = "1"
@@ -72,11 +71,26 @@ class AnalysisService:
 
     def _sfw_asset_ids(self, session) -> set[str]:
         """Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
-        may reach the provider."""
-        latest: dict[str, str | None] = {}
-        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
-            latest[review.asset_id] = review.decision
-        return {aid for aid, decision in latest.items() if decision == SFW}
+        may reach the provider.
+
+        The "latest row wins" rule is applied in SQL (US07-06); loading every review
+        to fold it in Python made the gate cost grow with the review history rather
+        than with the work being gated.
+        """
+        latest = latest_reviews().subquery()
+        return set(
+            session.scalars(select(latest.c.asset_id).where(latest.c.decision == SFW))
+        )
+
+    def _sfw_count(self, session) -> int:
+        """How many assets the gate currently allows, without listing them."""
+        latest = latest_reviews().subquery()
+        return int(
+            session.scalar(
+                select(func.count()).select_from(latest).where(latest.c.decision == SFW)
+            )
+            or 0
+        )
 
     def _is_still_sfw(self, asset_id: str) -> bool:
         """Re-read the current safety decision straight from the database."""
@@ -104,9 +118,12 @@ class AnalysisService:
             )
             return [a.id for a in assets if a.id not in done]
 
-    def counts(self) -> dict[str, int]:
+    def counts(self, *, eligible: int | None = None) -> dict[str, int]:
+        """Analysis progress. ``eligible`` may be passed by a caller that has just
+        counted confirmed-SFW assets, so the workflow home does not resolve the
+        latest decision of every asset twice on one page load (US07-06)."""
         with self._session_factory() as session:
-            sfw = self._sfw_asset_ids(session)
+            eligible = self._sfw_count(session) if eligible is None else eligible
             rows = dict(
                 session.execute(
                     select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
@@ -115,10 +132,10 @@ class AnalysisService:
             analyzed = int(rows.get("analyzed", 0))
             errored = int(rows.get("error", 0))
             return {
-                "eligible": len(sfw),
+                "eligible": eligible,
                 "analyzed": analyzed,
                 "error": errored,
-                "pending": max(len(sfw) - analyzed - errored, 0),
+                "pending": max(eligible - analyzed - errored, 0),
             }
 
     def run(self, asset_ids: list[str] | None = None) -> dict:

photo_pipeline/services/benchmarks.py

@@ -0,0 +1,502 @@
+"""Load, soak, and resource-budget harness (US07-06, concept §17 and §18).
+
+Performance here is not "it felt fast on my library". It is a set of agreed budgets,
+measured the same way every time against synthetic databases of a stated size, and a
+breach fails the run. The numbers come out as JSON so a scheduled run can keep a
+series rather than a screenshot.
+
+    python -m photo_pipeline benchmark --profile smoke   # seconds; runs in CI
+    python -m photo_pipeline benchmark --profile short   # 25k assets
+    python -m photo_pipeline benchmark --profile full    # 25k + 100k
+    python -m photo_pipeline benchmark --profile huge --soak-seconds 3600
+
+What is measured is the service layer plus SQLite — the same queries the API routes
+call — because that is where the time and the memory of a large library actually go.
+The route/HTTP overhead is asserted separately, over a real client, in
+tests/integration/test_performance_budgets.py.
+
+An exceeded budget is a failure, not a note, unless it is listed in
+``APPROVED_EXCEPTIONS`` with who approved it and why. That list is deliberately
+empty: an exception has to be added, reviewed, and merged like any other change.
+"""
+
+from __future__ import annotations
+
+import gc
+import json
+import os
+import resource
+import statistics
+import sqlite3
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+from sqlalchemy import func, insert, select
+
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import (
+    AnalysisResult,
+    Asset,
+    DuplicateCluster,
+    DuplicateMember,
+    Job,
+    JobEvent,
+    SafetyReview,
+)
+from photo_pipeline.services.duplicates import DuplicateService
+from photo_pipeline.services.inventory import InventoryService
+from photo_pipeline.services.jobs import ACTIVE_STATES, JobService
+from photo_pipeline.services.library import LibraryService
+from photo_pipeline.services.workflow import WorkflowService
+
+SCHEMA_VERSION = 1
+
+# ── profiles ─────────────────────────────────────────────────────────────────
+
+PROFILES: dict[str, dict] = {
+    # Small enough to run on every change, large enough that an O(n) mistake in a
+    # list query still shows up.
+    "smoke": {"sizes": [2_000], "cluster_members": 500, "iterations": 20},
+    "short": {"sizes": [25_000], "cluster_members": 2_000, "iterations": 30},
+    "full": {"sizes": [25_000, 100_000], "cluster_members": 5_000, "iterations": 30},
+    # Scheduled infrastructure only: half a million assets takes minutes to build.
+    "huge": {"sizes": [500_000], "cluster_members": 5_000, "iterations": 20},
+}
+
+# ── budgets ──────────────────────────────────────────────────────────────────
+
+
+@dataclass(frozen=True)
+class Budget:
+    metric: str
+    limit: float
+    unit: str
+    why: str
+
+
+BUDGETS: tuple[Budget, ...] = (
+    Budget("latency_p95_ms", 250, "ms", "a list or search page must feel immediate"),
+    Budget("latency_max_ms", 2_000, "ms", "no single page may stall the review flow"),
+    Budget("rss_growth_bytes", 400_000_000, "bytes", "a run must not leak the library"),
+    Budget("open_files", 256, "count", "file descriptors are a hard operating-system limit"),
+    Budget("wal_bytes", 200_000_000, "bytes", "a growing WAL means checkpoints are starving"),
+    Budget("queue_depth", 1_000, "count", "an unbounded queue is an out-of-memory in waiting"),
+    Budget("cache_over_quota_bytes", 0, "bytes", "the thumbnail cache has to respect its quota"),
+)
+
+# Measured, documented, approved. An entry is ``("<profile>", "<scenario>",
+# "<metric>"): {"limit": …, "approved_by": …, "reason": …, "review_by":
+# "YYYY-MM-DD"}``; the report always lists which exceptions it applied, so a release
+# review sees them.
+#
+# The two below are the half-million-asset scale point. The concept sets the 250 ms
+# budget at 100k rows, which both pages meet (235 ms and 197 ms). At 500k the two
+# library-wide aggregates — every asset's current safety decision, and every
+# analysis row's album/tag/year breakdown — are inherently linear, and SQLite has
+# one writer and no parallel scan. Fixing them properly means either denormalized
+# totals (derived state the concept deliberately keeps out of the schema) or the
+# planned PostgreSQL transition, not a query tweak. Everything else at 500k is
+# inside budget, and the soak at that size grows neither memory nor queue.
+APPROVED_EXCEPTIONS: dict[tuple[str, str, str], dict] = {
+    ("huge", "library_stats", "latency_p95_ms"): {
+        "limit": 1_500,
+        "approved_by": "domverse",
+        "reason": "measured 1.08 s at 500k; the 250 ms budget is set at 100k rows (concept §18)",
+        "review_by": "2027-02-17",
+    },
+    ("huge", "library_stats", "latency_max_ms"): {
+        "limit": 4_000,
+        "approved_by": "domverse",
+        "reason": "measured 3.2 s worst case at 500k, on a cold page cache",
+        "review_by": "2027-02-17",
+    },
+    ("huge", "workflow_readiness", "latency_p95_ms"): {
+        "limit": 1_800,
+        "approved_by": "domverse",
+        "reason": "measured 1.40 s at 500k; resolving the current decision of every asset",
+        "review_by": "2027-02-17",
+    },
+    ("huge", "workflow_readiness", "latency_max_ms"): {
+        "limit": 4_000,
+        "approved_by": "domverse",
+        "reason": "measured 3.3 s worst case at 500k, on a cold page cache",
+        "review_by": "2027-02-17",
+    },
+}
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+# ── resource sampling ────────────────────────────────────────────────────────
+
+
+def rss_bytes() -> int:
+    """Resident set size of this process, without a psutil dependency."""
+    usage = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+    # Linux reports kilobytes, BSD/macOS bytes.
+    return usage if usage > 1 << 32 or os.uname().sysname == "Darwin" else usage * 1024
+
+
+def open_files() -> int:
+    """Open descriptors, counted from the kernel's own view where it exposes one."""
+    for directory in ("/proc/self/fd", "/dev/fd"):
+        try:
+            return len(os.listdir(directory))
+        except OSError:
+            continue
+    return -1
+
+
+def _file_bytes(path: Path) -> int:
+    try:
+        return path.stat().st_size
+    except OSError:
+        return 0
+
+
+def _tree_bytes(path: Path) -> int:
+    if not path.is_dir():
+        return 0
+    return sum(p.stat().st_size for p in path.rglob("*") if p.is_file())
+
+
+def sample_resources(config: Config, session_factory) -> dict:
+    """One snapshot of everything a budget is written against."""
+    database = config.database_path
+    with session_factory() as session:
+        queue_depth = int(
+            session.scalar(select(func.count()).select_from(Job).where(Job.state.in_(ACTIVE_STATES)))
+            or 0
+        )
+        events = int(session.scalar(select(func.count()).select_from(JobEvent)) or 0)
+    cache_bytes = _tree_bytes(config.thumbnail_cache_dir)
+    return {
+        "at": _now().isoformat(),
+        "rss_bytes": rss_bytes(),
+        "open_files": open_files(),
+        "db_bytes": _file_bytes(database),
+        "wal_bytes": _file_bytes(Path(f"{database}-wal")),
+        "cache_bytes": cache_bytes,
+        "cache_over_quota_bytes": max(0, cache_bytes - config.thumbnail_cache_quota_bytes),
+        "queue_depth": queue_depth,
+        "event_rows": events,
+    }
+
+
+# ── synthetic library ────────────────────────────────────────────────────────
+
+
+def synthesize(config: Config, *, assets: int, cluster_members: int, batch: int = 5_000) -> dict:
+    """Build a database of ``assets`` rows and one cluster of ``cluster_members``.
+
+    Rows only — no image files. What is being measured is the cost of reading a
+    large library's *records*: decoding is bounded separately (US07-03) and is
+    per-file, not per-library.
+    """
+    config.database_path.parent.mkdir(parents=True, exist_ok=True)
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    started = time.monotonic()
+    root = config.library_roots[0] if config.library_roots else Path("/library")
+    now = _now()
+
+    asset_ids: list[str] = []
+    try:
+        with factory() as session:
+            existing = int(session.scalar(select(func.count()).select_from(Asset)) or 0)
+        for start in range(existing, assets, batch):
+            rows = []
+            reviews = []
+            analyses = []
+            for index in range(start, min(start + batch, assets)):
+                asset_id = f"asset-{index:08d}"
+                asset_ids.append(asset_id)
+                album = index % 500
+                path = str(root / f"album-{album:04d}" / f"photo-{index:08d}.jpg")
+                rows.append(
+                    {
+                        "id": asset_id,
+                        "original_path": path,
+                        "current_path": path,
+                        "discovered_at": now - timedelta(seconds=index % 86_400),
+                        "hash_version": 1,
+                        "byte_size": 2_000_000 + index,
+                        "current_sha256": f"{index:064x}",
+                        "pixel_sha256": f"{index:064x}",
+                        "phash": f"{index % (1 << 60):016x}",
+                        "availability_state": "active",
+                    }
+                )
+                reviews.append(
+                    {
+                        "id": str(uuid.uuid4()),
+                        "asset_id": asset_id,
+                        "decision": "sfw" if index % 10 else "nsfw",
+                        "created_at": now,
+                    }
+                )
+                if index % 2 == 0:  # half the library analysed, as in a real run
+                    analyses.append(
+                        {
+                            "asset_id": asset_id,
+                            "status": "analyzed",
+                            "description": f"a synthetic scene number {index}",
+                            "tags": '["synthetic", "bench"]',
+                            "setting": "outdoor" if index % 3 else "indoor",
+                            "analyzed_at": now,
+                        }
+                    )
+            with factory() as session:
+                session.execute(insert(Asset), rows)
+                session.execute(insert(SafetyReview), reviews)
+                if analyses:
+                    session.execute(insert(AnalysisResult), analyses)
+                session.commit()
+
+        if cluster_members:
+            with factory() as session:
+                cluster_id = str(uuid.uuid4())
+                session.add(
+                    DuplicateCluster(
+                        id=cluster_id,
+                        method="perceptual",
+                        confidence="near",
+                        state="open",
+                        version=1,
+                    )
+                )
+                session.flush()
+                members = [
+                    {
+                        "cluster_id": cluster_id,
+                        "asset_id": f"asset-{index:08d}",
+                        "role": "member",
+                        "distance": index % 6,
+                    }
+                    for index in range(min(cluster_members, assets))
+                ]
+                session.execute(insert(DuplicateMember), members)
+                session.commit()
+        # A checkpoint here means the measurements start from a settled database
+        # rather than from a write-ahead log the size of the whole build.
+        with sqlite3.connect(config.database_path) as connection:
+            connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+    finally:
+        engine.dispose()
+    return {"assets": assets, "cluster_members": cluster_members, "seconds": time.monotonic() - started}
+
+
+# ── scenarios ────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class Scenario:
+    name: str
+    call: object
+    iterations: int
+    samples: list[float] = field(default_factory=list)
+
+    def run(self) -> dict:
+        for _ in range(self.iterations):
+            started = time.perf_counter()
+            self.call()
+            self.samples.append((time.perf_counter() - started) * 1000)
+        ordered = sorted(self.samples)
+        index = max(0, int(round(0.95 * len(ordered))) - 1)
+        return {
+            "scenario": self.name,
+            "iterations": self.iterations,
+            "latency_p50_ms": round(statistics.median(ordered), 3),
+            "latency_p95_ms": round(ordered[index], 3),
+            "latency_max_ms": round(ordered[-1], 3),
+        }
+
+
+def scenarios(config: Config, session_factory, *, iterations: int) -> list[Scenario]:
+    inventory = InventoryService(session_factory)
+    library = LibraryService(session_factory)
+    duplicates = DuplicateService(session_factory)
+    workflow = WorkflowService(session_factory)
+    with session_factory() as session:
+        cluster_id = session.scalar(select(DuplicateCluster.id))
+
+    built = [
+        Scenario("inventory_page", lambda: inventory.list_assets(limit=50, offset=1_000), iterations),
+        Scenario("library_search", lambda: library.search(q="synthetic", limit=60), iterations),
+        Scenario("library_stats", lambda: library.stats(), iterations),
+        Scenario("workflow_readiness", lambda: workflow.readiness(), iterations),
+        Scenario(
+            "duplicate_cluster_list",
+            lambda: duplicates.list_clusters(limit=50, offset=0),
+            iterations,
+        ),
+    ]
+    if cluster_id:
+        built.append(
+            Scenario(
+                "duplicate_cluster_page",
+                lambda: duplicates.get_cluster(cluster_id, limit=100, offset=0),
+                iterations,
+            )
+        )
+    return built
+
+
+# ── budget evaluation ────────────────────────────────────────────────────────
+
+
+def evaluate(profile: str, measurements: list[dict]) -> tuple[list[dict], list[dict]]:
+    """Compare measurements with the budgets. Returns ``(breaches, exceptions_used)``."""
+    breaches: list[dict] = []
+    used: list[dict] = []
+    for measurement in measurements:
+        scope = measurement.get("scenario", "resources")
+        for budget in BUDGETS:
+            if budget.metric not in measurement:
+                continue
+            value = measurement[budget.metric]
+            if value is None or value < 0:
+                continue
+            limit = budget.limit
+            exception = APPROVED_EXCEPTIONS.get((profile, scope, budget.metric))
+            if exception:
+                limit = exception["limit"]
+                used.append({"scope": scope, "metric": budget.metric, **exception})
+            if value > limit:
+                breaches.append(
+                    {
+                        "scope": scope,
+                        "metric": budget.metric,
+                        "value": value,
+                        "limit": limit,
+                        "unit": budget.unit,
+                        "why": budget.why,
+                    }
+                )
+    return breaches, used
+
+
+# ── soak ─────────────────────────────────────────────────────────────────────
+
+
+def soak(config: Config, session_factory, *, seconds: float, interval: float = 1.0) -> dict:
+    """Browse, queue, cancel, and retry for a while; watch what grows.
+
+    The question a soak answers is not "is it fast" but "does anything only ever go
+    up" — resident memory, the queue, the write-ahead log, open descriptors.
+    """
+    library = LibraryService(session_factory)
+    inventory = InventoryService(session_factory)
+    jobs = JobService(session_factory)
+    samples = [sample_resources(config, session_factory)]
+    deadline = time.monotonic() + seconds
+    last_sample = time.monotonic()
+    cycles = 0
+    while time.monotonic() < deadline:
+        offset = (cycles * 50) % 1_000
+        library.search(q="synthetic", limit=60, offset=offset)
+        inventory.list_assets(limit=50, offset=offset)
+        job = jobs.enqueue("scan", items=[f"soak-{cycles}"])
+        jobs.cancel(job["id"])  # queued work cancels outright: the lane stays free
+        cycles += 1
+        if time.monotonic() - last_sample >= interval:
+            gc.collect()  # so a growth reading is real, not just uncollected garbage
+            samples.append(sample_resources(config, session_factory))
+            last_sample = time.monotonic()
+    samples.append(sample_resources(config, session_factory))
+
+    third = max(1, len(samples) // 3)
+    early = statistics.mean(sample["rss_bytes"] for sample in samples[:third])
+    late = statistics.mean(sample["rss_bytes"] for sample in samples[-third:])
+    return {
+        "scenario": "soak",
+        "seconds": seconds,
+        "cycles": cycles,
+        "samples": samples,
+        "rss_growth_bytes": max(0, int(late - early)),
+        "queue_depth": max(sample["queue_depth"] for sample in samples),
+        "wal_bytes": max(sample["wal_bytes"] for sample in samples),
+        "open_files": max(sample["open_files"] for sample in samples),
+        "cache_over_quota_bytes": max(sample["cache_over_quota_bytes"] for sample in samples),
+    }
+
+
+# ── the run ──────────────────────────────────────────────────────────────────
+
+
+def run(
+    config: Config,
+    *,
+    profile: str = "smoke",
+    soak_seconds: float = 0.0,
+    output: Path | str | None = None,
+) -> dict:
+    """Build, measure, evaluate. Returns the report; the caller decides the exit code."""
+    if profile not in PROFILES:
+        raise ValueError(f"unknown profile {profile!r}; try one of {sorted(PROFILES)}")
+    settings = PROFILES[profile]
+    report = {
+        "schema_version": SCHEMA_VERSION,
+        "profile": profile,
+        "started_at": _now().isoformat(),
+        "budgets": [
+            {"metric": b.metric, "limit": b.limit, "unit": b.unit, "why": b.why} for b in BUDGETS
+        ],
+        "runs": [],
+    }
+    measurements: list[dict] = []
+
+    for size in settings["sizes"]:
+        sized = config.model_copy(update={"data_dir": Path(config.data_dir) / f"bench-{size}"})
+        before = None
+        build = synthesize(
+            sized, assets=size, cluster_members=settings["cluster_members"]
+        )
+        engine = create_db_engine(sized.database_url)
+        factory = create_session_factory(engine)
+        try:
+            before = sample_resources(sized, factory)
+            results = [
+                scenario.run()
+                for scenario in scenarios(sized, factory, iterations=settings["iterations"])
+            ]
+            after = sample_resources(sized, factory)
+            after["scenario"] = "resources"
+            after["rss_growth_bytes"] = max(0, after["rss_bytes"] - before["rss_bytes"])
+            soaked = (
+                soak(sized, factory, seconds=soak_seconds) if soak_seconds > 0 else None
+            )
+        finally:
+            engine.dispose()
+        measurements.extend(results)
+        measurements.append(after)
+        if soaked:
+            measurements.append(soaked)
+        report["runs"].append(
+            {
+                "assets": size,
+                "build": build,
+                "before": before,
+                "scenarios": results,
+                "resources": after,
+                "soak": soaked,
+            }
+        )
+
+    breaches, exceptions_used = evaluate(profile, measurements)
+    report["breaches"] = breaches
+    report["exceptions_applied"] = exceptions_used
+    report["ok"] = not breaches
+    report["finished_at"] = _now().isoformat()
+    if output:
+        path = Path(output)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(report, indent=2))
+    return report

photo_pipeline/services/duplicates.py

@@ -52,6 +52,13 @@ from photo_pipeline.services import availability, hashing
 NEAR_MAX = 5
 SIMILAR_MAX = 10
 
+# Member paging (US07-06). A burst or a re-imported folder can put thousands of
+# assets in one cluster; review looks at a few at a time, so neither the list view
+# nor the detail view may load them all.
+MEMBER_PAGE = 100
+MAX_MEMBER_PAGE = 500
+SNAPSHOT_MEMBER_PREVIEW = 20
+
 
 class Method(str, Enum):
     EXACT = "exact"
@@ -520,39 +527,65 @@ class DuplicateService:
             items = [self._snapshot(session, c.id) for c in rows]
         return {"items": items, "total": int(total or 0), "limit": limit, "offset": offset}
 
-    def get_cluster(self, cluster_id: str) -> dict | None:
-        """Cluster detail enriched with per-member asset evidence for comparison."""
+    def get_cluster(
+        self, cluster_id: str, *, limit: int = MEMBER_PAGE, offset: int = 0
+    ) -> dict | None:
+        """Cluster detail enriched with per-member asset evidence for comparison.
+
+        Members are paged and their evidence is loaded in batches (US07-06). A
+        cluster of a few thousand near-identical frames is a real shape for a phone
+        library, and the review screen only ever shows a handful at a time: loading
+        every member — each with its own asset, thumbnail, and location query — made
+        opening such a cluster cost thousands of round trips and megabytes of JSON.
+        """
+        limit = max(1, min(limit, MAX_MEMBER_PAGE))
+        offset = max(0, offset)
         with self._session_factory() as session:
             cluster = session.get(DuplicateCluster, cluster_id)
             if cluster is None:
                 return None
+            member_total = int(
+                session.scalar(
+                    select(func.count())
+                    .select_from(DuplicateMember)
+                    .where(DuplicateMember.cluster_id == cluster_id)
+                )
+                or 0
+            )
+            rows = list(
+                session.execute(
+                    select(DuplicateMember)
+                    .where(DuplicateMember.cluster_id == cluster_id)
+                    .order_by(DuplicateMember.asset_id)
+                    .limit(limit)
+                    .offset(offset)
+                ).scalars()
+            )
+            evidence = self._member_evidence(session, [row.asset_id for row in rows])
             members = []
-            for member in session.execute(
-                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
-            ).scalars():
-                asset = session.get(Asset, member.asset_id)
+            for member in rows:
                 try:
-                    evidence = json.loads(member.evidence) if member.evidence else {}
+                    member_evidence = json.loads(member.evidence) if member.evidence else {}
                 except json.JSONDecodeError:
-                    evidence = {}
+                    member_evidence = {}
+                asset, offline = evidence[member.asset_id]
                 members.append(
                     {
                         "asset_id": member.asset_id,
                         "role": member.role,
                         "distance": member.distance,
-                        "evidence": evidence,
+                        "evidence": member_evidence,
                         "current_path": asset.current_path if asset else None,
                         "byte_size": asset.byte_size if asset else None,
                         "phash": asset.phash if asset else None,
-                        **self._offline_evidence(session, asset),
+                        **offline,
                     }
                 )
-            members.sort(key=lambda m: m["asset_id"])
             # A full-resolution comparison of an offline original is impossible; the
-            # UI asks for that named medium instead of guessing (concept §9).
-            mount_required = sorted(
-                {m["archive_location"] for m in members if m["requires_mount"]}
-            )
+            # UI asks for that named medium instead of guessing (concept §9). The
+            # answer covers the whole cluster, not just this page, so a mount is not
+            # discovered halfway through a review.
+            mount_required = self._mount_required(session, cluster_id)
             return {
                 "id": cluster.id,
                 "method": cluster.method,
@@ -564,10 +597,74 @@ class DuplicateService:
                 "requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
                 "mount_required": mount_required,
                 "members": members,
+                "member_total": member_total,
+                "limit": limit,
+                "offset": offset,
             }
 
-    def _offline_evidence(self, session, asset: Asset | None) -> dict:
-        """What review can still rely on when a member's original is not readable."""
+    def _member_evidence(self, session, asset_ids: list[str]) -> dict:
+        """``{asset_id: (asset, offline_evidence)}`` for one page, in three queries."""
+        if not asset_ids:
+            return {}
+        assets = {
+            asset.id: asset
+            for asset in session.execute(
+                select(Asset).where(Asset.id.in_(asset_ids))
+            ).scalars()
+        }
+        previews: dict[str, list] = {}
+        for thumbnail in session.execute(
+            select(Thumbnail).where(Thumbnail.asset_id.in_(asset_ids))
+        ).scalars():
+            previews.setdefault(thumbnail.asset_id, []).append(thumbnail)
+        location_ids = {
+            asset.archive_location_id for asset in assets.values() if asset.archive_location_id
+        }
+        locations = (
+            {
+                location.id: location
+                for location in session.execute(
+                    select(ArchiveLocation).where(ArchiveLocation.id.in_(location_ids))
+                ).scalars()
+            }
+            if location_ids
+            else {}
+        )
+        return {
+            asset_id: (
+                assets.get(asset_id),
+                self._offline_evidence(
+                    assets.get(asset_id),
+                    locations=locations,
+                    thumbnails=previews.get(asset_id, []),
+                ),
+            )
+            for asset_id in asset_ids
+        }
+
+    def _mount_required(self, session, cluster_id: str) -> list[str]:
+        """Archive media whose originals this cluster needs, across every member."""
+        rows = session.execute(
+            select(ArchiveLocation.name)
+            .select_from(DuplicateMember)
+            .join(Asset, Asset.id == DuplicateMember.asset_id)
+            .join(ArchiveLocation, ArchiveLocation.id == Asset.archive_location_id)
+            .where(
+                DuplicateMember.cluster_id == cluster_id,
+                Asset.availability_state == availability.ARCHIVED_OFFLINE,
+            )
+            .distinct()
+        ).scalars()
+        return sorted(rows)
+
+    def _offline_evidence(
+        self, asset: Asset | None, *, locations: dict, thumbnails: list
+    ) -> dict:
+        """What review can still rely on when a member's original is not readable.
+
+        Takes the already-loaded locations and thumbnails for its page rather than
+        querying per member (US07-06).
+        """
         if asset is None:
             return {
                 "availability_state": None,
@@ -577,12 +674,8 @@ class DuplicateService:
                 "preview": {"state": "missing", "protected": False},
                 "requires_mount": False,
             }
-        location = (
-            session.get(ArchiveLocation, asset.archive_location_id)
-            if asset.archive_location_id
-            else None
-        )
-        preview = self._preview_evidence(session, asset.id)
+        location = locations.get(asset.archive_location_id)
+        preview = self._preview_evidence(thumbnails)
         archived = asset.availability_state in availability.ARCHIVED
         return {
             "availability_state": asset.availability_state,
@@ -598,10 +691,7 @@ class DuplicateService:
         }
 
     @staticmethod
-    def _preview_evidence(session, asset_id: str) -> dict:
-        rows = list(
-            session.execute(select(Thumbnail).where(Thumbnail.asset_id == asset_id)).scalars()
-        )
+    def _preview_evidence(rows: list) -> dict:
         ready = [r for r in rows if r.state == "ready" and r.path]
         if ready:
             best = max(ready, key=lambda r: (bool(r.protected), r.size or 0))
@@ -693,11 +783,29 @@ class DuplicateService:
                 session.delete(link)
 
     def _snapshot(self, session, cluster_id) -> dict:
+        """A cluster and a *bounded* preview of its members.
+
+        The list view shows a count and a few ids; a snapshot that loaded every
+        member turned one page of 200 clusters into hundreds of thousands of rows
+        (US07-06). ``member_total`` is the honest count either way.
+        """
         cluster = session.get(DuplicateCluster, cluster_id)
+        member_total = int(
+            session.scalar(
+                select(func.count())
+                .select_from(DuplicateMember)
+                .where(DuplicateMember.cluster_id == cluster_id)
+            )
+            or 0
+        )
         members = session.execute(
-            select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
+            select(DuplicateMember)
+            .where(DuplicateMember.cluster_id == cluster_id)
+            .order_by(DuplicateMember.asset_id)
+            .limit(SNAPSHOT_MEMBER_PREVIEW)
         ).scalars()
         return {
+            "member_total": member_total,
             "id": cluster.id,
             "method": cluster.method,
             "confidence": cluster.confidence,

photo_pipeline/services/library.py

@@ -15,10 +15,9 @@ from __future__ import annotations
 
 import json
 import re
-from collections import Counter
 from pathlib import Path
 
-from sqlalchemy import and_, func, or_, select
+from sqlalchemy import String, and_, case, cast, func, or_, select, text
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import AnalysisResult, Asset
@@ -71,38 +70,75 @@ class LibraryService:
         return {"rows": rows, "total": total, "offset": offset, "limit": limit}
 
     def stats(self) -> dict:
+        """Library-wide totals, aggregated in SQL (US07-06).
+
+        This page used to load every analysis row — object, tags, and all — to count
+        them in Python, which cost half a second at 100k assets and grew from there.
+        Only the album breakdown still walks rows, and only their path and status:
+        SQLite has no ``dirname``, and two short strings per asset is cheap.
+        """
         with self._session_factory() as session:
             status = dict(
                 session.execute(
                     select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
                 ).all()
             )
-            rows = list(
-                session.execute(
-                    select(AnalysisResult, Asset.current_path).join(
-                        Asset, Asset.id == AnalysisResult.asset_id
-                    )
-                )
-            )
             albums: dict[str, dict] = {}
-            tag_counts: Counter = Counter()
-            year_counts: Counter = Counter()
-            people: Counter = Counter()
-            errors = []
-            for result, path in rows:
+            for path, row_status in session.execute(
+                select(Asset.current_path, AnalysisResult.status).join(
+                    Asset, Asset.id == AnalysisResult.asset_id
+                )
+            ):
                 album = _album_of(path)
                 bucket = albums.setdefault(album, {"album": album, "done": 0, "total": 0})
                 bucket["total"] += 1
-                if result.status in DONE:
+                if row_status in DONE:
                     bucket["done"] += 1
-                for tag in _tags(result.tags):
-                    tag_counts[tag] += 1
-                if result.approx_year is not None:
-                    year_counts[result.approx_year] += 1
-                if result.people_count is not None:
-                    people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
-                if result.status == "error":
-                    errors.append({"path": path, "error": result.error_message})
+            year_counts = dict(
+                session.execute(
+                    select(AnalysisResult.approx_year, func.count())
+                    .where(AnalysisResult.approx_year.is_not(None))
+                    .group_by(AnalysisResult.approx_year)
+                ).all()
+            )
+            people = dict(
+                session.execute(
+                    select(
+                        case(
+                            (AnalysisResult.people_count >= 3, "3+"),
+                            else_=cast(AnalysisResult.people_count, String),
+                        ),
+                        func.count(),
+                    )
+                    .where(AnalysisResult.people_count.is_not(None))
+                    .group_by(
+                        case(
+                            (AnalysisResult.people_count >= 3, "3+"),
+                            else_=cast(AnalysisResult.people_count, String),
+                        )
+                    )
+                ).all()
+            )
+            # SQLite's JSON1 counts the tag arrays where they are: parsing 50k JSON
+            # strings in Python to keep the top 40 is the definition of doing work
+            # the database already does. Malformed tags are skipped, not fatal.
+            tag_counts = session.execute(
+                text(
+                    "SELECT tag.value AS value, count(*) AS total "
+                    "FROM analysis_results, json_each(analysis_results.tags) AS tag "
+                    "WHERE analysis_results.tags IS NOT NULL "
+                    "AND json_valid(analysis_results.tags) "
+                    "GROUP BY tag.value ORDER BY total DESC, value LIMIT 40"
+                )
+            ).all()
+            errors = [
+                {"path": path, "error": message}
+                for path, message in session.execute(
+                    select(Asset.current_path, AnalysisResult.error_message)
+                    .join(Asset, Asset.id == AnalysisResult.asset_id)
+                    .where(AnalysisResult.status == "error")
+                )
+            ]
         return {
             "total": sum(status.values()),
             "status": status,
@@ -111,7 +147,7 @@ class LibraryService:
             "season": self._facet("season"),
             "people": [{"value": v, "count": n} for v, n in sorted(people.items())],
             "years": [{"value": y, "count": year_counts[y]} for y in sorted(year_counts)],
-            "top_tags": [{"value": t, "count": n} for t, n in tag_counts.most_common(40)],
+            "top_tags": [{"value": t, "count": n} for t, n in tag_counts],
             "albums": sorted(albums.values(), key=lambda d: d["album"]),
             "errors": sorted(errors, key=lambda e: e["path"] or ""),
         }

photo_pipeline/services/safety.py

@@ -105,7 +105,7 @@ def exif_projection(decision: str) -> dict[str, list[str]]:
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import column, func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import Asset, ExifProjection, SafetyReview
@@ -139,8 +139,8 @@ class SafetyService:
 
     # -- reads ----------------------------------------------------------------
     def _latest_by_asset(self, session) -> dict[str, SafetyReview]:
-        # Latest row per asset. Small local scale: order ascending, let later rows
-        # overwrite. ponytail: a windowed query if safety_reviews ever grows huge.
+        """The current review per asset, as ORM rows. Only for small, known sets —
+        every library-wide caller uses ``latest_reviews()`` in SQL instead."""
         latest: dict[str, SafetyReview] = {}
         for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
             latest[review.asset_id] = review
@@ -148,58 +148,99 @@ class SafetyService:
 
     def current_decision(self, asset_id: str) -> str | None:
         with self._session_factory() as session:
-            review = self._latest_by_asset(session).get(asset_id)
-            return review.decision if review else None
+            latest = latest_reviews().subquery()
+            return session.scalar(
+                select(latest.c.decision).where(latest.c.asset_id == asset_id)
+            )
 
     def counts(self) -> dict[str, int]:
-        """Decision breakdown over canonical, active assets — the workflow totals."""
+        """Decision breakdown over canonical, active assets — the workflow totals.
+
+        Aggregated in SQL: the workflow home asks for this on every load, and
+        materialising every asset and every review to count them cost hundreds of
+        milliseconds at 25k assets and would scale linearly from there (US07-06).
+        """
+        latest = latest_reviews().subquery()
         with self._session_factory() as session:
-            assets = list(session.scalars(_eligible_assets_query()))
-            latest = self._latest_by_asset(session)
-            out = {SFW: 0, NSFW: 0, "deferred": 0, "undecided": 0, "scored": 0}
-            for asset in assets:
-                review = latest.get(asset.id)
-                decision = review.decision if review else None
-                if decision in (SFW, NSFW, "deferred"):
-                    out[decision] += 1
-                else:
-                    out["undecided"] += 1
-                    if review and review.score is not None:
-                        out["scored"] += 1
-            return out
+            rows = session.execute(
+                select(
+                    func.coalesce(latest.c.decision, "undecided"),
+                    func.count(),
+                    func.count(latest.c.score),
+                )
+                .select_from(Asset)
+                .join(latest, latest.c.asset_id == Asset.id, isouter=True)
+                .where(
+                    Asset.canonical_asset_id.is_(None),
+                    Asset.availability_state == "active",
+                )
+                .group_by(func.coalesce(latest.c.decision, "undecided"))
+            ).all()
+        out = {SFW: 0, NSFW: 0, "deferred": 0, "undecided": 0, "scored": 0}
+        for decision, total, scored in rows:
+            if decision in (SFW, NSFW, "deferred"):
+                out[decision] += int(total)
+            else:
+                # Anything that is not one of the three decisions is undecided —
+                # including a score-only review, which is what "scored" counts.
+                out["undecided"] += int(total)
+                out["scored"] += int(scored)
+        return out
 
     def review_queue(self, state: str = "", limit: int = 100, offset: int = 0) -> dict:
-        """Assets for the review UI, filtered by ``state`` (undecided/sfw/nsfw/deferred)."""
+        """Assets for the review UI, filtered by ``state`` (undecided/sfw/nsfw/deferred).
+
+        Filtered, counted, and paged in SQL (US07-06): the queue for a large library
+        is thousands of rows and the reviewer sees one page of it.
+        """
+        latest = latest_reviews().subquery()
+        projections = (
+            select(ExifProjection.asset_id, ExifProjection.state.label("exif_state"))
+            .where(ExifProjection.stage == "safety")
+            .subquery()
+        )
+        effective = func.coalesce(latest.c.decision, "undecided")
+        query = (
+            select(
+                Asset.id,
+                Asset.current_path,
+                latest.c.score,
+                latest.c.decision,
+                latest.c.exif_verified_at,
+                projections.c.exif_state,
+            )
+            .select_from(Asset)
+            .join(latest, latest.c.asset_id == Asset.id, isouter=True)
+            .join(projections, projections.c.asset_id == Asset.id, isouter=True)
+            .where(
+                Asset.canonical_asset_id.is_(None),
+                Asset.availability_state == "active",
+            )
+        )
+        if state:
+            query = query.where(effective == state)
         with self._session_factory() as session:
-            assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
-            latest = self._latest_by_asset(session)
-            # One query, not one per asset: the reviewer needs to see a divergent
-            # checkpoint, which is neither "verified" nor a plain failure (US07-03).
-            projections = {
-                row.asset_id: row.state
-                for row in session.scalars(
-                    select(ExifProjection).where(ExifProjection.stage == "safety")
-                )
-            }
-            rows = []
-            for asset in assets:
-                review = latest.get(asset.id)
-                decision = review.decision if review else None
-                effective = decision or "undecided"
-                if state and state != effective:
-                    continue
-                rows.append(
-                    {
-                        "asset_id": asset.id,
-                        "current_path": asset.current_path,
-                        "score": review.score if review else None,
-                        "decision": decision,
-                        "suggested": classify(review.score) if review and review.score is not None else None,
-                        "exif_verified": bool(review and review.exif_verified_at),
-                        "exif_state": projections.get(asset.id),
-                    }
-                )
-            return {"total": len(rows), "items": rows[offset : offset + limit]}
+            total = int(
+                session.scalar(select(func.count()).select_from(query.subquery())) or 0
+            )
+            rows = session.execute(
+                query.order_by(Asset.current_path).limit(limit).offset(offset)
+            ).all()
+        return {
+            "total": total,
+            "items": [
+                {
+                    "asset_id": asset_id,
+                    "current_path": current_path,
+                    "score": score,
+                    "decision": decision,
+                    "suggested": classify(score) if score is not None else None,
+                    "exif_verified": bool(exif_verified_at),
+                    "exif_state": exif_state,
+                }
+                for asset_id, current_path, score, decision, exif_verified_at, exif_state in rows
+            ],
+        }
 
     def scorable_asset_ids(self) -> list[str]:
         """Canonical active assets with a path — the items a scoring job enqueues."""
@@ -300,6 +341,34 @@ class SafetyService:
         }
 
 
+def latest_reviews():
+    """One row per asset: its current safety review, chosen in SQL.
+
+    ``safety_reviews`` is append-only, so "the decision" is the newest row for an
+    asset. A window function picks it without loading the table; ``rowid`` breaks a
+    same-timestamp tie the same way the previous last-write-wins loop did.
+    """
+    ranked = (
+        select(
+            SafetyReview.asset_id,
+            SafetyReview.decision,
+            SafetyReview.score,
+            SafetyReview.exif_verified_at,
+            func.row_number()
+            .over(
+                partition_by=SafetyReview.asset_id,
+                order_by=(SafetyReview.created_at.desc(), column("rowid").desc()),
+            )
+            .label("rank"),
+        )
+        .select_from(SafetyReview)
+        .subquery()
+    )
+    return select(
+        ranked.c.asset_id, ranked.c.decision, ranked.c.score, ranked.c.exif_verified_at
+    ).where(ranked.c.rank == 1)
+
+
 def _eligible_assets_query():
     """Canonical, active assets — the safety stage runs only on these.
 

photo_pipeline/services/workflow.py

@@ -47,7 +47,9 @@ class WorkflowService:
             active = self._active_job(session)
 
         safety = SafetyService(self._session_factory).counts()
-        analysis = AnalysisService(self._session_factory).counts()
+        # Reuse the confirmed-SFW total just computed: resolving the current decision
+        # of every asset is the expensive part of this page (US07-06).
+        analysis = AnalysisService(self._session_factory).counts(eligible=safety[_SFW])
         undecided_clusters = cluster_states.get("open", 0) + cluster_states.get("reopened", 0)
 
         stages = [

tests/integration/test_performance_budgets.py

@@ -0,0 +1,324 @@
+"""Performance budgets, paging, and resource bounds (US07-06).
+
+The harness itself is the deliverable, so this suite proves the harness: that it
+builds a synthetic library, measures the same scenarios every time, exports
+machine-readable metrics, and — the part that matters — *fails* when a budget is
+exceeded rather than printing a number nobody reads.
+
+It runs the ``smoke`` profile. The 25k/100k/500k matrix is a scheduled command
+(README "Performance budgets"), because minutes of build time do not belong in the
+suite that runs on every change.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import func, select
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+from photo_pipeline.models import Asset, DuplicateCluster, DuplicateMember, SafetyReview
+from photo_pipeline.services import benchmarks
+from photo_pipeline.services.duplicates import MAX_MEMBER_PAGE, DuplicateService
+from photo_pipeline.services.safety import SafetyService
+
+NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+def _config(tmp_path, **extra) -> Config:
+    data = tmp_path / "data"
+    data.mkdir(parents=True, exist_ok=True)
+    lib = tmp_path / "lib"
+    lib.mkdir(exist_ok=True)
+    return Config.from_env(
+        {
+            "PHOTO_PIPELINE_DATA_DIR": str(data),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
+            **extra,
+        }
+    )
+
+
+# ── the harness ──────────────────────────────────────────────────────────────
+
+
+def test_the_smoke_profile_measures_every_scenario_and_passes_its_budgets(tmp_path):
+    config = _config(tmp_path)
+    output = tmp_path / "report.json"
+
+    report = benchmarks.run(config, profile="smoke", output=output)
+
+    assert report["ok"] is True, report["breaches"]
+    assert json.loads(output.read_text())["profile"] == "smoke"  # machine-readable
+    run = report["runs"][0]
+    measured = {scenario["scenario"] for scenario in run["scenarios"]}
+    assert measured == {
+        "inventory_page",
+        "library_search",
+        "library_stats",
+        "workflow_readiness",
+        "duplicate_cluster_list",
+        "duplicate_cluster_page",
+    }
+    for scenario in run["scenarios"]:
+        assert scenario["latency_p95_ms"] >= scenario["latency_p50_ms"]
+        assert scenario["iterations"] == benchmarks.PROFILES["smoke"]["iterations"]
+    for metric in ("rss_bytes", "open_files", "db_bytes", "wal_bytes", "queue_depth"):
+        assert metric in run["resources"]
+    assert run["build"]["assets"] == 2_000
+
+
+def test_a_breached_budget_fails_the_run_and_names_what_broke(tmp_path, monkeypatch):
+    config = _config(tmp_path)
+    # A budget nothing can meet: the run must fail, not shrug.
+    monkeypatch.setattr(
+        benchmarks,
+        "BUDGETS",
+        (benchmarks.Budget("latency_p95_ms", 0.0000001, "ms", "deliberately impossible"),),
+    )
+
+    report = benchmarks.run(config, profile="smoke")
+
+    assert report["ok"] is False
+    breach = report["breaches"][0]
+    assert {"scope", "metric", "value", "limit", "unit", "why"} <= set(breach)
+    assert breach["metric"] == "latency_p95_ms" and breach["value"] > breach["limit"]
+
+
+def test_an_approved_exception_raises_the_limit_and_is_recorded(tmp_path, monkeypatch):
+    config = _config(tmp_path)
+    monkeypatch.setattr(
+        benchmarks,
+        "BUDGETS",
+        (benchmarks.Budget("latency_p95_ms", 0.0000001, "ms", "deliberately impossible"),),
+    )
+    monkeypatch.setattr(
+        benchmarks,
+        "APPROVED_EXCEPTIONS",
+        {
+            ("smoke", "library_stats", "latency_p95_ms"): {
+                "limit": 10_000,
+                "approved_by": "test",
+                "reason": "documenting the mechanism",
+                "review_by": "2026-12-31",
+            }
+        },
+    )
+
+    report = benchmarks.run(config, profile="smoke")
+
+    scopes = {breach["scope"] for breach in report["breaches"]}
+    assert "library_stats" not in scopes, "the approved exception was not applied"
+    assert report["exceptions_applied"][0]["approved_by"] == "test"
+
+
+def test_an_unknown_profile_is_refused(tmp_path):
+    with pytest.raises(ValueError, match="unknown profile"):
+        benchmarks.run(_config(tmp_path), profile="enormous")
+
+
+def test_the_soak_reports_growth_queue_depth_and_leaves_no_backlog(tmp_path):
+    config = _config(tmp_path)
+    benchmarks.synthesize(config, assets=200, cluster_members=0)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    try:
+        result = benchmarks.soak(config, factory, seconds=1.5, interval=0.25)
+    finally:
+        engine.dispose()
+
+    assert result["cycles"] > 0 and len(result["samples"]) >= 2
+    # Every cycle enqueues and cancels a job: the lane must end empty, which is the
+    # difference between "busy" and "growing without bound".
+    assert result["queue_depth"] == 0
+    assert result["rss_growth_bytes"] >= 0
+    assert result["open_files"] <= 256
+
+
+# ── paging large clusters ────────────────────────────────────────────────────
+
+
+def _cluster(config: Config, members: int) -> tuple[str, object]:
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    cluster_id = str(uuid.uuid4())
+    with factory() as session:
+        session.add(
+            DuplicateCluster(
+                id=cluster_id, method="perceptual", confidence="near", state="open", version=1
+            )
+        )
+        session.flush()
+        for index in range(members):
+            asset_id = f"member-{index:06d}"
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=f"/lib/photo-{index}.jpg",
+                    current_path=f"/lib/photo-{index}.jpg",
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=1024,
+                )
+            )
+            session.add(
+                DuplicateMember(cluster_id=cluster_id, asset_id=asset_id, role="member", distance=1)
+            )
+        session.commit()
+    return cluster_id, factory
+
+
+def test_a_cluster_of_thousands_is_paged_not_dumped(tmp_path):
+    config = _config(tmp_path)
+    cluster_id, factory = _cluster(config, 3_000)
+    service = DuplicateService(factory)
+
+    first = service.get_cluster(cluster_id)
+
+    assert first["member_total"] == 3_000
+    assert len(first["members"]) == 100, "the default page, not the whole cluster"
+    second = service.get_cluster(cluster_id, limit=100, offset=100)
+    assert [m["asset_id"] for m in second["members"]][0] == "member-000100"
+    assert not {m["asset_id"] for m in first["members"]} & {
+        m["asset_id"] for m in second["members"]
+    }
+    # The last page is short and the pages together cover the cluster exactly.
+    tail = service.get_cluster(cluster_id, limit=MAX_MEMBER_PAGE, offset=2_900)
+    assert len(tail["members"]) == 100
+
+
+def test_the_cluster_list_carries_counts_without_loading_every_member(tmp_path):
+    config = _config(tmp_path)
+    cluster_id, factory = _cluster(config, 3_000)
+
+    listed = DuplicateService(factory).list_clusters(limit=50)
+
+    entry = listed["items"][0]
+    assert entry["id"] == cluster_id
+    assert entry["member_total"] == 3_000
+    assert len(entry["members"]) <= 20, "the list view shows a preview, never the cluster"
+
+
+def test_the_api_pages_cluster_members_and_bounds_the_page_size(tmp_path):
+    config = _config(tmp_path)
+    cluster_id, _ = _cluster(config, 1_200)
+    with TestClient(create_app(config)) as client:
+        default = client.get(f"/api/v1/duplicates/clusters/{cluster_id}").json()
+        assert default["member_total"] == 1_200 and len(default["members"]) == 100
+
+        paged = client.get(
+            f"/api/v1/duplicates/clusters/{cluster_id}", params={"limit": 250, "offset": 1_000}
+        ).json()
+        assert len(paged["members"]) == 200 and paged["offset"] == 1_000
+
+        # A caller cannot ask for the whole cluster by asking for a huge page.
+        assert (
+            client.get(
+                f"/api/v1/duplicates/clusters/{cluster_id}",
+                params={"limit": MAX_MEMBER_PAGE + 1},
+            ).status_code
+            == 422
+        )
+
+
+# ── the queries behind the pages ─────────────────────────────────────────────
+
+
+def test_the_review_queue_is_filtered_and_paged_in_the_database(tmp_path):
+    """A queue that loads every asset to slice 100 of them is the shape this story
+    exists to remove; the totals must stay exact while it pages."""
+    config = _config(tmp_path)
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    with factory() as session:
+        for index in range(500):
+            asset_id = f"asset-{index:04d}"
+            session.add(
+                Asset(
+                    id=asset_id,
+                    original_path=f"/lib/{index:04d}.jpg",
+                    current_path=f"/lib/{index:04d}.jpg",
+                    discovered_at=NOW,
+                    hash_version=1,
+                    byte_size=1,
+                )
+            )
+            if index % 2 == 0:
+                session.add(
+                    SafetyReview(
+                        id=str(uuid.uuid4()),
+                        asset_id=asset_id,
+                        decision="sfw",
+                        created_at=NOW,
+                    )
+                )
+        session.commit()
+    service = SafetyService(factory)
+
+    page = service.review_queue(state="undecided", limit=10, offset=0)
+    assert page["total"] == 250 and len(page["items"]) == 10
+    assert all(item["decision"] is None for item in page["items"])
+
+    second = service.review_queue(state="undecided", limit=10, offset=10)
+    assert not {item["asset_id"] for item in page["items"]} & {
+        item["asset_id"] for item in second["items"]
+    }
+    assert service.counts() == {
+        "sfw": 250,
+        "nsfw": 0,
+        "deferred": 0,
+        "undecided": 250,
+        "scored": 0,
+    }
+    engine.dispose()
+
+
+def test_the_latest_review_still_wins_after_a_revision(tmp_path):
+    """The counts are aggregated in SQL now; the rule they aggregate is unchanged."""
+    config = _config(tmp_path)
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    factory = create_session_factory(engine)
+    with factory() as session:
+        session.add(
+            Asset(
+                id="a",
+                original_path="/lib/a.jpg",
+                current_path="/lib/a.jpg",
+                discovered_at=NOW,
+                hash_version=1,
+                byte_size=1,
+            )
+        )
+        session.add(
+            SafetyReview(id="r1", asset_id="a", decision="sfw", score=0.1, created_at=NOW)
+        )
+        session.commit()
+    service = SafetyService(factory)
+    assert service.current_decision("a") == "sfw"
+
+    with factory() as session:
+        session.add(
+            SafetyReview(
+                id="r2",
+                asset_id="a",
+                decision="nsfw",
+                prior_decision="sfw",
+                created_at=NOW.replace(hour=2),
+            )
+        )
+        session.commit()
+
+    assert service.current_decision("a") == "nsfw"
+    assert service.counts()["nsfw"] == 1 and service.counts()["sfw"] == 0
+    with factory() as session:  # the history itself is never rewritten
+        assert session.scalar(select(func.count()).select_from(SafetyReview)) == 2
+    engine.dispose()

tests/story_traceability.json

@@ -166,6 +166,9 @@
     "US07-05": [
       "tests/integration/test_backup_recovery.py",
       "tests/integration/test_diagnostics.py"
+    ],
+    "US07-06": [
+      "tests/integration/test_performance_budgets.py"
     ]
   }
 }