US06-06: Automate Phase F End-to-End Acceptance

AGENTS.md

@@ -80,8 +80,7 @@ are the authoritative backlog.
    uncertain. Work only on the claimed story and its generated feature branch.
 4. Read the entire issue, linked specification, dependencies, and acceptance criteria.
    Reconcile them with the concept before designing or changing code.
-5. Inspect the legacy CLI donors — frozen in `legacy_cli_archive/` since US07-01,
-   with their ledger — before replacing applicable behavior. Update the donor
+5. Inspect the legacy CLI donors before replacing applicable behavior. Update the donor
    ledger and characterization tests required by the story.
 6. Implement every acceptance criterion and its automated tests.
 7. Run story-specific tests and the accumulated regression suite required by the epic.

README.md

@@ -17,25 +17,6 @@ python -m photo_pipeline serve     # start the API + static review UI (127.0.0.1
 Configuration comes from `PHOTO_PIPELINE_*` environment variables (see
 `photo_pipeline/config.py`); secrets are referenced, never logged.
 
-### API access (US07-02)
-
-The app listens on loopback, so its attacker is another page in the same browser.
-Every `/api/v1` route except `health/live`, `health/ready`, and `session` requires
-the application session, and every mutation requires its CSRF token as well:
-
-```bash
-BASE=http://127.0.0.1:8000
-TOKEN=$(curl -sc /tmp/pp.jar $BASE/api/v1/session | python -c 'import json,sys; print(json.load(sys.stdin)["csrf_token"])')
-curl -sb /tmp/pp.jar -H "X-CSRF-Token: $TOKEN" -X POST $BASE/api/v1/albums/proposals -d '{}' -H 'Content-Type: application/json'
-```
-
-The session is per server process — restarting `serve` invalidates it, and the
-browser client re-bootstraps by itself. Requests are also refused when the `Host` is
-not a loopback name (DNS rebinding), when `Origin` is any other origin, when
-`Sec-Fetch-Site` says the request came from another site (an `<img>` pointed at a
-thumbnail), or when the body exceeds `PHOTO_PIPELINE_MAX_REQUEST_BYTES`. There is no
-CORS middleware at all, so no other origin can read a response.
-
 ## Testing
 
 One offline command runs the whole suite (unit, integration, and browser
@@ -227,36 +208,3 @@ work_item/scripts/python -m pytest -m phase_f -q
   offline browsing, restore, collision, keyboard confirmation, and reload).
 
 Phases A–E remain green in the full run above.
-
-## Legacy CLI archive
-
-The command-line tools this application was extracted from are frozen in
-`legacy_cli_archive/` (US07-01): the original sources, their docs, the dependency
-lock they were last verified against, schema notes, a redacted sample
-configuration, the donor ledger, and a checksum for every file.
-
-```bash
-cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256   # verify the archive
-work_item/scripts/python -m pytest tests/unit/test_legacy_archive.py -q   # lint it
-```
-
-They are reference material and rollback evidence only. No module under
-`photo_pipeline/` imports or executes them, the archive is not on the application's
-import path, and `tests/unit/test_legacy_archive.py` enforces that along with the
-checksums and the redaction. Only the two suites that compare *against* the donors —
-`tests/characterization/` and `tests/integration/test_safety_parity.py` — put the
-archived sources on `sys.path`.
-
-The last path-keyed state they owned, `nsfw_scores.csv`, is imported once and then
-left alone:
-
-```bash
-work_item/scripts/python -m photo_pipeline import-legacy-scores /path/to/nsfw_scores.csv --dry-run
-```
-
-The import writes scored-but-unreviewed `safety_reviews` rows onto stable asset ids,
-never invents an asset for an unknown path, never overwrites a human decision, and
-writes a reconciliation report to the data directory saying exactly what it did.
-`legacy_cli_archive/donor_ledger.yaml` records every migrated behavior with its
-target, the tests that pin the donor, the tests that prove the replacement, and each
-intentional delta; rows still marked `pending` name the story that will resolve them.

donor_ledger.yaml

@@ -12,15 +12,7 @@
 #                            (if anything) carries over
 # Every row needs either `tests` (existing test IDs, module::function) or
 # `pending_story` (the backlog story that will characterize/deliver it).
-# status: characterized — donor behavior pinned by characterization tests
-#         resolved      — replacement shipped; `parity` names the tests that prove
-#                         it, and `delta` states every intentional difference
-#         pending       — not migrated yet; `pending_story` says which story will
-# parity: test ids (path::function) in any suite, proving the replacement
-# delta:  what the replacement deliberately does differently, or not at all
-#
-# Archived by US07-01: the sources referenced below now live beside this file in
-# src/ and are frozen (see README.md). Nothing in photo_pipeline imports them.
+# status: characterized | pending
 
 rows:
   # ── photo_analyzer.py ──────────────────────────────────────────────────────
@@ -88,12 +80,8 @@ rows:
       Copying a primary's analysis into variant rows survives, but keyed by
       asset_id and recorded as stage state instead of raw row copies.
     target: photo_pipeline/services/duplicates.py
-    parity:
-      - tests/integration/test_duplicate_engine.py::test_exact_copies_form_auto_decided_cluster
-      - tests/integration/test_duplicate_engine.py::test_perceptual_variant_is_review_only
-    delta: >
-      The donor propagated variant links implicitly while writing rows; the replacement makes the canonical link a reviewable, reversible cluster decision, so a propagated link can always be undone.
-    status: resolved
+    pending_story: US01-04
+    status: pending
 
   - id: pa-hashing
     area: hashing
@@ -153,12 +141,8 @@ rows:
     classification: replace
     rationale: Console report; superseded by the duplicate-review API/UI (US01-06).
     target: photo_pipeline/api/routes + frontend duplicate review
-    parity:
-      - tests/integration/test_review_api.py::test_clusters_list_and_detail
-      - tests/e2e/test_review_ui.py::test_fuzzy_decision_requires_confirmation
-    delta: >
-      The text listing became the paged cluster API and the comparison UI; no textual report is produced.
-    status: resolved
+    pending_story: US01-06
+    status: pending
 
   - id: pa-reconcile
     area: database
@@ -218,7 +202,7 @@ rows:
       RGB-normalize (drops alpha, converts HEIC), LANCZOS resize to 2048px
       long-edge, JPEG q85 base64 — the provider-input contract. Truncated-image
       tolerance (ImageFile.LOAD_TRUNCATED_IMAGES) carries with it.
-    target: photo_pipeline/services/analysis.py
+    target: photo_pipeline/integrations/vision.py
     tests:
       - test_pa_imaging::test_prepare_image_small_passthrough_jpeg
       - test_pa_imaging::test_prepare_image_resizes_to_max_long_edge
@@ -246,12 +230,9 @@ rows:
       response validation, 429/503 retry with exponential backoff. Prompt and
       model/config version must be persisted per analysis_runs. Characterized
       against a deterministic fake provider when the analysis service is ported.
-    target: photo_pipeline/services/analysis.py
-    parity:
-      - tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
-    delta: >
-      The prompt and response schema carry over; the provider is an injected adapter so the privacy gate is testable, and results are keyed to asset ids rather than paths.
-    status: resolved
+    target: photo_pipeline/integrations/vision.py
+    pending_story: US02-06
+    status: pending
 
   - id: pa-throttle
     area: logging
@@ -260,8 +241,8 @@ rows:
     rationale: >
       Rolling throttle window + persistent throttle_events.jsonl + RPD day
       counter become job metrics/events on the durable job model.
-    target: photo_pipeline/services/jobs.py + photo_pipeline/services/analysis.py
-    pending_story: US07-04
+    target: photo_pipeline/jobs/coordinator.py
+    pending_story: US02-02
     status: pending
 
   - id: pa-nsfw-filter
@@ -331,12 +312,8 @@ rows:
       checked between items, double-SIGINT force quit — becomes the durable
       JobRunner worker loop with the same drain-and-resume semantics.
     target: photo_pipeline/jobs/worker.py
-    parity:
-      - tests/integration/test_worker.py::test_worker_processes_all_items
-      - tests/integration/test_worker.py::test_cooperative_cancellation_leaves_items_resumable
-    delta: >
-      The in-process folder loop with SIGINT handling became durable jobs claimed by a worker: cancellation is a persisted request, not a signal, and an interrupted run resumes from the database instead of restarting.
-    status: resolved
+    pending_story: US02-02
+    status: pending
 
   - id: pa-ui-terminal
     area: ui
@@ -394,12 +371,8 @@ rows:
       the JSONL history logger become structured JSON logging with job_id/
       asset_id and job_events rows; per-photo history maps to job events.
     target: photo_pipeline structured logging + jobs/job_events
-    parity:
-      - tests/integration/test_jobs.py::test_enqueue_persists_items_and_event
-      - tests/integration/test_jobs_sse.py::test_sse_streams_all_events_then_closes
-    delta: >
-      The JSONL history file and rich console handler are replaced by structured JSON logs plus durable job_events; the browser reads events over SSE rather than tailing a file.
-    status: resolved
+    pending_story: US02-02
+    status: pending
 
   - id: pa-balance
     area: vision
@@ -409,11 +382,8 @@ rows:
       Provider balance/quota probes (report 'unsupported' on providers without
       the endpoint). Network-bound; characterized against the fake provider.
     target: photo_pipeline/services/analysis.py
-    parity:
-      - tests/integration/test_safety_analysis.py::test_provider_called_only_for_confirmed_sfw
-    delta: >
-      Not carried over: balance/quota polling was provider-specific (Gemini/OpenAI billing endpoints) and key-scoped. Cost reporting, when a story asks for it, comes from the per-run usage recorded with each analysis result rather than from a vendor endpoint.
-    status: resolved
+    pending_story: US02-06
+    status: pending
 
   - id: pa-cli
     area: configuration
@@ -423,13 +393,9 @@ rows:
       argparse surface is superseded by the API; flags map to job configs
       (documented in WEBAPP_CONCEPT.md §8 parity table). Transitional CLI calls
       the shared services until archival (E07).
-    target: photo_pipeline/__main__.py (serve | worker | migrate | import-legacy-scores) + /api/v1
-    parity:
-      - tests/integration/test_app_lifecycle.py::test_restart_preserves_data_and_reruns_migrations
-      - tests/unit/test_legacy_archive.py::test_production_code_never_imports_an_archived_module
-    delta: >
-      The argparse surface is not reproduced. Every flag that drove work became an API command or a job configuration; the CLI keeps only what an application needs to be operated (serve, worker, migrate) plus the one-off legacy CSV import.
-    status: resolved
+    target: photo_pipeline/api + transitional CLI
+    pending_story: US07-01
+    status: pending
 
   # ── nsfwtag/ ───────────────────────────────────────────────────────────────
   - id: nt-discovery
@@ -456,17 +422,9 @@ rows:
       nsfw_scores.csv stops being the source of truth (concept: DB state).
       Format characterized (4-decimal scores, bad rows dropped) because the
       existing CSV must migrate into assets.safety_score.
-    target: photo_pipeline/services/legacy_import.py (one-off import into safety_reviews)
+    target: photo_pipeline/repositories (safety), CSV import in US01-02 migration
     tests: [test_nsfwtag::test_score_cache_roundtrip_and_tolerance]
-    parity:
-      - tests/integration/test_legacy_import.py::test_scores_are_imported_onto_asset_identity
-      - tests/integration/test_legacy_import.py::test_a_reviewed_asset_is_never_overwritten_by_the_csv
-    delta: >
-      The CSV is no longer read at runtime at all: it is imported once into
-      scored-but-unreviewed safety_reviews rows and left on disk untouched. A path
-      that matches nothing is reported, never turned into an asset, and a human
-      decision always outranks an imported score.
-    status: resolved
+    status: characterized
 
   - id: nt-score-model
     area: nsfw
@@ -527,11 +485,8 @@ rows:
       Newline-list bulk tagging (nsfw_confirmed.txt flow) is superseded by DB
       review decisions; the existing list is a one-time migration input.
     target: photo_pipeline/services/safety.py (decision import in US01-02)
-    parity:
-      - tests/integration/test_safety_parity.py::test_extracted_marks_and_partition_match_donor
-    delta: >
-      Bulk keyword application from a file list is replaced by decisions against asset ids; the EXIF write itself is the extracted, read-back-verified one.
-    status: resolved
+    pending_story: US01-02
+    status: pending
 
   - id: nt-ui
     area: ui
@@ -543,11 +498,8 @@ rows:
       threshold/score review flow, lightbox and keyboard model are the frontend
       donor for the Safety view (preserved per concept §10; ported in US02-01).
     target: photo_pipeline/api + frontend Safety view
-    parity:
-      - tests/e2e/test_workflow_views.py::test_safety_review_decide_persists_across_reload
-    delta: >
-      The stdlib review server is replaced by the API plus the Safety view; the donor's layout, thresholds, and keyboard flow carry over, its embedded HTML generation does not.
-    status: resolved
+    pending_story: US02-01
+    status: pending
 
   - id: nt-bench
     area: nsfw
@@ -557,11 +509,8 @@ rows:
       Dev-only model benchmark; archived without webapp replacement (recorded
       basis of the AdamCodd model choice). No production caller.
     target: none (archive as reference)
-    parity:
-      - tests/unit/test_legacy_archive.py::test_every_archived_source_matches_its_checksum
-    delta: >
-      No replacement: a dev-only benchmark whose result (the AdamCodd model choice) is already recorded. Kept in the archive as the basis of that choice.
-    status: resolved
+    pending_story: US07-01
+    status: pending
 
   # ── webapp/ ────────────────────────────────────────────────────────────────
   - id: wa-query-search
@@ -630,13 +579,9 @@ rows:
       Subprocess-driving-the-CLI job control is superseded by durable DB jobs
       with a worker process. Two ideas carry over: progress derived from DB
       counts (not job-private state) and single-mutating-job enforcement.
