US07-05: Deliver Backup and Operational Recovery

.gitignore

@@ -20,12 +20,3 @@ _IGNORE/
 
 # Test failure evidence (US07-04)
 .artifacts/
-
-# Any dotenv, not only the default name.
-*.env
-
-# Local virtualenv for running the app.
-.venv/
-
-# setuptools editable-install metadata.
-*.egg-info/

README.md

@@ -7,49 +7,16 @@ archive workflow. Planning lives in `INTEGRATED_PIPELINE_CONCEPT.md` and
 ## Application (`photo_pipeline`)
 
 The target application lives in `photo_pipeline/` (FastAPI + SQLAlchemy + Alembic).
-Install it into a virtualenv once:
+Run it with:
 
 ```bash
-python3.12 -m venv .venv
+python -m photo_pipeline migrate   # apply database migrations
-.venv/bin/pip install -e ".[vision]"   # drop [vision] for a review-only install
+python -m photo_pipeline serve     # start the API + static review UI (127.0.0.1:8000)
 ```
 
-Then run the two processes:
-
-```bash
-.venv/bin/python -m photo_pipeline migrate   # apply database migrations
-.venv/bin/python -m photo_pipeline serve     # API + review UI at 127.0.0.1:8000/app/
-.venv/bin/python -m photo_pipeline worker    # second terminal: runs the jobs
-```
-
-The server enqueues work and serves the UI; nothing actually scans, scores,
-analyses, uploads, or archives without a worker. `work_item/scripts/python` is the
-*helper's* launcher — it prefers Conda base and falls back to a bare system
-interpreter, so it is not how the application is run.
-
 Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
 `photo_pipeline/config.py`); secrets are referenced, never logged.
 
-### Configuration file
-
-`.env` in the working directory is read at startup, or any path named by
-`PHOTO_PIPELINE_ENV_FILE`. It is parsed, never executed: `KEY=value` lines,
-`#` comments, optional quotes — no interpolation and no `export`. **Anything already
-exported wins**, so the file is the standing configuration and the shell is the
-override for one run.
-
-The archived CLI's variable names still work, so an existing `photo_analyzer.env`
-can be used as-is:
-
-| in the file | applied as |
-|---|---|
-| `LLM_API_KEY` / `GEMINI_API_KEY` | `OPENAI_API_KEY` |
-| `LLM_BASE_URL` | `OPENAI_BASE_URL` |
-| `LIBRARY` | `PHOTO_PIPELINE_LIBRARY_ROOTS` |
-
-`.env` and `*.env` are gitignored and denied by the work-item safety checks: the
-file holds a real key and must never be committed.
-
 ### API access (US07-02)
 
 The app listens on loopback, so its attacker is another page in the same browser.
@@ -314,110 +281,6 @@ 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.
 
-## Release gate (US07-07)
-
-One command runs every suite in an isolated stack and keeps the evidence:
-
-```bash
-work_item/scripts/python -m photo_pipeline release-gate --output data/release/$(date -u +%Y%m%dT%H%M%SZ)
-```
-
-It fails — and exits non-zero — when any stage fails, when a suite skips a test for
-a reason that is not a documented environment limit (`exiftool not installed`,
-`root ignores directory permissions`), or when the story matrix has a hole. The
-evidence directory holds `release-report.json` (revision, per-stage result, timings,
-summaries), `logs/<stage>.log`, and `CHECKSUMS.sha256` over both.
-
-**The story matrix** lives in `tests/story_traceability.json`: every story under
-`delivery_backlog/stories/` is either mapped to test files that exist, or listed in
-`planned` as an accepted but unimplemented story. A story that is neither, or a
-mapping to a file that has been deleted, fails the gate.
-
-**The journey** (`tests/e2e/test_release_journey.py`) takes one fresh library through
-discovery, duplicate review, safety, analysis, EXIF verification, album proposal,
-guarded rename, rescan, upload with server-side verification, archive, offline
-deduplication, and restore — over HTTP against real server and worker processes,
-with a full restart in the middle and at the end.
-
-### Real-library dry run and approval
-
-Before the application is pointed at photos that cannot be replaced:
-
-```bash
-work_item/scripts/python -m photo_pipeline dry-run --output dry-run.json
-work_item/scripts/python -m photo_pipeline approve-dry-run dry-run.json --approver "$(whoami)"
-```
-
-The dry run is strictly read-only: it opens no file for writing, writes no database
-row, and reports what it found — file counts by extension, folders, bytes, unreadable
-files, excluded directories, and a reconciliation against what the database already
-knows (already registered, new, recorded but absent). Set
-`PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL=1` and **every mutating API request is
-refused with `403 dry_run_not_approved`** until a report for exactly those library
-roots has been approved. Reading stays open — you have to be able to see what was
-found in order to approve it — and so does taking a backup. Change the library roots
-and the approval no longer applies: it approves that reconciliation, not the idea of
-mutating.
-
-## 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,10 +74,8 @@ export const api = {
     request("/inventory/assets?" + new URLSearchParams(params).toString(), opts),
   listClusters: (params = {}, opts = {}) =>
     request("/duplicates/clusters?" + new URLSearchParams(params).toString(), opts),