-    target: photo_pipeline/services/jobs.py + photo_pipeline/jobs/worker.py
-    parity:
-      - tests/integration/test_jobs.py::test_idempotency_key_returns_same_job
-      - tests/integration/test_worker.py::test_handler_failure_fails_the_job
-    delta: >
-      Subprocess supervision of a CLI is replaced by durable jobs in the same process family: there is no subprocess to supervise, and progress is persisted rather than scraped from stdout.
-    status: resolved
+    target: photo_pipeline/jobs/coordinator.py
+    pending_story: US02-02
+    status: pending
 
   - id: wa-server
     area: ui
@@ -648,9 +593,5 @@ rows:
       browser. analyzer.html + page.py design (dark OLED tokens, Library/
       Analyze/Stats views) is frontend donor material per concept §10.
     target: photo_pipeline/api/app.py + frontend
-    parity:
-      - tests/e2e/test_frontend_shell.py::test_shell_loads_assets_without_console_or_network_errors
-      - tests/e2e/test_frontend_shell.py::test_deep_link_and_reload_restore_view_and_filters
-    delta: >
-      The stdlib HTTP routes become the versioned FastAPI API and the static frontend shell; Python no longer interpolates HTML.
-    status: resolved
+    pending_story: US02-05
+    status: pending

frontend/js/api.js

@@ -2,43 +2,14 @@
 // cancellation. Every method accepts an optional { signal } from cancellable().
 export const BASE = "/api/v1";
 
-// The API refuses every request without the session cookie, and every mutation
-// without this token echoed back. The token is readable only same-origin, which is
-// what makes it proof that the caller is this app and not another page.
-let csrfToken = null;
-
-async function session() {
-  if (csrfToken === null) {
-    const response = await fetch(BASE + "/session", { credentials: "same-origin" });
-    const body = await response.json().catch(() => null);
-    csrfToken = (body && body.csrf_token) || null;
-  }
-  return csrfToken || "";
-}
-
-async function send(path, { signal, ...options }) {
-  return fetch(BASE + path, {
-    credentials: "same-origin",
-    signal,
-    ...options,
-    headers: {
-      "Content-Type": "application/json",
-      "X-CSRF-Token": await session(),
-      ...(options.headers || {}),
-    },
-  });
-}
-
 async function request(path, { signal, ...options } = {}) {
   let response;
   try {
-    response = await send(path, { signal, ...options });
-    // A restarted server issues a new session; re-bootstrap once rather than
-    // stranding an open tab on 401.
-    if (response.status === 401) {
-      csrfToken = null;
-      response = await send(path, { signal, ...options });
-    }
+    response = await fetch(BASE + path, {
+      headers: { "Content-Type": "application/json" },
+      signal,
+      ...options,
+    });
   } catch (error) {
     // A caller-cancelled fetch is not a failure; tag it so views can ignore it.
     if (error.name === "AbortError") {

frontend/js/tests/unit.js

@@ -29,10 +29,6 @@ function jsonResponse(status, body) {
 const tick = (ms = 0) => new Promise((r) => setTimeout(r, ms));
 
 async function run() {
-  // The client fetches its CSRF token once, lazily (US07-02). Do that against the
-  // real server first, so the stubbed fetch below only ever sees the call under test.
-  await api.workflow().catch(() => {});
-
   // ── store ────────────────────────────────────────────────────────────────
   {
     const store = createStore({ n: 0 });

legacy_cli_archive/CHECKSUMS.sha256

@@ -1,22 +0,0 @@
-d098bbde13d2ebc872ca781e244cc1e48b97cc551fedaca3251d6a5278a15234  src/compare_models.py
-7d0f68cb95fbd6603e4c558620098b46929ebdb6fd91a598bbf84f26eea35e6c  src/nsfw_tag.py
-35329c53570e215cef15f59429c1a5251a0448b90522b0910ab9d68efe1bc307  src/nsfwtag/__init__.py
-7f52b55e4f3b77eda3657d7cd2272c7cca8422100e241671de495603c1126ee2  src/nsfwtag/__main__.py
-6f5a97114e0d87d272ce22d065dc31fa0dd72ff8590d11ad14cb1c1f486d339c  src/nsfwtag/bench.py
-29498fac1d73ba2b7420ffe1ffe49a87f684ec299e2c3b9c6e14e63e95643ba4  src/nsfwtag/exif.py
-934e82c402813ebf503e5a20eb03d5103f84df13117bc4dd1fd95deaa01ab263  src/nsfwtag/README.md
-68125e6184c7e4d2a5b0153f2155675753ab0b1329d1dbc4933d4769492ec4be  src/nsfwtag/review.html
-a597eab74803dd31452b5d70abf7d6d6320eb17a61596c891b009da15d624e98  src/nsfwtag/scoring.py
-674969a18e58ee511a2abf574a875db92f2f524769b1b614d0d3662ab879a709  src/nsfwtag/server.py
-24a7c8d029da6e97d46b110a9fe9dbb3900127f8ae3c142f634fdd91d6c45243  src/nsfwtag/webapp.py
-2c2ea558f1b9095c1150f078ea29c8e0f180cc045a9d1097b83f61bfb145295b  src/photo_analyzer.py
-5fb7ce2f977a17da1f501c45d7985328318ae14298a3f347eab46a6e3f02aab3  src/test_dedup.py
-1673dc76cc60aba56a5b065d9bc9f5dda00342cf2a79d000a3f74686fb7cffec  src/test_nsfw_skip.py
-55aa82f348e0e90be0163ba6aa278e5fc8e445996e5335774917e5ef59ad3563  src/webapp/__init__.py
-ba4342bd0175591a2121f063f14a78a602d5679bcfcf24aa14c8d4b8cd5da1a2  src/webapp/__main__.py
-a60f24035a989909467778f888d62854c1afc2702f47b8d3cca32bf3cf04d5cf  src/webapp/analyzer.html
-018bcc2f770d716444b456db58d6b4a41800138a98598b3c27ba503ab22d939e  src/webapp/page.py
-5f599b107b2b117ca118b6fbdec5e5786a9ab4eea424aa9af1bfea7bf87cebab  src/webapp/query.py
-875cce697caa02717c85a2707ba935c46dd9ace18428fd4abb13105b7a8e24d9  src/webapp/README.md
-af7d0d72d245b4bbb1e0e30f9708239697543212d4171defd1bfe26dd24a4f05  src/webapp/runner.py
-cddf0555b06fe7e4daa14309baaaf91130f8506922cb0ffb6507d76cbe39add4  src/webapp/server.py

legacy_cli_archive/README.md

@@ -1,80 +0,0 @@
-# Legacy CLI archive (US07-01)
-
-Frozen, read-only sources of the command-line tools this application was extracted
-from. They are **reference material and rollback evidence** — provenance for
-behavior that now lives in `photo_pipeline/`, and the only way to answer "what did
-the original actually do?" once the replacement has drifted.
-
-> **Nothing here is production code.** No module under `photo_pipeline/` imports or
-> executes anything in this directory, and this directory is not on the application's
-> import path. `tests/unit/test_legacy_archive.py` enforces both, along with the
-> checksums and the redaction below.
-
-## What is here
-
-| Path | Role |
-|---|---|
-| `src/photo_analyzer.py` | the analysis CLI: discovery, hashing/dedup, vision analysis, EXIF writing, SQLite schema, album naming |
-| `src/nsfwtag/` | NSFW scoring, EXIF safety keywords, and the review server (`__init__` 1.1.0) |
-| `src/webapp/` | the stdlib review web app: FTS search, stats, subprocess runner, HTML shell (`__init__` 0.1.0) |
-| `src/nsfw_tag.py` | thin backwards-compatible entry point for `nsfwtag` |
-| `src/compare_models.py` | dev-only model comparison script |
-| `src/test_dedup.py`, `src/test_nsfw_skip.py` | the CLIs' own standalone self-checks (never pytest suites) |
-| `donor_ledger.yaml` | the donor ledger: every migrated behavior, its target, its tests, and every intentional delta |
-| `requirements-lock.txt` | the dependency versions the frozen sources were last verified against |
-| `photo_analyzer.env.sample` | the CLI's configuration surface, with every value replaced by a placeholder |
-| `CHECKSUMS.sha256` | SHA-256 of every archived source file |
-
-`photo_analyzer.py` carries no `__version__`; its identity is its checksum, recorded
-in `CHECKSUMS.sha256` and taken at commit `9b7ee6b` (the merge of US06-06, the last
-commit before archival).
-
-## Verifying the archive
-
-```bash
-cd legacy_cli_archive && shasum -a 256 -c CHECKSUMS.sha256
-```
-
-Any edit to an archived source must be accompanied by a regenerated checksum file
-and a note here explaining why a *frozen* archive changed — the normal answer being
-that it should not.
-
-## Schema notes
-
-`photo_analyzer.py` owned a path-keyed SQLite database (`SCHEMA`, near the top of the
-file):
-
-- `photos(id, path UNIQUE, status, phash, file_sha1, dup_of, description, tags,
-  people_count, setting, time_of_day, season, mood, location_hint, approx_year,
-  raw_response, error_message, analyzed_at, exif_written_at)`;
-- `photos_fts` — an FTS5 external-content index over `path, description, tags, mood,
-  location_hint`, kept in sync by insert/update/delete triggers;
-- late columns (`phash`, `file_sha1`, `dup_of`) were added by an in-code
-  `_migrate_schema()` rather than a migration tool, and their indexes are created
-  only after the `ALTER`.
-
-The replacement keeps the same analysis fields but re-keys everything to a stable
-`assets.id` (Alembic migrations `0001`…), because a path is not an identity: the
-donor's `path UNIQUE` is exactly what broke on every move and rename.
-
-`nsfwtag` kept its safety scores outside the database in `nsfw_scores.csv`
-(`path,nsfw_score`, four decimals, unreadable rows dropped). That file is no longer a
-source of truth; `photo_pipeline/services/legacy_import.py` imports it into
-`assets.safety_score` and reports exactly what matched, what did not, and why.
-
-## Redaction
-
-The archive contains no credentials. `photo_analyzer.env.sample` documents the
-configuration surface (`LLM_API_KEY`, `LLM_BASE_URL`, `LLM_MODEL`, and the tuning
-variables) with placeholder values only; the CLI itself never contained a key, it
-read one from `photo_analyzer.env` or the environment. No `.env`, database, log, CSV,
-or photo from the author's library is archived.
-
-## Why these tools were retired
-
-Each behavior's fate is recorded per row in `donor_ledger.yaml`: `reuse`, `extract`,
-`refactor`, or `replace`, with the target module, the characterization tests that
-pinned the donor's behavior, the parity tests the replacement passes, and — where the
-replacement deliberately does something else or nothing at all — a `delta` saying so.
-Rows still marked `pending` name the backlog story that will resolve them; they are
-the honest list of what has *not* been carried over yet.

legacy_cli_archive/photo_analyzer.env.sample

@@ -1,18 +0,0 @@
-# Sample configuration for the archived photo_analyzer.py CLI (US07-01).
-#
-# REDACTED: no value below is real. Each line names a variable the CLI read and
-# what belongs there; the placeholders are deliberately not key-shaped, so this
-# file can never be mistaken for — or scanned as — a credential.
-
-LLM_API_KEY=<paste your provider key here>
-LLM_BASE_URL=<provider base url, e.g. the OpenAI-compatible Gemini endpoint>
-LLM_MODEL=<model id, e.g. a Gemini Flash release>
-
-# Optional tuning the CLI read from the same file:
-PHASH_THRESHOLD=8
-MAX_WORKERS=4
-RETRY_ATTEMPTS=3
-RPD_LIMIT=0
-
-# The CLI ignored its own shipped placeholder (a literal "sk-REPLACE..." string)
-# until it was replaced, and a shell variable always won over this file.

legacy_cli_archive/requirements-lock.txt

@@ -1,20 +0,0 @@
-# Final dependency lock of the archived CLIs (US07-01).
-#
-# These are the versions present in the environment the archive was taken from —
-# what the frozen sources were last verified against by the characterization
-# suite. Restoring a donor for forensics means pinning these, not "latest".
-#
-# Python 3.14.6
-
-openai==3.0.0          # photo_analyzer: OpenAI-compatible vision client
-numpy==2.4.6           # photo_analyzer, nsfwtag: pixel work
-Pillow==12.3.0         # photo_analyzer, nsfwtag: decode/resize
-rich==15.0.0           # photo_analyzer: console output
-scipy==1.18.0          # photo_analyzer: perceptual-hash DCT
-PyYAML==6.0.3          # tooling that reads the donor ledger
-
-# NSFW inference (nsfwtag/scoring.py, nsfwtag/bench.py) was never installed in the
-# archiving environment; the model stack is recorded here from the sources so a
-# forensic run can reproduce it, not from a resolved lock:
-#   torch, transformers, timm  — AdamCodd/vit-base-nsfw-detector (see nsfwtag/README.md)
-# exiftool is an external binary, not a Python package.

photo_pipeline/__main__.py

@@ -1,4 +1,4 @@
-"""Application management CLI: ``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores}``."""
+"""Application management CLI: ``python -m photo_pipeline {serve,migrate}``."""
 
 from __future__ import annotations
 
@@ -16,17 +16,6 @@ def main(argv: Sequence[str] | None = None) -> int:
     commands.add_parser("migrate", help="Upgrade the database to the latest revision")
     worker_cmd = commands.add_parser("worker", help="Run a durable-job worker")
     worker_cmd.add_argument("--id", default="worker-1", help="Worker id (lease owner)")
-    import_cmd = commands.add_parser(
-        "import-legacy-scores",
-        help="Import the archived CLI's nsfw_scores.csv into the database (US07-01)",
-    )
-    import_cmd.add_argument("csv", help="Path to nsfw_scores.csv")
-    import_cmd.add_argument(
-        "--overwrite", action="store_true", help="Replace differing imported scores"
-    )
-    import_cmd.add_argument(
-        "--dry-run", action="store_true", help="Report what would happen and change nothing"
-    )
     args = parser.parse_args(argv)
 
     config = Config.from_env()
@@ -36,24 +25,6 @@ def main(argv: Sequence[str] | None = None) -> int:
         run_migrations(config.database_url)
         return 0
 
-    if args.command == "import-legacy-scores":
-        import json
-
-        from photo_pipeline.db import create_db_engine, create_session_factory
-        from photo_pipeline.services.legacy_import import LegacyImportService, write_report
-
-        run_migrations(config.database_url)
-        engine = create_db_engine(config.database_url)
-        service = LegacyImportService(create_session_factory(engine))
-        report = service.import_nsfw_scores(
-            args.csv, overwrite=args.overwrite, dry_run=args.dry_run
-        )
-        # The report is the point: an import nobody can audit is not a migration.
-        if not args.dry_run:
-            write_report(report, config.data_dir)
-        print(json.dumps(report.counts, indent=2))
-        return 0
-
     if args.command == "worker":
         from photo_pipeline.db import create_db_engine, create_session_factory
 

photo_pipeline/api/app.py

@@ -8,15 +8,11 @@ and exposes the versioned ``/api/v1`` surface; US01-02 ships only health.
 
 from __future__ import annotations
 
-import logging
 from contextlib import asynccontextmanager
 from pathlib import Path
 
-from fastapi import FastAPI, Request
-from fastapi.exceptions import RequestValidationError
-from fastapi.responses import JSONResponse
+from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
-from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from photo_pipeline.api.routes import (
     albums,
@@ -29,12 +25,10 @@ from photo_pipeline.api.routes import (
     library,
     renames,
     safety,
-    session as session_routes,
     thumbnails,
     uploads,
     workflow,
 )
-from photo_pipeline.api.security import DEFAULT_HEADERS, SecurityMiddleware, Session
 
 # Registers the safety_score / analysis job handlers on import.
 import photo_pipeline.jobs.domain_handlers  # noqa: F401
@@ -45,41 +39,6 @@ from photo_pipeline.services.upload_batches import UploadBatchService
 
 FRONTEND_DIR = Path(__file__).resolve().parents[2] / "frontend"
 
-log = logging.getLogger(__name__)
-
-
-def _envelope(status: int, code: str, message: str) -> JSONResponse:
-    return JSONResponse(
-        status_code=status,
-        content={"error": {"code": code, "message": message}},
-        headers=DEFAULT_HEADERS,
-    )
-
-
-def _install_error_handlers(app: FastAPI) -> None:
-    """One JSON error envelope everywhere, and nothing behind it.
-
-    An unhandled exception carries the library's absolute paths, SQL, and sometimes
-    a credential in its text; the client gets a code, the operator gets the traceback
-    in the server log (US07-02).
-    """
-
-    @app.exception_handler(StarletteHTTPException)
-    async def _http_error(request: Request, exc: StarletteHTTPException):
-        return _envelope(exc.status_code, "http_error", str(exc.detail))
-
-    @app.exception_handler(RequestValidationError)
-    async def _validation_error(request: Request, exc: RequestValidationError):
-        # Field locations only: the echoed input can be the caller's own data, but it
-        # is also what ends up in shared logs and screenshots.
-        fields = sorted(".".join(str(part) for part in error["loc"]) for error in exc.errors())
-        return _envelope(422, "invalid_request", f"invalid request fields: {', '.join(fields)}")
-
-    @app.exception_handler(Exception)
-    async def _unhandled(request: Request, exc: Exception):
-        log.exception("unhandled error serving %s", request.url.path)
-        return _envelope(500, "internal_error", "internal error")
-
 
 def create_app(config: Config | None = None) -> FastAPI:
     config = config or Config.from_env()
@@ -103,12 +62,6 @@ def create_app(config: Config | None = None) -> FastAPI:
             app.state.engine = None
 
     app = FastAPI(title="Photo Pipeline", version="0.1.0", lifespan=lifespan)
-    # One session per process: the browser exchanges it for a cookie + CSRF token,
-    # and every other origin is refused before a route ever runs (US07-02).
-    app.state.session = Session.create()
-    app.add_middleware(SecurityMiddleware, session=app.state.session, config=config)
-    _install_error_handlers(app)
-    app.include_router(session_routes.router, prefix="/api/v1")
     app.include_router(health.router, prefix="/api/v1")
     app.include_router(inventory.router, prefix="/api/v1")
     app.include_router(duplicates.router, prefix="/api/v1")

photo_pipeline/api/routes/analysis.py

@@ -18,10 +18,7 @@ router = APIRouter(tags=["analysis"])
 
 
 def _service(request: Request) -> AnalysisService:
-    return AnalysisService(
-        request.app.state.session_factory,
-        library_roots=tuple(request.app.state.config.library_roots),
-    )
+    return AnalysisService(request.app.state.session_factory)
 
 
 def _error(status: int, code: str, message: str) -> JSONResponse:

photo_pipeline/api/routes/session.py

@@ -1,30 +0,0 @@
-"""Session bootstrap: the one endpoint reachable without a session (US07-02).
-
-It sets the ``HttpOnly``/``SameSite=Strict`` session cookie and returns the CSRF
-token in the body. A foreign page can call this — it just cannot read the answer,
-because the app sends no CORS headers — and the cookie it received is never attached
-to a request that foreign page initiates.
-"""
-
-from __future__ import annotations
-
-from fastapi import APIRouter, Request
-from fastapi.responses import JSONResponse
-
-from photo_pipeline.api.security import SESSION_COOKIE
-
-router = APIRouter(tags=["session"])
-
-
-@router.get("/session")
-def start_session(request: Request) -> JSONResponse:
-    session = request.app.state.session
-    response = JSONResponse({"csrf_token": session.csrf_token})
-    response.set_cookie(
-        SESSION_COOKIE,
-        session.id,
-        httponly=True,
-        samesite="strict",
-        path="/",
-    )
-    return response

photo_pipeline/api/routes/thumbnails.py

@@ -31,7 +31,5 @@ def get_thumbnail(asset_id: str, request: Request, size: int = Query(512)):
     return FileResponse(
         path,
         media_type="image/webp",
-        # private: the URL is versioned and immutable, but these bytes are the user's
-        # photos and must never sit in a shared cache (US07-02).
-        headers={"Cache-Control": "private, max-age=31536000, immutable"},
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
     )

photo_pipeline/api/security.py

@@ -1,192 +0,0 @@
-"""Local-web attack surface: session, CSRF, Origin/Host checks, default headers.
-
-The app binds to 127.0.0.1, so the attacker is not a remote client but another page
-in the user's browser (concept §15, "Local web attack"): any site can issue requests
-to ``http://127.0.0.1:8000`` and can embed ``<img src=...>`` against media endpoints.
-The defenses stack, because each one alone has a hole:
-
-* **Host** must be a loopback name — a DNS rebinding host that resolves to 127.0.0.1
-  passes the browser's origin rules but not this check.
-* **Origin**, when the browser sends one, must be this exact origin (scheme, host,
-  port). There is no CORS middleware at all, so a foreign page can never *read* a
-  response even if it manages to send a request.
-* **Sec-Fetch-Site** rejects cross-site loads that carry no Origin, which is what an
-  ``<img>`` or ``<script>`` against a media endpoint looks like.
-* A **session cookie** (``SameSite=Strict``, ``HttpOnly``) is required by every
-  ``/api/v1`` route except liveness/readiness and the bootstrap itself. Strict means
-  the browser never attaches it to a request another site initiated.
-* A **CSRF token** must be echoed in a header on every mutation. It is handed out
-  only in the bootstrap response body, which a foreign page cannot read (no CORS) —
-  so possessing it proves the caller is same-origin.
-
-``evaluate`` is a pure function over the request metadata: the whole policy is one
-table that a unit test can enumerate, and the middleware only applies its verdict.
-"""
-
-from __future__ import annotations
-
-import secrets
-from collections.abc import Mapping
-from dataclasses import dataclass
-from urllib.parse import urlsplit
-
-from starlette.datastructures import Headers, MutableHeaders
-from starlette.responses import JSONResponse
-
-SESSION_COOKIE = "pp_session"
-CSRF_HEADER = "x-csrf-token"
-API_PREFIX = "/api/v1"
-SAFE_METHODS = frozenset({"GET", "HEAD", "OPTIONS"})
-# Reachable without a session: liveness/readiness (an orchestrator has no cookie)
-# and the bootstrap that issues the session in the first place.
-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]"})
-
-# 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
-# embedding the app or its thumbnails.
-DEFAULT_HEADERS = {
-    "x-content-type-options": "nosniff",
-    "x-frame-options": "DENY",
-    "referrer-policy": "no-referrer",
-    "cross-origin-resource-policy": "same-origin",
-    "cross-origin-opener-policy": "same-origin",
-    "content-security-policy": (
-        "default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; "
-        "connect-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'none'"
-    ),
-}
-
-
-@dataclass(frozen=True)
-class Session:
-    """One process, one session. A local app has exactly one user; a session store
-    would be bookkeeping without a second subject to distinguish.
-
-    ponytail: per-session rows if the app ever serves more than one operator.
-    """
-
-    id: str
-    csrf_token: str
-
-    @classmethod
-    def create(cls) -> Session:
-        return cls(secrets.token_urlsafe(32), secrets.token_urlsafe(32))
-
-
-@dataclass(frozen=True)
-class Refusal:
-    status: int
-    code: str
-    message: str
-
-
-def split_host(value: str) -> tuple[str, str]:
-    """``"127.0.0.1:8000"`` -> ``("127.0.0.1", "8000")``; bracketed IPv6 aware."""
-    value = value.strip()
-    if value.startswith("["):
-        host, _, port = value.partition("]")
-        return host + "]", port.lstrip(":")
-    host, _, port = value.partition(":")
-    return host, port
-
-
-def evaluate(
-    *,
-    method: str,
-    path: str,
-    headers: Mapping[str, str],
-    session: Session,
-    allowed_hosts: frozenset[str] = LOOPBACK_HOSTS,
-    max_request_bytes: int,
-) -> Refusal | None:
-    """Why this request must be refused, or ``None`` when it may proceed."""
-    host_header = headers.get("host", "")
-    host, port = split_host(host_header)
-    if host.lower() not in allowed_hosts:
-        return Refusal(403, "host_not_allowed", "request host is not a local address")
-
-    origin = headers.get("origin")
-    if origin is not None and origin != "":
-        parts = urlsplit(origin)
-        origin_host, origin_port = split_host(parts.netloc)
-        if (
-            parts.scheme not in ("http", "https")
-            or origin_host.lower() not in allowed_hosts
-            or origin_port != port
-        ):
-            return Refusal(403, "origin_not_allowed", "request origin is not this application")
-
-    # Absent means a non-browser client; "none" is a user-initiated navigation.
-    fetch_site = headers.get("sec-fetch-site")
-    if fetch_site is not None and fetch_site not in ("same-origin", "none"):
-        return Refusal(403, "cross_site_blocked", "cross-site requests are not accepted")
-
-    length = headers.get("content-length")
-    if length and length.isdigit() and int(length) > max_request_bytes:
-        return Refusal(413, "payload_too_large", "request body exceeds the configured limit")
-
-    protected = path.startswith(API_PREFIX) and path not in PUBLIC_PATHS
-    if not protected:
-        return None
-
-    if headers.get("cookie-session") != session.id:
-        return Refusal(401, "unauthenticated", "a valid application session is required")
-    if method.upper() not in SAFE_METHODS and headers.get(CSRF_HEADER) != session.csrf_token:
-        return Refusal(403, "csrf_failed", "missing or invalid CSRF token")
-    return None
-
-
-class SecurityMiddleware:
-    """Pure-ASGI so the SSE stream keeps streaming (BaseHTTPMiddleware buffers)."""
-
-    def __init__(self, app, *, session: Session, config) -> None:
-        self.app = app
-        self.session = session
-        self.max_request_bytes = config.max_request_bytes
-        self.allowed_hosts = frozenset(LOOPBACK_HOSTS | {str(config.host).lower()})
-
-    async def __call__(self, scope, receive, send) -> None:
-        if scope["type"] != "http":
-            await self.app(scope, receive, send)
-            return
-        headers = Headers(scope=scope)
-        # The cookie is read here and handed to the pure policy as one value, so the
-        # policy never has to parse a Cookie header.
-        lookup = dict(headers)
-        lookup["cookie-session"] = _cookie(headers.get("cookie", ""), SESSION_COOKIE)
-        refusal = evaluate(
-            method=scope.get("method", "GET"),
-            path=scope.get("path", "/"),
-            headers=lookup,
-            session=self.session,
-            allowed_hosts=self.allowed_hosts,
-            max_request_bytes=self.max_request_bytes,
-        )
-        if refusal is not None:
-            response = JSONResponse(
-                status_code=refusal.status,
-                content={"error": {"code": refusal.code, "message": refusal.message}},
-                headers=DEFAULT_HEADERS,
-            )
-            await response(scope, receive, send)
-            return
-
-        async def send_with_headers(message):
-            if message["type"] == "http.response.start":
-                out = MutableHeaders(scope=message)
-                for name, value in DEFAULT_HEADERS.items():
-                    out.setdefault(name, value)
-            await send(message)
-
-        await self.app(scope, receive, send_with_headers)
-
-
-def _cookie(header: str, name: str) -> str:
-    for part in header.split(";"):
-        key, _, value = part.strip().partition("=")
-        if key == name:
-            return value
-    return ""

photo_pipeline/config.py

@@ -30,10 +30,6 @@ class Config(BaseModel):
     log_level: str = "INFO"
     log_format: str = "json"  # "json" or "text"
 
-    # Largest request body the API accepts. Every endpoint takes small JSON commands;
-    # anything larger is a mistake or an attempt to exhaust memory (US07-02).
-    max_request_bytes: int = 1_048_576
-
     # Library boundary for path validation (os.pathsep-separated in the env var).
     library_roots: tuple[Path, ...] = ()
     thumbnail_cache_quota_bytes: int = 500_000_000

photo_pipeline/jobs/domain_handlers.py

@@ -38,8 +38,7 @@ def _safety_score_item(asset_id: str, ctx: JobContext) -> None:
 def _analysis_item(asset_id: str, ctx: JobContext) -> None:
     from photo_pipeline.services.analysis import AnalysisService
 
-    roots = tuple(getattr(ctx.config, "library_roots", ()) or ())
-    AnalysisService(ctx.session_factory, library_roots=roots).run([asset_id])
+    AnalysisService(ctx.session_factory).run([asset_id])
 
 
 def _upload_batch_item(batch_id: str, ctx: JobContext) -> None:

photo_pipeline/path_policy.py

@@ -49,29 +49,6 @@ def resolve_within(root: Path, path: os.PathLike | str) -> Path:
     return resolved
 
 
-def resolve_in_roots(roots: Iterable[os.PathLike | str], path: os.PathLike | str) -> Path:
-    """The resolved path, proven to be inside one of ``roots`` and not excluded.
-
-    Callers must use the **returned** path for whatever they do next: validating one
-    name and then opening another is the symlink race this exists to close (US07-02).
-    The message names no path — it reaches API responses.
-
-    With no roots configured there is no boundary to check; that is a property of the
-    configuration, not permission granted to this call.
-    """
-    if is_excluded(path):
-        raise PathPolicyError("path is inside an excluded (_IGNORE/) tree")
-    roots = list(roots)
-    if not roots:
-        return Path(path)
-    for root in roots:
-        try:
-            return resolve_within(Path(root), path)
-        except PathPolicyError:
-            continue
-    raise PathPolicyError("path is outside the configured library roots")
-
-
 def iter_supported_files(root: os.PathLike | str) -> Iterator[Path]:
     """Yield supported, non-excluded files under ``root`` in deterministic order.
 

photo_pipeline/services/analysis.py

@@ -25,7 +25,6 @@ from typing import Protocol
 from sqlalchemy import func, select
 from sqlalchemy.orm import sessionmaker
 
-from photo_pipeline import path_policy
 from photo_pipeline.integrations import exiftool
 from photo_pipeline.models import AnalysisResult, Asset, SafetyReview
 from photo_pipeline.services import hashing
@@ -60,16 +59,9 @@ def _now() -> datetime:
 
 
 class AnalysisService:
-    def __init__(
-        self,
-        session_factory: sessionmaker,
-        *,
-        provider: VisionProvider | None = None,
-        library_roots: tuple = (),
-    ) -> None:
+    def __init__(self, session_factory: sessionmaker, *, provider: VisionProvider | None = None) -> None:
         self._session_factory = session_factory
         self._provider = provider
-        self._roots = tuple(library_roots)
 
     def _sfw_asset_ids(self, session) -> set[str]:
         """Asset ids whose latest safety decision is ``sfw`` — the ONLY assets that
@@ -144,18 +136,6 @@ class AnalysisService:
             if not path:
                 skipped += 1
                 continue
-            # Second gate, at the moment of use: the database says where the file
-            # was, the filesystem decides what that name means now. A link swapped
-            # under an asset after the scan would otherwise send bytes from outside
-            # the library — the one place that leaves this machine (US07-02).
-            try:
-                path = str(path_policy.resolve_in_roots(self._roots, path))
-            except path_policy.PathPolicyError as error:
-                self._store(
-                    asset_id, status="error", result=None, error=str(error), tokens=0, raw=""
-                )
-                errors += 1
-                continue
             try:
                 result = provider.analyze(path, album_hint=_album_hint(path))
             except Exception as error:  # provider/validation failure is per-asset

photo_pipeline/services/legacy_import.py

@@ -1,220 +0,0 @@
-"""Import the last path-keyed CSV state into the database (US07-01).
-
-``nsfwtag`` cached its safety scores in ``nsfw_scores.csv`` next to the library:
-one ``path,nsfw_score`` row per photo, four decimals, unreadable rows dropped. That
-file stops being a source of truth when the CLI is archived, so its scores are
-imported once — as scored-but-unreviewed ``safety_reviews`` rows on the stable
-``assets.id`` each path resolves to — and the CSV is left untouched on disk as its
-own backup.
-
-The import is deliberately conservative, because a score is evidence about a photo
-and a path is not an identity:
-
-- a row whose path matches no known asset is **unmatched**, never a new asset;
-- an asset a human already reviewed is **never** touched: a score is evidence, a
-  decision is a judgement, and an import may not overwrite the second with the
-  first — the difference is reported as a conflict instead;
-- a row for an asset that already carries a score is **skipped** unless
-  ``overwrite`` is asked for, and a differing score is reported either way;
-- an unparsable score or a duplicate path is **rejected** with its reason;
-- rerunning changes nothing that is already imported.
-
-Everything it did — and everything it refused to do — comes back as a
-reconciliation report, which the caller can persist next to the database. The
-donor's own reader is the specification for the format (donor ledger:
-``nt-score-cache``).
-"""
-
-from __future__ import annotations
-
-import csv
-import json
-import uuid
-from collections import Counter
-from dataclasses import asdict, dataclass, field
-from datetime import datetime, timezone
-from pathlib import Path
-
-from sqlalchemy import select
-from sqlalchemy.orm import sessionmaker
-
-from photo_pipeline.models import Asset, AssetPath, SafetyReview
-
-REPORT_VERSION = 1
-# The donor wrote scores as four-decimal strings; anything outside 0..1 was never
-# something it produced, so it is data corruption rather than a score.
-SCORE_RANGE = (0.0, 1.0)
-# Recorded as the reviewer so an imported score is never mistaken for a judgement.
-REVIEWER = "legacy-csv-import"
-
-
-@dataclass
-class ImportReport:
-    source: str
-    rows: int = 0
-    imported: int = 0
-    skipped_existing: int = 0
-    unchanged: int = 0
-    reviewed: int = 0
-    unmatched: list[str] = field(default_factory=list)
-    rejected: list[dict] = field(default_factory=list)
-    conflicts: list[dict] = field(default_factory=list)
-    generated_at: str = ""
-    schema_version: int = REPORT_VERSION
-
-    def to_dict(self) -> dict:
-        return asdict(self)
-
-    @property
-    def counts(self) -> dict[str, int]:
-        return {
-            "rows": self.rows,
-            "imported": self.imported,
-            "skipped_existing": self.skipped_existing,
-            "unchanged": self.unchanged,
-            "reviewed": self.reviewed,
-            "unmatched": len(self.unmatched),
-            "rejected": len(self.rejected),
-            "conflicts": len(self.conflicts),
-        }
-
-
-class LegacyImportService:
-    def __init__(self, session_factory: sessionmaker) -> None:
-        self._session_factory = session_factory
-
-    def import_nsfw_scores(
-        self, csv_path: Path | str, *, overwrite: bool = False, dry_run: bool = False
-    ) -> ImportReport:
-        """Import ``nsfw_scores.csv`` onto asset identity and report what happened."""
-        path = Path(csv_path)
-        report = ImportReport(source=str(path))
-        if not path.is_file():
-            report.rejected.append({"path": str(path), "reason": "csv_missing"})
-            return self._stamp(report)
-
-        with self._session_factory() as session:
-            # Paths are matched against every occurrence an asset ever had, so a
-            # photo scored before a rename is still recognised.
-            by_path = self._path_index(session)
-            latest = self._latest_reviews(session)
-            seen: Counter[str] = Counter()
-
-            for row in self._rows(path, report):
-                report.rows += 1
-                raw_path, raw_score = row
-                seen[raw_path] += 1
-                if seen[raw_path] > 1:
-                    report.rejected.append({"path": raw_path, "reason": "duplicate_path"})
-                    continue
-                score = _parse_score(raw_score)
-                if score is None:
-                    report.rejected.append(
-                        {"path": raw_path, "reason": "unparsable_score", "value": raw_score}
-                    )
-                    continue
-                asset_id = by_path.get(raw_path)
-                if asset_id is None:
-                    report.unmatched.append(raw_path)
-                    continue
-
-                current = latest.get(asset_id)
-                if current is not None and current.decision:
-                    # A human decided this one; the CSV is older evidence.
-                    report.reviewed += 1
-                    if current.score is None or abs(current.score - score) >= 1e-9:
-                        report.conflicts.append(
-                            {
-                                "path": raw_path,
-                                "asset_id": asset_id,
-                                "current": current.score,
-                                "decision": current.decision,
-                                "csv": score,
-                            }
-                        )
-                    continue
-                if current is not None and current.score is not None:
-                    if abs(current.score - score) < 1e-9:
-                        report.unchanged += 1
-                        continue
-                    report.conflicts.append(
-                        {
-                            "path": raw_path,
-                            "asset_id": asset_id,
-                            "current": current.score,
-                            "decision": None,
-                            "csv": score,
-                        }
-                    )
-                    if not overwrite:
-                        report.skipped_existing += 1
-                        continue
-                session.add(
-                    SafetyReview(
-                        id=str(uuid.uuid4()),
-                        asset_id=asset_id,
-                        score=score,
-                        reviewer=REVIEWER,
-                    )
-                )
-                latest[asset_id] = SafetyReview(asset_id=asset_id, score=score)
-                report.imported += 1
-
-            if dry_run:
-                session.rollback()
-            else:
-                session.commit()
-        return self._stamp(report)
-
-    @staticmethod
-    def _rows(path: Path, report: ImportReport):
-        """Yield ``(path, score)`` pairs, tolerating the donor's own sloppiness."""
-        with open(path, newline="", encoding="utf-8", errors="replace") as handle:
-            for row in csv.DictReader(handle):
-                raw_path = (row.get("path") or "").strip()
-                if not raw_path:
-                    report.rejected.append({"path": "", "reason": "missing_path"})
-                    continue
-                yield raw_path, row.get("nsfw_score")
-
-    @staticmethod
-    def _latest_reviews(session) -> dict[str, SafetyReview]:
-        """The current safety row per asset — latest wins, as everywhere else."""
-        latest: dict[str, SafetyReview] = {}
-        for review in session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)):
-            latest[review.asset_id] = review
-        return latest
-
-    @staticmethod
-    def _path_index(session) -> dict[str, str]:
-        """Every path an asset is or was known by → its stable id."""
-        index: dict[str, str] = {}
-        for asset_id, path in session.execute(select(AssetPath.asset_id, AssetPath.path)):
-            index.setdefault(path, asset_id)
-        for asset_id, path in session.execute(select(Asset.id, Asset.current_path)):
-            if path:
-                index[path] = asset_id  # the current path wins over a closed one
-        return index
-
-    @staticmethod
-    def _stamp(report: ImportReport) -> ImportReport:
-        report.generated_at = datetime.now(timezone.utc).isoformat()
-        return report
-
-
-def write_report(report: ImportReport, directory: Path) -> Path:
-    """Persist the reconciliation report; the import is not evidence until it is."""
-    directory.mkdir(parents=True, exist_ok=True)
-    path = directory / "legacy-nsfw-import.json"
-    path.write_text(json.dumps(report.to_dict(), indent=2, sort_keys=True), encoding="utf-8")
-    return path
-
-
-def _parse_score(value) -> float | None:
-    try:
-        score = float(value)
-    except (TypeError, ValueError):
-        return None
-    if not SCORE_RANGE[0] <= score <= SCORE_RANGE[1]:
-        return None
-    return score

photo_pipeline/services/thumbnails.py

@@ -129,10 +129,7 @@ class ThumbnailService:
                 raise ThumbnailUnavailable(f"asset {asset_id} has no readable file")
             source = str(source)
             if source == asset.current_path:
-                # Render the *resolved* path the check approved: revalidating and then
-                # reopening the original name would let a symlink swapped in between
-                # the two steps decide which bytes are served (US07-02).
-                source = str(self._validate_path(source))  # archive roots lie outside
+                self._validate_path(source)  # archive roots lie outside the library
 
         # Rendering happens outside the DB session (no transaction held during I/O).
         try:
@@ -198,16 +195,20 @@ class ThumbnailService:
                 session.commit()
 
     # ── path safety ──────────────────────────────────────────────────────────
-    def _validate_path(self, current_path: str) -> Path:
-        """The resolved path to read, or ``PathNotAllowed``.
-
-        The message names no path: a refusal is returned to the browser, and where
-        the library lives is not the caller's business (US07-02).
-        """
+    def _validate_path(self, current_path: str) -> None:
+        path = Path(current_path)
+        if path_policy.is_excluded(path):
+            raise PathNotAllowed(f"excluded path: {current_path}")
+        roots = self._config.library_roots
+        if not roots:
+            return
+        for root in roots:
             try:
-            return path_policy.resolve_in_roots(self._config.library_roots, current_path)
-        except path_policy.PathPolicyError as error:
-            raise PathNotAllowed(str(error)) from None
+                path_policy.resolve_within(Path(root), path)
+                return
+            except path_policy.PathPolicyError:
+                continue
+        raise PathNotAllowed(f"path outside configured roots: {current_path}")
 
     # ── cache key + rendering ──────────────────────────────────────────────────
     @staticmethod

pyproject.toml

@@ -24,8 +24,6 @@ test = [
 
 [tool.ruff]
 line-length = 100
-# The frozen CLI archive is evidence, not code under maintenance (US07-01).
-extend-exclude = ["legacy_cli_archive"]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]

tests/characterization/conftest.py

@@ -14,11 +14,6 @@ import pytest
 from PIL import Image
 
 REPO = Path(__file__).resolve().parents[2]
-# The donors are frozen in the read-only archive (US07-01). Only this suite — and
-# the parity test that compares against them — puts that directory on sys.path;
-# production never does, which tests/unit/test_legacy_archive.py enforces.
-ARCHIVED_SOURCES = REPO / "legacy_cli_archive" / "src"
-sys.path.insert(0, str(ARCHIVED_SOURCES))
 sys.path.insert(0, str(REPO))
 
 EXIFTOOL = shutil.which("exiftool")

tests/characterization/test_donor_ledger.py

@@ -1,20 +1,13 @@
-"""Ledger lint (US01-01, extended by US07-01).
-
-Every donor-ledger row must carry a real source reference — now inside the frozen
-archive — a target location, and either existing characterization test IDs or a
-real pending backlog story. Since archival (US07-01) a row may also be ``resolved``:
-its replacement has shipped, ``parity`` names tests that exist and prove it, and
-``delta`` states every intentional difference. Nothing may quietly become
-"finished" without one of those two."""
+"""Ledger lint (US01-01): every donor-ledger row must carry a real source
+reference, a target location, and either existing characterization test IDs or
+a real pending backlog story."""
 import re
 from pathlib import Path
 
 import yaml
 
 REPO = Path(__file__).resolve().parents[2]
-ARCHIVE = REPO / "legacy_cli_archive"
-LEDGER = ARCHIVE / "donor_ledger.yaml"
-ARCHIVED_SOURCES = ARCHIVE / "src"
+LEDGER = REPO / "donor_ledger.yaml"
 STORIES = REPO / "delivery_backlog" / "stories"
 TESTS_DIR = Path(__file__).resolve().parent
 
@@ -22,7 +15,7 @@ CLASSIFICATIONS = {"reuse", "extract", "refactor", "replace"}
 REQUIRED_AREAS = {"discovery", "hashing", "imaging", "nsfw", "vision", "exif",
                   "database", "ui", "configuration", "logging", "cancellation",
                   "error"}
-STATUSES = {"characterized", "resolved", "pending"}
+STATUSES = {"characterized", "pending"}
 
 
 def load_rows():
@@ -56,11 +49,10 @@ def test_rows_have_required_fields_and_unique_ids():
 
 
 def test_source_references_resolve():
-    """Source paths are relative to the archive: the donors moved there, whole."""
     for r in load_rows():
         src = r["source"]
-        f = ARCHIVED_SOURCES / src["file"]
-        assert f.is_file(), f"{r['id']}: source file {src['file']} missing from the archive"
+        f = REPO / src["file"]
+        assert f.is_file(), f"{r['id']}: source file {src['file']} missing"
         text = f.read_text(encoding="utf-8")
         for sym in src["symbols"]:
             assert sym in text, f"{r['id']}: symbol {sym!r} not found in {src['file']}"
@@ -71,9 +63,7 @@ def test_rows_have_tests_or_pending_story():
     for r in load_rows():
         tests = r.get("tests", [])
         pending = r.get("pending_story")
-        parity = r.get("parity", [])
-        assert tests or pending or parity, \
-            f"{r['id']}: neither characterization tests, parity tests, nor a pending story"
+        assert tests or pending, f"{r['id']}: neither tests nor pending_story"
         for t in tests:
             assert t in known_tests, f"{r['id']}: unknown test id {t}"
         if pending:
@@ -83,58 +73,16 @@ def test_rows_have_tests_or_pending_story():
             assert tests, f"{r['id']}: characterized rows need test ids"
 
 
-def test_resolved_rows_name_their_parity_or_their_delta():
-    """A resolved row is a claim that the behavior is handled. It has to say how:
-    tests that prove the replacement, or a stated difference — usually both."""
-    for r in load_rows():
-        if r["status"] != "resolved":
-            assert "parity" not in r, f"{r['id']}: parity on a non-resolved row"
-            continue
-        parity = r.get("parity", [])
-        delta = r.get("delta")
-        assert parity or delta, f"{r['id']}: resolved without parity tests or a delta"
-        if delta:
-            assert len(str(delta).strip()) >= 20, f"{r['id']}: delta too thin to be a reason"
-        for ref in parity:
-            rel, _, func = ref.partition("::")
-            path = REPO / rel
-            assert path.is_file(), f"{r['id']}: parity test file {rel} missing"
-            assert f"def {func}" in path.read_text(encoding="utf-8"), \
-                f"{r['id']}: parity test {ref} not found"
-
-
-def test_pending_rows_are_the_only_unfinished_work():
-    """The ledger is the honest list of what has not been carried over: a pending
-    row names the story that will, and that story must still be open work."""
-    for r in load_rows():
-        if r["status"] != "pending":
-            continue
-        story = r.get("pending_story")
-        assert story, f"{r['id']}: pending without a story"
-        assert list(STORIES.glob(f"{story}-*.md")), f"{r['id']}: unknown story {story}"
-
-
-def test_every_target_module_exists():
-    """A row is only finished if the thing it points at is really there."""
-    for r in load_rows():
-        if r["status"] == "pending":
-            continue
-        modules = re.findall(r"photo_pipeline/[\w/]+\.py", str(r["target"]))
-        for module in modules:
-            assert (REPO / module).is_file(), f"{r['id']}: target {module} does not exist"
-
-
 def test_all_required_areas_covered():
     covered = {r["area"] for r in load_rows()}
     assert REQUIRED_AREAS <= covered, f"uncovered areas: {REQUIRED_AREAS - covered}"
     assert covered <= REQUIRED_AREAS, f"unknown areas: {covered - REQUIRED_AREAS}"
 
 
-def test_every_donor_is_in_the_archive_and_nowhere_else():
-    """US01-01 forbade archiving before characterization; US07-01 requires it after.
-    Each donor exists exactly once — frozen, in the archive."""
+def test_no_legacy_file_moved():
+    # US01-01 explicitly forbids moving/archiving donors; the ledger's source
+    # files must all still exist at their original locations.
     for donor in ("photo_analyzer.py", "nsfwtag/scoring.py", "nsfwtag/exif.py",
                   "nsfwtag/server.py", "webapp/query.py", "webapp/runner.py",
                   "webapp/server.py"):
-        assert (ARCHIVED_SOURCES / donor).is_file(), f"donor missing from archive: {donor}"
-        assert not (REPO / donor).exists(), f"donor still live at the repo root: {donor}"
+        assert (REPO / donor).is_file(), f"donor moved: {donor}"

tests/conftest.py

@@ -1,15 +1,4 @@
-"""Make the repository root importable for the pipeline test suites, and give every
-suite the application session the API requires since US07-02.
-
-The suites drive the API the way the browser does — module-level ``httpx`` calls and
-``TestClient`` — so instead of threading a cookie through several hundred call sites,
-both clients bootstrap the session themselves exactly like ``frontend/js/api.js``:
-fetch ``/api/v1/session`` once, then send the cookie plus the CSRF header, and
-re-bootstrap once on 401 (a restarted server issues a new session).
-
-Security tests deliberately bypass this by constructing their own ``httpx.Client``;
-only the module-level helpers are wrapped.
-"""
+"""Make the repository root importable for the pipeline test suites."""
 
 import sys
 from pathlib import Path
@@ -17,85 +6,3 @@ from pathlib import Path
 REPO = Path(__file__).resolve().parents[1]
 if str(REPO) not in sys.path:
     sys.path.insert(0, str(REPO))
-
-import httpx
-import httpx._api  # httpx.get/post resolve `request` in this module
-import pytest
-from starlette.testclient import TestClient
-
-from photo_pipeline.api.security import CSRF_HEADER, SESSION_COOKIE
-
-_SESSIONS: dict[str, tuple[str, str]] = {}  # base url -> (session id, csrf token)
-
-
-def _base(url) -> str:
-    parts = httpx.URL(str(url))
-    return f"{parts.scheme}://{parts.netloc.decode()}"
-
-
-def _bootstrap(base: str) -> tuple[str, str]:
-    if base not in _SESSIONS:
-        response = httpx.Client(timeout=10).get(f"{base}/api/v1/session")
-        _SESSIONS[base] = (
-            response.cookies.get(SESSION_COOKIE, ""),
-            response.json().get("csrf_token", ""),
-        )
-    return _SESSIONS[base]
-
-
-def session_client(base_url: str, **kwargs) -> httpx.Client:
-    """An ``httpx.Client`` that has bootstrapped its own application session.
-
-    For suites that keep one client for a whole journey; it holds its own cookie, so
-    a client made after a server restart picks up the new session by construction.
-    """
-    client = httpx.Client(base_url=base_url, **kwargs)
-    client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
-    return client
-
-
-def _authorized(url, headers, cookies):
-    session_id, token = _bootstrap(_base(url))
-    headers = dict(headers or {})
-    headers.setdefault(CSRF_HEADER, token)
-    cookies = dict(cookies or {})
-    cookies.setdefault(SESSION_COOKIE, session_id)
-    return headers, cookies
-
-
-@pytest.fixture(autouse=True, scope="session")
-def _api_session():
-    real_request, real_stream = httpx._api.request, httpx._api.stream
-    real_client_request, real_client_init = TestClient.request, TestClient.__init__
-
-    def request(method, url, *, headers=None, cookies=None, **kwargs):
-        sent, jar = _authorized(url, headers, cookies)
-        response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
-        if response.status_code == 401:
-            _SESSIONS.pop(_base(url), None)
-            sent, jar = _authorized(url, headers, cookies)
-            response = real_request(method, url, headers=sent, cookies=jar, **kwargs)
-        return response
-
-    def stream(method, url, *, headers=None, cookies=None, **kwargs):
-        sent, jar = _authorized(url, headers, cookies)
-        return real_stream(method, url, headers=sent, cookies=jar, **kwargs)
-
-    def client_init(self, app, *args, base_url="http://127.0.0.1", **kwargs):
-        # The default "http://testserver" is not a local host, which is exactly what
-        # the Host check refuses; in-process tests are still same-origin callers.
-        real_client_init(self, app, *args, base_url=base_url, **kwargs)
-
-    def client_request(self, method, url, *, headers=None, **kwargs):
-        if CSRF_HEADER not in self.headers:
-            response = real_client_request(self, "GET", "/api/v1/session")
-            self.headers[CSRF_HEADER] = response.json()["csrf_token"]
-        return real_client_request(self, method, url, headers=headers, **kwargs)
-
-    httpx._api.request, httpx.request = request, request
-    httpx._api.stream, httpx.stream = stream, stream
-    TestClient.request, TestClient.__init__ = client_request, client_init
-    yield
-    httpx._api.request, httpx.request = real_request, real_request
-    httpx._api.stream, httpx.stream = real_stream, real_stream
-    TestClient.request, TestClient.__init__ = real_client_request, real_client_init

tests/e2e/test_phase_a_pipeline.py

@@ -22,8 +22,6 @@ import numpy as np
 import pytest
 from PIL import Image
 
-from tests.conftest import session_client
-
 REPO = Path(__file__).resolve().parents[2]
 
 
@@ -94,7 +92,7 @@ class ServerController:
                 pytest.fail(f"server exited: {err.decode(errors='replace')}")
             try:
                 if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
-                    self.client = session_client(self.base, timeout=10)
+                    self.client = httpx.Client(base_url=self.base, timeout=10)
                     return
             except httpx.HTTPError:
                 time.sleep(0.2)

tests/e2e/test_review_ui.py

@@ -17,9 +17,6 @@ import httpx
 import numpy as np
 import pytest
 from PIL import Image
-from playwright.sync_api import expect
-
-from tests.conftest import session_client
 
 REPO = Path(__file__).resolve().parents[2]
 
@@ -114,7 +111,7 @@ def server(tmp_path):
         proc.terminate()
         pytest.fail("server never became ready")
 
-    client = session_client(base, timeout=5)
+    client = httpx.Client(base_url=base, timeout=5)
 
     def cluster_by_method(method):
         clusters = client.get("/api/v1/duplicates/clusters").json()["items"]
@@ -151,9 +148,9 @@ def test_fuzzy_decision_requires_confirmation(page, server):
     page.get_by_test_id("confirm").wait_for()
     assert "open" in page.get_by_test_id("cluster-state").inner_text()
     page.get_by_test_id("confirm-yes").click()
-    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
-    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
-    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
+    page.wait_for_function(
+        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
+    )
 
 
 def test_stale_version_shows_conflict(page, server):
@@ -179,9 +176,9 @@ def test_decision_persists_after_reload(page, server):
     page.goto(f"{server.base}/app/#/duplicates/{cluster['id']}")
     page.get_by_test_id("not-duplicate").click()
     page.get_by_test_id("confirm-yes").click()
-    # A locator assertion, not wait_for_function: the app's CSP forbids eval, and
-    # Playwright's polling predicate is evaluated as a string in the page (US07-02).
-    expect(page.get_by_test_id("cluster-state")).to_contain_text("dismissed")
+    page.wait_for_function(
+        "document.querySelector('[data-testid=cluster-state]').innerText.includes('dismissed')"
+    )
     page.reload()
     page.get_by_test_id("cluster-state").wait_for()
     assert "dismissed" in page.get_by_test_id("cluster-state").inner_text()

tests/e2e/test_security.py

@@ -1,280 +0,0 @@
-"""US07-02 black-box security tests: authorization and path boundaries.
-
-Everything here talks to a real ``photo_pipeline serve`` child process over HTTP with
-its own ``httpx.Client``, deliberately outside the session helper the rest of the
-suite uses — an attacker does not get a bootstrapped client.
-
-The threat is a page in the user's browser, not a remote attacker: the app listens on
-127.0.0.1, so any site the user visits can send requests to it and can point an
-``<img>`` at its media endpoints. Each journey below is one of those attempts.
-"""
-
-from __future__ import annotations
-
-import json
-from contextlib import contextmanager
-from pathlib import Path
-
-import httpx
-import pytest
-
-from tests.e2e._pipeline_harness import Server, image, seed_library
-
-TIMEOUT = 20
-SENTINEL_KEY = "immich-sentinel-9f3a2b"
-SESSION_COOKIE = "pp_session"
-CSRF_HEADER = "X-CSRF-Token"
-
-# A GET, a mutation, and a media endpoint: the three shapes the policy must cover.
-PROTECTED = [
-    ("GET", "/api/v1/workflow", None),
-    ("POST", "/api/v1/albums/proposals", {}),
-    ("GET", "/api/v1/assets/{asset}/thumbnail?size=256", None),
-]
-
-
-@pytest.fixture(scope="module")
-def stack(tmp_path_factory):
-    """One server, one album of two photos, and a credential sentinel in its config."""
-    tmp_path = tmp_path_factory.mktemp("security")
-    seeded = seed_library(tmp_path, {"a": 1, "b": 2}, {"a": "sfw", "b": "sfw"})
-    outside = tmp_path / "outside"
-    outside.mkdir()
-    image(outside / "secret.jpg", 99)
-    server = Server(
-        seeded,
-        extra_env={
-            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
-            "PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
-        },
-    ).start()
-    yield server, seeded, outside
-    server.stop()
-
-
-@contextmanager
-def anonymous(server):
-    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
-        yield client
-
-
-@contextmanager
-def authenticated(server):
-    """A browser that has loaded the app: session cookie in the jar, token in a header."""
-    with httpx.Client(base_url=server.base, timeout=TIMEOUT) as client:
-        client.headers[CSRF_HEADER] = client.get("/api/v1/session").json()["csrf_token"]
-        yield client
-
-
-def call(client, method, path, body, asset):
-    return client.request(method, path.format(asset=asset), json=body)
-
-
-def test_the_api_refuses_every_caller_without_a_session(stack):
-    server, seeded, _ = stack
-    with anonymous(server) as client:
-        for method, path, body in PROTECTED:
-            response = call(client, method, path, body, seeded.asset_ids["a"])
-            assert response.status_code == 401, path
-            assert response.json()["error"]["code"] == "unauthenticated"
-        # Liveness and readiness stay open: an orchestrator holds no session.
-        assert client.get("/api/v1/health/ready").status_code == 200
-        assert client.get("/api/v1/health/live").status_code == 200
-
-
-def test_a_guessed_session_cookie_is_refused(stack):
-    server, _, _ = stack
-    with anonymous(server) as client:
-        client.cookies.set(SESSION_COOKIE, "guessed", domain="127.0.0.1")
-        response = client.get("/api/v1/workflow")
-        assert response.status_code == 401
-        assert response.json()["error"]["code"] == "unauthenticated"
-
-
-def test_the_bootstrap_issues_a_strict_httponly_cookie(stack):
-    server, _, _ = stack
-    with anonymous(server) as client:
-        response = client.get("/api/v1/session")
-        cookie = response.headers["set-cookie"].lower()
-        assert "httponly" in cookie and "samesite=strict" in cookie and "path=/" in cookie
-        assert response.json()["csrf_token"]
-        # The token is in the body, which no other origin may read: no CORS header
-        # grants access to it.
-        assert "access-control-allow-origin" not in response.headers
-
-
-def test_a_session_without_the_csrf_token_may_read_but_not_mutate(stack):
-    server, _, _ = stack
-    with authenticated(server) as client:
-        del client.headers[CSRF_HEADER]
-        assert client.get("/api/v1/albums/proposals").status_code == 200
-
-        response = client.post("/api/v1/albums/proposals", json={})
-        assert response.status_code == 403
-        assert response.json()["error"]["code"] == "csrf_failed"
-
-        response = client.post(
-            "/api/v1/albums/proposals", json={}, headers={CSRF_HEADER: "guessed"}
-        )
-        assert response.status_code == 403
-    # And nothing was created behind the refusal.
-    with authenticated(server) as client:
-        assert client.get("/api/v1/albums/proposals").json()["items"] == []
-
-
-def test_a_foreign_origin_cannot_mutate_even_with_a_session(stack):
-    server, _, _ = stack
-    with authenticated(server) as client:
-        for origin in ("http://evil.example", "http://127.0.0.1:1", "null"):
-            response = client.post(
-                "/api/v1/albums/proposals", json={}, headers={"Origin": origin}
-            )
-            assert response.status_code == 403, origin
-            assert response.json()["error"]["code"] == "origin_not_allowed"
-        # This app's own origin is accepted, so the check is not simply refusing all.
-        allowed = client.post(
-            "/api/v1/albums/proposals", json={}, headers={"Origin": server.base}
-        )
-        assert allowed.status_code == 200
-
-
-def test_a_rebinding_host_is_refused(stack):
-    """A name that resolves to 127.0.0.1 makes the browser treat the attacker's page
-    as same-origin. The Host header still carries that name, so it is checked."""
-    server, _, _ = stack
-    with authenticated(server) as client:
-        response = client.get("/api/v1/workflow", headers={"Host": "photos.evil.example"})
-        assert response.status_code == 403
-        assert response.json()["error"]["code"] == "host_not_allowed"
-
-
-def test_media_cannot_be_embedded_by_another_page(stack):
-    server, seeded, _ = stack
-    url = f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"
-    with authenticated(server) as client:
-        # What an <img> on another site produces: no Origin, but a cross-site marker.
-        blocked = client.get(url, headers={"Sec-Fetch-Site": "cross-site"})
-        assert blocked.status_code == 403
-        assert blocked.json()["error"]["code"] == "cross_site_blocked"
-
-        served = client.get(url, headers={"Sec-Fetch-Site": "same-origin"})
-        assert served.status_code == 200
-        assert served.headers["content-type"] == "image/webp"
-        assert served.headers["cross-origin-resource-policy"] == "same-origin"
-        # Photos must never land in a shared cache.
-        assert served.headers["cache-control"].startswith("private")
-
-
-def test_every_response_carries_the_default_headers_and_no_cors(stack):
-    server, seeded, _ = stack
-    with authenticated(server) as client:
-        responses = [
-            client.get("/api/v1/workflow"),
-            client.get(f"/api/v1/assets/{seeded.asset_ids['a']}/thumbnail?size=256"),
-            client.get("/app/"),
-            client.get("/api/v1/does-not-exist"),
-        ]
-    for response in responses:
-        assert response.headers["x-content-type-options"] == "nosniff"
-        assert response.headers["x-frame-options"] == "DENY"
-        assert response.headers["referrer-policy"] == "no-referrer"
-        assert "frame-ancestors 'none'" in response.headers["content-security-policy"]
-        assert "access-control-allow-origin" not in response.headers
-        assert "access-control-allow-credentials" not in response.headers
-
-
-def test_a_traversal_attempt_addresses_nothing(stack):
-    """Identifiers are database keys, not paths: traversal has nowhere to land."""
-    server, _, _ = stack
-    with authenticated(server) as client:
-        for path in (
-            "/api/v1/albums/proposals/..%2F..%2F..%2Fetc%2Fpasswd",
-            "/api/v1/albums/proposals/../../../etc/passwd",
-            "/api/v1/assets/..%2F..%2Fetc%2Fpasswd/thumbnail?size=256",
-            "/api/v1/assets/%2Fetc%2Fpasswd/thumbnail?size=256",
-        ):
-            response = client.get(path)
-            assert response.status_code in (404, 422), path
-            assert "root:" not in response.text, path
-
-
-def test_an_oversized_request_is_refused_before_it_is_parsed(stack):
-    server, _, _ = stack
-    with authenticated(server) as client:
-        response = client.post(
-            "/api/v1/albums/proposals",
-            content=json.dumps({"albums": ["x" * 2_000_000]}),
-            headers={"Content-Type": "application/json"},
-        )
-        assert response.status_code == 413
-        assert response.json()["error"]["code"] == "payload_too_large"
-
-
-def test_a_malformed_request_reports_the_field_and_nothing_else(stack):
-    server, _, _ = stack
-    with authenticated(server) as client:
-        for content in ("{", '{"albums": 5}', ""):
-            response = client.post(
-                "/api/v1/albums/proposals",
-                content=content,
-                headers={"Content-Type": "application/json"},
-            )
-            assert response.status_code == 422, content
-            error = response.json()["error"]
-            assert error["code"] == "invalid_request"
-            assert "Traceback" not in response.text and "photo_pipeline/" not in response.text
-
-
-def test_errors_reveal_neither_the_credential_nor_an_internal(stack):
-    """The API key is configured but must appear nowhere; failures additionally say
-    nothing about where the library lives or how the server is built.
-
-    Successful responses are a different matter: path previews are the point of the
-    rename and archive views, and the operator is the one who owns those paths.
-    """
-    server, seeded, _ = stack
-    with authenticated(server) as client:
-        responses = [
-            client.post("/api/v1/upload-preflight", json={}),
-            client.get("/api/v1/workflow"),
-            client.get("/api/v1/upload-batches"),
-            client.get("/api/v1/assets/unknown-asset/thumbnail?size=256"),
-            client.get("/api/v1/archive-plans/unknown-plan"),
-            client.post("/api/v1/archive-locations", json={"name": "x", "root": "/nope"}),
-            client.get("/api/v1/does-not-exist"),
-        ]
-    for response in responses:
-        assert SENTINEL_KEY not in response.text
-        assert "Traceback" not in response.text
-        if response.status_code >= 400:
-            assert str(seeded.lib) not in response.text
-            assert "photo_pipeline/" not in response.text
-            assert "sqlite" not in response.text.lower()
-
-
-def test_a_symlink_swapped_under_an_asset_cannot_be_served(stack):
-    """TOCTOU on the media path: the file the database points at is replaced by a
-    link to something outside the library between the scan and the request."""
-    server, seeded, outside = stack
-    secret = outside / "secret.jpg"
-    original = seeded.lib / "b.jpg"
-    original.unlink()
-    original.symlink_to(secret)
-
-    with authenticated(server) as client:
-        response = client.get(f"/api/v1/assets/{seeded.asset_ids['b']}/thumbnail?size=256")
-    assert response.status_code == 403
-    assert response.json()["error"]["code"] == "path_not_allowed"
-    # The refusal names no filesystem location, and no bytes escaped with it.
-    assert str(outside) not in response.text and str(seeded.lib) not in response.text
-    assert secret.read_bytes()[:16] not in response.content
-
-
-def test_the_frontend_shell_stays_reachable_without_a_session(stack):
-    """It must load before any JavaScript can ask for a session."""
-    server, _, _ = stack
-    with anonymous(server) as client:
-        response = client.get("/app/", headers={"Sec-Fetch-Site": "none"})
-        assert response.status_code == 200
-        assert response.headers["content-type"].startswith("text/html")
-        assert Path("frontend/index.html").exists()

tests/integration/test_legacy_import.py

@@ -1,238 +0,0 @@
-"""Importing the archived CLI's CSV state into the database (US07-01).
-
-The donor's ``nsfw_scores.csv`` was keyed by path; the database is keyed by a
-stable asset id. Every case here is about that gap: a path that moved, a path that
-matches nothing, a score a human has already overruled, and a file that is simply
-malformed. The import may add evidence and must never invent an asset, overwrite a
-judgement, or fail silently — whatever it does ends up in the report.
-"""
-
-from __future__ import annotations
-
-import json
-import uuid
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
-
-import pytest
-from sqlalchemy import select
-
-from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
-from photo_pipeline.models import Asset, AssetPath, SafetyReview
-from photo_pipeline.services.legacy_import import (
-    REVIEWER,
-    LegacyImportService,
-    write_report,
-)
-
-NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
-
-
-@pytest.fixture
-def factory(tmp_path):
-    url = f"sqlite:///{tmp_path / 'legacy.db'}"
-    run_migrations(url)
-    engine = create_db_engine(url)
-    yield create_session_factory(engine)
-    engine.dispose()
-
-
-def _asset(sf, path: Path, *, previous: Path | None = None) -> str:
-    asset_id = str(uuid.uuid4())
-    with sf() as session:
-        session.add(
-            Asset(
-                id=asset_id,
-                original_path=str(previous or path),
-                current_path=str(path),
-                discovered_at=NOW,
-                hash_version=1,
-            )
-        )
-        session.add(AssetPath(asset_id=asset_id, path=str(path), valid_from=NOW))
-        if previous is not None:
-            session.add(
-                AssetPath(
-                    asset_id=asset_id, path=str(previous), valid_from=NOW, valid_until=NOW
-                )
-            )
-        session.commit()
-    return asset_id
-
-
-def _review(sf, asset_id: str, *, score=None, decision=None, at=NOW) -> None:
-    with sf() as session:
-        session.add(
-            SafetyReview(
-                id=str(uuid.uuid4()),
-                asset_id=asset_id,
-                score=score,
-                decision=decision,
-                reviewer="dom",
-                created_at=at,
-            )
-        )
-        session.commit()
-
-
-def _csv(tmp_path, rows: str) -> Path:
-    path = tmp_path / "nsfw_scores.csv"
-    path.write_text("path,nsfw_score\n" + rows, encoding="utf-8")
-    return path
-
-
-def _scores(sf) -> dict[str, float]:
-    with sf() as session:
-        return {
-            review.asset_id: review.score
-            for review in session.scalars(
-                select(SafetyReview).order_by(SafetyReview.created_at)
-            )
-        }
-
-
-def test_scores_are_imported_onto_asset_identity(tmp_path, factory):
-    a = _asset(factory, tmp_path / "a.jpg")
-    # This one was scored under its old path and has since moved.
-    b = _asset(factory, tmp_path / "moved" / "b.jpg", previous=tmp_path / "b.jpg")
-    csv_path = _csv(
-        tmp_path, f"{tmp_path / 'a.jpg'},0.9123\n{tmp_path / 'b.jpg'},0.0100\n"
-    )
-
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
-
-    assert report.counts == {
-        "rows": 2,
-        "imported": 2,
-        "skipped_existing": 0,
-        "unchanged": 0,
-        "reviewed": 0,
-        "unmatched": 0,
-        "rejected": 0,
-        "conflicts": 0,
-    }
-    assert _scores(factory) == {a: 0.9123, b: 0.0100}
-    # The imported rows are evidence, not judgements: no decision is invented.
-    with factory() as session:
-        rows = list(session.scalars(select(SafetyReview)))
-        assert {r.decision for r in rows} == {None}
-        assert {r.reviewer for r in rows} == {REVIEWER}
-    assert csv_path.exists(), "the CSV is left on disk as its own backup"
-
-
-def test_a_reviewed_asset_is_never_overwritten_by_the_csv(tmp_path, factory):
-    asset_id = _asset(factory, tmp_path / "a.jpg")
-    _review(factory, asset_id, score=0.2, decision="sfw")
-    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9999\n")
-
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path, overwrite=True)
-
-    assert (report.imported, report.reviewed) == (0, 1)
-    assert report.conflicts[0]["decision"] == "sfw"
-    assert report.conflicts[0]["csv"] == 0.9999
-    with factory() as session:
-        rows = list(session.scalars(select(SafetyReview)))
-        assert len(rows) == 1 and rows[0].decision == "sfw"
-
-
-def test_an_unknown_path_is_reported_never_turned_into_an_asset(tmp_path, factory):
-    _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(
-        tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n{tmp_path / 'ghost.jpg'},0.5000\n"
-    )
-
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
-
-    assert report.unmatched == [str(tmp_path / "ghost.jpg")]
-    with factory() as session:
-        assert session.scalar(select(Asset).where(Asset.current_path.like("%ghost%"))) is None
-
-
-def test_malformed_rows_are_rejected_with_their_reason(tmp_path, factory):
-    _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(
-        tmp_path,
-        f"{tmp_path / 'a.jpg'},0.5000\n"
-        f"{tmp_path / 'a.jpg'},0.5000\n"  # the same path twice
-        f"{tmp_path / 'b.jpg'},not-a-number\n"
-        f"{tmp_path / 'c.jpg'},7.5\n"  # outside 0..1: corruption, not a score
-        ",0.5\n",  # no path at all
-    )
-
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
-
-    reasons = sorted(r["reason"] for r in report.rejected)
-    assert reasons == ["duplicate_path", "missing_path", "unparsable_score", "unparsable_score"]
-    assert report.imported == 1
-
-
-def test_rerunning_changes_nothing_and_a_differing_score_needs_overwrite(tmp_path, factory):
-    asset_id = _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
-    service = LegacyImportService(factory)
-    service.import_nsfw_scores(csv_path)
-
-    again = service.import_nsfw_scores(csv_path)
-    assert (again.imported, again.unchanged) == (0, 1)
-    assert len(_scores(factory)) == 1
-
-    changed = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.8000\n")
-    refused = service.import_nsfw_scores(changed)
-    assert (refused.imported, refused.skipped_existing) == (0, 1)
-    assert refused.conflicts[0]["current"] == 0.5
-
-    forced = service.import_nsfw_scores(changed, overwrite=True)
-    assert forced.imported == 1
-    with factory() as session:
-        latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
-        assert (latest.asset_id, latest.score) == (asset_id, 0.8)
-
-
-def test_a_dry_run_reports_without_writing(tmp_path, factory):
-    _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
-
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path, dry_run=True)
-
-    assert report.imported == 1
-    assert _scores(factory) == {}, "a dry run must leave the database alone"
-
-
-def test_a_missing_csv_is_a_reported_outcome_not_a_crash(tmp_path, factory):
-    report = LegacyImportService(factory).import_nsfw_scores(tmp_path / "nothing.csv")
-    assert report.rejected == [{"path": str(tmp_path / "nothing.csv"), "reason": "csv_missing"}]
-    assert report.rows == 0
-
-
-def test_the_report_is_written_where_it_can_be_audited(tmp_path, factory):
-    _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.5000\n")
-    report = LegacyImportService(factory).import_nsfw_scores(csv_path)
-
-    written = write_report(report, tmp_path / "data")
-
-    payload = json.loads(written.read_text(encoding="utf-8"))
-    assert payload["imported"] == 1
-    assert payload["source"] == str(csv_path)
-    assert payload["schema_version"] == 1
-    assert payload["generated_at"]
-
-
-def test_a_later_review_still_wins_after_an_import(tmp_path, factory):
-    """The import is a floor, not a ceiling: a human decision made afterwards is
-    the current state, exactly as everywhere else in the app."""
-    asset_id = _asset(factory, tmp_path / "a.jpg")
-    csv_path = _csv(tmp_path, f"{tmp_path / 'a.jpg'},0.9000\n")
-    LegacyImportService(factory).import_nsfw_scores(csv_path)
-    # The import stamps itself with the wall clock, so "afterwards" has to be too.
-    _review(
-        factory,
-        asset_id,
-        score=0.9,
-        decision="sfw",
-        at=datetime.now(timezone.utc) + timedelta(hours=1),
-    )
-
-    with factory() as session:
-        latest = list(session.scalars(select(SafetyReview).order_by(SafetyReview.created_at)))[-1]
-        assert latest.decision == "sfw"

tests/integration/test_safety_parity.py

@@ -3,8 +3,6 @@ same fixture files (real exiftool-written keywords)."""
 
 import shutil
 import subprocess
-import sys
-from pathlib import Path
 
 import numpy as np
 import pytest
@@ -13,10 +11,6 @@ from PIL import Image
 from photo_pipeline.integrations import exiftool
 from photo_pipeline.services import safety
 
-# The donor is frozen in the read-only archive (US07-01); parity still compares
-# against it, so this suite is the second and last place that may import it.
-ARCHIVED_SOURCES = Path(__file__).resolve().parents[2] / "legacy_cli_archive" / "src"
-
 EXIFTOOL = shutil.which("exiftool")
 pytestmark = pytest.mark.skipif(EXIFTOOL is None, reason="exiftool not installed")
 
@@ -37,7 +31,6 @@ def _tag(path, keyword):
 
 
 def test_extracted_marks_and_partition_match_donor(tmp_path):
-    sys.path.insert(0, str(ARCHIVED_SOURCES))  # the donor lives in the archive now
     donor_exif = pytest.importorskip("nsfwtag.exif")
 
     nsfw = _jpeg(tmp_path / "nsfw.jpg", 1)

tests/integration/test_security_boundaries.py

@@ -1,136 +0,0 @@
-"""US07-02: what an unhandled failure says, and where the library ends.
-
-Two boundaries that only show up below the HTTP surface:
-
-* Exception text is where internals leak — absolute paths, SQL, and occasionally a
-  credential passed to the call that blew up. The first tests drive the real
-  application with a route that raises such an exception (no production route does)
-  and assert the client sees only a code.
-* The path the database recorded is not the path the filesystem will open a moment
-  later. Analysis is the one stage whose bytes leave this machine, so it resolves
-  the source against the library roots immediately before the provider call.
-"""
-
-from __future__ import annotations
-
-import logging
-import uuid
-from datetime import datetime, timezone
-
-import pytest
-from PIL import Image
-from starlette.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 AnalysisResult, Asset
-from photo_pipeline.services.analysis import AnalysisService
-from photo_pipeline.services.safety import SafetyService
-
-BOOM = "sqlite:///Users/someone/Pictures/private.db failed with key sk-secret-123"
-
-
-@pytest.fixture
-def client(tmp_path):
-    app = create_app(Config(data_dir=tmp_path / "data"))
-
-    @app.get("/api/v1/boom")
-    def boom():
-        raise RuntimeError(BOOM)
-
-    with TestClient(app, raise_server_exceptions=False) as test_client:
-        yield test_client
-
-
-def test_an_unhandled_error_returns_a_bare_envelope(client, caplog):
-    with caplog.at_level(logging.ERROR):
-        response = client.get("/api/v1/boom")
-
-    assert response.status_code == 500
-    assert response.json() == {"error": {"code": "internal_error", "message": "internal error"}}
-    assert BOOM not in response.text and "Traceback" not in response.text
-    # The operator still gets the whole story, on the server side.
-    assert BOOM in caplog.text
-
-
-def test_a_refusal_response_still_carries_the_default_headers(client):
-    """A 500 escaping the middleware's response path would also escape its headers."""
-    response = client.get("/api/v1/boom")
-    assert response.headers["x-content-type-options"] == "nosniff"
-    assert response.headers["x-frame-options"] == "DENY"
-
-
-# ── the library boundary, revalidated at the moment of use ───────────────────
-
-
-class RecordingProvider:
-    def __init__(self):
-        self.calls = []
-
-    def analyze(self, path, *, album_hint):
-        self.calls.append(path)
-        return {"description": "a photo", "tags": []}
-
-
-def _library(tmp_path):
-    lib = tmp_path / "lib"
-    lib.mkdir()
-    outside = tmp_path / "outside"
-    outside.mkdir()
-    for path in (lib / "inside.jpg", outside / "private.jpg"):
-        Image.new("RGB", (8, 8), "blue").save(path)
-    (tmp_path / "data").mkdir()
-    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
-    run_migrations(config.database_url)
-    return lib, outside, create_session_factory(create_db_engine(config.database_url))
-
-
-def _sfw_asset(sf, path):
-    asset = Asset(
-        id=str(uuid.uuid4()),
-        original_path=str(path),
-        current_path=str(path),
-        discovered_at=datetime.now(timezone.utc),
-        hash_version=1,
-    )
-    with sf() as session:
-        session.add(asset)
-        session.commit()
-    SafetyService(sf).decide(asset.id, "sfw", write_exif=False)
-    return asset.id
-
-
-def test_analysis_will_not_send_a_file_that_left_the_library(tmp_path):
-    """A link swapped under an asset after the scan points at something the user
-    never put in the library. Those bytes must not reach the vision provider — it is
-    the one place in the pipeline where content leaves this machine."""
-    lib, outside, sf = _library(tmp_path)
-    inside = lib / "inside.jpg"
-    asset_id = _sfw_asset(sf, inside)
-    inside.unlink()
-    inside.symlink_to(outside / "private.jpg")
-
-    provider = RecordingProvider()
-    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
-
-    assert provider.calls == [], "the provider must never have been constructed a request"
-    assert result == {"analyzed": 0, "skipped": 0, "errors": 1}
-    with sf() as session:
-        row = session.get(AnalysisResult, asset_id)
-    # The failure is visible and names no path.
-    assert row.status == "error"
-    assert "outside the configured library roots" in row.error_message
-    assert str(outside) not in row.error_message
-
-
-def test_analysis_still_reads_a_file_that_stayed_inside(tmp_path):
-    """The guard must resolve real paths, not refuse everything."""
-    lib, _, sf = _library(tmp_path)
-    asset_id = _sfw_asset(sf, lib / "inside.jpg")
-
-    provider = RecordingProvider()
-    result = AnalysisService(sf, provider=provider, library_roots=(lib,)).run([asset_id])
-
-    assert result["analyzed"] == 1
-    assert provider.calls == [str((lib / "inside.jpg").resolve())]

tests/story_traceability.json

@@ -143,15 +143,6 @@
     ],
     "US06-06": [
       "tests/e2e/test_phase_f_pipeline.py"
-    ],
-    "US07-01": [
-      "tests/unit/test_legacy_archive.py",
-      "tests/integration/test_legacy_import.py"
-    ],
-    "US07-02": [
-      "tests/unit/test_security_policy.py",
-      "tests/integration/test_security_boundaries.py",
-      "tests/e2e/test_security.py"
     ]
   }
 }