-  getCluster: (id, params = {}, opts = {}) => {
+  getCluster: (id, opts = {}) =>
-    const query = new URLSearchParams(params).toString();
+    request(`/duplicates/clusters/${encodeURIComponent(id)}`, opts),
-    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.member_total ?? cluster.members.length} members · confidence ${cluster.confidence}`)
+      el("div", { class: "muted" }, `${cluster.members.length} members · confidence ${cluster.confidence}`)
     )
   );
 
@@ -207,11 +207,8 @@ 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, shown ? { limit: shown } : {});
+    cluster = await api.getCluster(id);
   } catch (error) {
     show(errorBanner(`Failed to load cluster: ${error.message}`));
     return;
@@ -301,19 +298,6 @@ 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

@@ -1,45 +0,0 @@
-"""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

@@ -11,7 +11,6 @@ from __future__ import annotations
 
 import argparse
 import json
-from pathlib import Path
 from typing import Sequence
 
 from photo_pipeline.config import Config
@@ -55,28 +54,6 @@ 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")
-
-    gate_cmd = commands.add_parser(
-        "release-gate", help="Run every suite in an isolated stack and keep the evidence"
-    )
-    gate_cmd.add_argument("--output", help="Evidence directory (default: data/release/<stamp>)")
-    dry_cmd = commands.add_parser(
-        "dry-run", help="Read-only reconciliation of the configured library (US07-07)"
-    )
-    dry_cmd.add_argument("--output", help="Write the report here as well as to stdout")
-    approve_cmd = commands.add_parser(
-        "approve-dry-run", help="Approve a dry-run report, which is what enables mutation"
-    )
-    approve_cmd.add_argument("report", help="Path to the dry-run report")
-    approve_cmd.add_argument("--approver", required=True, help="Who is accepting this")
 
     args = parser.parse_args(argv)
 
@@ -112,59 +89,6 @@ 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 == "release-gate":
-        from photo_pipeline.services import release
-
-        report = release.run_gate(config, output=args.output)
-        print(
-            json.dumps(
-                {k: v for k, v in report.items() if k not in ("stages", "matrix")}, indent=2
-            )
-        )
-        return 0 if report["ok"] else 1
-
-    if args.command == "dry-run":
-        from photo_pipeline.services import release
-
-        try:
-            report = release.dry_run(config)
-        except release.ReleaseError as error:
-            print(str(error))
-            return 1
-        if args.output:
-            Path(args.output).write_text(json.dumps(report, indent=2))
-        print(json.dumps(report, indent=2))
-        return 0
-
-    if args.command == "approve-dry-run":
-        from photo_pipeline.services import release
-
-        try:
-            record = release.approve(config, args.report, approver=args.approver)
-        except (release.ReleaseError, OSError, ValueError) as error:
-            print(str(error))
-            return 1
-        print(json.dumps(record, indent=2))
-        return 0
-
     if args.command == "diagnostics":
         from photo_pipeline.services import diagnostics
 

photo_pipeline/api/routes/duplicates.py

@@ -11,13 +11,7 @@ from fastapi import APIRouter, Query, Request
 from fastapi.responses import JSONResponse
 
 from photo_pipeline.schemas import DecisionRequest
-from photo_pipeline.services.duplicates import (
+from photo_pipeline.services.duplicates import ConflictError, DuplicateError, DuplicateService
-    MAX_MEMBER_PAGE,
-    MEMBER_PAGE,
-    ConflictError,
-    DuplicateError,
-    DuplicateService,
-)
 
 router = APIRouter(tags=["duplicates"])
 
@@ -48,13 +42,8 @@ def list_clusters(
 
 
 @router.get("/duplicates/clusters/{cluster_id}")
-def get_cluster(
+def get_cluster(cluster_id: str, request: Request):
-    cluster_id: str,
+    detail = _service(request).get_cluster(cluster_id)
-    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/api/security.py

@@ -43,9 +43,6 @@ PUBLIC_PATHS = frozenset(
     {f"{API_PREFIX}/health/live", f"{API_PREFIX}/health/ready", f"{API_PREFIX}/session"}
 )
 LOOPBACK_HOSTS = frozenset({"127.0.0.1", "localhost", "::1", "[::1]"})
-# Mutating endpoints that must stay reachable while mutation itself is gated: the
-# backup a careful operator takes first, and its retention (US07-07).
-MUTATION_EXEMPT_PATHS = frozenset({f"{API_PREFIX}/backups", f"{API_PREFIX}/backups/prune"})
 
 # Applied to every response. No inline script/style is used by the frontend, so the
 # policy can stay strict; `frame-ancestors 'none'` and CORP keep other pages from
@@ -148,7 +145,6 @@ class SecurityMiddleware:
     def __init__(self, app, *, session: Session, config) -> None:
         self.app = app
         self.session = session
-        self.config = config
         self.max_request_bytes = config.max_request_bytes
         self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
 
@@ -169,8 +165,6 @@ class SecurityMiddleware:
             allowed_hosts=self.allowed_hosts,
             max_request_bytes=self.max_request_bytes,
         )
-        if refusal is None:
-            refusal = self._mutation_refusal(scope)
         if refusal is not None:
             response = JSONResponse(
                 status_code=refusal.status,
@@ -189,26 +183,6 @@ class SecurityMiddleware:
 
         await self.app(scope, receive, send_with_headers)
 
-    def _mutation_refusal(self, scope) -> Refusal | None:
-        """Refuse every mutating request while the library's dry run is unapproved.
-
-        One choke point for the whole API: every mutation the browser can start is a
-        non-safe method under ``/api/v1``. Reading stays open — an operator has to be
-        able to look at what the application found in order to approve it (US07-07).
-        """
-        method = scope.get("method", "GET").upper()
-        path = scope.get("path", "/")
-        if method in SAFE_METHODS or not path.startswith(API_PREFIX):
-            return None
-        if path in MUTATION_EXEMPT_PATHS:
-            return None
-        from photo_pipeline.services.release import mutation_blockers
-
-        blockers = mutation_blockers(self.config)
-        if not blockers:
-            return None
-        return Refusal(403, blockers[0]["code"], blockers[0]["message"])
-
 
 def _cookie(header: str, name: str) -> str:
     for part in header.split(";"):

photo_pipeline/config.py

@@ -18,61 +18,6 @@ from typing import Mapping
 from pydantic import BaseModel, ConfigDict, SecretStr
 
 ENV_PREFIX = "PHOTO_PIPELINE_"
-ENV_FILE_VAR = f"{ENV_PREFIX}ENV_FILE"
-DEFAULT_ENV_FILE = Path(".env")
-
-# The archived CLI's variable names, so the configuration file an operator already
-# has keeps working. The vision provider reads the OpenAI SDK's names, and the
-# library root is configuration here rather than a bare path (US07-01 donor).
-LEGACY_ALIASES = {
-    "LLM_API_KEY": "OPENAI_API_KEY",
-    "GEMINI_API_KEY": "OPENAI_API_KEY",
-    "LLM_BASE_URL": "OPENAI_BASE_URL",
-    "LIBRARY": f"{ENV_PREFIX}LIBRARY_ROOTS",
-}
-
-
-def parse_env_file(text: str) -> dict[str, str]:
-    """``KEY=value`` lines into a mapping. Comments, blanks, and quotes handled.
-
-    Deliberately not a shell: no interpolation, no ``export``, no multi-line values.
-    A configuration file that can run code is a configuration file that can be a
-    vulnerability.
-    """
-    values: dict[str, str] = {}
-    for line in text.splitlines():
-        line = line.strip()
-        if not line or line.startswith("#") or "=" not in line:
-            continue
-        key, _, raw = line.partition("=")
-        key = key.strip()
-        if not key or key.startswith("#"):
-            continue
-        value = raw.strip().strip('"').strip("'")
-        values[key] = value
-        alias = LEGACY_ALIASES.get(key)
-        if alias:
-            values.setdefault(alias, value)
-    return values
-
-
-def load_env_file(path: Path | str | None = None) -> dict[str, str]:
-    """Load ``PHOTO_PIPELINE_ENV_FILE`` (or ``./.env``) into the environment.
-
-    Anything already exported wins: a file is the standing configuration, the shell
-    is what you meant *this time*. Returns what it applied, which is what the CLI
-    prints — names only, never values.
-    """
-    candidate = path or os.environ.get(ENV_FILE_VAR) or DEFAULT_ENV_FILE
-    candidate = Path(candidate)
-    if not candidate.is_file():
-        return {}
-    applied = {}
-    for key, value in parse_env_file(candidate.read_text()).items():
-        if key not in os.environ:
-            os.environ[key] = value
-            applied[key] = value
-    return applied
 
 
 class Config(BaseModel):
@@ -97,12 +42,6 @@ class Config(BaseModel):
     # Free space an archive destination must keep beyond the transfer itself.
     archive_free_space_reserve_bytes: int = 1_000_000_000
 
-    # Refuse every mutating request until a read-only dry run of the configured
-    # library has been produced and explicitly approved (US07-07). Off by default so
-    # a development setup is unchanged; turn it on before pointing the application at
-    # a library whose photos cannot be replaced.
-    require_dry_run_approval: bool = False
-
     vision_api_key: SecretStr | None = None
     immich_api_key: SecretStr | None = None
     immich_server_url: str = ""
@@ -123,9 +62,6 @@ class Config(BaseModel):
     @classmethod
     def from_env(cls, environ: Mapping[str, str] | None = None) -> "Config":
         env = os.environ if environ is None else environ
-        if environ is None:
-            load_env_file()  # a file never overrides what the shell already set
-            env = os.environ
         data: dict = {}
         for name in cls.model_fields:
             raw = env.get(ENV_PREFIX + name.upper())

photo_pipeline/services/analysis.py

@@ -18,6 +18,7 @@ from __future__ import annotations
 
 import json
 import os
+import uuid
 from datetime import datetime, timezone
 from typing import Protocol
 
@@ -25,9 +26,9 @@ from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline import path_policy
-from photo_pipeline.models import AnalysisResult, Asset
+from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
 from photo_pipeline.services import exif_checkpoint
-from photo_pipeline.services.safety import SFW, latest_reviews
+from photo_pipeline.services.safety import SFW
 
 MODEL = "gemini-2.5-flash"
 PROMPT_VERSION = "1"
@@ -71,26 +72,11 @@ 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.
+        may reach the provider."""
-
+        latest: dict[str, str | None] = {}
-        The "latest row wins" rule is applied in SQL (US07-06); loading every review
+        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
-        to fold it in Python made the gate cost grow with the review history rather
+            latest[review.asset_id] = review.decision
-        than with the work being gated.
+        return {aid for aid, decision in latest.items() if decision == SFW}
-        """
-        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."""
@@ -118,12 +104,9 @@ class AnalysisService:
             )
             return [a.id for a in assets if a.id not in done]
 
-    def counts(self, *, eligible: int | None = None) -> dict[str, int]:
+    def counts(self) -> 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:
-            eligible = self._sfw_count(session) if eligible is None else eligible
+            sfw = self._sfw_asset_ids(session)
             rows = dict(
                 session.execute(
                     select(AnalysisResult.status, func.count()).group_by(AnalysisResult.status)
@@ -132,10 +115,10 @@ class AnalysisService:
             analyzed = int(rows.get("analyzed", 0))
             errored = int(rows.get("error", 0))
             return {
-                "eligible": eligible,
+                "eligible": len(sfw),
                 "analyzed": analyzed,
                 "error": errored,
-                "pending": max(eligible - analyzed - errored, 0),
+                "pending": max(len(sfw) - analyzed - errored, 0),
             }
 
     def run(self, asset_ids: list[str] | None = None) -> dict:
@@ -264,11 +247,8 @@ class AnalysisService:
             asset = session.get(Asset, asset_id)
             if asset is not None and checkpoint.sha256:
                 # The bytes changed when the container was rewritten; upload must use
-                # the hash of what is actually on disk now (concept §3), and the
+                # the hash of what is actually on disk now (concept §3).
-                # recorded size has to move with it (US07-07).
                 asset.current_sha256 = checkpoint.sha256
-                if checkpoint.byte_size is not None:
-                    asset.byte_size = checkpoint.byte_size
             session.commit()
 
     def get(self, asset_id: str) -> dict | None:

photo_pipeline/services/benchmarks.py

@@ -1,502 +0,0 @@
-"""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,13 +52,6 @@ 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"
@@ -527,65 +520,39 @@ 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(
+    def get_cluster(self, cluster_id: str) -> dict | None:
-        self, cluster_id: str, *, limit: int = MEMBER_PAGE, offset: int = 0
+        """Cluster detail enriched with per-member asset evidence for comparison."""
-    ) -> 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 rows:
+            for member in session.execute(
+                select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
+            ).scalars():
+                asset = session.get(Asset, member.asset_id)
                 try:
-                    member_evidence = json.loads(member.evidence) if member.evidence else {}
+                    evidence = json.loads(member.evidence) if member.evidence else {}
                 except json.JSONDecodeError:
-                    member_evidence = {}
+                    evidence = {}
-                asset, offline = evidence[member.asset_id]
                 members.append(
                     {
                         "asset_id": member.asset_id,
                         "role": member.role,
                         "distance": member.distance,
-                        "evidence": member_evidence,
+                        "evidence": 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,
-                        **offline,
+                        **self._offline_evidence(session, asset),
                     }
                 )
+            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). The
+            # UI asks for that named medium instead of guessing (concept §9).
-            # answer covers the whole cluster, not just this page, so a mount is not
+            mount_required = sorted(
-            # discovered halfway through a review.
+                {m["archive_location"] for m in members if m["requires_mount"]}
-            mount_required = self._mount_required(session, cluster_id)
+            )
             return {
                 "id": cluster.id,
                 "method": cluster.method,
@@ -597,74 +564,10 @@ class DuplicateService:
                 "requires_confirmation": cluster.method == Method.PERCEPTUAL.value,
                 "mount_required": mount_required,
                 "members": members,
-                "member_total": member_total,
-                "limit": limit,
-                "offset": offset,
             }
 
-    def _member_evidence(self, session, asset_ids: list[str]) -> dict:
+    def _offline_evidence(self, session, asset: Asset | None) -> dict:
-        """``{asset_id: (asset, offline_evidence)}`` for one page, in three queries."""
+        """What review can still rely on when a member's original is not readable."""
-        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,
@@ -674,8 +577,12 @@ class DuplicateService:
                 "preview": {"state": "missing", "protected": False},
                 "requires_mount": False,
             }
-        location = locations.get(asset.archive_location_id)
+        location = (
-        preview = self._preview_evidence(thumbnails)
+            session.get(ArchiveLocation, asset.archive_location_id)
+            if asset.archive_location_id
+            else None
+        )
+        preview = self._preview_evidence(session, asset.id)
         archived = asset.availability_state in availability.ARCHIVED
         return {
             "availability_state": asset.availability_state,
@@ -691,7 +598,10 @@ class DuplicateService:
         }
 
     @staticmethod
-    def _preview_evidence(rows: list) -> dict:
+    def _preview_evidence(session, asset_id: str) -> dict:
+        rows = list(
+            session.execute(select(Thumbnail).where(Thumbnail.asset_id == asset_id)).scalars()
+        )
         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))
@@ -783,29 +693,11 @@ 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)
+            select(DuplicateMember).where(DuplicateMember.cluster_id == cluster_id)
-            .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/exif_checkpoint.py

@@ -21,7 +21,6 @@ unreadable file, or a write that did not take is not evidence that metadata is f
 from __future__ import annotations
 
 import json
-import os
 import uuid
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -59,11 +58,6 @@ class CheckpointResult:
     state: str  # verified | divergent | failed
     changed_fields: tuple[str, ...] = ()
     sha256: str | None = None
-    # exiftool rewrites the container, so the file's size moves with its hash. Both
-    # are inventory facts about the current bytes and both have to be refreshed
-    # together, or the next stage compares against a size that no longer exists
-    # (US07-07: a rename plan blocked itself forever after any EXIF write).
-    byte_size: int | None = None
     verified_at: datetime | None = None
     reason: str | None = None
 
@@ -143,12 +137,9 @@ def run(
 
     changed = compare(before, after)
     sha256 = hashing.sha256_file(path)
-    byte_size = os.path.getsize(path)
     if changed:
-        return CheckpointResult(
+        return CheckpointResult(DIVERGENT, changed_fields=changed, sha256=sha256)
-            DIVERGENT, changed_fields=changed, sha256=sha256, byte_size=byte_size
+    return CheckpointResult(VERIFIED, sha256=sha256, verified_at=_now())
-        )
-    return CheckpointResult(VERIFIED, sha256=sha256, byte_size=byte_size, verified_at=_now())
 
 
 def record(

photo_pipeline/services/library.py

@@ -15,9 +15,10 @@ from __future__ import annotations
 
 import json
 import re
+from collections import Counter
 from pathlib import Path
 
-from sqlalchemy import String, and_, case, cast, func, or_, select, text
+from sqlalchemy import and_, func, or_, select
 from sqlalchemy.orm import sessionmaker
 
 from photo_pipeline.models import AnalysisResult, Asset
@@ -70,75 +71,38 @@ 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()
             )
-            albums: dict[str, dict] = {}
+            rows = list(
-            for path, row_status in session.execute(
+                session.execute(
-                select(Asset.current_path, AnalysisResult.status).join(
+                    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:
                 album = _album_of(path)
                 bucket = albums.setdefault(album, {"album": album, "done": 0, "total": 0})
                 bucket["total"] += 1
-                if row_status in DONE:
+                if result.status in DONE:
                     bucket["done"] += 1
-            year_counts = dict(
+                for tag in _tags(result.tags):
-                session.execute(
+                    tag_counts[tag] += 1
-                    select(AnalysisResult.approx_year, func.count())
+                if result.approx_year is not None:
-                    .where(AnalysisResult.approx_year.is_not(None))
+                    year_counts[result.approx_year] += 1
-                    .group_by(AnalysisResult.approx_year)
+                if result.people_count is not None:
-                ).all()
+                    people["3+" if result.people_count >= 3 else str(result.people_count)] += 1
-            )
+                if result.status == "error":
-            people = dict(
+                    errors.append({"path": path, "error": result.error_message})
-                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,
@@ -147,7 +111,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],
+            "top_tags": [{"value": t, "count": n} for t, n in tag_counts.most_common(40)],
             "albums": sorted(albums.values(), key=lambda d: d["album"]),
             "errors": sorted(errors, key=lambda e: e["path"] or ""),
         }

photo_pipeline/services/release.py

@@ -1,391 +0,0 @@
-"""The release gate, the real-library dry run, and the approval that unlocks
-mutation (US07-07, concept §18 release gates).
-
-Three things live here because they are one decision:
-
-1. **The gate** — one command that provisions an isolated stack, runs every suite in
-   a fixed order, and retains versioned evidence with checksums. A release is not
-   "the tests passed on my machine last Tuesday"; it is a report that says which
-   revision, which suites, how long, and what the artefacts hash to.
-2. **The dry run** — a strictly read-only pass over the real photo library that
-   answers "what would this application do to it?" before it is allowed to do
-   anything. It opens no file for writing, creates no database rows, and touches no
-   metadata; it counts, classifies, and reconciles against whatever the database
-   already knows.
-3. **The approval** — a person reads that report and signs it off for exactly the
-   library roots it describes. Until then, with
-   ``PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL`` set, every mutating request is
-   refused. Change the roots, or produce a newer report, and the approval no longer
-   matches: it approves *that* reconciliation, not the idea of mutating.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import json
-import os
-import subprocess
-import sys
-import time
-from collections import Counter
-from dataclasses import dataclass
-from datetime import datetime, timezone
-from pathlib import Path
-
-from photo_pipeline import path_policy
-from photo_pipeline.config import Config
-
-SCHEMA_VERSION = 1
-APPROVAL_NAME = "dry-run-approval.json"
-CHECKSUMS_NAME = "CHECKSUMS.sha256"
-REPORT_NAME = "release-report.json"
-
-# The suites, in the order a failure is cheapest to read: units before the stacks
-# they compose. ``label`` is what the report and the operator see.
-STAGES: tuple[tuple[str, tuple[str, ...]], ...] = (
-    ("unit", ("tests/unit",)),
-    ("characterization", ("tests/characterization",)),
-    ("integration", ("tests/integration",)),
-    ("browser", ("tests/e2e",)),
-)
-
-# Skips the gate accepts, because they describe the machine rather than the code.
-ALLOWED_SKIP_REASONS = ("exiftool not installed", "root ignores directory permissions")
-
-
-class ReleaseError(RuntimeError):
-    pass
-
-
-def _now() -> datetime:
-    return datetime.now(timezone.utc)
-
-
-def sha256_file(path: Path) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as handle:
-        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
-            digest.update(chunk)
-    return digest.hexdigest()
-
-
-def sha256_bytes(payload: bytes) -> str:
-    return hashlib.sha256(payload).hexdigest()
-
-
-def revision() -> str | None:
-    """The commit this gate ran against, when the tree is a git checkout."""
-    try:
-        result = subprocess.run(
-            ["git", "rev-parse", "HEAD"],
-            capture_output=True,
-            text=True,
-            timeout=10,
-            cwd=str(Path(__file__).resolve().parents[2]),
-        )
-    except (OSError, subprocess.SubprocessError):
-        return None
-    return result.stdout.strip() or None
-
-
-# ── the story matrix ─────────────────────────────────────────────────────────
-
-
-def story_matrix(repo: Path | None = None) -> dict:
-    """Every backlog story, and how it is covered.
-
-    A story is ``delivered`` (mapped to test files that exist) or ``planned`` (an
-    accepted, not-yet-implemented story). Anything else — a story file nobody
-    mapped, or a mapping to a file that is gone — is a hole in the matrix, and the
-    gate fails on it rather than reporting a green run over missing coverage.
-    """
-    repo = repo or Path(__file__).resolve().parents[2]
-    traceability = json.loads((repo / "tests" / "story_traceability.json").read_text())
-    mapped: dict[str, list[str]] = traceability["stories"]
-    planned: list[str] = traceability.get("planned", [])
-    stories = sorted(
-        "-".join(path.stem.split("-")[:2])
-        for path in (repo / "delivery_backlog" / "stories").glob("US*.md")
-    )
-
-    missing_tests = [
-        f"{story}: {rel}"
-        for story, files in mapped.items()
-        for rel in files
-        if not (repo / rel).is_file()
-    ]
-    unmapped = [s for s in stories if s not in mapped and s not in planned]
-    unknown = [s for s in list(mapped) + planned if s not in stories]
-    overlap = sorted(set(mapped) & set(planned))
-    return {
-        "stories": len(stories),
-        "delivered": sorted(mapped),
-        "planned": sorted(planned),
-        "problems": [
-            *(f"story with no tests and not planned: {s}" for s in unmapped),
-            *(f"mapped test file is missing — {entry}" for entry in missing_tests),
-            *(f"mapped story is not in the backlog: {s}" for s in unknown),
-            *(f"story is both delivered and planned: {s}" for s in overlap),
-        ],
-    }
-
-
-# ── the gate ─────────────────────────────────────────────────────────────────
-
-
-@dataclass
-class StageResult:
-    label: str
-    command: list[str]
-    returncode: int
-    seconds: float
-    summary: str
-    skipped: list[str]
-
-    def as_dict(self) -> dict:
-        return {
-            "stage": self.label,
-            "command": self.command,
-            "returncode": self.returncode,
-            "seconds": round(self.seconds, 2),
-            "summary": self.summary,
-            "skipped": self.skipped,
-            "ok": self.returncode == 0,
-        }
-
-
-def _run_stage(label: str, paths: tuple[str, ...], *, repo: Path, log_dir: Path) -> StageResult:
-    command = [sys.executable, "-m", "pytest", *paths, "-q", "-rs"]
-    started = time.monotonic()
-    result = subprocess.run(command, cwd=str(repo), capture_output=True, text=True)
-    elapsed = time.monotonic() - started
-    output = result.stdout + result.stderr
-    (log_dir / f"{label}.log").write_text(output)
-    lines = [line for line in output.splitlines() if line.strip()]
-    summary = lines[-1] if lines else ""
-    skipped = [line for line in lines if line.startswith("SKIPPED")]
-    return StageResult(label, command, result.returncode, elapsed, summary, skipped)
-
-
-def unexpected_skips(results: list[StageResult]) -> list[str]:
-    """Skips the gate will not accept: everything but the documented environment ones."""
-    return [
-        line
-        for result in results
-        for line in result.skipped
-        if not any(reason in line for reason in ALLOWED_SKIP_REASONS)
-    ]
-
-
-def run_gate(
-    config: Config,
-    *,
-    output: Path | str | None = None,
-    stages: tuple[tuple[str, tuple[str, ...]], ...] = STAGES,
-    repo: Path | None = None,
-) -> dict:
-    """Run every suite in an isolated stack and retain checksummed evidence.
-
-    The stack is isolated by construction: each pytest run builds its own temporary
-    data directories and libraries, so the gate never reads or writes the operator's
-    photos. What it keeps afterwards is the report, the per-stage logs, and a
-    checksum file over both.
-    """
-    repo = repo or Path(__file__).resolve().parents[2]
-    directory = Path(output) if output else Path(config.data_dir) / "release" / _now().strftime(
-        "%Y%m%dT%H%M%SZ"
-    )
-    logs = directory / "logs"
-    logs.mkdir(parents=True, exist_ok=True)
-
-    matrix = story_matrix(repo)
-    results = [_run_stage(label, paths, repo=repo, log_dir=logs) for label, paths in stages]
-    skips = unexpected_skips(results)
-
-    report = {
-        "schema_version": SCHEMA_VERSION,
-        "started_at": _now().isoformat(),
-        "revision": revision(),
-        "python": sys.version.split()[0],
-        "platform": os.uname().sysname,
-        "matrix": matrix,
-        "stages": [result.as_dict() for result in results],
-        "unexpected_skips": skips,
-        "failures": [result.label for result in results if result.returncode != 0],
-    }
-    report["ok"] = not report["failures"] and not matrix["problems"] and not skips
-    report["finished_at"] = _now().isoformat()
-
-    (directory / REPORT_NAME).write_text(json.dumps(report, indent=2))
-    # The evidence is only evidence if it can be shown to be the evidence that was
-    # produced. Checksums are the honest version of "signed" without a key: a real
-    # signature belongs to whatever key management the release actually has.
-    checksums = "\n".join(
-        f"{sha256_file(path)}  {path.relative_to(directory)}"
-        for path in sorted(directory.rglob("*"))
-        if path.is_file() and path.name != CHECKSUMS_NAME
-    )
-    (directory / CHECKSUMS_NAME).write_text(checksums + "\n")
-    report["evidence"] = str(directory)
-    return report
-
-
-# ── the real-library dry run ─────────────────────────────────────────────────
-
-
-def dry_run(config: Config, *, roots: tuple[Path, ...] | None = None) -> dict:
-    """Read-only reconciliation of the configured library. Changes nothing.
-
-    Opens no file for writing, writes no database row, and reads only what
-    ``os.stat`` and the existing database already say. The point is to be able to
-    look at a real library — the one with the irreplaceable photos in it — and see
-    what the application believes about it before it is allowed to act.
-    """
-    roots = roots or tuple(Path(root) for root in config.library_roots)
-    if not roots:
-        raise ReleaseError("no library roots are configured")
-
-    by_extension: Counter = Counter()
-    folders: set[str] = set()
-    files: list[str] = []
-    unreadable: list[str] = []
-    excluded = 0
-    total_bytes = 0
-    for root in roots:
-        if not Path(root).is_dir():
-            raise ReleaseError(f"library root {root} is not a directory")
-        for path in sorted(Path(root).rglob("*")):
-            if path.is_dir():
-                # Never traverse into an excluded directory, and never report its
-                # contents: proving exclusion must not require opening it.
-                if path_policy.is_excluded(path):
-                    excluded += 1
-                continue
-            if path_policy.is_excluded(path):
-                continue
-            try:
-                stat = path.stat()
-            except OSError:
-                unreadable.append(str(path))
-                continue
-            files.append(str(path))
-            folders.add(str(path.parent))
-            by_extension[path.suffix.lower() or "(none)"] += 1
-            total_bytes += stat.st_size
-
-    known = _known_paths(config)
-    on_disk = set(files)
-    report = {
-        "schema_version": SCHEMA_VERSION,
-        "generated_at": _now().isoformat(),
-        "revision": revision(),
-        "library_roots": [str(root) for root in roots],
-        "files": len(files),
-        "folders": len(folders),
-        "bytes": total_bytes,
-        "excluded_directories": excluded,
-        "unreadable": unreadable,
-        "by_extension": dict(sorted(by_extension.items())),
-        "reconciliation": {
-            "known_to_database": len(known),
-            "already_registered": len(on_disk & known),
-            "new_to_the_application": len(on_disk - known),
-            "recorded_but_absent": sorted(known - on_disk)[:100],
-            "recorded_but_absent_total": len(known - on_disk),
-        },
-        "mutation": "none — this pass is read-only",
-    }
-    report["checksum"] = sha256_bytes(
-        json.dumps(report, sort_keys=True).encode("utf-8")
-    )
-    return report
-
-
-def _known_paths(config: Config) -> set[str]:
-    """Current asset paths the database holds, or an empty set if there is none."""
-    if not config.database_path.exists():
-        return set()
-    from sqlalchemy import select
-
-    from photo_pipeline.db import create_db_engine, create_session_factory
-    from photo_pipeline.models import Asset
-
-    engine = create_db_engine(config.database_url)
-    try:
-        with create_session_factory(engine)() as session:
-            return {
-                path
-                for path in session.scalars(select(Asset.current_path))
-                if path is not None
-            }
-    except Exception:
-        return set()
-    finally:
-        engine.dispose()
-
-
-# ── the approval ─────────────────────────────────────────────────────────────
-
-
-def approval_path(config: Config) -> Path:
-    return Path(config.data_dir) / APPROVAL_NAME
-
-
-def approve(config: Config, report: dict | Path | str, *, approver: str) -> dict:
-    """Record that a person read this reconciliation and accepts mutation for it."""
-    if isinstance(report, (str, Path)):
-        report = json.loads(Path(report).read_text())
-    if "checksum" not in report:
-        raise ReleaseError("this is not a dry-run report: it has no checksum")
-    record = {
-        "schema_version": SCHEMA_VERSION,
-        "approved_at": _now().isoformat(),
-        "approved_by": approver,
-        "report_checksum": report["checksum"],
-        "library_roots": report["library_roots"],
-        "files": report["files"],
-        "revision": report.get("revision"),
-    }
-    path = approval_path(config)
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(record, indent=2))
-    return record
-
-
-def mutation_blockers(config: Config) -> list[dict]:
-    """Why mutation must stay refused, or an empty list.
-
-    Only enforced when ``require_dry_run_approval`` is configured — the loopback
-    developer setup keeps working unchanged, and an operator turns this on before
-    pointing the application at the library they cannot replace.
-    """
-    if not config.require_dry_run_approval:
-        return []
-    path = approval_path(config)
-    if not path.exists():
-        return [
-            {
-                "code": "dry_run_not_approved",
-                "message": (
-                    "run `python -m photo_pipeline dry-run` and approve its report "
-                    "before mutation is enabled"
-                ),
-            }
-        ]
-    try:
-        record = json.loads(path.read_text())
-    except ValueError:
-        return [{"code": "approval_unreadable", "message": f"{path} is not readable JSON"}]
-    approved_roots = [str(root) for root in record.get("library_roots", [])]
-    configured = [str(root) for root in config.library_roots]
-    if sorted(approved_roots) != sorted(configured):
-        return [
-            {
-                "code": "approval_scope_mismatch",
-                "message": (
-                    f"the approval covers {approved_roots}, but the configured library "
-                    f"is {configured}; run a new dry run"
-                ),
-            }
-        ]
-    return []

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 column, func, select
+from sqlalchemy import 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]:
-        """The current review per asset, as ORM rows. Only for small, known sets —
+        # Latest row per asset. Small local scale: order ascending, let later rows
-        every library-wide caller uses ``latest_reviews()`` in SQL instead."""
+        # overwrite. ponytail: a windowed query if safety_reviews ever grows huge.
         latest: dict[str, SafetyReview] = {}
         for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
             latest[review.asset_id] = review
@@ -148,99 +148,58 @@ class SafetyService:
 
     def current_decision(self, asset_id: str) -> str | None:
         with self._session_factory() as session:
-            latest = latest_reviews().subquery()
+            review = self._latest_by_asset(session).get(asset_id)
-            return session.scalar(
+            return review.decision if review else None
-                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:
-            rows = session.execute(
+            assets = list(session.scalars(_eligible_assets_query()))
-                select(
+            latest = self._latest_by_asset(session)
-                    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:
+            for asset in assets:
+                review = latest.get(asset.id)
+                decision = review.decision if review else None
                 if decision in (SFW, NSFW, "deferred"):
-                out[decision] += int(total)
+                    out[decision] += 1
                 else:
-                # Anything that is not one of the three decisions is undecided —
+                    out["undecided"] += 1
-                # including a score-only review, which is what "scored" counts.
+                    if review and review.score is not None:
-                out["undecided"] += int(total)
+                        out["scored"] += 1
-                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:
-            total = int(
+            assets = list(session.scalars(_eligible_assets_query().order_by(Asset.current_path)))
-                session.scalar(select(func.count()).select_from(query.subquery())) or 0
+            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 = session.execute(
+            }
-                query.order_by(Asset.current_path).limit(limit).offset(offset)
+            rows = []
-            ).all()
+            for asset in assets:
-        return {
+                review = latest.get(asset.id)
-            "total": total,
+                decision = review.decision if review else None
-            "items": [
+                effective = decision or "undecided"
+                if state and state != effective:
+                    continue
+                rows.append(
                     {
-                    "asset_id": asset_id,
+                        "asset_id": asset.id,
-                    "current_path": current_path,
+                        "current_path": asset.current_path,
-                    "score": score,
+                        "score": review.score if review else None,
                         "decision": decision,
-                    "suggested": classify(score) if score is not None else None,
+                        "suggested": classify(review.score) if review and review.score is not None else None,
-                    "exif_verified": bool(exif_verified_at),
+                        "exif_verified": bool(review and review.exif_verified_at),
-                    "exif_state": exif_state,
+                        "exif_state": projections.get(asset.id),
-                }
-                for asset_id, current_path, score, decision, exif_verified_at, exif_state in rows
-            ],
                     }
+                )
+            return {"total": len(rows), "items": rows[offset : offset + limit]}
 
     def scorable_asset_ids(self) -> list[str]:
         """Canonical active assets with a path — the items a scoring job enqueues."""
@@ -294,7 +253,6 @@ class SafetyService:
 
         exif_verified_at = None
         result_sha256 = None
-        result_byte_size = None
         if write_exif and decision in (SFW, NSFW) and path:
             ops = exif_projection(decision)
             # The full checkpoint: write the owned keyword, read the whole file back,
@@ -315,7 +273,6 @@ class SafetyService:
             if result.verified:
                 exif_verified_at = result.verified_at
                 result_sha256 = result.sha256
-                result_byte_size = result.byte_size
 
         now = _now()
         with self._session_factory() as session:
@@ -334,8 +291,6 @@ class SafetyService:
             if result_sha256:
                 asset = session.get(Asset, asset_id)
                 asset.current_sha256 = result_sha256
-                if result_byte_size is not None:
-                    asset.byte_size = result_byte_size
             session.commit()
         return {
             "asset_id": asset_id,
@@ -345,34 +300,6 @@ 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,9 +47,7 @@ class WorkflowService:
             active = self._active_job(session)
 
         safety = SafetyService(self._session_factory).counts()
-        # Reuse the confirmed-SFW total just computed: resolving the current decision
+        analysis = AnalysisService(self._session_factory).counts()
-        # 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 = [

pyproject.toml

@@ -8,32 +8,18 @@ dependencies = [
     "sqlalchemy>=2.0",
     "alembic>=1.13",
     "pydantic>=2.7",
-    # Imaging is runtime, not test-only: thumbnails decode through Pillow, and the
-    # perceptual hash is a DCT over the decoded pixels (services/hashing.py).
-    "pillow>=10",
-    "numpy>=1.26",
-    "scipy>=1.11",
 ]
 
 [project.optional-dependencies]
-# The cloud vision provider. Optional because the analysis stage is the only thing
-# that needs it, and a local review-only install should not pull an API client.
-vision = ["openai>=1.30"]
 test = [
     "pytest>=8",
     "httpx>=0.27",
+    "pillow>=10",
+    "numpy>=1.26",
+    "scipy>=1.11",
     "playwright>=1.40",
     "pytest-playwright>=0.4",
 ]
-
-[build-system]
-requires = ["setuptools>=68"]
-build-backend = "setuptools.build_meta"
-
-[tool.setuptools]
-# The importable application. ``migrations`` and ``work_item`` live beside it but
-# are not part of the package; without this, an editable install cannot guess.
-packages = ["photo_pipeline"]
 # Browser end-to-end tests also require: python -m playwright install chromium
 
 [tool.ruff]

tests/e2e/test_release_gate.py

@@ -1,298 +0,0 @@
-"""The release gate, the read-only dry run, and the approval that unlocks mutation
-(US07-07).
-
-The gate itself is exercised with a tiny stage set — running the whole suite from
-inside the suite would be a fork bomb with better manners. What is proven here is
-the machinery a release depends on: the story matrix is complete, a failing stage
-fails the gate, an unexpected skip fails the gate, and the evidence is written with
-checksums that match what was written.
-
-The dry run is proven to be read-only against a real temporary library, and the
-approval is proven to be what stands between a configured library and any mutation.
-"""
-
-from __future__ import annotations
-
-import json
-import os
-import uuid
-from datetime import datetime, timezone
-from pathlib import Path
-
-import pytest
-from fastapi.testclient import TestClient
-
-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
-from photo_pipeline.services import release
-
-REPO = Path(__file__).resolve().parents[2]
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-# Two throwaway stages: one that passes, one the test can point at a failure.
-PASSING = ("tests/e2e/test_traceability.py",)
-
-
-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 story matrix ─────────────────────────────────────────────────────────
-
-
-def test_every_backlog_story_is_delivered_or_explicitly_planned():
-    matrix = release.story_matrix(REPO)
-
-    assert matrix["problems"] == [], "the story matrix has holes"
-    assert len(matrix["delivered"]) + len(matrix["planned"]) == matrix["stories"]
-    assert "US01-01" in matrix["delivered"] and "US07-07" in matrix["delivered"]
-
-
-def test_a_story_without_tests_is_a_gate_failure(tmp_path):
-    """A story file nobody covered must not pass quietly as 'no tests ran'."""
-    fake = tmp_path / "repo"
-    (fake / "delivery_backlog" / "stories").mkdir(parents=True)
-    (fake / "tests").mkdir()
-    (fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
-    (fake / "tests" / "story_traceability.json").write_text(json.dumps({"stories": {}}))
-
-    matrix = release.story_matrix(fake)
-
-    assert matrix["problems"] == ["story with no tests and not planned: US99-01"]
-
-
-def test_a_mapping_to_a_deleted_test_file_is_a_gate_failure(tmp_path):
-    fake = tmp_path / "repo"
-    (fake / "delivery_backlog" / "stories").mkdir(parents=True)
-    (fake / "tests").mkdir()
-    (fake / "delivery_backlog" / "stories" / "US99-01-invented.md").write_text("# US99-01")
-    (fake / "tests" / "story_traceability.json").write_text(
-        json.dumps({"stories": {"US99-01": ["tests/gone.py"]}})
-    )
-
-    assert release.story_matrix(fake)["problems"] == [
-        "mapped test file is missing — US99-01: tests/gone.py"
-    ]
-
-
-# ── the gate ─────────────────────────────────────────────────────────────────
-
-
-def test_the_gate_runs_its_stages_and_keeps_checksummed_evidence(tmp_path):
-    config = _config(tmp_path)
-    evidence = tmp_path / "evidence"
-
-    report = release.run_gate(config, output=evidence, stages=(("smoke", PASSING),))
-
-    assert report["ok"] is True and report["failures"] == []
-    assert report["stages"][0]["stage"] == "smoke" and report["stages"][0]["ok"] is True
-    assert report["revision"], "the evidence must say which commit it covers"
-    assert report["matrix"]["problems"] == []
-
-    written = json.loads((evidence / release.REPORT_NAME).read_text())
-    assert written["ok"] is True
-    assert (evidence / "logs" / "smoke.log").exists()
-    checksums = (evidence / release.CHECKSUMS_NAME).read_text().splitlines()
-    assert len(checksums) >= 2
-    for line in checksums:
-        digest, name = line.split("  ", 1)
-        assert release.sha256_file(evidence / name) == digest
-
-
-def test_a_failing_stage_fails_the_gate(tmp_path):
-    config = _config(tmp_path)
-    failing = tmp_path / "failing_test.py"
-    failing.write_text("def test_no():\n    assert False\n")
-
-    report = release.run_gate(
-        config, output=tmp_path / "evidence", stages=(("broken", (str(failing),)),)
-    )
-
-    assert report["ok"] is False and report["failures"] == ["broken"]
-    assert report["stages"][0]["returncode"] != 0
-
-
-def test_an_unexpected_skip_fails_the_gate_but_an_environment_skip_does_not():
-    environment = release.StageResult(
-        "unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: exiftool not installed"]
-    )
-    silent = release.StageResult(
-        "unit", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: flaky, look later"]
-    )
-
-    assert release.unexpected_skips([environment]) == []
-    assert release.unexpected_skips([silent, environment]) == [
-        "SKIPPED [1] x.py:1: flaky, look later"
-    ]
-
-
-# ── the real-library dry run ─────────────────────────────────────────────────
-
-
-def _library(root: Path) -> None:
-    (root / "album").mkdir(parents=True)
-    (root / "album" / "a.jpg").write_bytes(b"a" * 128)
-    (root / "album" / "b.png").write_bytes(b"b" * 64)
-    (root / "loose.JPG").write_bytes(b"c" * 32)
-    excluded = root / "_IGNORE" / "private"
-    excluded.mkdir(parents=True)
-    (excluded / "secret.jpg").write_bytes(b"never read")
-
-
-def test_the_dry_run_describes_the_library_without_touching_it(tmp_path):
-    config = _config(tmp_path)
-    root = Path(config.library_roots[0])
-    _library(root)
-    before = {
-        str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
-    }
-
-    report = release.dry_run(config)
-
-    assert report["files"] == 3, "the excluded sentinel is not counted"
-    assert report["by_extension"] == {".jpg": 2, ".png": 1}
-    assert report["excluded_directories"] >= 1
-    assert report["mutation"] == "none — this pass is read-only"
-    assert report["checksum"]
-    assert not any("secret" in json.dumps(report) for _ in [0]), "excluded content never appears"
-    after = {
-        str(p): (p.stat().st_mtime_ns, p.read_bytes()) for p in root.rglob("*") if p.is_file()
-    }
-    assert after == before, "a read-only pass changed the library"
-
-
-def test_the_dry_run_reconciles_against_what_the_database_already_knows(tmp_path):
-    config = _config(tmp_path)
-    root = Path(config.library_roots[0])
-    _library(root)
-    run_migrations(config.database_url)
-    engine = create_db_engine(config.database_url)
-    with create_session_factory(engine)() as session:
-        session.add(
-            Asset(
-                id=str(uuid.uuid4()),
-                original_path=str(root / "album" / "a.jpg"),
-                current_path=str(root / "album" / "a.jpg"),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=128,
-            )
-        )
-        session.add(
-            Asset(
-                id=str(uuid.uuid4()),
-                original_path=str(root / "album" / "gone.jpg"),
-                current_path=str(root / "album" / "gone.jpg"),
-                discovered_at=NOW,
-                hash_version=1,
-                byte_size=1,
-            )
-        )
-        session.commit()
-    engine.dispose()
-
-    reconciliation = release.dry_run(config)["reconciliation"]
-
-    assert reconciliation["known_to_database"] == 2
-    assert reconciliation["already_registered"] == 1
-    assert reconciliation["new_to_the_application"] == 2
-    assert reconciliation["recorded_but_absent_total"] == 1
-    assert reconciliation["recorded_but_absent"][0].endswith("gone.jpg")
-
-
-def test_a_library_root_that_is_not_there_is_refused(tmp_path):
-    config = _config(tmp_path, PHOTO_PIPELINE_LIBRARY_ROOTS=str(tmp_path / "nowhere"))
-    with pytest.raises(release.ReleaseError, match="not a directory"):
-        release.dry_run(config)
-
-
-# ── the approval ─────────────────────────────────────────────────────────────
-
-
-def test_mutation_is_refused_until_the_dry_run_is_approved(tmp_path):
-    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
-    _library(Path(config.library_roots[0]))
-
-    blockers = release.mutation_blockers(config)
-    assert [blocker["code"] for blocker in blockers] == ["dry_run_not_approved"]
-
-    record = release.approve(config, release.dry_run(config), approver="domverse")
-    assert record["approved_by"] == "domverse" and record["report_checksum"]
-    assert release.mutation_blockers(config) == []
-
-
-def test_an_approval_covers_the_library_it_was_written_for(tmp_path):
-    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
-    _library(Path(config.library_roots[0]))
-    release.approve(config, release.dry_run(config), approver="domverse")
-
-    other = tmp_path / "other-library"
-    other.mkdir()
-    moved = config.model_copy(update={"library_roots": (other,)})
-
-    assert [b["code"] for b in release.mutation_blockers(moved)] == ["approval_scope_mismatch"]
-
-
-def test_without_the_requirement_nothing_changes(tmp_path):
-    config = _config(tmp_path)  # the loopback development default
-    assert release.mutation_blockers(config) == []
-
-
-def test_the_api_refuses_every_mutation_until_the_report_is_approved(tmp_path):
-    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
-    _library(Path(config.library_roots[0]))
-    with TestClient(create_app(config)) as client:
-        # Reading stays open: an operator has to see what was found to approve it.
-        assert client.get("/api/v1/workflow").status_code == 200
-        refused = client.post("/api/v1/inventory/scan", json={})
-        assert refused.status_code == 403
-        assert refused.json()["error"]["code"] == "dry_run_not_approved"
-        # A backup is the one mutation a careful operator takes first.
-        assert client.post("/api/v1/backups", json={}).status_code == 201
-
-        release.approve(config, release.dry_run(config), approver="domverse")
-        assert client.post("/api/v1/inventory/scan", json={}).status_code in (200, 201, 202)
-
-
-def test_approving_something_that_is_not_a_report_is_refused(tmp_path):
-    config = _config(tmp_path)
-    with pytest.raises(release.ReleaseError, match="not a dry-run report"):
-        release.approve(config, {"files": 3}, approver="domverse")
-
-
-def test_the_cli_runs_the_dry_run_and_the_approval(tmp_path):
-    config = _config(tmp_path, PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL="1")
-    _library(Path(config.library_roots[0]))
-    from photo_pipeline.__main__ import main
-
-    environment = {
-        "PHOTO_PIPELINE_DATA_DIR": str(config.data_dir),
-        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(config.library_roots[0]),
-        "PHOTO_PIPELINE_REQUIRE_DRY_RUN_APPROVAL": "1",
-    }
-    previous = {key: os.environ.get(key) for key in environment}
-    os.environ.update(environment)
-    try:
-        report_path = tmp_path / "dry-run.json"
-        assert main(["dry-run", "--output", str(report_path)]) == 0
-        assert json.loads(report_path.read_text())["files"] == 3
-        assert main(["approve-dry-run", str(report_path), "--approver", "domverse"]) == 0
-    finally:
-        for key, value in previous.items():
-            if value is None:
-                os.environ.pop(key, None)
-            else:
-                os.environ[key] = value
-    assert release.mutation_blockers(config) == []

tests/e2e/test_release_journey.py

@@ -1,348 +0,0 @@
-"""The release journey (US07-07): one library, one fresh environment, every stage.
-
-This is the acceptance the whole backlog builds up to — discovery, duplicate review,
-safety, analysis, EXIF verification, album proposal, guarded rename, rescan and
-reconciliation, upload, archive, offline deduplication, restore — driven over HTTP
-against real ``photo_pipeline serve`` and worker child processes, with full process
-restarts in the middle and at the end.
-
-Nothing is reached into. External services are the deterministic fakes the earlier
-phases already use, invoked through the real integration layer: a vision fake that
-records every path it was given, a real fake ``immich-go`` executable, and an
-archive medium that is an ordinary directory whose marker file is its identity.
-
-The invariants asserted along the way are the ones the concept calls non-negotiable:
-
-- an asset's identity survives a rename, an upload, an archive, and a restore;
-- an ``_IGNORE`` sentinel is never discovered, counted, analysed, or uploaded;
-- an NSFW asset never reaches the vision provider but still reaches Immich;
-- no photo's bytes are lost at any point — every hash is still reachable somewhere;
-- every stage's durable state survives a restart of both processes.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import shutil
-from pathlib import Path
-
-import httpx
-import pytest
-
-from tests.e2e._pipeline_harness import (
-    SENTINEL_KEY,
-    FakeImmich,
-    Server,
-    fake_uploader,
-    image,
-    seed_library,
-    start_worker,
-    wait_until,
-)
-
-TIMEOUT = 30
-ALBUM = "rome"
-UPLOADER = 'echo "INFO  uploaded $6"\necho "Uploaded 2, duplicates 0"\nexit 0\n'
-
-
-def _sha256(path: Path) -> str:
-    return hashlib.sha256(path.read_bytes()).hexdigest()
-
-
-def _hashes(*roots: Path) -> set[str]:
-    return {
-        _sha256(path)
-        for root in roots
-        for path in root.rglob("*.jpg")
-        if path.is_file() and not path.name.startswith(".")
-    }
-
-
-def _post(base: str, path: str, **kwargs) -> httpx.Response:
-    response = httpx.post(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
-    response.raise_for_status()
-    return response
-
-
-def _get(base: str, path: str, **kwargs) -> dict:
-    response = httpx.get(f"{base}/api/v1{path}", timeout=TIMEOUT, **kwargs)
-    response.raise_for_status()
-    return response.json()
-
-
-def _await_job(base: str, job_id: str, *, states=("succeeded",)) -> dict:
-    return wait_until(
-        lambda: (
-            snapshot
-            if (snapshot := _get(base, f"/jobs/{job_id}"))["state"] in states
-            else None
-        ),
-        timeout=90,
-    )
-
-
-@pytest.fixture
-def library(tmp_path):
-    """A fresh library: an album, an exact duplicate, and an excluded sentinel."""
-    seeded = seed_library(tmp_path, {}, {})
-    album = seeded.lib / ALBUM
-    image(album / "a.jpg", 11)
-    image(album / "b.jpg", 12)
-    shutil.copyfile(album / "a.jpg", album / "a-copy.jpg")  # exact duplicate
-    ignored = seeded.lib / "_IGNORE" / "private"
-    ignored.mkdir(parents=True)
-    image(ignored / "sentinel-9f3a2b.jpg", 99)
-    return seeded
-
-
-@pytest.mark.skipif(shutil.which("exiftool") is None, reason="exiftool not installed")
-def test_the_full_release_journey_survives_every_stage_and_two_restarts(library, tmp_path):
-    immich = FakeImmich()
-    uploader = fake_uploader(tmp_path, UPLOADER)
-    vision_log = tmp_path / "vision.log"
-    archive_root = tmp_path / "medium"
-    archive_root.mkdir()
-    environment = {
-        "PHOTO_PIPELINE_IMMICH_SERVER_URL": immich.url,
-        "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
-        "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(uploader),
-        "PHOTO_PIPELINE_ARCHIVE_FREE_SPACE_RESERVE_BYTES": "0",
-        "PHOTO_PIPELINE_FAKE_VISION_LOG": str(vision_log),
-    }
-    server = Server(library, extra_env=environment).start()
-    worker = start_worker(library, extra_env=environment)
-    base = server.base
-
-    try:
-        # ── 0. discovery ─────────────────────────────────────────────────────
-        _post(base, "/inventory/scan")
-        assets = _get(base, "/inventory/assets", params={"limit": 200})["items"]
-        assert len(assets) == 3, "the sentinel under _IGNORE is not an asset"
-        paths = {asset["current_path"] for asset in assets}
-        assert not any("_IGNORE" in path or "sentinel" in path for path in paths)
-        identity = {asset["id"]: Path(asset["current_path"]).name for asset in assets}
-
-        # ── 1. duplicate review ──────────────────────────────────────────────
-        _post(base, "/duplicates/detect")
-        clusters = _get(base, "/duplicates/clusters")["items"]
-        assert len(clusters) == 1 and clusters[0]["member_total"] == 2
-        cluster = _get(base, f"/duplicates/clusters/{clusters[0]['id']}")
-        canonical = sorted(member["asset_id"] for member in cluster["members"])[0]
-        _post(
-            base,
-            f"/duplicates/clusters/{cluster['id']}/decision",
-            json={
-                "decision": "canonical",
-                "canonical_asset_id": canonical,
-                "expected_version": cluster["version"],
-            },
-        )
-
-        # ── 2. safety, with its EXIF checkpoint ──────────────────────────────
-        queue = _get(base, "/safety/queue", params={"limit": 100})["items"]
-        assert len(queue) == 2, "a non-canonical variant is not reviewed twice"
-        decisions = {}
-        for index, item in enumerate(sorted(queue, key=lambda row: row["current_path"])):
-            decision = "nsfw" if index == 0 else "sfw"
-            decisions[item["asset_id"]] = decision
-            result = _post(
-                base, "/safety/decisions", json={"asset_id": item["asset_id"], "decision": decision}
-            ).json()
-            assert result["exif_verified"] is True, "the safety checkpoint must verify"
-
-        # ── restart: everything so far has to be durable ─────────────────────
-        server.stop()
-        server.start()
-        base = server.base
-        assert _get(base, "/safety/counts")["nsfw"] == 1
-        assert {a["id"] for a in _get(base, "/inventory/assets", params={"limit": 200})["items"]} == set(
-            identity
-        )
-
-        # ── 3. analysis, gated to confirmed-SFW assets ───────────────────────
-        job = _post(base, "/analysis/jobs").json()
-        _await_job(base, job["id"])
-        analysed = [
-            name
-            for name, decision in (
-                (identity[asset_id], decision) for asset_id, decision in decisions.items()
-            )
-            if decision == "sfw"
-        ]
-        seen = vision_log.read_text().splitlines()
-        assert len(seen) == len(analysed) == 1
-        assert not any("sentinel" in line or "_IGNORE" in line for line in seen)
-        nsfw_id = next(aid for aid, decision in decisions.items() if decision == "nsfw")
-        assert all(identity[nsfw_id] not in line for line in seen), "NSFW reached the provider"
-
-        # From here on no stage may change a photo's bytes: the metadata stages are
-        # done, and moving, uploading, archiving, and restoring only relocate them.
-        stable_hashes = _hashes(library.lib)
-
-        # ── 4. album proposal and guarded rename ─────────────────────────────
-        _post(base, "/albums/proposals", json={})
-        proposal = _get(base, f"/albums/proposals/{ALBUM}")
-        _post(
-            base,
-            f"/albums/proposals/{ALBUM}/edit",
-            json={"name": "2019 Rome", "expected_version": proposal["version"]},
-        )
-        proposal = _get(base, f"/albums/proposals/{ALBUM}")
-        _post(
-            base,
-            f"/albums/proposals/{ALBUM}/approve",
-            json={"expected_version": proposal["version"]},
-        )
-        plan = _post(base, "/rename-plans").json()
-        assert plan["blockers"] == [], [
-            (issue["code"], issue["message"])
-            for op in plan["operations"]
-            for issue in op["issues"]
-        ]
-        response = httpx.post(
-            f"{base}/api/v1/rename-plans/{plan['id']}/apply",
-            json={"expected_version": plan["version"], "expected_checksum": plan["checksum"]},
-            timeout=TIMEOUT,
-        )
-        assert response.status_code == 200, response.text
-        applied = response.json()
-        assert applied["failed"] == 0 and applied["applied"] == 1
-        assert (library.lib / "2019 Rome").is_dir() and not (library.lib / ALBUM).exists()
-
-        # ── 5. rescan and reconciliation: identity survives the move ─────────
-        _post(base, "/inventory/scan")
-        after_rename = _get(base, "/inventory/assets", params={"limit": 200})["items"]
-        assert {asset["id"] for asset in after_rename} == set(identity)
-        assert all("2019 Rome" in asset["current_path"] for asset in after_rename)
-        assert _hashes(library.lib) == stable_hashes, "a rename changed a photo's bytes"
-
-        # ── 6. upload ────────────────────────────────────────────────────────
-        report = _post(base, "/upload-preflight", json={"albums": ["2019 Rome"]}).json()
-        assert report["state"] == "ready", report["blockers"]
-        batch = _post(
-            base,
-            "/upload-batches",
-            json={"albums": ["2019 Rome"], "token": report["token"]},
-        ).json()["batches"][0]
-        started = _post(base, f"/upload-batches/{batch['id']}/start").json()
-        _await_job(base, started["job"]["id"])
-        uploaded = _get(base, f"/upload-batches/{batch['id']}")
-        assert uploaded["state"] == "succeeded"
-        # The uploader said nothing per file, so the outcome is uncertain until the
-        # server itself is asked whether it holds those exact bytes (US05-04).
-        assert uploaded["outcome_state"] == "requires_verification"
-        verified = _post(base, f"/upload-batches/{batch['id']}/verify").json()
-        assert verified["outcome_state"] == "verified", verified
-        uploaded = _get(base, f"/upload-batches/{batch['id']}")
-        # Reviewed NSFW is uploaded; it simply never reached the analyser.
-        assert {item["asset_id"] for item in uploaded["items"]} >= {nsfw_id}
-
-        # ── 7. archive ───────────────────────────────────────────────────────
-        location = _post(
-            base, "/archive-locations", json={"name": "external", "root": str(archive_root)}
-        ).json()
-        preflight = _post(
-            base, "/archive-preflight", json={"location_id": location["id"]}
-        ).json()
-        assert preflight["state"] == "ready", [
-            (asset["asset_id"], asset["blockers"])
-            for album in preflight["albums"]
-            for asset in album["assets"]
-            if asset["blockers"]
-        ] or preflight
-        archive_plan = _post(
-            base,
-            "/archive-plans",
-            json={"location_id": location["id"], "token": preflight["token"]},
-        ).json()
-        uploaded_ids = {item["asset_id"] for item in uploaded["items"]}
-        _post(base, f"/archive-plans/{archive_plan['id']}/apply")
-        wait_until(
-            lambda: all(
-                asset["availability_state"].startswith("archived")
-                for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
-                if asset["id"] in uploaded_ids
-            ),
-            timeout=90,
-        )
-        assert _hashes(library.lib, archive_root) == stable_hashes, "archiving lost bytes"
-
-        # ── 8. offline deduplication ─────────────────────────────────────────
-        (archive_root / ".photo-pipeline-archive.json").rename(
-            archive_root / ".photo-pipeline-archive.json.away"
-        )
-        # A copy of an archived photo turns up in the library under its own name —
-        # the real shape of "I re-imported an old card" — so nothing occupies the
-        # path the archived original would be restored to.
-        returned = library.lib / "2019 Rome" / "rediscovered.jpg"
-        returned.parent.mkdir(parents=True, exist_ok=True)
-        archived_copy = next(archive_root.rglob("*.jpg"))
-        shutil.copyfile(archived_copy, returned)
-        _post(base, "/inventory/scan")
-        _post(base, "/duplicates/detect")
-        offline = _get(base, "/inventory/assets", params={"limit": 200})["items"]
-        archived = [a for a in offline if a["availability_state"].startswith("archived")]
-        assert archived, "an unmounted medium must not make assets missing"
-        assert all(a["availability_state"] != "missing_unexpected" for a in offline)
-        assert any(
-            cluster["member_total"] >= 2 for cluster in _get(base, "/duplicates/clusters")["items"]
-        ), "the rediscovered copy did not meet its archived original"
-
-        # ── 9. restore ───────────────────────────────────────────────────────
-        (archive_root / ".photo-pipeline-archive.json.away").rename(
-            archive_root / ".photo-pipeline-archive.json"
-        )
-        restore_report = _post(
-            base, "/restore-preflight", json={"location_id": location["id"]}
-        ).json()
-        restore_plan = _post(
-            base,
-            "/restore-plans",
-            json={"location_id": location["id"], "token": restore_report["token"]},
-        ).json()
-        _post(base, f"/restore-plans/{restore_plan['id']}/apply")
-        wait_until(
-            lambda: all(
-                asset["availability_state"] == "active"
-                for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
-                if asset["id"] in identity
-            ),
-            timeout=90,
-        )
-
-        # ── 10. the final restart proves every stage was durable ─────────────
-        worker.kill()
-        worker.wait(timeout=20)
-        server.stop()
-        server.start()
-        base = server.base
-        final = {
-            asset["id"]: asset
-            for asset in _get(base, "/inventory/assets", params={"limit": 200})["items"]
-        }
-        assert set(identity) <= set(final), "an asset id did not survive the journey"
-        assert _get(base, "/safety/counts")["nsfw"] == 1
-        assert _get(base, "/upload-batches")["batches"][0]["state"] == "succeeded"
-        reachable = {
-            _sha256(path): str(path)
-            for root in (library.lib, archive_root)
-            for path in root.rglob("*.jpg")
-            if path.is_file() and not path.name.startswith(".")
-        }
-        assert stable_hashes <= set(reachable), (
-            "a photo was lost",
-            sorted(stable_hashes - set(reachable)),
-            sorted(reachable.values()),
-        )
-        workflow = _get(base, "/workflow")
-        assert {stage["key"] for stage in workflow["stages"]} >= {
-            "inventory",
-            "duplicates",
-            "safety",
-            "analysis",
-        }
-    finally:
-        worker.kill()
-        worker.wait(timeout=20)
-        server.stop()
-        immich.stop()

tests/integration/test_performance_budgets.py

@@ -1,324 +0,0 @@
-"""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

@@ -15,10 +15,9 @@
       "tests/characterization/test_webapp_query.py"
     ],
     "US01-02": [
-      "tests/integration/test_app_lifecycle.py",
-      "tests/integration/test_migrations.py",
       "tests/unit/test_config.py",
-      "tests/unit/test_env_file.py"
+      "tests/integration/test_migrations.py",
+      "tests/integration/test_app_lifecycle.py"
     ],
     "US01-03": [
       "tests/unit/test_path_policy.py",
@@ -167,21 +166,6 @@
     "US07-05": [
       "tests/integration/test_backup_recovery.py",
       "tests/integration/test_diagnostics.py"
-    ],
-    "US07-06": [
-      "tests/integration/test_performance_budgets.py"
-    ],
-    "US07-07": [
-      "tests/e2e/test_release_gate.py",
-      "tests/e2e/test_release_journey.py"
     ]
-  },
+  }
-  "planned": [
-    "US08-01",
-    "US08-02",
-    "US08-03",
-    "US08-04",
-    "US08-05"
-  ],
-  "_planned_comment": "Accepted backlog stories that are not implemented yet. The release gate (US07-07) requires every story file to be either mapped to tests or listed here, so an unimplemented story is a visible decision rather than a hole in the matrix."
 }

tests/unit/test_env_file.py

@@ -1,69 +0,0 @@
-"""Configuration from a dotenv file, including the archived CLI's variable names.
-
-An operator who already has a ``photo_analyzer.env`` should not have to rewrite it
-to run the application it was replaced by. The file is standing configuration; the
-shell is what you meant this time, so the shell always wins.
-"""
-
-from __future__ import annotations
-
-import os
-
-from photo_pipeline.config import Config, load_env_file, parse_env_file
-
-SAMPLE = """
-# The archived CLI's shape, comments and all.
-LLM_API_KEY=not-real
-LLM_BASE_URL="https://example.invalid/v1beta/openai/"
-LLM_MODEL='gemini-2.5-flash'
-LIBRARY=/tmp/pictures
-
-MAX_WORKERS=4
-# commented=ignored
-malformed line without an equals sign
-"""
-
-
-def test_the_file_is_parsed_and_never_executed():
-    values = parse_env_file(SAMPLE)
-
-    assert values["LLM_BASE_URL"] == "https://example.invalid/v1beta/openai/"  # quotes stripped
-    assert values["LLM_MODEL"] == "gemini-2.5-flash"
-    assert values["MAX_WORKERS"] == "4"
-    assert "commented" not in values and "malformed line without an equals sign" not in values
-
-
-def test_the_archived_cli_names_still_configure_the_application():
-    values = parse_env_file(SAMPLE)
-
-    assert values["OPENAI_API_KEY"] == "not-real"
-    assert values["OPENAI_BASE_URL"] == "https://example.invalid/v1beta/openai/"
-    assert values["PHOTO_PIPELINE_LIBRARY_ROOTS"] == "/tmp/pictures"
-
-
-def test_an_explicit_shell_variable_beats_the_file(tmp_path, monkeypatch):
-    path = tmp_path / "photo_analyzer.env"
-    path.write_text(SAMPLE)
-    monkeypatch.setenv("OPENAI_API_KEY", "from-the-shell")
-    monkeypatch.delenv("PHOTO_PIPELINE_LIBRARY_ROOTS", raising=False)
-
-    applied = load_env_file(path)
-
-    assert "OPENAI_API_KEY" not in applied, "the file overrode an exported value"
-    assert os.environ["OPENAI_API_KEY"] == "from-the-shell"
-    assert os.environ["PHOTO_PIPELINE_LIBRARY_ROOTS"] == "/tmp/pictures"
-    assert Config.from_env().library_roots[0].name == "pictures"
-
-
-def test_the_file_is_found_through_its_variable(tmp_path, monkeypatch):
-    path = tmp_path / "custom.env"
-    path.write_text("PHOTO_PIPELINE_PORT=9123\n")
-    monkeypatch.delenv("PHOTO_PIPELINE_PORT", raising=False)
-    monkeypatch.setenv("PHOTO_PIPELINE_ENV_FILE", str(path))
-    monkeypatch.chdir(tmp_path)  # no ./.env here, so only the variable can find it
-
-    assert Config.from_env().port == 9123
-
-
-def test_a_missing_file_is_not_an_error(tmp_path):
-    assert load_env_file(tmp_path / "nothing-here.env") == {}