tests/unit/test_legacy_archive.py

@@ -1,173 +0,0 @@
-"""Archive lint (US07-01): the frozen CLI archive is complete, honest, and inert.
-
-Freezing the donors is only worth something if three things stay true: the archive
-still holds exactly the bytes it claims, it holds no secret, and production cannot
-reach it. Each of those is a one-line mistake away — a helpful `sys.path` insert, a
-copied `.env`, an edited "just this once" source — so each is asserted here.
-"""
-
-from __future__ import annotations
-
-import hashlib
-import re
-import subprocess
-import sys
-from pathlib import Path
-
-import pytest
-import yaml
-
-REPO = Path(__file__).resolve().parents[2]
-ARCHIVE = REPO / "legacy_cli_archive"
-SOURCES = ARCHIVE / "src"
-LEDGER = ARCHIVE / "donor_ledger.yaml"
-CHECKSUMS = ARCHIVE / "CHECKSUMS.sha256"
-
-# Everything the concept requires an archive to carry (§3 "Donor-first CLI
-# migration and archival"): the sources, their docs, a dependency lock, schema
-# notes, a redacted sample configuration, the ledger, and recorded checksums.
-REQUIRED_ARTIFACTS = (
-    "README.md",
-    "donor_ledger.yaml",
-    "CHECKSUMS.sha256",
-    "requirements-lock.txt",
-    "photo_analyzer.env.sample",
-)
-ARCHIVED_MODULES = ("photo_analyzer", "nsfwtag", "webapp", "nsfw_tag", "compare_models")
-# Only the two suites that compare against the donors may put the archive on the
-# import path; every other tree must not name it at all.
-IMPORT_ALLOWED = {
-    REPO / "tests" / "characterization" / "conftest.py",
-    REPO / "tests" / "characterization" / "test_donor_ledger.py",  # lints the ledger there
-    REPO / "tests" / "integration" / "test_safety_parity.py",
-    Path(__file__),
-}
-SECRET_PATTERNS = (
-    re.compile(r"sk-(?!REPLACE_WITH_YOUR_KEY)[A-Za-z0-9_\-]{16,}"),
-    re.compile(r"AIza[0-9A-Za-z_\-]{20,}"),  # Google API keys
-    re.compile(r"(?i)api[_-]?key\s*[=:]\s*['\"][A-Za-z0-9_\-]{16,}['\"]"),
-)
-
-
-def _archived_files() -> list[Path]:
-    return sorted(p for p in SOURCES.rglob("*") if p.is_file() and "__pycache__" not in p.parts)
-
-
-# ── completeness ─────────────────────────────────────────────────────────────
-
-
-def test_the_archive_carries_every_required_artifact():
-    for artifact in REQUIRED_ARTIFACTS:
-        path = ARCHIVE / artifact
-        assert path.is_file(), f"archive is missing {artifact}"
-        assert path.stat().st_size > 0, f"archive artifact {artifact} is empty"
-    assert (SOURCES / "photo_analyzer.py").is_file()
-    assert (SOURCES / "nsfwtag" / "README.md").is_file(), "donor docs must be archived too"
-    assert (SOURCES / "webapp" / "README.md").is_file()
-
-
-def test_every_ledger_source_is_present_in_the_archive():
-    rows = yaml.safe_load(LEDGER.read_text(encoding="utf-8"))["rows"]
-    for row in rows:
-        source = SOURCES / row["source"]["file"]
-        assert source.is_file(), f"{row['id']}: {row['source']['file']} is not archived"
-
-
-def test_the_readme_records_provenance_and_the_no_import_rule():
-    readme = (ARCHIVE / "README.md").read_text(encoding="utf-8")
-    for expected in ("CHECKSUMS.sha256", "requirements-lock.txt", "donor_ledger.yaml"):
-        assert expected in readme, f"README does not point at {expected}"
-    assert "Schema notes" in readme, "the donor's schema must be documented"
-    assert "nsfw_scores.csv" in readme, "the CSV's fate must be documented"
-
-
-# ── integrity ────────────────────────────────────────────────────────────────
-
-
-def test_every_archived_source_matches_its_checksum():
-    """A frozen archive that silently drifts is not evidence of anything."""
-    recorded = {}
-    for line in CHECKSUMS.read_text(encoding="utf-8").splitlines():
-        digest, _, rel = line.partition("  ")
-        if rel:
-            recorded[rel.strip()] = digest
-    actual = {
-        str(path.relative_to(ARCHIVE)): hashlib.sha256(path.read_bytes()).hexdigest()
-        for path in _archived_files()
-    }
-    assert actual == recorded, "archived sources and CHECKSUMS.sha256 disagree"
-
-
-# ── redaction ────────────────────────────────────────────────────────────────
-
-
-def test_the_archive_contains_no_credential():
-    for path in [*_archived_files(), *(ARCHIVE / a for a in REQUIRED_ARTIFACTS)]:
-        if path.suffix in (".png", ".jpg", ".webp"):
-            continue
-        text = path.read_text(encoding="utf-8", errors="replace")
-        for pattern in SECRET_PATTERNS:
-            assert not pattern.search(text), f"possible secret in {path.relative_to(REPO)}"
-
-
-def test_the_sample_configuration_is_a_placeholder_only():
-    sample = (ARCHIVE / "photo_analyzer.env.sample").read_text(encoding="utf-8")
-    assert "REDACTED" in sample
-    # Placeholders are not key-shaped, so neither a scanner nor a reader can
-    # mistake the sample for a credential.
-    assert "LLM_API_KEY=<" in sample
-    # No real env file, database, log, or CSV may ride along in the archive.
-    strays = [
-        p.name
-        for p in ARCHIVE.rglob("*")
-        if p.is_file() and p.suffix in (".env", ".db", ".sqlite", ".sqlite3", ".log", ".csv")
-    ]
-    assert strays == [], f"unexpected runtime files archived: {strays}"
-
-
-# ── inertness ────────────────────────────────────────────────────────────────
-
-
-def test_production_code_never_imports_an_archived_module():
-    for path in (REPO / "photo_pipeline").rglob("*.py"):
-        text = path.read_text(encoding="utf-8")
-        for module in ARCHIVED_MODULES:
-            assert not re.search(rf"^\s*(import|from)\s+{module}\b", text, re.MULTILINE), \
-                f"{path.relative_to(REPO)} imports the archived {module}"
-        assert "legacy_cli_archive" not in text, \
-            f"{path.relative_to(REPO)} names the archive"
-
-
-def test_only_the_parity_suites_put_the_archive_on_the_import_path():
-    for path in (REPO / "tests").rglob("*.py"):
-        if path in IMPORT_ALLOWED or "__pycache__" in path.parts:
-            continue
-        assert "legacy_cli_archive" not in path.read_text(encoding="utf-8"), \
-            f"{path.relative_to(REPO)} reaches into the archive"
-
-
-def test_the_archived_modules_are_unimportable_from_a_clean_interpreter():
-    """The real check: a fresh process with the repo on its path cannot load them."""
-    script = (
-        "import importlib.util, sys; "
-        f"sys.path.insert(0, {str(REPO)!r}); "
-        "print([m for m in "
-        f"{list(ARCHIVED_MODULES)!r}"
-        " if importlib.util.find_spec(m) is not None])"
-    )
-    result = subprocess.run(
-        [sys.executable, "-c", script], capture_output=True, text=True, cwd=str(REPO), check=True
-    )
-    assert result.stdout.strip() == "[]", f"still importable: {result.stdout.strip()}"
-
-
-def test_the_application_starts_without_the_archive(tmp_path):
-    """Nothing in the runtime path may need the frozen sources to exist."""
-    pytest.importorskip("fastapi")
-    from photo_pipeline.api.app import create_app
-    from photo_pipeline.config import Config
-
-    config = Config.from_env(
-        {"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"), "PHOTO_PIPELINE_LIBRARY_ROOTS": ""}
-    )
-    assert create_app(config) is not None

tests/unit/test_path_policy.py

@@ -81,42 +81,3 @@ def test_symlink_within_root_is_allowed(tmp_path):
         pytest.skip("cannot create symlink on this platform")
     found = path_policy.discover([root])
     assert target in found and link in found
-
-
-def test_resolve_in_roots_returns_the_path_the_caller_must_use(tmp_path):
-    """The resolved path is the answer, not a yes/no: a caller that revalidates one
-    name and then opens another has an open symlink race (US07-02)."""
-    root = tmp_path / "lib"
-    (root / "sub").mkdir(parents=True)
-    target = root / "sub" / "real.jpg"
-    target.write_bytes(b"x")
-    link = root / "alias.jpg"
-    os.symlink(target, link)
-
-    assert path_policy.resolve_in_roots([root], link) == target.resolve()
-    assert path_policy.resolve_in_roots([root], target) == target.resolve()
-
-
-def test_resolve_in_roots_refuses_escapes_without_naming_them(tmp_path):
-    root = tmp_path / "lib"
-    root.mkdir()
-    outside = tmp_path / "outside.jpg"
-    outside.write_bytes(b"x")
-    link = root / "alias.jpg"
-    os.symlink(outside, link)
-
-    for candidate in (link, outside, root / "_IGNORE" / "a.jpg"):
-        with pytest.raises(path_policy.PathPolicyError) as raised:
-            path_policy.resolve_in_roots([root], candidate)
-        assert str(outside) not in str(raised.value)
-
-
-def test_resolve_in_roots_checks_every_configured_root(tmp_path):
-    first, second = tmp_path / "one", tmp_path / "two"
-    first.mkdir()
-    second.mkdir()
-    photo = second / "b.jpg"
-    photo.write_bytes(b"x")
-    assert path_policy.resolve_in_roots([first, second], photo) == photo.resolve()
-    # No configured boundary means nothing to check against.
-    assert path_policy.resolve_in_roots([], photo) == photo

tests/unit/test_security_policy.py

@@ -1,169 +0,0 @@
-"""US07-02: the request-admission policy, enumerated.
-
-``evaluate`` decides every refusal the API can make before a route runs, so the
-whole local-web threat model is one table here: who may call, from where, with what
-proof. The middleware and the endpoints are covered black box in
-``tests/e2e/test_security.py``; this file pins the rules themselves, including the
-combinations a browser can produce but a test client rarely does.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-from photo_pipeline.api.security import (
-    CSRF_HEADER,
-    PUBLIC_PATHS,
-    Session,
-    evaluate,
-    split_host,
-)
-
-SESSION = Session(id="session-id", csrf_token="csrf-token")
-HOST = "127.0.0.1:8000"
-LIMIT = 1024
-
-
-def check(method="GET", path="/api/v1/workflow", **headers):
-    """Evaluate a request that is authenticated and same-origin unless overridden."""
-    sent = {
-        "host": HOST,
-        "cookie-session": SESSION.id,
-        CSRF_HEADER: SESSION.csrf_token,
-    }
-    sent.update({name.replace("_", "-"): value for name, value in headers.items()})
-    sent = {name: value for name, value in sent.items() if value is not None}
-    return evaluate(
-        method=method,
-        path=path,
-        headers=sent,
-        session=SESSION,
-        max_request_bytes=LIMIT,
-    )
-
-
-def test_an_authenticated_same_origin_request_is_admitted():
-    assert check() is None
-    assert check(method="POST", origin="http://127.0.0.1:8000") is None
-    assert check(sec_fetch_site="same-origin") is None
-
-
-@pytest.mark.parametrize("host", ["evil.example", "evil.example:8000", "192.168.1.10:8000", ""])
-def test_a_non_loopback_host_is_refused(host):
-    """DNS rebinding: the browser thinks it is talking to the attacker's name, which
-    resolves to 127.0.0.1. The name is the evidence, so the name is checked."""
-    refusal = check(host=host)
-    assert (refusal.status, refusal.code) == (403, "host_not_allowed")
-
-
-@pytest.mark.parametrize("host", ["127.0.0.1:8000", "localhost:8000", "[::1]:8000", "localhost"])
-def test_loopback_hosts_are_accepted(host):
-    assert check(host=host) is None
-
-
-@pytest.mark.parametrize(
-    "origin",
-    [
-        "http://evil.example",
-        "https://evil.example:8000",
-        "http://127.0.0.1:9999",  # another local app is still another origin
-        "http://localhost.evil.example:8000",
-        "null",
-        "file://",
-    ],
-)
-def test_a_foreign_origin_is_refused(origin):
-    refusal = check(method="POST", origin=origin)
-    assert (refusal.status, refusal.code) == (403, "origin_not_allowed")
-
-
-@pytest.mark.parametrize("origin", ["http://127.0.0.1:8000", "http://localhost:8000"])
-def test_this_applications_origin_is_accepted(origin):
-    assert check(method="POST", origin=origin) is None
-
-
-@pytest.mark.parametrize("site", ["cross-site", "same-site"])
-def test_a_cross_site_fetch_is_refused_even_without_an_origin(site):
-    """What ``<img src="http://127.0.0.1:8000/...">`` on another page looks like."""
-    refusal = check(path="/api/v1/assets/a1/thumbnail", sec_fetch_site=site)
-    assert (refusal.status, refusal.code) == (403, "cross_site_blocked")
-
-
-def test_a_user_initiated_navigation_is_accepted():
-    assert check(sec_fetch_site="none") is None
-
-
-def test_a_request_without_a_session_is_unauthenticated():
-    for method, path in [
-        ("GET", "/api/v1/workflow"),
-        ("POST", "/api/v1/jobs"),
-        ("GET", "/api/v1/assets/a1/thumbnail"),
-    ]:
-        refusal = check(method=method, path=path, cookie_session=None)
-        assert (refusal.status, refusal.code) == (401, "unauthenticated"), path
-
-
-def test_a_forged_session_is_unauthenticated():
-    refusal = check(cookie_session="guessed")
-    assert (refusal.status, refusal.code) == (401, "unauthenticated")
-
-
-@pytest.mark.parametrize("path", sorted(PUBLIC_PATHS))
-def test_health_and_the_bootstrap_stay_reachable_without_a_session(path):
-    assert check(path=path, cookie_session=None) is None
-
-
-def test_the_static_shell_needs_no_session():
-    """It has to load before any JavaScript can ask for one."""
-    assert check(path="/app/index.html", cookie_session=None) is None
-
-
-@pytest.mark.parametrize("method", ["POST", "PUT", "PATCH", "DELETE"])
-def test_a_mutation_without_a_valid_csrf_token_is_refused(method):
-    for token in (None, "guessed"):
-        refusal = check(method=method, path="/api/v1/jobs", **{CSRF_HEADER: token})
-        assert (refusal.status, refusal.code) == (403, "csrf_failed")
-
-
-@pytest.mark.parametrize("method", ["GET", "HEAD", "OPTIONS"])
-def test_reads_need_no_csrf_token(method):
-    assert check(method=method, **{CSRF_HEADER: None}) is None
-
-
-def test_an_oversized_body_is_refused_before_it_is_read():
-    refusal = check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT + 1))
-    assert (refusal.status, refusal.code) == (413, "payload_too_large")
-    assert check(method="POST", path="/api/v1/jobs", content_length=str(LIMIT)) is None
-
-
-def test_the_host_check_precedes_authentication():
-    """A refusal must not tell a foreign caller whether its session guess was right."""
-    refusal = check(host="evil.example", cookie_session="guessed")
-    assert refusal.code == "host_not_allowed"
-
-
-def test_refusals_name_no_path_secret_or_internal():
-    refusals = [
-        check(host="evil.example"),
-        check(method="POST", origin="http://evil.example"),
-        check(cookie_session=None),
-        check(method="POST", **{CSRF_HEADER: None}),
-    ]
-    for refusal in refusals:
-        assert SESSION.id not in refusal.message
-        assert SESSION.csrf_token not in refusal.message
-        assert "/" not in refusal.message
-
-
-@pytest.mark.parametrize(
-    "value,expected",
-    [
-        ("127.0.0.1:8000", ("127.0.0.1", "8000")),
-        ("localhost", ("localhost", "")),
-        ("[::1]:8000", ("[::1]", "8000")),
-        ("[::1]", ("[::1]", "")),
-        ("", ("", "")),
-    ],
-)
-def test_split_host(value, expected):
-    assert split_host(value) == expected