docs: add BACKLOG.md for near-term open tasks

Track loose ends between milestones: the pgvector spike's final coverage
snapshot (pending CLIP re-run completion), the M1 validation-gate run on a hard
sample, and the M1.5 least-privilege read-only role. Linked from ROADMAP.

.env.example

@@ -1,6 +1,14 @@
 IMMICH_URL=http://your-immich-host:2283
 IMMICH_API_KEY=your-immich-api-key
+# Optional: Postgres DSN for read-only access to Immich's pgvector embeddings.
+# Only needed by the pgvector spike (scripts/pgvector_spike.py) / M1.5 visual similarity.
+# Format: postgresql://USER:PASSWORD@HOST:PORT/DBNAME  (Immich defaults: user=postgres, db=immich)
+# IMMICH_DB_URL=postgresql://postgres:your-db-password@your-immich-host:5432/immich
 ANTHROPIC_API_KEY=
 DATA_DIR=./data
 UID=1000
 GID=1000
+
+GITEA_HOST=
+GITEA_USER=
+GITEA_TOKEN=

CLAUDE.md

@@ -0,0 +1,38 @@
+# CLAUDE.md
+
+## What this is
+
+A monorepo (M1) of an Immich photo-flow foundation + the `trip-cluster` app. See
+`docs/ROADMAP.md` for milestones and `docs/superpowers/specs/2026-06-27-immich-photo-flow-design.md`
+for the M1 design.
+
+## Architecture map
+
+- `shared/photoflow/immich` — **the only** Immich client (`client.py`) + `_pipeline/` tag
+  conventions (`pipeline.py`).
+- `shared/photoflow/core` — **the only** SQLite owner: `store.py` (data-access) + `models.py`.
+- `shared/photoflow/ui` — `base.html`, shared Jinja macros, `shared.js` grid+lightbox.
+- `apps/trip-cluster/app` — `config.py`, `cli.py` (ingest|cluster|serve|apply),
+  `ingest.py`, `clustering.py` (pure), `coverage.py` (pure), `cluster_run.py`,
+  `review.py`, `writeback.py`, `routes/`, `templates/`, `static/app.js`.
+
+## Key invariants
+
+- **Immich is the source of truth**; SQLite is rebuildable. Only **applied** decisions
+  survive loss of SQLite (via `_pipeline/processed` + `writeback_log`).
+- Content/trip tags are **never namespaced**; pipeline meta-tags nest under `_pipeline/`.
+- Trip detection is **timestamp-first, density-adaptive**, anchored by existing trip tags,
+  refined by GPS. Existing trip tags are authoritative seeds.
+- Write-back is **idempotent** (`writeback_log`) and needs **explicit confirmation**.
+- Ingest is **scopeable** (`--from/--to`, `--tag`, `--subset`) and incremental (`updatedAfter`).
+- trip-cluster serves on **8084**.
+
+## Dev commands
+
+```bash
+.venv/bin/python -m pytest                     # unit + route (Immich mocked)
+.venv/bin/python -m pytest apps/trip-cluster/tests/ui   # Playwright
+docker compose up                              # UI on :8084
+```
+
+Write a failing test first (TDD). Mirrors the sibling apps in `/home/mischa/Projects/`.

README.md

@@ -0,0 +1,37 @@
+# immich-photo-flow
+
+A monorepo of small tools that clean up and structure a large [Immich](https://immich.app)
+library into travel "memories". **M1** ships the shared foundation
+(`shared/photoflow/{immich,core,ui}`) and the `trip-cluster` app.
+
+## Layout
+
+- `shared/photoflow/immich` — the one Immich REST client + `_pipeline/` tag conventions.
+- `shared/photoflow/core` — SQLite store + domain models (the only SQL).
+- `shared/photoflow/ui` — base template, DaisyUI/Tailwind/Alpine/HTMX, shared grid+lightbox.
+- `apps/trip-cluster` — CLI + Flask review UI (port 8084).
+
+## Dev setup
+
+```bash
+python3.12 -m venv .venv
+.venv/bin/pip install -e ./shared -e ./apps/trip-cluster \
+    pytest==8.3.4 pytest-httpserver==1.1.0 pytest-playwright==0.6.2
+.venv/bin/python -m pytest                 # unit + route tests
+.venv/bin/python -m playwright install chromium
+.venv/bin/python -m pytest apps/trip-cluster/tests/ui   # Playwright UI
+```
+
+## trip-cluster workflow
+
+```bash
+cp .env.example .env       # fill IMMICH_URL + IMMICH_API_KEY
+cd apps/trip-cluster
+python categorize.py ingest --tag "Italy 2019"     # or --from/--to/--subset
+python categorize.py cluster
+python categorize.py serve                          # review at http://localhost:8084
+python categorize.py apply                          # write approved tags back (asks to confirm)
+```
+
+Immich is the source of truth; SQLite is a rebuildable working layer. Content/trip tags
+are never namespaced; pipeline meta-tags live under `_pipeline/`.

apps/trip-cluster/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.12-slim
+WORKDIR /srv
+COPY shared/ ./shared/
+COPY apps/trip-cluster/ ./apps/trip-cluster/
+RUN pip install --no-cache-dir ./shared ./apps/trip-cluster
+ENV DATA_DIR=/data
+EXPOSE 8084
+WORKDIR /srv/apps/trip-cluster
+CMD ["python", "categorize.py", "serve"]

apps/trip-cluster/app/cli.py

@@ -18,6 +18,53 @@ def _immich(deps, cfg):
     return ImmichClient(cfg.immich_url, cfg.immich_api_key)
 
 
+def cmd_ingest(deps, *, date_from, date_to, tag, subset, full) -> int:
+    cfg = deps["config"]
+    store = _store(cfg)
+    client = _immich(deps, cfg)
+    from app.ingest import run_ingest
+    res = run_ingest(client, store, cfg.thumbs_dir, date_from=date_from,
+                     date_to=date_to, tag=tag, subset=subset, full=full)
+    store.close()
+    print(f"Ingested {res['fetched']} asset(s); "
+          f"{res['processed_marked']} already-processed.")
+    return 0
+
+
+def cmd_cluster(deps, *, gap_factor) -> int:
+    cfg = deps["config"]
+    store = _store(cfg)
+    from app.cluster_run import run_cluster
+    res = run_cluster(store, gap_factor=gap_factor)
+    store.close()
+    print(f"Built {res['clusters']} candidate cluster(s). Run 'categorize serve' to review.")
+    return 0
+
+
+def cmd_apply(deps, *, yes) -> int:
+    cfg = deps["config"]
+    store = _store(cfg)
+    from app.writeback import apply_all, APPLYABLE
+    pending = [c for c in store.all_clusters() if c.status in APPLYABLE]
+    if not pending:
+        print("Nothing to apply.")
+        store.close()
+        return 0
+    if not yes:
+        ans = input(f"Apply {len(pending)} cluster decision(s) to Immich? [y/N] ")
+        if ans.strip().lower() not in ("y", "yes"):
+            print("Aborted.")
+            store.close()
+            return 1
+    client = _immich(deps, cfg)
+    results = apply_all(client, store)
+    store.close()
+    ok = sum(len(r["succeeded"]) for r in results)
+    bad = sum(len(r["failed"]) for r in results)
+    print(f"Applied {ok} tag write(s); {bad} failure(s) across {len(results)} cluster(s).")
+    return 0 if bad == 0 else 1
+
+
 def cmd_serve(deps) -> int:
     from app import create_app
     create_app(deps["config"]).run(host="0.0.0.0", port=8084)
@@ -51,7 +98,13 @@ def main(argv=None) -> int:
     deps = {"config": load_config()}
     if args.command == "serve":
         return cmd_serve(deps)
-    # ingest / cluster / apply are wired in later tasks.
+    if args.command == "ingest":
+        return cmd_ingest(deps, date_from=args.date_from, date_to=args.date_to,
+                          tag=args.tag, subset=args.subset, full=args.full)
+    if args.command == "cluster":
+        return cmd_cluster(deps, gap_factor=args.gap_factor)
+    if args.command == "apply":
+        return cmd_apply(deps, yes=args.yes)
     print(f"Command '{args.command}' is not implemented yet.")
     return 1
 

apps/trip-cluster/app/cluster_run.py

@@ -0,0 +1,57 @@
+from collections import defaultdict
+
+from photoflow.core.models import Cluster
+from photoflow.immich import pipeline
+from app.clustering import cluster_assets, _epoch
+from app.coverage import coverage_members
+
+# A tag seeds only if its photos dominate the time window they span: among all
+# assets whose taken_at falls in [min, max] of the tag's photos, at least this
+# fraction must carry the tag. A name/event tag sprinkled among many untagged
+# photos in the same window fails this gate.
+SEED_DOMINANCE_MIN = 0.5
+
+
+def _seed_tags(assets, tags_by_asset, max_span_days: int,
+               dominance_min: float = SEED_DOMINANCE_MIN) -> set:
+    epochs = {a.immich_id: _epoch(a.taken_at) for a in assets}
+    tagged = defaultdict(list)  # tag -> epochs of its assets
+    for a in assets:
+        for t in tags_by_asset.get(a.immich_id, []):
+            if not pipeline.is_pipeline_tag(t):
+                tagged[t].append(epochs[a.immich_id])
+
+    seeds = set()
+    for tag, ts in tagged.items():
+        if len(ts) < 2:
+            continue
+        lo, hi = min(ts), max(ts)
+        # Loose safety cap only — reject absurdly long spans, but keep genuine
+        # long trips (this library has one > 60 days), so the default is high.
+        if hi - lo > max_span_days * 86400:
+            continue
+        # Dominance / temporal contiguity gate.
+        in_window = sum(1 for e in epochs.values() if lo <= e <= hi)
+        if in_window and len(ts) / in_window >= dominance_min:
+            seeds.add(tag)
+    return seeds
+
+
+def run_cluster(store, *, gap_factor: float = 6.0, seed_max_span_days: int = 400) -> dict:
+    assets = store.all_assets(include_processed=False)
+    tags_by_asset = {a.immich_id: store.asset_tags(a.immich_id) for a in assets}
+    seed_tags = _seed_tags(assets, tags_by_asset, seed_max_span_days)
+    candidates = cluster_assets(assets, tags_by_asset, seed_tags, gap_factor=gap_factor)
+    assets_by_id = {a.immich_id: a for a in assets}
+
+    store.clear_clusters()
+    n = 0
+    for cand in candidates:
+        members = coverage_members(cand, assets_by_id)
+        store.insert_cluster(Cluster(
+            start_at=cand.start_at, end_at=cand.end_at,
+            count=sum(1 for m in members if m.included),
+            suggested_name=cand.suggested_name, confidence=cand.confidence,
+            kind_guess=cand.kind_guess, status="pending"), members)
+        n += 1
+    return {"clusters": n}

apps/trip-cluster/app/config.py

@@ -17,6 +17,12 @@ class Config:
     immich_api_key: str
     anthropic_api_key: str
     data_dir: str
+    # Postgres DSN for read-only access to Immich's pgvector embeddings.
+    # Optional: only the pgvector spike / M1.5 visual-similarity work needs it;
+    # REST creds (immich_url/api_key) stay required.
+    # M1.5 TODO: the actual pgvector reader belongs in shared/photoflow/immich
+    # (the only Immich client, per CLAUDE.md) — this field only carries the DSN.
+    immich_db_url: Optional[str] = None
 
     @property
     def db_path(self) -> str:
@@ -33,9 +39,11 @@ def load_config(env: Optional[Mapping] = None) -> Config:
     if missing:
         raise ConfigError(missing)
     data_dir = (env.get("DATA_DIR") or "").strip() or os.path.join(os.getcwd(), "data")
+    immich_db_url = (env.get("IMMICH_DB_URL") or "").strip() or None
     return Config(
         immich_url=env["IMMICH_URL"].strip().rstrip("/"),
         immich_api_key=env["IMMICH_API_KEY"].strip(),
         anthropic_api_key=(env.get("ANTHROPIC_API_KEY") or "").strip(),
         data_dir=data_dir,
+        immich_db_url=immich_db_url,
     )

apps/trip-cluster/app/coverage.py

@@ -0,0 +1,35 @@
+from photoflow.core.models import ClusterMember
+from app.clustering import _epoch, _median
+
+
+def coverage_members(candidate, assets_by_id, *, outlier_factor: float = 8.0) -> list:
+    member_ids = [i for i in candidate.member_ids if i in assets_by_id]
+    members = sorted((assets_by_id[i] for i in member_ids), key=lambda a: a.taken_at)
+    epochs = [_epoch(a.taken_at) for a in members]
+    gaps = [b - a for a, b in zip(epochs, epochs[1:])]
+    base = _median(gaps) if gaps else 0.0
+
+    out = []
+    for idx, a in enumerate(members):
+        is_outlier = False
+        if candidate.seed_tag and base > 0:
+            left = epochs[idx] - epochs[idx - 1] if idx > 0 else 0
+            right = epochs[idx + 1] - epochs[idx] if idx < len(members) - 1 else 0
+            nearest = min([g for g in (left, right) if g > 0], default=0)
+            if nearest > outlier_factor * base:
+                is_outlier = True
+        out.append(ClusterMember(
+            cluster_id=0, immich_id=a.immich_id,
+            member_confidence=candidate.confidence, is_outlier=is_outlier,
+            included=True, flagged_coverage=False))
+
+    if candidate.seed_tag and candidate.start_at and candidate.end_at:
+        mset = set(member_ids)
+        for a in sorted(assets_by_id.values(), key=lambda x: x.taken_at):
+            if a.immich_id in mset:
+                continue
+            if candidate.start_at <= a.taken_at <= candidate.end_at:
+                out.append(ClusterMember(
+                    cluster_id=0, immich_id=a.immich_id, member_confidence=0.0,
+                    is_outlier=False, included=False, flagged_coverage=True))
+    return out

apps/trip-cluster/app/ingest.py

@@ -0,0 +1,102 @@
+import datetime
+import logging
+import os
+
+from photoflow.core.models import Asset
+from photoflow.immich import pipeline
+
+log = logging.getLogger(__name__)
+
+
+def _now() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None,
+               tag=None, subset=None, full=False) -> dict:
+    os.makedirs(thumbs_dir, exist_ok=True)
+
+    tag_ids = None
+    if tag:
+        tid = client.resolve_tag_id(tag)
+        if not tid:
+            raise ValueError(f"Tag {tag!r} not found in Immich")
+        tag_ids = [tid]
+
+    updated_after = None if full else store.get_meta("last_ingest_at")
+
+    assets = client.search_assets(taken_after=date_from, taken_before=date_to,
+                                  tag_ids=tag_ids, updated_after=updated_after)
+    if subset is not None:
+        assets = assets[:subset]
+
+    processed_marked = 0
+    tag_counts: dict = {}
+    successful_updated: list = []   # updated_at of assets ingested this run
+    failed_updated: list = []       # updated_at of assets we could NOT ingest
+
+    for a in assets:
+        # Revision (2): sanitize the Immich-provided id before using it as a
+        # path component, guarding against path traversal (mirror /thumb proxy).
+        safe_id = os.path.basename(a["id"])
+        thumb_path = os.path.join(thumbs_dir, f"{safe_id}.jpg")
+        # Revision (1): treat an existing 0-byte file as missing.
+        need_thumb = (not os.path.exists(thumb_path)
+                      or os.path.getsize(thumb_path) == 0)
+        if need_thumb:
+            try:
+                data = client.download_thumbnail(a["id"])
+            except Exception:
+                log.exception("Thumbnail download failed for asset %s", a["id"])
+                failed_updated.append(a["updated_at"])
+                continue
+            # Write to a temp path and os.replace into place so a failure
+            # never leaves a 0-byte <id>.jpg behind; clean up the temp file
+            # if the write itself fails (e.g. disk full).
+            tmp_path = f"{thumb_path}.tmp"
+            try:
+                with open(tmp_path, "wb") as f:
+                    f.write(data)
+                os.replace(tmp_path, thumb_path)
+            except Exception:
+                log.exception("Thumbnail write failed for asset %s", a["id"])
+                try:
+                    os.unlink(tmp_path)
+                except OSError:
+                    pass
+                failed_updated.append(a["updated_at"])
+                continue
+
+        store.upsert_asset(Asset(
+            immich_id=a["id"], taken_at=a["taken_at"], gps_lat=a["gps_lat"],
+            gps_lon=a["gps_lon"], place_city=a["place_city"],
+            place_country=a["place_country"], type=a["type"],
+            thumb_path=thumb_path, ingested_at=_now(), updated_at=a["updated_at"]))
+        store.set_asset_tags(a["id"], a["tags"])
+        for t in a["tags"]:
+            tag_counts[t] = tag_counts.get(t, 0) + 1
+
+        if pipeline.PROCESSED in a["tags"]:
+            store.mark_processed(a["id"])
+            processed_marked += 1
+
+        successful_updated.append(a["updated_at"])
+
+    for name, count in tag_counts.items():
+        store.upsert_tag(name, count=count)
+
+    # Advance the incremental cursor, but never past an asset we failed to ingest
+    # (a later success must not strand an earlier failure on the next run), never
+    # below the prior cursor, and not at all on a --subset sampling run.
+    if subset is None:
+        new_watermark = updated_after or ""
+        floor = min((u for u in failed_updated if u), default=None)
+        for u in successful_updated:
+            if not u or (floor is not None and u >= floor):
+                continue
+            if u > new_watermark:
+                new_watermark = u
+        if new_watermark:
+            store.set_meta("last_ingest_at", new_watermark)
+
+    return {"fetched": len(assets), "processed_marked": processed_marked}

apps/trip-cluster/app/review.py

@@ -0,0 +1,41 @@
+import datetime
+
+
+def _now() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def approve(store, cluster_id, name=None) -> None:
+    c = store.get_cluster(cluster_id)
+    decided = (name or "").strip() or c.suggested_name
+    store.update_cluster(cluster_id, status="approved",
+                         decided_name=decided, reviewed_at=_now())
+
+
+def mark_non_trip(store, cluster_id) -> None:
+    store.update_cluster(cluster_id, status="non_trip", reviewed_at=_now())
+
+
+def skip(store, cluster_id) -> None:
+    store.update_cluster(cluster_id, status="skipped", reviewed_at=_now())
+
+
+def split(store, cluster_id, boundary_immich_id):
+    return store.split_cluster(cluster_id, boundary_immich_id)
+
+
+def merge(store, cluster_id_a, cluster_id_b):
+    return store.merge_clusters(cluster_id_a, cluster_id_b)
+
+
+def set_member(store, cluster_id, immich_id, included: bool) -> None:
+    store.set_member_inclusion(cluster_id, immich_id, included)
+
+
+def approve_high_confidence(store, threshold: float = 0.75) -> int:
+    count = 0
+    for c in store.all_clusters():
+        if c.status == "pending" and c.kind_guess == "trip" and c.confidence >= threshold:
+            approve(store, c.id)
+            count += 1
+    return count

apps/trip-cluster/app/routes/nav.py

@@ -1,8 +1,14 @@
-from flask import Blueprint
+from flask import Blueprint, current_app, render_template
+from photoflow.core import Store
 
 bp = Blueprint("nav", __name__)
 
 
+def _store():
+    cfg = current_app.config["APP_CONFIG"]
+    return Store(cfg.db_path).connect()
+
+
 @bp.route("/health")
 def health():
     return "ok"
@@ -10,4 +16,9 @@ def health():
 
 @bp.route("/")
 def index():
-    return "trip-cluster"
+    s = _store()
+    clusters = s.clusters_by_attention()
+    stats = {"assets": len(s.all_assets()), "clusters": len(clusters),
+             "pending": sum(1 for c in clusters if c.status == "pending")}
+    s.close()
+    return render_template("review.html", clusters=clusters, stats=stats)

apps/trip-cluster/app/routes/review.py

@@ -1,3 +1,127 @@
-from flask import Blueprint
+from flask import Blueprint, current_app, render_template, request, jsonify, abort
+from photoflow.core import Store
+from app import review
 
 bp = Blueprint("review", __name__)
+
+
+def _store():
+    cfg = current_app.config["APP_CONFIG"]
+    return Store(cfg.db_path).connect()
+
+
+def _client():
+    factory = current_app.config.get("IMMICH_FACTORY")
+    if factory:
+        return factory()
+    from photoflow.immich import ImmichClient
+    cfg = current_app.config["APP_CONFIG"]
+    return ImmichClient(cfg.immich_url, cfg.immich_api_key)
+
+
+def _body():
+    return request.get_json(silent=True) or {}
+
+
+@bp.route("/cluster/<int:cid>")
+def detail(cid):
+    s = _store()
+    c = s.get_cluster(cid)
+    if c is None:
+        s.close()
+        abort(404)
+    members = s.cluster_members(cid)
+    prev_id, next_id = s.chronological_neighbors(cid)
+    s.close()
+    return render_template("_detail.html", c=c, members=members,
+                           prev_id=prev_id, next_id=next_id)
+
+
+@bp.route("/cluster/<int:cid>/approve", methods=["POST"])
+def approve(cid):
+    s = _store()
+    review.approve(s, cid, _body().get("name"))
+    c = s.get_cluster(cid)
+    s.close()
+    return jsonify({"ok": True, "status": c.status, "name": c.decided_name})
+
+
+@bp.route("/cluster/<int:cid>/non-trip", methods=["POST"])
+def non_trip(cid):
+    s = _store()
+    review.mark_non_trip(s, cid)
+    s.close()
+    return jsonify({"ok": True, "status": "non_trip"})
+
+
+@bp.route("/cluster/<int:cid>/skip", methods=["POST"])
+def skip(cid):
+    s = _store()
+    review.skip(s, cid)
+    s.close()
+    return jsonify({"ok": True, "status": "skipped"})
+
+
+@bp.route("/cluster/<int:cid>/split", methods=["POST"])
+def split(cid):
+    s = _store()
+    boundary = _body().get("boundary_asset_id")
+    if not boundary:
+        s.close()
+        return jsonify({"error": "boundary_asset_id required"}), 400
+    try:
+        id1, id2 = review.split(s, cid, boundary)
+    except ValueError as e:
+        s.close()
+        return jsonify({"error": str(e)}), 400
+    s.close()
+    return jsonify({"ok": True, "ids": [id1, id2]})
+
+
+@bp.route("/cluster/<int:cid>/merge", methods=["POST"])
+def merge(cid):
+    s = _store()
+    other = _body().get("other_id")
+    if other is None:
+        s.close()
+        return jsonify({"error": "other_id required"}), 400
+    new_id = review.merge(s, cid, int(other))
+    s.close()
+    return jsonify({"ok": True, "id": new_id})
+
+
+@bp.route("/cluster/<int:cid>/member", methods=["POST"])
+def member(cid):
+    s = _store()
+    body = _body()
+    review.set_member(s, cid, body["asset_id"], bool(body.get("included")))
+    s.close()
+    return jsonify({"ok": True})
+
+
+@bp.route("/approve-high-confidence", methods=["POST"])
+def approve_high_confidence():
+    s = _store()
+    threshold = float(_body().get("threshold", 0.75))
+    n = review.approve_high_confidence(s, threshold)
+    s.close()
+    return jsonify({"approved": n})
+
+
+@bp.route("/cluster/<int:cid>/apply", methods=["POST"])
+def apply_one(cid):
+    # Lazy import: app.writeback is owned by a later task and may be absent
+    # at app-startup; importing here keeps the blueprint importable regardless.
+    from app.writeback import apply_cluster
+    with _store() as s:               # close even if the Immich write-back raises
+        res = apply_cluster(_client(), s, cid)
+    return jsonify(res)
+
+
+@bp.route("/apply-all", methods=["POST"])
+def apply_everything():
+    # Lazy import: see apply_one above.
+    from app.writeback import apply_all
+    with _store() as s:               # close even if the Immich write-back raises
+        results = apply_all(_client(), s)
+    return jsonify({"results": results})

apps/trip-cluster/app/static/app.js

@@ -0,0 +1,184 @@
+function clusterReview() {
+  return {
+    ...photoGrid(),
+    selected: null,
+    clusterIds: [],
+    failures: [],
+
+    init() {
+      this.clusterIds = [...document.querySelectorAll('.cluster-row')]
+        .map(el => parseInt(el.dataset.clusterId, 10));
+      this.selected = this.clusterIds.length ? this.clusterIds[0] : null;
+      document.body.addEventListener('htmx:afterSwap', (e) => {
+        if (e.target.id === 'detail') {
+          this.hideIndicator();
+          this.selectFirst();
+        }
+      });
+    },
+
+    // --- F16: loading indicator for programmatic swaps -----------------
+    showIndicator() {
+      const el = document.getElementById('detail-indicator');
+      if (el) el.classList.add('htmx-request');
+    },
+    hideIndicator() {
+      const el = document.getElementById('detail-indicator');
+      if (el) el.classList.remove('htmx-request');
+    },
+
+    selectCluster(id) {
+      this.selected = id;
+      this.showIndicator();
+      htmx.ajax('GET', `/cluster/${id}`, { target: '#detail' });
+    },
+
+    moveCluster(dir) {
+      const n = this.clusterIds.length;
+      if (!n) return;
+      const i = this.clusterIds.indexOf(this.selected);
+      const j = ((i + dir) % n + n) % n;   // wrap around (mirrors nextPendingId)
+      if (this.clusterIds[j] != null) this.selectCluster(this.clusterIds[j]);
+    },
+
+    // --- F3: in-place rail status + advance to next pending ------------
+    statusClass(status) {
+      return ({
+        pending: 'badge-ghost', approved: 'badge-success', non_trip: 'badge-neutral',
+        skipped: 'badge-warning', merged: 'badge-info', split: 'badge-info',
+      })[status] || 'badge-ghost';
+    },
+    railStatusEl(id) {
+      return document.querySelector(`.cluster-status[data-cluster-id="${id}"]`);
+    },
+    railStatusText(id) {
+      const el = this.railStatusEl(id);
+      return el ? el.textContent.trim() : '';
+    },
+    updateRailStatus(id, status) {
+      const el = this.railStatusEl(id);
+      if (el) el.innerHTML =
+        `<span class="badge badge-sm ${this.statusClass(status)}">${status}</span>`;
+    },
+    nextPendingId(fromId) {
+      const n = this.clusterIds.length;
+      const start = Math.max(0, this.clusterIds.indexOf(fromId));
+      for (let k = 1; k <= n; k++) {
+        const id = this.clusterIds[(start + k) % n];
+        if (id !== fromId && this.railStatusText(id) === 'pending') return id;
+      }
+      return null;
+    },
+    // Update the acted cluster's rail badge in place (keeping the new status
+    // text visible) and advance the detail pane to the next pending cluster
+    // instead of bouncing back to the attention-queue head.
+    afterDecision(id, status) {
+      this.updateRailStatus(id, status);
+      const next = this.nextPendingId(id);
+      if (next != null) this.selectCluster(next);
+      else this.selectCluster(id);  // no pending left: refresh current detail
+    },
+
+    nameValue() {
+      const el = document.getElementById('cluster-name');
+      return el ? el.value : '';
+    },
+
+    async post(path, body) {
+      const res = await fetch(path, {
+        method: 'POST', headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify(body || {}),
+      });
+      return res.json();
+    },
+
+    async approve() {
+      const acted = this.selected;
+      const r = await this.post(`/cluster/${acted}/approve`, { name: this.nameValue() });
+      this.afterDecision(acted, r.status || 'approved');
+    },
+    async nonTrip() {
+      const acted = this.selected;
+      const r = await this.post(`/cluster/${acted}/non-trip`);
+      this.afterDecision(acted, r.status || 'non_trip');
+    },
+    async skip() {
+      const acted = this.selected;
+      const r = await this.post(`/cluster/${acted}/skip`);
+      this.afterDecision(acted, r.status || 'skipped');
+    },
+    async split() {
+      // F7: require a focused boundary photo before splitting.
+      if (!this.focused) { alert('Focus the first photo of the second trip, then Split.'); return; }
+      await this.post(`/cluster/${this.selected}/split`, { boundary_asset_id: this.focused.dataset.assetId });
+      location.reload();  // split restructures the rail: full refresh is correct
+    },
+    async merge(otherId) {
+      // F17: merge terminally marks both clusters merged with no UI un-merge.
+      if (!confirm('Merge these two clusters? This cannot be undone in the UI.')) return;
+      await this.post(`/cluster/${this.selected}/merge`, { other_id: otherId });
+      location.reload();  // merge restructures the rail: full refresh is correct
+    },
+    async setMember(assetId, included) {
+      await this.post(`/cluster/${this.selected}/member`, { asset_id: assetId, included });
+      this.selectCluster(this.selected);
+    },
+    async approveHighConfidence() {
+      // F19: show a pre-action count and confirm.
+      const threshold = 0.75;
+      const n = this.clusterIds.filter(id => {
+        const row = document.querySelector(`.cluster-row[data-cluster-id="${id}"]`);
+        return row && row.querySelector('.badge-success[title]');  // confidence "high"
+      }).length;
+      if (!confirm(`Approve ${n} high-confidence cluster(s) (>= ${threshold})?`)) return;
+      const r = await this.post('/approve-high-confidence', { threshold });
+      alert(`Approved ${r.approved} cluster(s).`);
+      location.reload();
+    },
+    async apply() {
+      // Write-back needs explicit confirmation (mirrors applyAll/merge).
+      if (!confirm('Apply this cluster’s decision to Immich?')) return;
+      const r = await this.post(`/cluster/${this.selected}/apply`, {});
+      alert(`Applied ${r.succeeded.length}, failed ${r.failed.length}.`);
+    },
+    async applyAll() {
+      if (!confirm('Apply all approved decisions to Immich?')) return;
+      const r = await this.post('/apply-all', {});
+      const results = r.results || [];
+      let ok = 0, bad = 0;
+      this.failures = [];
+      for (const x of results) {
+        const succeeded = x.succeeded || [];
+        const failed = x.failed || [];
+        ok += succeeded.length;
+        bad += failed.length;
+        if (failed.length) {
+          const id = x.cluster_id != null ? x.cluster_id : x.id;
+          const row = document.querySelector(`.cluster-row[data-cluster-id="${id}"]`);
+          if (row) row.classList.add('badge-error', 'ring-1', 'ring-error');
+          this.failures.push({
+            id,
+            name: row ? row.dataset.clusterName : `cluster ${id}`,
+            failed: failed.length,
+          });
+        }
+      }
+      alert(`Applied ${ok} write(s), ${bad} failure(s).`);
+    },
+
+    onKey(e) {
+      if (e.target.tagName === 'INPUT') return;
+      const k = e.key.toLowerCase();
+      if (e.key === '[') { e.preventDefault(); this.moveCluster(-1); }
+      else if (e.key === ']') { e.preventDefault(); this.moveCluster(1); }
+      else if (e.key === 'ArrowLeft') { e.preventDefault(); this.navigate(-1); }
+      else if (e.key === 'ArrowRight') { e.preventDefault(); this.navigate(1); }
+      else if (e.key === 'Enter') { if (this.focused) this.openLightbox(this.focused); }
+      else if (e.key === 'Escape') { this.closeLightbox(); }
+      else if (k === 'a') { this.approve(); }
+      else if (k === 'n') { this.nonTrip(); }
+      else if (k === 's') { if (this.focused) this.split(); }  // F7: suppress until focused
+      else if (k === 'x') { this.skip(); }
+    },
+  };
+}

apps/trip-cluster/app/templates/_detail.html

@@ -0,0 +1,41 @@
+{% from "macros.html" import lightbox %}
+<div data-cluster-id="{{ c.id }}">
+  <div class="flex flex-wrap items-center gap-2 mb-3">
+    <input id="cluster-name" class="input input-bordered input-sm"
+           value="{{ c.decided_name or c.suggested_name }}">
+    <button class="btn btn-sm btn-success" @click="approve()">Approve (A)</button>
+    <button class="btn btn-sm" @click="nonTrip()">Non-trip (N)</button>
+    <!-- F7: Split disabled until a boundary photo is focused -->
+    <button class="btn btn-sm" :disabled="!focused" @click="split()">Split (S)</button>
+    <button class="btn btn-sm" @click="skip()">Skip (X)</button>
+    <button class="btn btn-sm btn-ghost" @click="apply()">Apply</button>
+    {% if prev_id %}<button class="btn btn-xs" @click="merge({{ prev_id }})">⤺ merge prev</button>{% endif %}
+    {% if next_id %}<button class="btn btn-xs" @click="merge({{ next_id }})">merge next ⤻</button>{% endif %}
+  </div>
+  <!-- F7: visible instruction for Split -->
+  <div class="text-xs opacity-60 mb-2" x-show="!focused">
+    Focus the first photo of the second trip (click it or use ← →), then Split.
+  </div>
+  <div class="text-sm opacity-60 mb-2">
+    {{ c.start_at[:10] }} → {{ c.end_at[:10] }} · {{ members | length }} assets · status {{ c.status }}
+  </div>
+  <div class="grid grid-cols-3 sm:grid-cols-4 md:grid-cols-6 gap-2">
+    {% for a, m in members %}
+    <div class="photo-card relative cursor-pointer rounded-lg overflow-hidden border-2
+                {% if m.flagged_coverage %}border-info{% elif m.is_outlier %}border-warning{% else %}border-transparent{% endif %}"
+         data-asset-id="{{ a.immich_id }}" data-included="{{ '1' if m.included else '' }}"
+         tabindex="0" @click="openLightbox($el)" @focus="select($el)">
+      <img src="/thumb/{{ a.immich_id }}" class="w-full aspect-square object-cover" loading="lazy" alt="">
+      {% if m.flagged_coverage %}
+      <button class="absolute bottom-1 left-1 badge badge-xs badge-info"
+              @click.stop="setMember('{{ a.immich_id }}', true)">+ include</button>
+      {% elif m.is_outlier %}
+      <button class="absolute bottom-1 left-1 badge badge-xs badge-warning"
+              @click.stop="setMember('{{ a.immich_id }}', false)">– exclude</button>
+      {% endif %}
+      {% if not m.included %}<div class="absolute inset-0 bg-black/50 pointer-events-none"></div>{% endif %}
+    </div>
+    {% endfor %}
+  </div>
+  {{ lightbox() }}
+</div>

apps/trip-cluster/app/templates/review.html

@@ -0,0 +1,71 @@
+{% extends "base.html" %}
+{% from "macros.html" import confidence_badge, status_badge %}
+{% block title %}trip-cluster{% endblock %}
+{% block navbar_title %}trip-cluster{% endblock %}
+{% block content %}
+{% if stats.clusters == 0 %}
+  <div class="hero py-20"><div class="hero-content text-center"><div>
+    <h1 class="text-2xl font-bold">No clusters yet</h1>
+    {% if stats.assets == 0 %}
+      <p class="opacity-70 mt-2">Nothing ingested. Run
+        <code>categorize ingest</code> then <code>categorize cluster</code>.</p>
+    {% else %}
+      <p class="opacity-70 mt-2">{{ stats.assets }} assets ingested, but this scope
+        produced no clusters. Try a wider <code>categorize ingest</code> scope.</p>
+    {% endif %}
+  </div></div></div>
+{% else %}
+<div x-data="clusterReview()" @keydown.window="onKey($event)" class="flex gap-4">
+  <div class="w-72 shrink-0 max-h-[85vh] overflow-y-auto">
+    <div class="flex items-center justify-between mb-1">
+      <span class="font-bold">Clusters</span>
+      <button class="btn btn-xs btn-primary" @click="approveHighConfidence()">Approve high-conf</button>
+    </div>
+    <!-- F8: explain the attention sort -->
+    <p class="text-xs opacity-60 mb-2">Needs attention first — lowest-confidence pending.</p>
+    {% if stats.pending == 0 %}
+      <div class="alert alert-success text-xs mb-2">All reviewed — “Apply all” or <code>categorize apply</code>.</div>
+    {% endif %}
+    <!-- F4: write-failed clusters surfaced here after Apply all -->
+    <div x-show="failures.length" class="alert alert-error text-xs mb-2 flex-col items-start">
+      <span class="font-semibold">Write failures:</span>
+      <ul class="list-disc ml-4">
+        <template x-for="f in failures" :key="f.id">
+          <li><span x-text="f.name"></span> — <span x-text="f.failed"></span> failed</li>
+        </template>
+      </ul>
+    </div>
+    {% for c in clusters %}
+    <div class="cluster-row block p-2 rounded cursor-pointer hover:bg-base-300"
+         :class="selected == {{ c.id }} && 'bg-base-300'"
+         data-cluster-id="{{ c.id }}" data-cluster-name="{{ c.decided_name or c.suggested_name }}"
+         @click="selectCluster({{ c.id }})">
+      <div class="flex items-center justify-between gap-1">
+        <span class="truncate text-sm font-medium">{{ c.decided_name or c.suggested_name }}</span>
+        {{ confidence_badge(c.confidence) }}
+      </div>
+      <div class="flex items-center gap-1 mt-1">
+        <span class="cluster-status" data-cluster-id="{{ c.id }}">{{ status_badge(c.status) }}</span>
+        <span class="badge badge-xs">{{ c.count }}</span>
+        <span class="text-xs opacity-50">{{ c.start_at[:10] }}</span>
+      </div>
+    </div>
+    {% endfor %}
+    <button class="btn btn-xs btn-block mt-3" @click="applyAll()">Apply all approved</button>
+  </div>
+  <div class="flex-1 relative">
+    <!-- F16: loading indicator for declarative + programmatic swaps -->
+    <div id="detail-indicator" class="htmx-indicator absolute top-2 right-2 z-10">
+      <span class="loading loading-spinner loading-md"></span>
+    </div>
+    <div id="detail" class="flex-1"
+         hx-get="/cluster/{{ clusters[0].id }}" hx-trigger="load" hx-target="#detail"
+         hx-indicator="#detail-indicator"></div>
+  </div>
+</div>
+<div class="fixed bottom-2 right-2 text-xs opacity-50">
+  [ ] cluster · ← → grid · Enter open · A approve · N non-trip · S split (focus a photo) · X skip · Esc close
+</div>
+{% endif %}
+{% endblock %}
+{% block extra_scripts %}<script src="/static/app.js"></script>{% endblock %}

apps/trip-cluster/app/writeback.py

@@ -0,0 +1,62 @@
+from photoflow.immich import pipeline
+
+APPLYABLE = ("approved", "non_trip", "skipped")
+
+
+def _apply_tag(client, store, asset_ids, action, tag):
+    todo = [a for a in asset_ids if not store.already_applied(a, action, tag)]
+    if not todo:
+        return [], []
+    try:
+        # upsert_tag is part of the write: a failure here must be caught and
+        # recorded like a tag_assets failure, not propagate out of apply_cluster
+        # (which would abort the whole apply_all batch and skip later clusters).
+        tag_id = client.upsert_tag(tag)
+        client.tag_assets(tag_id, todo)
+    except Exception as e:                       # noqa: BLE001 — recorded, surfaced
+        for a in todo:
+            store.log_writeback(a, action, tag, f"error:{e}")
+        return [], [(a, str(e)) for a in todo]
+    for a in todo:
+        store.log_writeback(a, action, tag, "ok")
+    return todo, []
+
+
+def apply_cluster(client, store, cluster_id) -> dict:
+    c = store.get_cluster(cluster_id)
+    if c is None or c.status not in APPLYABLE:
+        return {"cluster_id": cluster_id, "status": c.status if c else None,
+                "succeeded": [], "failed": []}
+
+    included = [a.immich_id for a, m in store.cluster_members(cluster_id) if m.included]
+    succeeded, failed = [], []
+
+    if c.status == "approved":
+        tag = c.decided_name or c.suggested_name
+        ok, fail = _apply_tag(client, store, included, "trip", tag)
+        succeeded += ok
+        failed += fail
+    elif c.status == "non_trip":
+        ok, fail = _apply_tag(client, store, included, "non-trip", pipeline.NON_TRIP)
+        succeeded += ok
+        failed += fail
+    # 'skipped': no content/non-trip tag, only processed below.
+
+    failed_ids = {i for i, _ in failed}
+    proc_targets = [a for a in included if a not in failed_ids]
+    if proc_targets:
+        # Surface processed-marker failures too — a discarded return here makes a
+        # failed _pipeline/processed write look like success in the CLI/UI summary.
+        _, proc_fail = _apply_tag(client, store, proc_targets, "processed", pipeline.PROCESSED)
+        failed += proc_fail
+        for a in proc_targets:
+            if store.already_applied(a, "processed", pipeline.PROCESSED):
+                store.mark_processed(a)
+
+    return {"cluster_id": cluster_id, "status": c.status,
+            "succeeded": succeeded, "failed": failed}
+
+
+def apply_all(client, store) -> list:
+    return [apply_cluster(client, store, c.id)
+            for c in store.all_clusters() if c.status in APPLYABLE]

apps/trip-cluster/pyproject.toml

@@ -11,3 +11,6 @@ dependencies = ["photoflow", "flask==3.1.0", "requests==2.32.3", "Pillow==11.0.0
 [tool.setuptools.packages.find]
 where = ["."]
 include = ["app*"]
+
+[tool.setuptools.package-data]
+"app" = ["templates/*.html", "static/*.js"]

apps/trip-cluster/tests/test_cluster_run.py

@@ -0,0 +1,61 @@
+from photoflow.core import Store
+from photoflow.core.models import Asset
+from photoflow.immich import pipeline
+from app.cluster_run import run_cluster
+
+
+def _store(tmp_path):
+    return Store(str(tmp_path / "t.db")).connect()
+
+
+def test_run_cluster_seeds_bounded_tag_not_people_tag(tmp_path):
+    s = _store(tmp_path)
+    # "Italy 2019" spans 2 days -> seeds; "Mum" spans years -> not a seed.
+    rows = [("a", "2019-06-01T10:00:00", ["Italy 2019", "Mum"]),
+            ("b", "2019-06-02T10:00:00", ["Italy 2019"]),
+            ("c", "2010-01-01T10:00:00", ["Mum"]),
+            ("d", "2022-01-01T10:00:00", ["Mum"])]
+    for i, t, tags in rows:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+        s.set_asset_tags(i, tags)
+    res = run_cluster(s)
+    seeded = [c for c in s.all_clusters() if c.suggested_name == "Italy 2019"]
+    assert len(seeded) == 1
+    assert sorted(x.immich_id for x, _ in s.cluster_members(seeded[0].id)) == ["a", "b"]
+    assert res["clusters"] >= 1
+    s.close()
+
+
+def test_run_cluster_excludes_processed_and_replaces(tmp_path):
+    s = _store(tmp_path)
+    s.upsert_asset(Asset(immich_id="a", taken_at="2019-06-01T10:00:00"))
+    s.upsert_asset(Asset(immich_id="p", taken_at="2019-06-01T11:00:00"))
+    s.set_asset_tags("a", [])
+    s.set_asset_tags("p", [pipeline.PROCESSED])
+    s.mark_processed("p")
+    run_cluster(s)
+    run_cluster(s)  # idempotent replace — not doubled
+    all_member_ids = [x.immich_id for c in s.all_clusters()
+                      for x, _ in s.cluster_members(c.id)]
+    assert "p" not in all_member_ids
+    assert all_member_ids.count("a") == 1
+    s.close()
+
+
+def test_run_cluster_short_span_non_dominant_tag_does_not_seed(tmp_path):
+    s = _store(tmp_path)
+    # "Lunch" is on 2 photos within a ~30-min window, but that same window is
+    # full of untagged photos -> the tag does NOT dominate its window
+    # (2/5 = 0.4 < 0.5), so it must not seed even though its span is tiny.
+    rows = [("x", "2020-03-01T12:00:00", ["Lunch"]),
+            ("y", "2020-03-01T12:30:00", ["Lunch"]),
+            ("u1", "2020-03-01T12:05:00", []),
+            ("u2", "2020-03-01T12:15:00", []),
+            ("u3", "2020-03-01T12:25:00", [])]
+    for i, t, tags in rows:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+        s.set_asset_tags(i, tags)
+    run_cluster(s)
+    seeded = [c for c in s.all_clusters() if c.suggested_name == "Lunch"]
+    assert seeded == []
+    s.close()

apps/trip-cluster/tests/test_config.py

@@ -16,3 +16,13 @@ def test_anthropic_optional_and_paths(tmp_path):
     assert cfg.anthropic_api_key == ""           # optional in M1
     assert cfg.db_path == os.path.join(str(tmp_path), "trip-cluster.db")
     assert cfg.thumbs_dir == os.path.join(str(tmp_path), "thumbs")
+
+
+def test_immich_db_url_optional():
+    # DSN is optional — only the pgvector spike / M1.5 need it; REST creds stay required.
+    cfg = load_config({"IMMICH_URL": "http://x", "IMMICH_API_KEY": "k"})
+    assert cfg.immich_db_url is None
+
+    cfg = load_config({"IMMICH_URL": "http://x", "IMMICH_API_KEY": "k",
+                       "IMMICH_DB_URL": "  postgresql://u:p@h:5432/immich  "})
+    assert cfg.immich_db_url == "postgresql://u:p@h:5432/immich"  # trimmed

apps/trip-cluster/tests/test_coverage.py

@@ -0,0 +1,48 @@
+from photoflow.core.models import Asset
+from app.clustering import CandidateCluster
+from app.coverage import coverage_members
+
+
+def _a(i, taken):
+    return Asset(immich_id=i, taken_at=taken)
+
+
+def test_coverage_candidate_inside_seeded_window():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("b", "2019-06-01T12:00:00"),
+               _a("c", "2019-06-02T10:00:00")]
+    intruder = _a("x", "2019-06-01T13:00:00")          # in window, untagged
+    outside = _a("y", "2019-07-01T10:00:00")           # out of window
+    by_id = {m.immich_id: m for m in members + [intruder, outside]}
+    cand = CandidateCluster(member_ids=["a", "b", "c"], start_at="2019-06-01T10:00:00",
+                            end_at="2019-06-02T10:00:00", suggested_name="Italy 2019",
+                            confidence=0.95, kind_guess="trip", seed_tag="Italy 2019")
+    out = coverage_members(cand, by_id)
+    flagged = {m.immich_id for m in out if m.flagged_coverage}
+    assert flagged == {"x"}                            # only the in-window intruder
+    assert all(not m.included for m in out if m.flagged_coverage)
+
+
+def test_outlier_member_far_from_bulk():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("a2", "2019-06-01T11:00:00"),
+               _a("a3", "2019-06-01T12:00:00"),
+               _a("z", "2019-09-01T10:00:00")]          # tagged but months away
+    by_id = {m.immich_id: m for m in members}
+    cand = CandidateCluster(member_ids=["a", "a2", "a3", "z"],
+                            start_at="2019-06-01T10:00:00", end_at="2019-09-01T10:00:00",
+                            suggested_name="Italy 2019", confidence=0.95,
+                            kind_guess="trip", seed_tag="Italy 2019")
+    out = coverage_members(cand, by_id)
+    outliers = {m.immich_id for m in out if m.is_outlier}
+    assert outliers == {"z"}
+
+
+def test_non_seeded_cluster_has_no_coverage_or_outliers():
+    members = [_a("a", "2019-06-01T10:00:00"), _a("b", "2019-06-01T12:00:00")]
+    intruder = _a("x", "2019-06-01T11:00:00")
+    by_id = {m.immich_id: m for m in members + [intruder]}
+    cand = CandidateCluster(member_ids=["a", "b"], start_at="2019-06-01T10:00:00",
+                            end_at="2019-06-01T12:00:00", suggested_name="Trip",
+                            confidence=0.4, kind_guess="everyday", seed_tag=None)
+    out = coverage_members(cand, by_id)
+    assert {m.immich_id for m in out} == {"a", "b"}
+    assert not any(m.flagged_coverage or m.is_outlier for m in out)

apps/trip-cluster/tests/test_ingest.py

@@ -0,0 +1,140 @@
+import os
+from photoflow.core import Store
+from photoflow.immich import pipeline
+from app.ingest import run_ingest
+
+
+class FakeImmich:
+    def __init__(self, assets, tag_map=None, thumb=b"\xff\xd8\xffjpeg"):
+        self._assets = assets
+        self._tag_map = tag_map or {}
+        self._thumb = thumb
+        self.searches = []
+
+    def resolve_tag_id(self, name):
+        return self._tag_map.get(name)
+
+    def search_assets(self, **kwargs):
+        self.searches.append(kwargs)
+        return list(self._assets)
+
+    def download_thumbnail(self, asset_id):
+        return self._thumb
+
+
+def _asset(i, taken, tags=None, updated="2026-01-01T00:00:00Z"):
+    return {"id": i, "original_filename": f"{i}.jpg", "taken_at": taken,
+            "gps_lat": None, "gps_lon": None, "place_city": None,
+            "place_country": None, "type": "IMAGE", "tags": tags or [],
+            "rating": 0, "updated_at": updated}
+
+
+def test_ingest_upserts_assets_tags_and_thumbs(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs")
+    os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01", tags=["Italy 2019"]),
+                         _asset("b", "2019-06-02")])
+    res = run_ingest(client, store, thumbs, date_from="2019-01-01", date_to="2020-01-01")
+    assert res["fetched"] == 2
+    assert store.get_asset("a").taken_at == "2019-06-01"
+    assert store.asset_tags("a") == ["Italy 2019"]
+    assert os.path.exists(os.path.join(thumbs, "a.jpg"))
+    assert store.get_meta("last_ingest_at") == "2026-01-01T00:00:00Z"
+    store.close()
+
+
+def test_ingest_marks_processed_from_pipeline_tag(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01", tags=[pipeline.PROCESSED])])
+    res = run_ingest(client, store, thumbs)
+    assert res["processed_marked"] == 1
+    assert store.get_asset("a").processed is True
+    store.close()
+
+
+def test_ingest_incremental_passes_updated_after(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    store.set_meta("last_ingest_at", "2026-05-01T00:00:00Z")
+    client = FakeImmich([_asset("a", "2019-06-01")])
+    run_ingest(client, store, thumbs)
+    assert client.searches[0].get("updated_after") == "2026-05-01T00:00:00Z"
+    # --full ignores it
+    client2 = FakeImmich([_asset("a", "2019-06-01")])
+    run_ingest(client2, store, thumbs, full=True)
+    assert client2.searches[0].get("updated_after") is None
+    store.close()
+
+
+def test_ingest_subset_and_tag_resolution(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")],
+                        tag_map={"Italy 2019": "t1"})
+    res = run_ingest(client, store, thumbs, tag="Italy 2019", subset=1)
+    assert res["fetched"] == 1
+    assert client.searches[0].get("tag_ids") == ["t1"]
+    store.close()
+
+
+def test_ingest_resilient_to_thumb_failure_and_zero_byte(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+
+    class FailThenOk(FakeImmich):
+        def download_thumbnail(self, asset_id):
+            if asset_id == "a":
+                raise RuntimeError("boom")
+            return self._thumb
+
+    # Pre-existing 0-byte file for "b" must be treated as missing and refetched.
+    with open(os.path.join(thumbs, "b.jpg"), "wb"):
+        pass
+
+    client = FailThenOk([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")])
+    res = run_ingest(client, store, thumbs)
+    # The run completes despite "a" failing.
+    assert res["fetched"] == 2
+    # No 0-byte file left for the failed asset.
+    assert not os.path.exists(os.path.join(thumbs, "a.jpg"))
+    # The 0-byte file for "b" was replaced with real bytes.
+    assert os.path.getsize(os.path.join(thumbs, "b.jpg")) > 0
+    store.close()
+
+
+def test_ingest_watermark_not_advanced_past_failed_asset(tmp_path):
+    # A later successful asset must NOT advance last_ingest_at past an earlier
+    # asset whose thumbnail failed, or that asset is stranded on the next
+    # incremental run (review finding #9).
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+
+    class FailOld(FakeImmich):
+        def download_thumbnail(self, asset_id):
+            if asset_id == "old":
+                raise RuntimeError("boom")
+            return self._thumb
+
+    client = FailOld([_asset("old", "2008-06-01", updated="2026-01-01T00:00:00Z"),
+                      _asset("new", "2024-03-01", updated="2026-03-01T00:00:00Z")])
+    run_ingest(client, store, thumbs)
+    assert store.get_asset("old") is None and store.get_asset("new") is not None
+    wm = store.get_meta("last_ingest_at")
+    # cursor must stay below the failed asset (here: not advanced at all)
+    assert wm is None or wm < "2026-01-01T00:00:00Z"
+    assert wm != "2026-03-01T00:00:00Z"
+    store.close()
+
+
+def test_ingest_subset_run_does_not_advance_watermark(tmp_path):
+    # --subset is a sampling run; it must not move the incremental cursor or it
+    # would strand the un-fetched remainder.
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01", updated="2026-02-01T00:00:00Z"),
+                         _asset("b", "2019-06-02", updated="2026-02-02T00:00:00Z")])
+    run_ingest(client, store, thumbs, subset=1)
+    assert store.get_meta("last_ingest_at") is None
+    store.close()

apps/trip-cluster/tests/test_review.py

@@ -0,0 +1,59 @@
+from photoflow.core import Store
+from photoflow.core.models import Asset, Cluster, ClusterMember
+from app import review
+
+
+def _store(tmp_path):
+    s = Store(str(tmp_path / "t.db")).connect()
+    for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-06-03")]:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+    return s
+
+
+def _cluster(s, name="Trip", conf=0.9, kind="trip", ids=("a", "b")):
+    return s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-03", suggested_name=name,
+                confidence=conf, kind_guess=kind, status="pending"),
+        [ClusterMember(cluster_id=0, immich_id=i) for i in ids])
+
+
+def test_approve_uses_suggested_when_no_name(tmp_path):
+    s = _store(tmp_path)
+    cid = _cluster(s, name="Venice")
+    review.approve(s, cid)
+    c = s.get_cluster(cid)
+    assert c.status == "approved" and c.decided_name == "Venice" and c.reviewed_at
+    review.approve(s, cid, name="Venezia")
+    assert s.get_cluster(cid).decided_name == "Venezia"
+    s.close()
+
+
+def test_non_trip_skip(tmp_path):
+    s = _store(tmp_path)
+    cid = _cluster(s)
+    review.mark_non_trip(s, cid)
+    assert s.get_cluster(cid).status == "non_trip"
+    c2 = _cluster(s)
+    review.skip(s, c2)
+    assert s.get_cluster(c2).status == "skipped"
+    s.close()
+
+
+def test_set_member(tmp_path):
+    s = _store(tmp_path)
+    cid = _cluster(s, ids=("a", "b"))
+    review.set_member(s, cid, "b", False)
+    inc = {m.immich_id: m.included for _, m in s.cluster_members(cid)}
+    assert inc["b"] is False
+    s.close()
+
+
+def test_approve_high_confidence_only_trips(tmp_path):
+    s = _store(tmp_path)
+    hi = _cluster(s, conf=0.9, kind="trip")
+    _cluster(s, conf=0.3, kind="trip")              # too low
+    _cluster(s, conf=0.9, kind="everyday")          # everyday excluded
+    n = review.approve_high_confidence(s, threshold=0.75)
+    assert n == 1
+    assert s.get_cluster(hi).status == "approved"
+    s.close()

apps/trip-cluster/tests/test_routes.py

@@ -1,6 +1,25 @@
 import os
 from app import create_app
 from app.config import Config
+from photoflow.core import Store
+from photoflow.core.models import Asset, Cluster, ClusterMember
+
+
+def _seed_store(cfg):
+    s = Store(cfg.db_path).connect()
+    for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-07-10")]:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+    cid = s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name="Venice",
+                confidence=0.9, kind_guess="trip", status="pending"),
+        [ClusterMember(cluster_id=0, immich_id="a"),
+         ClusterMember(cluster_id=0, immich_id="b")])
+    other = s.insert_cluster(
+        Cluster(start_at="2019-07-10", end_at="2019-07-10", suggested_name="Rome",
+                confidence=0.3, kind_guess="trip", status="pending"),
+        [ClusterMember(cluster_id=0, immich_id="c")])
+    s.close()
+    return cid, other
 
 
 def _app(tmp_path):
@@ -24,3 +43,58 @@ def test_thumb_served(tmp_path):
     r = client.get("/thumb/a")
     assert r.status_code == 200 and r.mimetype == "image/jpeg"
     assert _app(tmp_path).test_client().get("/thumb/missing").status_code == 404
+
+
+def test_index_lists_clusters(tmp_path):
+    app = _app(tmp_path)
+    _seed_store(app.config["APP_CONFIG"])
+    r = app.test_client().get("/")
+    assert r.status_code == 200 and b"Venice" in r.data and b"Rome" in r.data
+
+
+def test_index_empty_state(tmp_path):
+    r = _app(tmp_path).test_client().get("/")
+    assert r.status_code == 200 and b"No clusters yet" in r.data
+
+
+def test_detail_and_approve(tmp_path):
+    app = _app(tmp_path)
+    cid, _ = _seed_store(app.config["APP_CONFIG"])
+    client = app.test_client()
+    d = client.get(f"/cluster/{cid}")
+    assert d.status_code == 200 and b"/thumb/a" in d.data
+    r = client.post(f"/cluster/{cid}/approve", json={"name": "Venezia"})
+    assert r.get_json()["status"] == "approved"
+    s = Store(app.config["APP_CONFIG"].db_path).connect()
+    assert s.get_cluster(cid).decided_name == "Venezia"
+    s.close()
+
+
+def test_member_toggle_and_high_confidence(tmp_path):
+    app = _app(tmp_path)
+    cid, _ = _seed_store(app.config["APP_CONFIG"])
+    client = app.test_client()
+    client.post(f"/cluster/{cid}/member", json={"asset_id": "b", "included": False})
+    s = Store(app.config["APP_CONFIG"].db_path).connect()
+    assert {m.immich_id: m.included for _, m in s.cluster_members(cid)}["b"] is False
+    s.close()
+    r = client.post("/approve-high-confidence", json={"threshold": 0.75})
+    assert r.get_json()["approved"] == 1     # only the 0.9 cluster
+
+
+def test_apply_all_with_injected_client(tmp_path):
+    app = _app(tmp_path)
+    cid, _ = _seed_store(app.config["APP_CONFIG"])
+
+    class FakeImmich:
+        def __init__(self): self.tagged = []
+        def upsert_tag(self, name): return f"id:{name}"
+        def tag_assets(self, tid, ids): self.tagged.append((tid, list(ids)))
+
+    fake = FakeImmich()
+    app.config["IMMICH_FACTORY"] = lambda: fake
+    client = app.test_client()
+    client.post(f"/cluster/{cid}/approve", json={"name": "Venice"})
+    r = client.post("/apply-all", json={})
+    assert r.status_code == 200
+    assert any(tid == "id:Venice" for tid, _ in fake.tagged)

apps/trip-cluster/tests/test_writeback.py

@@ -0,0 +1,148 @@
+from photoflow.core import Store
+from photoflow.core.models import Asset, Cluster, ClusterMember
+from photoflow.immich import pipeline
+from app.writeback import apply_cluster, apply_all
+
+
+class FakeImmich:
+    def __init__(self):
+        self.tagged = []
+        self.fail_tag_id = None
+        self.fail_upsert = None          # tag name whose upsert_tag should raise
+        self._ids = {}
+
+    def upsert_tag(self, name):
+        if self.fail_upsert is not None and name == self.fail_upsert:
+            raise RuntimeError("upsert boom")
+        self._ids.setdefault(name, f"id:{name}")
+        return self._ids[name]
+
+    def tag_assets(self, tag_id, ids):
+        if self.fail_tag_id is not None and tag_id == self.fail_tag_id:
+            raise RuntimeError("boom")
+        self.tagged.append((tag_id, list(ids)))
+
+
+def _store(tmp_path):
+    s = Store(str(tmp_path / "t.db")).connect()
+    for i in ("a", "b", "x"):
+        s.upsert_asset(Asset(immich_id=i, taken_at="2019-06-01"))
+    return s
+
+
+def _approved(s, name="Venice"):
+    cid = s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name=name,
+                status="approved", decided_name=name),
+        [ClusterMember(cluster_id=0, immich_id="a"),
+         ClusterMember(cluster_id=0, immich_id="b"),
+         ClusterMember(cluster_id=0, immich_id="x", included=False, flagged_coverage=True)])
+    return cid
+
+
+def test_apply_approved_tags_included_then_processed(tmp_path):
+    s = _store(tmp_path)
+    cid = _approved(s)
+    client = FakeImmich()
+    res = apply_cluster(client, s, cid)
+    assert sorted(res["succeeded"]) == ["a", "b"] and res["failed"] == []
+    # trip tag on a,b ; processed on a,b ; x (excluded) never tagged
+    assert ("id:Venice", ["a", "b"]) in client.tagged
+    assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged
+    assert all("x" not in ids for _, ids in client.tagged)
+    assert s.get_asset("a").processed is True
+    s.close()
+
+
+def test_apply_is_idempotent(tmp_path):
+    s = _store(tmp_path)
+    cid = _approved(s)
+    client = FakeImmich()
+    apply_cluster(client, s, cid)
+    before = len(client.tagged)
+    apply_cluster(client, s, cid)             # second run writes nothing new
+    assert len(client.tagged) == before
+    s.close()
+
+
+def test_partial_failure_leaves_retryable(tmp_path):
+    s = _store(tmp_path)
+    cid = _approved(s)
+    client = FakeImmich()
+    client.fail_tag_id = "id:Venice"          # trip tag write fails
+    res = apply_cluster(client, s, cid)
+    assert res["succeeded"] == [] and sorted(i for i, _ in res["failed"]) == ["a", "b"]
+    assert s.get_asset("a").processed is False         # not marked processed on failure
+    assert s.already_applied("a", "trip", "Venice") is False  # retryable
+    s.close()
+
+
+def test_apply_non_trip(tmp_path):
+    s = _store(tmp_path)
+    cid = s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-02", status="non_trip"),
+        [ClusterMember(cluster_id=0, immich_id="a")])
+    client = FakeImmich()
+    apply_cluster(client, s, cid)
+    assert ("id:_pipeline/non-trip", ["a"]) in client.tagged
+    assert ("id:_pipeline/processed", ["a"]) in client.tagged
+    s.close()
+
+
+def test_apply_all_reports_per_cluster(tmp_path):
+    s = _store(tmp_path)
+    _approved(s, name="Venice")
+    client = FakeImmich()
+    results = apply_all(client, s)
+    assert len(results) == 1 and results[0]["status"] == "approved"
+    s.close()
+
+
+def test_apply_skipped(tmp_path):
+    s = _store(tmp_path)
+    cid = s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-02", status="skipped"),
+        [ClusterMember(cluster_id=0, immich_id="a"),
+         ClusterMember(cluster_id=0, immich_id="b")])
+    client = FakeImmich()
+    res = apply_cluster(client, s, cid)
+    # skipped: only _pipeline/processed is written, no content/non-trip tag
+    assert res["status"] == "skipped"
+    assert res["succeeded"] == [] and res["failed"] == []
+    assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged
+    assert len(client.tagged) == 1            # nothing but the processed tag
+    assert s.get_asset("a").processed is True
+    assert s.get_asset("b").processed is True
+    s.close()
+
+
+def test_upsert_tag_failure_is_caught_and_does_not_abort_batch(tmp_path):
+    # A failing upsert_tag must be recorded as a per-asset failure (retryable),
+    # not propagate out of apply_cluster and abort apply_all (review finding R1).
+    s = _store(tmp_path)
+    _approved(s, name="Venice")
+    _approved(s, name="Rome")           # second cluster must still be applied
+    client = FakeImmich()
+    client.fail_upsert = "Venice"       # first cluster's trip-tag upsert fails
+    results = apply_all(client, s)
+    assert len(results) == 2            # batch was not aborted by the first failure
+    venice = next(r for r in results if r["cluster_id"] == 1)
+    assert venice["succeeded"] == [] and sorted(i for i, _ in venice["failed"]) == ["a", "b"]
+    assert s.already_applied("a", "trip", "Venice") is False   # retryable
+    rome = next(r for r in results if r["cluster_id"] == 2)
+    assert sorted(rome["succeeded"]) == ["a", "b"]
+    s.close()
+
+
+def test_processed_write_failure_is_surfaced_in_result(tmp_path):
+    # Trip tag succeeds but the _pipeline/processed write fails: the failure must
+    # appear in result["failed"] and the asset must NOT be marked processed.
+    s = _store(tmp_path)
+    cid = _approved(s, name="Venice")
+    client = FakeImmich()
+    client.fail_upsert = pipeline.PROCESSED
+    res = apply_cluster(client, s, cid)
+    assert sorted(res["succeeded"]) == ["a", "b"]              # trip tag still applied
+    assert sorted(i for i, _ in res["failed"]) == ["a", "b"]  # processed surfaced
+    assert s.get_asset("a").processed is False
+    s.close()

apps/trip-cluster/tests/ui/conftest.py

@@ -0,0 +1,60 @@
+import io
+import os
+import threading
+import time
+
+import pytest
+from PIL import Image
+from werkzeug.serving import make_server
+
+from app import create_app
+from app.config import Config
+from photoflow.core import Store
+from photoflow.core.models import Asset, Cluster, ClusterMember
+
+
+def _jpeg(color):
+    buf = io.BytesIO()
+    Image.new("RGB", (48, 48), color).save(buf, format="JPEG")
+    return buf.getvalue()
+
+
+@pytest.fixture(scope="session")
+def data_dir(tmp_path_factory):
+    d = str(tmp_path_factory.mktemp("data"))
+    thumbs = os.path.join(d, "thumbs")
+    os.makedirs(thumbs, exist_ok=True)
+    s = Store(os.path.join(d, "trip-cluster.db")).connect()
+    palette = {"a": (10, 20, 30), "b": (40, 80, 120), "c": (200, 50, 90)}
+    for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-07-10")]:
+        s.upsert_asset(Asset(immich_id=i, taken_at=t))
+        with open(os.path.join(thumbs, f"{i}.jpg"), "wb") as f:
+            f.write(_jpeg(palette[i]))
+    s.insert_cluster(
+        Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name="Venice",
+                confidence=0.9, kind_guess="trip", status="pending"),
+        [ClusterMember(cluster_id=0, immich_id="a"),
+         ClusterMember(cluster_id=0, immich_id="b")])
+    s.insert_cluster(
+        Cluster(start_at="2019-07-10", end_at="2019-07-10", suggested_name="Rome",
+                confidence=0.3, kind_guess="trip", status="pending"),
+        [ClusterMember(cluster_id=0, immich_id="c")])
+    s.close()
+    return d
+
+
+@pytest.fixture(scope="session")
+def flask_app(data_dir):
+    cfg = Config(immich_url="http://127.0.0.1:1", immich_api_key="k",
+                 anthropic_api_key="", data_dir=data_dir)
+    return create_app(cfg)
+
+
+@pytest.fixture(scope="session")
+def base_url(flask_app):
+    server = make_server("127.0.0.1", 8095, flask_app)
+    t = threading.Thread(target=server.serve_forever, daemon=True)
+    t.start()
+    time.sleep(0.2)
+    yield "http://127.0.0.1:8095"
+    server.shutdown()

apps/trip-cluster/tests/ui/test_review_ui.py

@@ -0,0 +1,45 @@
+def _open(page, base_url):
+    page.goto(base_url)
+    page.wait_for_selector(".photo-card")
+
+
+def test_grid_arrow_navigation_moves_focus_ring(base_url, page):
+    # Venice (2 photos) is high-confidence; click it to load its detail.
+    _open(page, base_url)
+    page.click("text=Venice")
+    page.wait_for_selector(".photo-card")
+    cards = page.query_selector_all(".photo-card")
+    assert len(cards) == 2
+    page.keyboard.press("ArrowRight")
+    # second card gains the focus ring class
+    assert "ring-4" in (page.query_selector_all(".photo-card")[1].get_attribute("class"))
+
+
+def test_lightbox_opens_and_closes(base_url, page):
+    _open(page, base_url)
+    page.click("text=Venice")
+    page.wait_for_selector(".photo-card")
+    page.keyboard.press("ArrowRight")
+    page.keyboard.press("Enter")
+    assert page.is_visible("#lb")
+    page.keyboard.press("Escape")
+    page.wait_for_selector("#lb", state="hidden")
+
+
+def test_cluster_switching_with_brackets(base_url, page):
+    _open(page, base_url)
+    page.click("text=Venice")
+    page.wait_for_selector(".photo-card")
+    page.keyboard.press("]")            # move to the next cluster
+    page.wait_for_function(
+        "document.querySelectorAll('.photo-card').length === 1")  # Rome has 1 photo
+
+
+def test_approve_updates_status(base_url, page):
+    _open(page, base_url)
+    page.click("text=Venice")
+    page.wait_for_selector("#cluster-name")
+    page.once("dialog", lambda d: d.accept())   # no dialog expected, but be safe
+    page.click("text=Approve (A)")
+    page.wait_for_selector("text=approved")
+    assert "approved" in page.inner_text("body")

apps/trip-cluster/tests/ui/test_smoke_ui.py

@@ -0,0 +1,9 @@
+def test_health(base_url, page):
+    page.goto(f"{base_url}/health")
+    assert "ok" in page.content()
+
+
+def test_index_lists_clusters(base_url, page):
+    page.goto(base_url)
+    assert "Venice" in page.inner_text("body")
+    assert "Rome" in page.inner_text("body")

docker-compose.yml

@@ -0,0 +1,13 @@
+services:
+  trip-cluster:
+    build:
+      context: .
+      dockerfile: apps/trip-cluster/Dockerfile
+    ports:
+      - "8084:8084"
+    volumes:
+      - ./data:/data
+    env_file: .env
+    environment:
+      - DATA_DIR=/data
+    user: "${UID}:${GID}"

docs/BACKLOG.md

@@ -0,0 +1,25 @@
+# Backlog
+
+Small, near-term open tasks that don't warrant a full milestone. Milestones live in
+`docs/ROADMAP.md`; this file tracks the loose ends between them. Remove an item when it's done.
+
+## Open
+
+- [ ] **pgvector spike — final coverage snapshot.** The feasibility spike has *passed* (embeddings
+  readable, `assetId → asset.id` join clean, shape pinned: `smart_search.embedding`, 1152-dim,
+  cosine `<=>`). The CLIP re-run with the stronger model is still in progress (coverage ~70% and
+  climbing as of 2026-06-27). When it reaches ~100%, run `scripts/pgvector_spike.py` once (no
+  flags) and commit the refreshed findings doc to record final coverage.
+  See the "Definition of done" in `docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`.
+
+- [ ] **M1 validation gate — run on a hard, GPS-poor sample.** M1 shipped, but the acceptance-bar
+  run on a hard sample (not the easy last trip) is still pending before widening.
+  See `docs/M1-validation-gate.md` and the M1 row in `docs/ROADMAP.md`.
+
+## Later / dependent
+
+- [ ] **M1.5 — provision a least-privilege read-only Postgres role.** When M1.5 builds the pgvector
+  reader, replace the spike's session-level read-only guard (using Immich's write-capable `postgres`
+  user) with a dedicated `SELECT`-only role. Recipe is in the findings doc. Also: the pgvector reader
+  belongs in `shared/photoflow/immich` (the only Immich client) and any SQLite access in
+  `shared/photoflow/core` (the only SQLite owner) — not in app-level code.

docs/M1-validation-gate.md

@@ -0,0 +1,26 @@
+# M1 Validation Gate
+
+Before widening past the hard sample, trip-cluster must clear a quantified bar on a
+**deliberately hard slice** — a GPS-poor, multi-year, low-density window (a well-remembered
+old trip plus its surrounding everyday photos), **not** the easy phone-era last trip.
+
+## Protocol
+
+1. `categorize ingest --from <hard-window-start> --to <hard-window-end>` (or `--tag <old-trip>`).
+2. `categorize cluster`, then `categorize serve`.
+3. Hand-label the slice: the true trip boundaries + which surrounding photos are non-trip.
+4. Compare candidate clusters against the labels.
+
+## Acceptance bar (record actual numbers per run)
+
+| Metric | Definition | Target |
+|--------|------------|--------|
+| Trip-boundary precision | proposed boundaries that are real | ≥ 0.8 |
+| Trip-boundary recall | real boundaries proposed | ≥ 0.8 |
+| Coverage-flag recall | in-window missing-tag assets surfaced | ≥ 0.9 |
+| Over-split rate | extra clusters per real trip | ≤ 0.5 |
+| False-cluster rate | clusters that are pure noise | ≤ 0.1 |
+
+If unmet, tune `--gap-factor` / seed-span / confidence thresholds — do **not** widen the
+backlog. See the F4 open question (everyday cluster blow-up): observe the real cluster
+count on a representative subset first, then choose the surfacing/collapsing strategy.

docs/ROADMAP.md

@@ -31,7 +31,8 @@ Each app is independently containerized (its own Dockerfile + port). `shared/*`
 
 | # | Name | Goal | Status |
 |---|------|------|--------|
-| **M1** | Foundation + `trip-cluster` | Stand up the monorepo, shared packages, SQLite store, Immich ingest, and the `trip-cluster` app end-to-end. Validate on the last trip's photos. **This is the POC that proves the foundation.** | Design approved; spec written; plan pending |
+| **M1** | Foundation + `trip-cluster` | Stand up the monorepo, shared packages (`shared/ai` deferred to M3), SQLite store, Immich ingest, and the `trip-cluster` app end-to-end. Validate on a **hard, GPS-poor sample against a quantified acceptance bar** (not the easy last trip) before widening. **This is the POC that proves the foundation.** | **Shipped** — implemented, reviewed, merged to `main` (2026-06-27); 66 tests green. Validation-gate run on a hard sample still pending. |
+| **M1.5** | Visual similarity | Trip-level CLIP clustering over Immich's pgvector embeddings (readability/shape verified by the pgvector spike — see findings) — the rescue signal for the GPS-poor old library where timestamp/GPS signals are weakest. | Not started; dependency spike **done** (2026-06-27, embeddings readable + joinable — see findings spec) |
 | **M2** | `tag-verify` | Verify/normalize existing tags, dedupe the tag vocabulary, find outliers — on the proven foundation. | Not started |
 | **M3** | `enrich` | Geocode existing location tags ("Kiev" → coordinates) and backfill GPS into Immich to improve future trip detection; AI captioning; content tags; noise detection. | Not started |
 | **M4** | Migrate `image-rater` | Move the existing app onto the shared packages (+ optional SQLite). Largely mechanical (delete local copies, import shared). | Not started |
@@ -57,9 +58,12 @@ These apply across all milestones (decided during the 2026-06-27 brainstorm):
 
 ## Enablement notes (things to set up to unlock later milestones)
 
-- **Visual similarity (post-M1, leading enhancement):** Immich's REST API doesn't cleanly expose CLIP embeddings. Viable path = **read-only access to Immich's Postgres pgvector** embeddings (verify table/column against the live version). User can provide DB access; user is re-running CLIP with a stronger model — both pure upside for trip-level similarity, especially valuable for the GPS-poor old library.
+- **Visual similarity (M1.5; dependency verified by the pgvector spike, 2026-06-27):** Immich's REST API doesn't cleanly expose CLIP embeddings. Viable path = **read-only access to Immich's Postgres pgvector** embeddings. A read-only feasibility spike (`scripts/pgvector_spike.py`, tracked separately from M1 — *not* run inside M1) **confirmed** embeddings are readable and pinned the shape against the live DB: `smart_search.embedding`, 1152-dim, cosine `<=>`, `assetId → asset.id` join clean. See the findings spec for the version-pinned contract. The trip-level clustering itself is **M1.5**. User provides DB access (`IMMICH_DB_URL`) and is re-running CLIP with a stronger model — both pure upside, especially for the GPS-poor old library.
 - **GPS facts:** Immich's reverse-geocoding only *labels* coordinates a photo already has; it does **not** invent GPS, and Immich cannot infer GPS from image content. For GPS-less old photos, coordinates come from manual map placement or the **M3 `enrich`** step (geocode location tags → write back as GPS). Reverse-geocoding + metadata-extraction jobs can be run anytime to strengthen location anchors for the GPS-having subset.
 
 ## Related specs
 
+- Open near-term tasks (loose ends between milestones): `docs/BACKLOG.md`
 - M1: `docs/superpowers/specs/2026-06-27-immich-photo-flow-design.md`
+- M1.5 pgvector spike — design: `docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`
+- M1.5 pgvector spike — findings (M1.5's contract): `docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md`

docs/solutions/tooling-decisions/gitea-pull-requests.md

@@ -0,0 +1,71 @@
+---
+title: Creating pull requests on the self-hosted Gitea remote
+date: 2026-06-27
+category: tooling-decisions
+module: dev-workflow / gitea
+problem_type: tooling_decision
+component: development_workflow
+severity: low
+applies_when:
+  - Opening or editing a PR for this repo from a non-interactive agent/CI shell
+  - "`gh pr create` fails with: none of the git remotes point to a known GitHub host"
+  - tea exits with "Failed to read SSH passphrase ... could not open TTY"
+tags: [gitea, pull-request, tea, gh, rest-api, ssh, dotenv]
+---
+
+# Creating pull requests on the self-hosted Gitea remote
+
+## Context
+
+This repo's `origin` is a **self-hosted Gitea** instance, not GitHub:
+
+- API/web base: `https://git.gorinskat.nl` (owner `m038`, repo `immich-photo-flow`)
+- Push remote (SSH): `ssh://git@m038-nas.tail63ee39.ts.net:222/m038/immich-photo-flow.git` — a *different host* than the API, reverse-proxied behind nginx.
+
+Opening a PR from an agent/non-interactive shell fails through the two obvious tools, which wastes a lot of back-and-forth if you don't know why.
+
+## Guidance
+
+Open PRs via the **Gitea REST API over HTTPS**, authenticated with a token from `.env`. Do not rely on `gh` or `tea` from a non-interactive shell.
+
+1. Make sure the **base branch already exists on the server** — a PR needs it. (Gitea sets the repo default branch to the *first* branch you push, so push your base branch, e.g. `main`/`master`, before or alongside the feature branch.)
+2. Read `GITEA_HOST`, `GITEA_USER`, `GITEA_TOKEN` from `.env` (a `write:repository`-scoped token). **Never print, echo, or commit `.env`** — `source` it; keep the token out of `argv` (use a `curl -K` config file, mode 0600) and shred the file after.
+3. `POST {base}/api/v1/repos/{GITEA_USER}/{repo}/pulls` with `{head, base, title, body}`.
+
+```bash
+set -a; . .env; set +a                       # never cat/echo this file
+case "$GITEA_HOST" in *://*) base="$GITEA_HOST";; *) base="https://$GITEA_HOST";; esac
+umask 077; cfg=$(mktemp); printf 'header = "Authorization: token %s"\n' "$GITEA_TOKEN" > "$cfg"
+# body via file -> JSON payload (avoids backtick/$ re-evaluation in the body)
+python3 - <<'PY'
+import json; json.dump({"head":"feat/my-branch","base":"main",
+  "title":"feat: ...","body":open("/path/to/body.md").read()}, open("/tmp/pr.json","w"))
+PY
+curl -sS -K "$cfg" -X POST -H "Content-Type: application/json" --data @/tmp/pr.json \
+  "${base%/}/api/v1/repos/$GITEA_USER/$repo/pulls" -w '\nHTTP %{http_code}\n'
+shred -u "$cfg"
+```
+
+Repo-level settings use the same API, e.g. set the default branch:
+`PATCH {base}/api/v1/repos/{owner}/{repo}` with `{"default_branch":"main"}` (only do this with explicit user consent — it's a persistent change to shared infra).
+
+## Why This Matters
+
+- **`gh` is GitHub-only.** It errors `none of the git remotes ... point to a known GitHub host` and cannot target a Gitea instance.
+- **`tea` (0.14) can't authenticate non-interactively here.** Its login is configured with an SSH key that has a passphrase, and it insists on reading the passphrase from `/dev/tty` — even with `ssh_agent: true` and the key already loaded in the agent. From an agent/CI shell there is no TTY, so it dies with `could not open TTY`. (Plain `git`-over-SSH still works because the agent answers the key; only `tea`'s own auth flow needs the TTY.) The user *can* run `tea` in their own terminal — it only fails for non-interactive callers.
+- The REST API needs neither GitHub nor a TTY, so it's the reliable path for automation.
+
+## When to Apply
+
+- Any time an agent needs to open or edit a PR (or change repo settings) on this Gitea remote.
+- Generalizes to any non-GitHub Gitea/Forgejo remote reached from a non-interactive shell.
+
+## Examples
+
+- **Symptom → cause:** `gh pr create` → wrong forge; `tea pulls create` → `Failed to read SSH passphrase: could not open TTY` → use the REST API instead.
+- **Gotcha:** after pushing only a feature branch first, the repo default branch became the feature branch; pushing `master` gave the PR a base, and a later `PATCH default_branch` fixed the default.
+
+## Related
+
+- Credentials live in `.env` (gitignored): `GITEA_HOST`, `GITEA_USER`, `GITEA_TOKEN` — documented in `.env.example`.
+- The two repo paths `~/Projects/immich-photo-flow` and `~/Nextcloud/Projects/immich-photo-flow` are the same repo (`~/Projects` is a symlink), not two clones.

docs/superpowers/specs/2026-06-27-immich-photo-flow-design.md

@@ -51,7 +51,7 @@ Both apps currently **duplicate** near-identical code (`immich.py`, `config.py`,
 
 ## Roadmap
 
-The full multi-milestone roadmap (M1–M6), end-state, and cross-cutting decisions live in the authoritative **[`docs/ROADMAP.md`](../../ROADMAP.md)**. In brief: **this spec is M1** — Foundation + `trip-cluster`, the POC that proves the shared foundation. M2 (`tag-verify`), M3 (`enrich`), M4/M5 (migrate the existing apps onto the foundation), and M6 (non-trip categorization) follow, each with its own spec→plan→build cycle.
+The full multi-milestone roadmap (M1–M6), end-state, and cross-cutting decisions live in the authoritative **[`docs/ROADMAP.md`](../../ROADMAP.md)**. In brief: **this spec is M1** — Foundation + `trip-cluster`, the POC that proves the shared foundation. M1.5 (visual-similarity clustering, dependency spiked in M1), M2 (`tag-verify`), M3 (`enrich`, which also introduces `shared/ai`), M4/M5 (migrate the existing apps onto the foundation), and M6 (non-trip categorization) follow, each with its own spec→plan→build cycle.
 
 ---
 
@@ -69,8 +69,8 @@ immich-photo-flow/
   shared/
     immich/               # the one true Immich client
     core/                 # SQLite store + domain models + persistence
-    ai/                   # Anthropic batch wrapper (scaffolded; light/unused in M1)
     ui/                   # base.html, Tailwind/DaisyUI/Alpine/HTMX, Jinja macros, app.js
+                          # (shared/ai is deferred to M3 — see Shared packages)
   apps/
     trip-cluster/
       Dockerfile
@@ -108,8 +108,8 @@ SQLite store + domain dataclasses + persistence (connection, schema/migrations,
 
 **SQLite tables:** `assets`, `asset_tags`, `tags`, `clusters`, `cluster_members`, `writeback_log`, `meta`. Single DB file on the mounted volume.
 
-### `shared/ai`
-Anthropic **batch** Messages API wrapper using the official `anthropic` SDK, carrying over image-rater's proven conventions (per-criterion judgments, `confidence`, score floors, chunking ≤50/batch, default model `claude-haiku-4-5`). **Scaffolded but barely used in M1** — trip-cluster is algorithm-first (near-zero AI cost). It earns its keep in M3 (enrich).
+### `shared/ai` — deferred to M3
+**Not built in M1.** trip-cluster is algorithm-first (near-zero AI cost), so M1 has no consumer for an Anthropic wrapper; building it now would freeze its interface before the M3 enrich requirements that actually shape it. `shared/ai` is introduced in **M3 (enrich)** — an Anthropic **batch** Messages API wrapper using the official `anthropic` SDK, carrying over image-rater's proven conventions (per-criterion judgments, `confidence`, score floors, chunking ≤50/batch, default model `claude-haiku-4-5`). The `ANTHROPIC_API_KEY` plumbing stays optional in M1.
 
 ### `shared/ui`
 The shared visual foundation, extracted from the proven travel-memories/image-rater templates:
@@ -141,28 +141,36 @@ Pull assets from Immich via `shared/immich`, upsert metadata into SQLite, downlo
 - `--tag NAME` → a single already-tagged trip;
 - `--subset N` → a cap.
 
-This lets the whole tool be validated on the last trip for near-zero cost before widening to the backlog over time.
+This lets the tool be validated cheaply before widening to the backlog over time. **Validation gate (M1):** validate on a deliberately *hard* sample — a GPS-poor, multi-year, low-density slice (e.g. a well-remembered old trip plus its surrounding everyday photos), **not** the most recent trip (the easy case: phone-era, GPS-rich, already tagged). Widening is gated on a **quantified acceptance bar** measured against a small hand-labelled set: trip-boundary precision/recall, coverage-flag recall, and an acceptable over-split / false-cluster rate. Until that bar is met on the hard sample, the backlog is not widened.
 
 ### 2. Cluster (automatic) — signal hierarchy
 Pure, algorithmic, **no API calls** (keeps the POC nearly free). Produces candidate trips, each with a `suggested_name`, `confidence`, and `kind_guess`:
 
 1. **Existing trip tag** → authoritative seed; its assets form a confirmed cluster (still shown, to verify *completeness*). Respects the user's existing trip-tag convention.
-2. **Timestamp gap clustering** → primary structure: sort by `taken_at`, split where the inter-photo gap exceeds a tunable threshold.
+2. **Timestamp gap clustering** → primary structure: sort by `taken_at`, split where the inter-photo gap is large relative to local cadence. A single global threshold fails across a 15-year density gradient (a sparse old trip has multi-day intra-trip gaps; a dense recent everyday period has hour-scale inter-day gaps), so the split is **density-adaptive** — threshold relative to local photo cadence / a per-era percentile — validated with unit fixtures spanning both a sparse-old and a dense-recent regime.
 3. **Location anchors** (existing location tags like "Kiev" + GPS when present) → refine boundaries, propose names.
 4. **Coverage detection** → assets *inside* a confirmed trip's time window but *missing* its tag are flagged "likely belongs here" (the completeness gap); assets carrying a trip tag but *outside* their cluster are flagged as outliers.
-5. **Visual similarity** → **deferred from M1; leading post-M1 enhancement.** Trips are defined by time + place, not visual likeness, so the signals above resolve the large majority; visual similarity only helps a narrow case (an ambiguous time gap where two bursts may be one trip). pHash is the wrong tool (it finds near-duplicates, not trip-level similarity). CLIP is the right tool, but Immich's REST API does not cleanly expose raw embedding vectors. **Viable path: read-only access to Immich's Postgres pgvector embeddings** (do nearest-neighbor/clustering ourselves), pending a feasibility check against the live Immich version. Especially valuable here because the old library is GPS-poor, so visual continuity may be one of the few secondary signals for those photos.
+5. **Visual similarity** → **clustering deferred to M1.5; dependency verified in M1.** Trips are defined by time + place, not visual likeness, so the signals above resolve the large majority; visual similarity helps the narrow-but-important case of the GPS-poor old library, where visual continuity may be one of the few secondary signals. pHash is the wrong tool (near-duplicates, not trip-level similarity); CLIP is right, but Immich's REST API does not cleanly expose raw embedding vectors. **Viable path: read-only access to Immich's Postgres pgvector embeddings** (do nearest-neighbor/clustering ourselves). Because this is the rescue signal for the hardest case, **M1 includes a read-only feasibility spike** — confirm the embeddings are readable and pin the table/column shape and DB access/credential against the live Immich version (no clustering build). The clustering itself is **M1.5** (see roadmap), so it sits in a scheduled near-term milestone rather than floating indefinitely.
 
 **Confidence & kind_guess** (echoing image-rater's confidence/floor approach): tight time window + existing trip tag + consistent location → high confidence; sparse, untagged, no GPS → low ("needs your eye"). Low-volume scattered clusters → `kind_guess = everyday` (suggested non-trip).
 
 ### 3. Review (human, cluster-level)
-A review screen lists **candidate trips sorted by "needs attention"** (low confidence first). Per cluster: thumbnail grid (with the shared grid+lightbox / arrow-key / full-screen component), editable suggested name, and actions:
+**Layout — master/detail.** A left rail lists **candidate clusters sorted by "needs attention"** (low confidence first), each with confidence / status / count badges; selecting one opens a detail pane with its editable suggested name, action controls, and the canonical **grid + lightbox** (ring-selection, arrow-key navigation *within the grid*, full-screen view, Esc to close). Moving between clusters uses its own shortcut so arrow keys stay bound to the grid.
+
+Per-cluster actions:
 - **Approve trip** (confirm + tweak name)
 - **Mark non-trip**
-- **Split** (break one cluster into two)
-- **Merge adjacent** (combine with a neighbor)
+- **Split** — select a boundary asset in the chronologically-ordered grid and choose "split before here"; the cluster partitions at that timestamp into two clusters, both with editable names. Reuses the existing ring-selection / arrow-key focus.
+- **Merge** — surfaces the cluster's **chronological** neighbour(s) (prev/next by time, computed from a temporal index, *independent of the needs-attention list sort*) with a preview of the combined range; the user confirms which to absorb.
 - **Skip**
 
-To keep it fast: high-confidence clusters arrive **pre-filled**, and an **"approve all high-confidence"** bulk action lets the user rubber-stamp the obvious cases, concentrating attention on the fuzzy ones. Every decision **persists immediately** to SQLite (resumable: closing and reopening resumes exactly where the user left off).
+**Per-asset refinement (coverage flags).** Within a cluster's grid, assets flagged "likely belongs here" (the completeness gap) appear with a distinct badge and an **include** toggle; flagged **outliers** appear with an **exclude** toggle. Include/exclude updates `cluster_members` before write-back, so the algorithm's completeness work is actionable rather than informational.
+
+**Keyboard-first.** Cluster-level actions mirror the sibling convention (e.g. `A` approve, `N` non-trip, `S` split, `M` merge, `X` skip; arrows navigate within the grid; `[` / `]` move between clusters), listed in a help footer as image-rater/travel-memories do. *(First-class requirement.)*
+
+**Empty states.** Because ingest is scopeable and assets can already be `_pipeline/processed`, a `cluster` run can legitimately yield nothing. `serve` distinguishes three cases, each naming the next step: *no clusters yet* (run ingest/cluster), *all clusters reviewed*, and *this scope produced no clusters*.
+
+To keep it fast: high-confidence clusters arrive **pre-filled**, and an **"approve all high-confidence"** bulk action lets the user rubber-stamp the obvious cases, concentrating attention on the fuzzy ones. Every decision **persists immediately** to SQLite, so review is resumable across sessions (see Write-back for what survives loss of the SQLite layer itself).
 
 ### 4. Write-back (to Immich, idempotent)
 On approval (per-cluster or batch `apply`):
@@ -170,8 +178,9 @@ On approval (per-cluster or batch `apply`):
 - **Mark non-trip** → apply `_pipeline/non-trip` so those assets are filtered out and never resurface as unreviewed.
 - **Mark processed** → every adjudicated asset (trip-assigned, non-trip, **or** reviewed-and-skipped) gets `_pipeline/processed`. This is the durable "done" flag in the source of truth: it captures the reviewed-but-untagged case, and lets a fresh `ingest` re-derive what's already handled even if the SQLite working DB is lost (see Ingest read-back).
 - **Idempotent**: `writeback_log` records applied changes; re-runs skip what's already done. **Explicit confirmation before any write** (mirrors image-rater's export safety).
+- **Partial-failure reporting**: a batch "apply all approved" reports **per-cluster results** (succeeded / failed with reason) from `writeback_log`; clusters that fail stay in their approved-but-unapplied state so a re-run retries only those. The UI surfaces this as inline status badges plus a summary, so the user always knows what landed in the library.
 
-All durable state lands in **Immich**; SQLite remains the working/review layer that can be rebuilt from Immich tags.
+All durable state lands in **Immich**; SQLite is the working/review layer. **Only applied (tag-written) decisions are rebuildable from Immich** — approved-but-unapplied clusters, in-progress split/merge, edited names and notes live only in SQLite. So the resumability guarantee holds against loss of SQLite only for applied work: decisions should be **applied promptly on approval** (or the SQLite working DB backed up), and the rebuild-from-Immich path recovers everything already written as tags.
 
 ### Tag conventions (shared across all apps)
 Two clearly separated kinds of tags:
@@ -206,3 +215,11 @@ M1 is algorithm-first, so trip-cluster makes **essentially no Anthropic calls**
 - Migrating the existing apps onto the shared foundation (M4/M5).
 - Geocoding / GPS backfill / AI captioning (M3).
 - Narrative text drafting (human-owned; downstream).
+
+---
+
+## Deferred / Open Questions
+
+### From 2026-06-27 ce-doc-review
+
+- **Cluster-count blow-up from everyday photos (F4).** Cluster-level QA assumes "a few dozen" candidate trips, but clustering everyday/non-trip photos with per-gap splitting could mint hundreds–thousands of low-confidence clusters that "approve all high-confidence" won't relieve. The concern is real, but the reviewer's proposed pre-pass (collapsing everyday spans into bulk buckets) is unconvincing; a candidate alternative is ordering/filtering clusters by **date-span** (a multi-week span ranks above a 2-day span). Crucially, the actual cluster output on real Immich data is unknown to both reviewer and author — resolve by **observing real behaviour on a representative subset first**, then choose the surfacing/collapsing strategy.

docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md

@@ -0,0 +1,84 @@
+# pgvector embedding feasibility spike — findings
+
+**Generated:** 2026-06-27 18:49 UTC by `scripts/pgvector_spike.py`
+**Status:** machine-probed against the live Immich Postgres.
+**This is M1.5's contract.** See the design spec:
+`docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`.
+
+> ⚠️ **Immich's undocumented, internal schema — no deprecation contract.**
+> Everything below is valid **only** for the model + Immich version observed and
+> can be renamed/restructured on any Immich upgrade. M1.5 must re-run this probe
+> (or version-guard) on every Immich upgrade. The "contract" is version-pinned,
+> not durable.
+
+## Requirement 0 — does M1.5 need raw vectors (vs. a REST query)?
+
+M1.5 clusters photos by *visual similarity* at trip level — it needs either the
+raw CLIP embedding vectors or an arbitrary asset→asset nearest-neighbour query.
+Immich's REST surface provides neither:
+
+- `POST /api/search/smart` — **text→image** CLIP search: takes a text query,
+  returns assets. It never returns embedding vectors and cannot do asset→asset
+  similarity without a text prompt. Insufficient.
+- `POST /api/search/metadata`, `/api/search/random` — metadata/random only; no
+  embeddings, no similarity.
+- Duplicate detection (`/api/duplicates`) consumes embeddings *internally* but only
+  surfaces near-duplicate groups above Immich's own threshold — not a tunable
+  pairwise similarity usable for trip-level clustering. Insufficient.
+- No documented endpoint returns raw CLIP vectors or arbitrary k-NN neighbours.
+
+**Conclusion:** the load-bearing "REST can't expose embeddings" premise holds for
+Immich's documented API → **read-only Postgres access (below) is the viable path.**
+(Re-confirm against the OpenAPI of the running version on upgrade.)
+
+## Requirements 1–4 — probed facts
+
+| # | Question | Answer |
+|---|----------|--------|
+| 1 | Embeddings readable? | **yes — read 20727 distinct embedding(s)** |
+| 2 | Join to asset table? | **20727/20727 embeddings join to asset via assetId** |
+| 3 | Table / column | `public.smart_search` / `embedding` |
+| 3 | Vector dimension | **1152** (live sample=1152, declared typmod=1152) |
+| 3 | Distance operator | **<=> (cosine) — from index opclass vector_cosine_ops** |
+| 3 | Operator sanity check | OK — nearest neighbour (a duplicate (identical-vector) asset) at distance 0 (seed self-distance 0) |
+| 4 | Coverage (raw) | **45.2%** (20727/45854 assets) |
+| 4 | Coverage (image-only) | **46.6%** (20323/43601 IMAGE assets) |
+
+- **Asset-key join:** FK smart_search.assetId -> asset.id
+- **Orphan embeddings** (no matching asset): 0
+- **SQLite cross-check (optional):** skipped — no SQLite store at data/trip-cluster.db (fresh checkout; not a failure)
+
+## The join M1.5 relies on
+
+`smart_search.assetId` → `asset.id` →
+SQLite `assets.immich_id` (`shared/photoflow/core`). The Immich asset UUID is the
+same key our store uses, so embeddings index straight onto tracked photos.
+
+> ⚠️ Immich's asset table is **`asset`** in the observed version
+> (it was `assets` in older versions). The probe discovers this from the FK; M1.5's
+> reader must not hardcode the name.
+
+## Environment observed
+
+- **Postgres:** 18.2 (Debian 18.2-1.pgdg12+1)
+- **pgvector extension:** 0.8.1
+- **Schema age proxy (latest migration):** unknown
+- **Immich server version:** _record from the Immich UI / `GET /api/server/version`_
+  — not reliably available in the DB.
+- **CLIP model:** _record the model from Immich's Machine-Learning settings_ — the
+  user is re-running CLIP with a stronger model, so dimension + coverage above are a
+  snapshot of whichever model was live at probe time.
+
+## Read-only access (recommended hardening for M1.5)
+
+The probe enforces read-only at the session layer
+(`SET default_transaction_read_only = on` + psycopg `read_only`), which neutralises
+write capability even with Immich's write/DDL-capable `postgres` user. For M1.5,
+provision a dedicated least-privilege role instead:
+
+```sql
+CREATE ROLE photoflow_ro LOGIN PASSWORD '...';
+GRANT CONNECT ON DATABASE immich TO photoflow_ro;
+GRANT USAGE ON SCHEMA public TO photoflow_ro;
+GRANT SELECT ON public.smart_search, public.asset TO photoflow_ro;
+```

docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md

@@ -0,0 +1,202 @@
+# pgvector embedding feasibility spike — design
+
+**Date:** 2026-06-27
+**Milestone dependency:** unblocks **M1.5** (visual similarity). See `docs/ROADMAP.md`.
+**Status:** passed 2026-06-27 — Req 0–4 answered (see findings doc). Final coverage snapshot pending CLIP re-run completion; see "Running the spike" / "Definition of done" below.
+
+## Why this exists
+
+The roadmap narrative claims M1 "runs a read-only feasibility spike" and that M1.5's
+dependency is "verified in M1." It is not: the M1 plan explicitly deferred the spike
+(`docs/superpowers/plans/2026-06-27-immich-photo-flow-m1.md:3710` — *"pgvector spike →
+out of this plan by decision (tracked separately; M1.5 dependency)"*). So M1.5 is blocked
+on a prerequisite that never ran. This spec defines that prerequisite.
+
+M1.5 wants to cluster photos by visual similarity using the CLIP embeddings Immich already
+computes — the rescue signal for the GPS-poor old library where timestamp/GPS are weakest.
+The working assumption is that Immich's REST API does not cleanly expose raw embedding
+vectors, so the viable path is **read-only access to Immich's Postgres pgvector embeddings**.
+That assumption is load-bearing — it must be *tested*, not asserted (see Requirement 0). Before
+M1.5 can build clustering on that path, this spike must prove the path exists and pin its shape.
+
+## Scope
+
+**Feasibility only.** Read-only. Throwaway probe. The spike:
+
+- does **not** build clustering, a tested module, or any `shared/` reader;
+- does **not** write to Immich or its database (only `SELECT`);
+- exists to turn unknowns into **pinned facts** and hand M1.5 a verified contract.
+
+## Requirements (pass/fail)
+
+The spike succeeds when it answers all of the following against the **live** database and
+records the answers in the findings doc. Phrased as plain questions:
+
+0. **Does M1.5 actually need raw vectors (vs. a REST similarity query)?** Before relying on
+   DB access, state what M1.5's clustering needs — raw embedding vectors or a pairwise
+   similarity-neighbor query — and record which Immich REST endpoints were checked (e.g.
+   `/api/search/smart`, any asset-similarity/duplicate endpoint) and *why each is
+   insufficient*. This turns the load-bearing "REST can't expose embeddings" premise into a
+   documented finding; if a REST path suffices, the DB-access path below is unnecessary.
+1. **Can we read the CLIP embeddings at all?** Connect read-only over the LAN and read the
+   embedding vectors Immich stores. *This de-risks **access** only: if the embeddings are not
+   readable, M1.5 needs a different plan. It does **not** prove the signal is useful (see the
+   signal-usefulness note below).*
+2. **Can we join each embedding back to a photo we already track?** Confirm **inside
+   Postgres** that each embedding row's key references `assets.id` — the join that must hold
+   for the vectors to be usable. That asset ID is the same key our SQLite store uses, so a
+   cross-check against SQLite (`shared/photoflow/core`) is *optional* confirmation and must
+   degrade gracefully when the store is unpopulated (a fresh checkout has no `ingest` run),
+   rather than failing this requirement for the wrong reason.
+3. **What are the exact shapes?** Record the table name, embedding column, **vector
+   dimension**, and the correct pgvector **distance operator** for similarity. These are
+   **model- and version-dependent** (and have drifted across Immich versions), so they must be
+   read from *this* DB, not assumed — and the recorded shape is valid only for the model +
+   Immich version observed (see Requirement 4 / Risks).
+4. **What is the coverage?** What fraction of the **embeddable** library currently has an
+   embedding. Compute against the image/embeddable population, not all assets —
+   `count(distinct embedding.assetId) / count(assets WHERE type = 'IMAGE')` (or Immich's
+   equivalent asset-type filter) — since videos and other non-image rows CLIP never embeds
+   would otherwise deflate the ratio. Record both the raw and image-only ratios. *Tells M1.5
+   how much it can lean on the signal — but the user is re-running CLIP with a stronger model,
+   so coverage (and the dimension in Requirement 3) is a snapshot of whichever model is live
+   when the probe runs.*
+
+**Signal-usefulness is M1.5's first task, not this spike's.** Passing Requirements 1–4 proves
+the embeddings are reachable and well-shaped; it does **not** prove visual similarity actually
+rescues trip detection in the GPS-poor library. Before building clustering, M1.5 must validate
+the signal (e.g. eyeball nearest-neighbour quality on a sample of the GPS-poor set).
+
+## What we already know (to verify, not assume)
+
+Immich historically stores CLIP vectors in a **`smart_search`** table with an **`embedding`**
+column of pgvector type `vector`, keyed by `assetId` referencing `assets.id`; similarity uses
+**cosine** distance (`<=>`); the legacy default model (`ViT-B-32`) produced **512-dim**
+vectors. Names and dimension have changed across versions and the user's re-run uses a
+stronger model, so the probe **discovers** these rather than trusting them — the list above is
+only the set of candidates to probe first.
+
+This is Immich's **undocumented, internal** schema with no deprecation contract: it can be
+renamed or restructured on any Immich upgrade. The probe therefore records the exact Immich
+version observed alongside the shape, and M1.5 must budget a re-probe / version-guard on every
+Immich upgrade — the "contract" the findings doc hands M1.5 is version-pinned, not durable.
+
+## Approach
+
+Disposable read-only probe + two durable artifacts (chosen over a throwaway-only or a
+build-the-module-now approach: feasibility-only honors the roadmap, but capturing the DSN and
+a written contract is the cheap part that saves M1.5 from guessing).
+
+### Connection
+
+- New optional env var **`IMMICH_DB_URL`** (a Postgres DSN), added to `.env.example` and
+  loaded by `config.py` as an **optional** field (REST creds stay required; the DSN is only
+  needed for the spike / M1.5). `config.py` reads only `os.environ`, and the repo loads `.env`
+  solely via docker-compose's `env_file`, so a standalone host-run probe must populate the
+  environment itself — either run it via `docker compose run` (so `env_file` applies) or load
+  `.env` explicitly first (e.g. `set -a; source .env; set +a`). If run in-container, confirm
+  container-to-Immich-Postgres network reachability.
+- Read-only must be **enforced at the DB/session layer**, not just by which statements the
+  script issues: open the session read-only (`SET default_transaction_read_only = on` /
+  `SET TRANSACTION READ ONLY`) or connect via a role granted only `SELECT`. The simplest
+  credentials that work are Immich's existing Postgres user, but that user is write/DDL-capable
+  against the **source-of-truth** DB, so the session-level guard is required to remove write
+  capability while the spike runs. A dedicated least-privilege read-only role remains
+  recommended hardening for M1.5.
+- Driver: `psycopg` (psycopg3), plus the **`pgvector`** Python package. psycopg3 returns a
+  `vector` column as a *string* unless the adapter is registered, so call
+  `pgvector.psycopg.register_vector(conn)` after connect — or read the dimension server-side
+  (`SELECT vector_dims(embedding) …` / catalog `atttypmod`) and run the `<=>` check in SQL,
+  which needs no adapter. Add these to the spike's dependencies; M1.5 will formalize them.
+
+### The probe
+
+A single disposable script, **`scripts/pgvector_spike.py`**, that is idempotent and
+read-only. It:
+
+1. connects using `IMMICH_DB_URL`;
+2. discovers candidate embedding tables/columns from the catalog (probe `smart_search` /
+   `embedding` first, then fall back to scanning `information_schema` for `vector`-typed
+   columns) — so it survives version drift;
+3. reads one embedding and reports its **dimension** — via the registered `pgvector` adapter
+   or server-side `vector_dims(embedding)`, **not** the length of a raw string — and a sample
+   of the **distance operator** working (e.g. a `... ORDER BY embedding <=> embedding LIMIT 5`
+   self-similarity sanity check, casting literals to `::vector` as needed);
+4. verifies the **asset-ID join inside Postgres**: the embedding row's key references
+   `assets.id`. As *optional* confirmation it also checks whether that ID is one we store in
+   SQLite (`shared/photoflow/core`), degrading gracefully (not failing the check) when the
+   store is unpopulated;
+5. computes **coverage** against the embeddable population:
+   `count(distinct embedding.assetId) / count(assets WHERE type = 'IMAGE')` (or Immich's
+   equivalent asset-type filter), recording **both** the raw and image-only ratios;
+6. prints a human-readable report **and** writes/refreshes the findings doc.
+
+It hardcodes nothing destructive, takes no write path, and is safe to re-run.
+
+## Deliverables
+
+1. `scripts/pgvector_spike.py` — disposable read-only probe (removed or left as a documented
+   one-off after M1.5 internalizes its findings).
+2. `IMMICH_DB_URL` in `.env.example`; optional `immich_db_url` field in `config.py`. Spike
+   dependencies added: `psycopg` (psycopg3) and the `pgvector` Python package.
+3. **`docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md`** — the one-page
+   schema that becomes M1.5's contract: table, embedding column, vector dimension, distance
+   operator, the embedding → `assetId` → SQLite `asset` join, coverage % (raw and image-only),
+   the model **and** Immich version observed, and the (optional) read-only-role recipe. The doc
+   must state that this is Immich's **unsupported internal schema**, valid only for the recorded
+   model + version, and that M1.5 re-probes / version-guards on every Immich upgrade.
+4. Roadmap correction: fix **both** false claims named in "Why this exists" — the narrative
+   that M1 "runs a read-only feasibility spike" (→ "spike defined/pending per this spec") **and**
+   M1.5's status note "verified in M1" (→ "dependency spike pending/this spec").
+
+## Running the spike
+
+Prereqs: read access to Immich's Postgres and `IMMICH_DB_URL` set — put it in `.env` at the
+repo root; the probe searches upward from the working directory to find it. DSN form:
+`postgresql://USER:PASSWORD@HOST:PORT/DBNAME` (Immich defaults: user `postgres`, db `immich`;
+URL-encode special chars in the password). Immich's stock compose does **not** publish Postgres
+to the host, so either map its port or run where the DB is reachable on the LAN.
+
+From the repo root (a fresh checkout / new worktree has no `.venv`):
+
+```bash
+python3 -m venv .venv
+.venv/bin/python -m pip install -r scripts/requirements-spike.txt
+.venv/bin/python scripts/pgvector_spike.py        # read-only, idempotent, safe to re-run
+```
+
+The probe prints a report and (re)writes the findings doc. Flags: `--no-write` (report only,
+leave the doc untouched), `--sqlite PATH` (optional SQLite cross-check target; skips gracefully
+when absent), `--findings PATH` (write the doc elsewhere). The DB session is opened read-only at
+the session layer, so the probe cannot mutate Immich even via the write-capable `postgres` user.
+
+## Definition of done
+
+The spike is **done** once Requirements 0–4 are answered and recorded in the findings doc:
+REST insufficiency documented (R0), embeddings readable (R1), join to `asset` confirmed (R2),
+shape pinned — table / column / dimension / distance operator (R3), and coverage recorded both
+raw and image-only (R4). **Coverage value does not gate done-ness** — per Risks, near-zero
+coverage is a recorded fact, not a failure. By these criteria the spike already passed on
+2026-06-27 (see the findings doc); `ROADMAP.md` reflects this.
+
+**Picking it up when the CLIP re-run reaches ~100%:** the only remaining action is to refresh the
+coverage snapshot in the contract doc. Once the re-run completes, run
+`scripts/pgvector_spike.py` (no flags) so the committed findings doc records the final
+full-library coverage, then commit the regenerated doc. Nothing else changes — readability, the
+`assetId → asset.id` join, and the vector shape are already locked and are re-verified on every
+run. (If a later Immich upgrade changes the schema, the probe's catalog discovery adapts; re-run
+and re-commit the doc — that is M1.5's version-guard duty.)
+
+## Out of scope (explicitly M1.5)
+
+CLIP clustering; a tested `shared/photoflow/immich/db.py` reader; nearest-neighbor queries at
+scale; any use of the embeddings beyond the feasibility checks above.
+
+## Risks
+
+- **Embeddings unreadable / not joinable** → M1.5 blocked; spike's whole point is to surface
+  this early and cheaply.
+- **Schema differs from the known shape** → expected and handled by catalog discovery, not
+  hardcoded names.
+- **Coverage near zero** (re-run unfinished) → not a spike failure; recorded as a fact so M1.5
+  can sequence around it.

scripts/pgvector_spike.py

@@ -0,0 +1,583 @@
+#!/usr/bin/env python3
+"""pgvector embedding feasibility spike — disposable read-only probe.
+
+Answers the M1.5 prerequisite questions against the *live* Immich Postgres and
+records them in a findings doc. See the design spec:
+    docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md
+
+Questions (pass/fail):
+  1. Can we read the CLIP embeddings at all?
+  2. Can each embedding join back to a photo we track (embedding key -> assets.id)?
+  3. Exact shapes: table, embedding column, vector dimension, distance operator.
+  4. Coverage over the *embeddable* (IMAGE) population — raw and image-only ratios.
+
+This is read-only and idempotent. The DB session is opened read-only at the
+session layer (not merely by which statements we issue), so even Immich's
+write-capable postgres user cannot mutate the source-of-truth DB while we probe.
+
+Usage (from the repo root; a fresh checkout / new worktree has no .venv):
+    python3 -m venv .venv
+    .venv/bin/python -m pip install -r scripts/requirements-spike.txt
+    .venv/bin/python scripts/pgvector_spike.py        # read-only, idempotent, safe to re-run
+
+IMMICH_DB_URL is read from the environment, else from the nearest .env searched upward from the
+working directory (postgresql://USER:PASS@HOST:PORT/DBNAME). Or pass it inline:
+    IMMICH_DB_URL=postgresql://user:pass@host:5432/immich .venv/bin/python scripts/pgvector_spike.py
+
+Flags: --sqlite PATH (optional SQLite cross-check), --findings PATH, --no-write (report only).
+See docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md for the definition of done.
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sqlite3
+import sys
+import textwrap
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Optional
+
+import psycopg
+from psycopg import sql
+
+# pgvector type candidates to probe first (verified against the catalog, not trusted).
+KNOWN_TABLE = "smart_search"
+KNOWN_COLUMN = "embedding"
+# pgvector opclass -> (distance operator, human name).
+OPCLASS_OPERATORS = {
+    "vector_cosine_ops": ("<=>", "cosine"),
+    "vector_l2_ops": ("<->", "L2 / euclidean"),
+    "vector_ip_ops": ("<#>", "negative inner product"),
+}
+DEFAULT_FINDINGS = "docs/superpowers/specs/2026-06-27-pgvector-embedding-findings.md"
+
+
+# --------------------------------------------------------------------------- env
+def find_dsn() -> Optional[str]:
+    """IMMICH_DB_URL from the environment, else from the nearest .env walking up."""
+    dsn = (os.environ.get("IMMICH_DB_URL") or "").strip()
+    if dsn:
+        return dsn
+    here = Path.cwd().resolve()
+    for base in [here, *here.parents, Path(__file__).resolve().parent.parent]:
+        env_file = base / ".env"
+        if not env_file.is_file():
+            continue
+        for line in env_file.read_text().splitlines():
+            line = line.strip()
+            if line.startswith("#") or "=" not in line:
+                continue
+            key, _, val = line.partition("=")
+            if key.strip() == "IMMICH_DB_URL":
+                val = val.strip().strip('"').strip("'")
+                if val:
+                    return val
+    return None
+
+
+def connect_readonly(dsn: str) -> psycopg.Connection:
+    """Open a connection and force the whole session read-only.
+
+    Belt-and-suspenders: psycopg's ``read_only`` wraps each transaction READ ONLY,
+    and ``default_transaction_read_only`` covers anything that slips outside it.
+    """
+    conn = psycopg.connect(dsn, autocommit=True)
+    conn.execute("SET default_transaction_read_only = on")
+    conn.read_only = True
+    return conn
+
+
+# ----------------------------------------------------------------------- discovery
+def discover_vector_columns(conn: psycopg.Connection) -> list[tuple[str, str, str]]:
+    """All (schema, table, column) holding a pgvector ``vector`` column.
+
+    Catalog-driven so it survives Immich's cross-version schema drift. The known
+    ``smart_search.embedding`` pair is sorted first when present.
+    """
+    rows = conn.execute(
+        """
+        SELECT n.nspname, c.relname, a.attname
+        FROM pg_attribute a
+        JOIN pg_class c ON c.oid = a.attrelid
+        JOIN pg_namespace n ON n.oid = c.relnamespace
+        JOIN pg_type t ON t.oid = a.atttypid
+        WHERE t.typname = 'vector'
+          AND a.attnum > 0 AND NOT a.attisdropped
+          AND c.relkind IN ('r', 'p')          -- ordinary + partitioned tables
+          AND n.nspname NOT IN ('pg_catalog', 'information_schema')
+        ORDER BY (c.relname = %s AND a.attname = %s) DESC, n.nspname, c.relname
+        """,
+        (KNOWN_TABLE, KNOWN_COLUMN),
+    ).fetchall()
+    return [(r[0], r[1], r[2]) for r in rows]
+
+
+def vector_dimension(conn, schema: str, table: str, column: str) -> tuple[Optional[int], str]:
+    """Vector dimension, preferring a live sample over the declared typmod."""
+    # atttypmod holds the declared dimension for pgvector (>0 when fixed).
+    typmod = conn.execute(
+        """
+        SELECT a.atttypmod
+        FROM pg_attribute a
+        JOIN pg_class c ON c.oid = a.attrelid
+        JOIN pg_namespace n ON n.oid = c.relnamespace
+        WHERE n.nspname = %s AND c.relname = %s AND a.attname = %s
+        """,
+        (schema, table, column),
+    ).fetchone()
+    declared = typmod[0] if typmod and typmod[0] and typmod[0] > 0 else None
+
+    sampled = None
+    try:
+        ident = sql.Identifier(schema, table)
+        col = sql.Identifier(column)
+        row = conn.execute(
+            sql.SQL("SELECT vector_dims({col}) FROM {tbl} WHERE {col} IS NOT NULL LIMIT 1")
+            .format(col=col, tbl=ident)
+        ).fetchone()
+        sampled = row[0] if row else None
+    except Exception as exc:  # noqa: BLE001 — record, don't abort the probe
+        return declared, f"declared typmod={declared}; live sample failed: {exc}"
+
+    if sampled is not None:
+        note = f"live sample={sampled}" + (f", declared typmod={declared}" if declared else "")
+        return sampled, note
+    return declared, f"declared typmod={declared}; table empty (no live vector to sample)"
+
+
+def distance_operator(conn, schema: str, table: str, column: str) -> tuple[str, str]:
+    """The distance operator Immich's index was built for, read from its opclass."""
+    row = conn.execute(
+        """
+        SELECT opc.opcname
+        FROM pg_index i
+        JOIN pg_class ic ON ic.oid = i.indexrelid
+        JOIN pg_class tc ON tc.oid = i.indrelid
+        JOIN pg_namespace n ON n.oid = tc.relnamespace
+        JOIN pg_opclass opc ON opc.oid = ANY (i.indclass)
+        JOIN pg_attribute at ON at.attrelid = i.indrelid AND at.attnum = ANY (i.indkey)
+        WHERE n.nspname = %s AND tc.relname = %s AND at.attname = %s
+          AND opc.opcname LIKE 'vector_%%'
+        LIMIT 1
+        """,
+        (schema, table, column),
+    ).fetchone()
+    if row and row[0] in OPCLASS_OPERATORS:
+        op, name = OPCLASS_OPERATORS[row[0]]
+        return op, f"{op} ({name}) — from index opclass {row[0]}"
+    return "<=>", "<=> (cosine) — assumed; no vector index opclass found on this column"
+
+
+def self_similarity_ok(conn, schema, table, column, key_col, op) -> str:
+    """Sanity-check the operator: a vector must be its own nearest neighbour (dist ~0).
+
+    Param-free (seed picked inside a CTE) so it works regardless of how psycopg
+    binds vector literals.
+    """
+    try:
+        tbl, col, key = sql.Identifier(schema, table), sql.Identifier(column), sql.Identifier(key_col)
+        rows = conn.execute(
+            sql.SQL(
+                "WITH seed AS (SELECT {key} AS k, {col} AS v FROM {tbl} "
+                "  WHERE {col} IS NOT NULL LIMIT 1) "
+                "SELECT (e.{key} = seed.k) AS is_seed, (e.{col} {op} seed.v) AS dist, "
+                "       (seed.v {op} seed.v) AS self_dist "
+                "FROM {tbl} e, seed WHERE e.{col} IS NOT NULL ORDER BY dist ASC LIMIT 5"
+            ).format(key=key, col=col, tbl=tbl, op=sql.SQL(op))
+        ).fetchall()
+        if not rows:
+            return "skipped — no embeddings present"
+        is_seed, top_dist, self_dist = rows[0]
+        # Operator works iff an identical vector sorts first at the operator's own
+        # self-distance. That value is operator-dependent (0 for cosine <=> / L2 <->,
+        # ~-1 for inner product <#>), so compare the nearest distance to the seed's
+        # self-distance rather than to a hardcoded 0. The nearest row may be a duplicate
+        # photo rather than the seed itself — still a pass.
+        good = abs(float(top_dist) - float(self_dist)) < 1e-6
+        who = "the seed itself" if is_seed else "a duplicate (identical-vector) asset"
+        return (
+            f"{'OK' if good else 'UNEXPECTED'} — nearest neighbour ({who}) "
+            f"at distance {float(top_dist):.6g} (seed self-distance {float(self_dist):.6g})"
+        )
+    except Exception as exc:  # noqa: BLE001
+        return f"failed: {exc}"
+
+
+def detect_asset_key(conn, schema, table) -> tuple[Optional[str], Optional[str], Optional[str], str]:
+    """Find the column + table joining the embedding row to the asset record.
+
+    Returns ``(key_col, asset_table, asset_pk, note)``. The asset table name is
+    *discovered from the FK*, never assumed — Immich renamed ``assets`` -> ``asset``
+    across versions, exactly the drift the spec warns about.
+    """
+    fks = conn.execute(
+        """
+        SELECT kcu.column_name, ccu.table_name, ccu.column_name
+        FROM information_schema.table_constraints tc
+        JOIN information_schema.key_column_usage kcu
+          ON tc.constraint_name = kcu.constraint_name AND tc.table_schema = kcu.table_schema
+        JOIN information_schema.constraint_column_usage ccu
+          ON ccu.constraint_name = tc.constraint_name AND ccu.table_schema = tc.table_schema
+        WHERE tc.constraint_type = 'FOREIGN KEY'
+          AND tc.table_schema = %s AND tc.table_name = %s
+        """,
+        (schema, table),
+    ).fetchall()
+    # Prefer the FK whose target table is the asset record (singular/plural tolerated).
+    for col, ftable, fcol in fks:
+        if ftable in ("asset", "assets"):
+            return col, ftable, fcol, f"FK {table}.{col} -> {ftable}.{fcol}"
+
+    cols = {
+        r[0]
+        for r in conn.execute(
+            "SELECT column_name FROM information_schema.columns "
+            "WHERE table_schema = %s AND table_name = %s",
+            (schema, table),
+        ).fetchall()
+    }
+    existing_tables = {
+        r[0]
+        for r in conn.execute(
+            "SELECT table_name FROM information_schema.tables "
+            "WHERE table_schema = %s AND table_name IN ('asset', 'assets')",
+            (schema,),
+        ).fetchall()
+    }
+    asset_table = "asset" if "asset" in existing_tables else ("assets" if "assets" in existing_tables else None)
+    for cand in ("assetId", "asset_id", "assetsId"):
+        if cand in cols and asset_table:
+            return cand, asset_table, "id", f"no FK; matched column {table}.{cand} -> {asset_table}.id (assumed)"
+    return None, None, None, "no asset-key column found (FK or assetId/asset_id)"
+
+
+def join_and_coverage(conn, schema, table, key_col, asset_table, assets_pk):
+    """Join validity + coverage (raw and image-only) inside Postgres.
+
+    Returns a dict (the contract its three callers read by key):
+      n_embeddings_distinct, n_joined_to_assets, n_orphan, n_assets_total: int
+      n_assets_image, n_embedded_image: int | None   (None when no asset 'type' column)
+      raw_ratio, image_ratio: float | None            (None when the denominator is 0/None)
+    """
+    tbl, key = sql.Identifier(schema, table), sql.Identifier(key_col)
+    atbl, apk = sql.Identifier(schema, asset_table), sql.Identifier(assets_pk)
+
+    n_emb = conn.execute(
+        sql.SQL("SELECT count(DISTINCT {key}) FROM {tbl} WHERE {key} IS NOT NULL")
+        .format(key=key, tbl=tbl)
+    ).fetchone()[0]
+    n_joined = conn.execute(
+        sql.SQL(
+            "SELECT count(DISTINCT e.{key}) FROM {tbl} e "
+            "JOIN {atbl} a ON a.{apk} = e.{key}"
+        ).format(key=key, tbl=tbl, atbl=atbl, apk=apk)
+    ).fetchone()[0]
+    n_assets_total = conn.execute(
+        sql.SQL("SELECT count(*) FROM {atbl}").format(atbl=atbl)
+    ).fetchone()[0]
+
+    # Image-only population — Immich never embeds video/audio/other with CLIP.
+    has_type = bool(
+        conn.execute(
+            "SELECT 1 FROM information_schema.columns "
+            "WHERE table_schema = %s AND table_name = %s AND column_name = 'type' LIMIT 1",
+            (schema, asset_table),
+        ).fetchone()
+    )
+    n_assets_image = n_emb_image = None
+    if has_type:
+        n_assets_image = conn.execute(
+            sql.SQL("SELECT count(*) FROM {atbl} WHERE type = 'IMAGE'").format(atbl=atbl)
+        ).fetchone()[0]
+        n_emb_image = conn.execute(
+            sql.SQL(
+                "SELECT count(DISTINCT a.{apk}) FROM {atbl} a "
+                "JOIN {tbl} e ON e.{key} = a.{apk} WHERE a.type = 'IMAGE'"
+            ).format(apk=apk, atbl=atbl, tbl=tbl, key=key)
+        ).fetchone()[0]
+
+    return {
+        "n_embeddings_distinct": n_emb,
+        "n_joined_to_assets": n_joined,
+        "n_orphan": n_emb - n_joined,
+        "n_assets_total": n_assets_total,
+        "n_assets_image": n_assets_image,
+        "n_embedded_image": n_emb_image,
+        "raw_ratio": (n_joined / n_assets_total) if n_assets_total else None,
+        "image_ratio": (n_emb_image / n_assets_image)
+        if (n_assets_image not in (None, 0)) else None,
+    }
+
+
+def sqlite_crosscheck(conn, schema, table, key_col, sqlite_path: Path) -> str:
+    """Optional confirmation that embedding asset IDs match our SQLite store.
+
+    Degrades gracefully: a fresh checkout has no `ingest` run, so an empty/absent
+    store is not a failure of the join requirement.
+
+    NOTE: this opens SQLite directly, a deliberate exception for this throwaway probe.
+    Production M1.5 code must NOT — it must go through shared/photoflow/core/store.py,
+    the only SQLite owner (CLAUDE.md). This raw read-only sample check stays in the spike.
+    """
+    if not sqlite_path.is_file():
+        return f"skipped — no SQLite store at {sqlite_path} (fresh checkout; not a failure)"
+    try:
+        sample = [
+            str(r[0])
+            for r in conn.execute(
+                sql.SQL("SELECT {key} FROM {tbl} WHERE {key} IS NOT NULL LIMIT 50")
+                .format(key=sql.Identifier(key_col), tbl=sql.Identifier(schema, table))
+            ).fetchall()
+        ]
+        if not sample:
+            return "skipped — no embeddings to sample"
+        sconn = sqlite3.connect(f"file:{sqlite_path}?mode=ro", uri=True)
+        try:
+            tbl_exists = sconn.execute(
+                "SELECT 1 FROM sqlite_master WHERE type='table' AND name='assets'"
+            ).fetchone()
+            if not tbl_exists:
+                return "skipped — SQLite store has no assets table (unpopulated)"
+            total = sconn.execute("SELECT count(*) FROM assets").fetchone()[0]
+            if total == 0:
+                return "skipped — SQLite assets table empty (no ingest run; not a failure)"
+            placeholders = ",".join("?" * len(sample))
+            hits = sconn.execute(
+                f"SELECT count(*) FROM assets WHERE immich_id IN ({placeholders})", sample
+            ).fetchone()[0]
+        finally:
+            sconn.close()
+        return f"{hits}/{len(sample)} sampled embedding asset IDs found in SQLite (store has {total} assets)"
+    except Exception as exc:  # noqa: BLE001
+        return f"failed (non-fatal): {exc}"
+
+
+def environment_meta(conn) -> dict:
+    """Best-effort version context. Immich's *server* version is not in the DB."""
+    out = {}
+    try:
+        out["postgres_version"] = conn.execute("SHOW server_version").fetchone()[0]
+    except Exception as exc:  # noqa: BLE001
+        out["postgres_version"] = f"unknown: {exc}"
+    try:
+        row = conn.execute(
+            "SELECT extversion FROM pg_extension WHERE extname IN ('vector', 'vectors') LIMIT 1"
+        ).fetchone()
+        out["pgvector_extension"] = row[0] if row else "not found"
+    except Exception as exc:  # noqa: BLE001
+        out["pgvector_extension"] = f"unknown: {exc}"
+    # Newest applied migration as a proxy for schema age (table name varies by version).
+    out["latest_migration"] = "unknown"
+    for mig_table in ("migrations", "kysely_migration", "typeorm_metadata"):
+        try:
+            row = conn.execute(
+                sql.SQL("SELECT name FROM {} ORDER BY 1 DESC LIMIT 1")
+                .format(sql.Identifier(mig_table))
+            ).fetchone()
+            if row:
+                out["latest_migration"] = f"{mig_table}: {row[0]}"
+                break
+        except Exception:  # noqa: BLE001 — table absent in this version
+            continue
+    return out
+
+
+# -------------------------------------------------------------------------- output
+def _pct(ratio: Optional[float]) -> str:
+    """Format a coverage ratio as a percentage, or 'n/a' when it could not be computed."""
+    return "n/a" if ratio is None else f"{ratio * 100:.1f}%"
+
+
+def render_findings(facts: dict) -> str:
+    cov = facts["coverage"]
+    raw_pct = _pct(cov["raw_ratio"])
+    img_pct = _pct(cov["image_ratio"])
+    meta = facts["meta"]
+    return textwrap.dedent(
+        f"""\
+        # pgvector embedding feasibility spike — findings
+
+        **Generated:** {facts['generated']} by `scripts/pgvector_spike.py`
+        **Status:** machine-probed against the live Immich Postgres.
+        **This is M1.5's contract.** See the design spec:
+        `docs/superpowers/specs/2026-06-27-pgvector-embedding-spike-design.md`.
+
+        > ⚠️ **Immich's undocumented, internal schema — no deprecation contract.**
+        > Everything below is valid **only** for the model + Immich version observed and
+        > can be renamed/restructured on any Immich upgrade. M1.5 must re-run this probe
+        > (or version-guard) on every Immich upgrade. The "contract" is version-pinned,
+        > not durable.
+
+        ## Requirement 0 — does M1.5 need raw vectors (vs. a REST query)?
+
+        M1.5 clusters photos by *visual similarity* at trip level — it needs either the
+        raw CLIP embedding vectors or an arbitrary asset→asset nearest-neighbour query.
+        Immich's REST surface provides neither:
+
+        - `POST /api/search/smart` — **text→image** CLIP search: takes a text query,
+          returns assets. It never returns embedding vectors and cannot do asset→asset
+          similarity without a text prompt. Insufficient.
+        - `POST /api/search/metadata`, `/api/search/random` — metadata/random only; no
+          embeddings, no similarity.
+        - Duplicate detection (`/api/duplicates`) consumes embeddings *internally* but only
+          surfaces near-duplicate groups above Immich's own threshold — not a tunable
+          pairwise similarity usable for trip-level clustering. Insufficient.
+        - No documented endpoint returns raw CLIP vectors or arbitrary k-NN neighbours.
+
+        **Conclusion:** the load-bearing "REST can't expose embeddings" premise holds for
+        Immich's documented API → **read-only Postgres access (below) is the viable path.**
+        (Re-confirm against the OpenAPI of the running version on upgrade.)
+
+        ## Requirements 1–4 — probed facts
+
+        | # | Question | Answer |
+        |---|----------|--------|
+        | 1 | Embeddings readable? | **{facts['readable']}** |
+        | 2 | Join to asset table? | **{facts['join_summary']}** |
+        | 3 | Table / column | `{facts['schema']}.{facts['table']}` / `{facts['column']}` |
+        | 3 | Vector dimension | **{facts['dimension']}** ({facts['dimension_note']}) |
+        | 3 | Distance operator | **{facts['operator']}** |
+        | 3 | Operator sanity check | {facts['self_similarity']} |
+        | 4 | Coverage (raw) | **{raw_pct}** ({cov['n_joined_to_assets']}/{cov['n_assets_total']} assets) |
+        | 4 | Coverage (image-only) | **{img_pct}** ({cov['n_embedded_image']}/{cov['n_assets_image']} IMAGE assets) |
+
+        - **Asset-key join:** {facts['asset_key_note']}
+        - **Orphan embeddings** (no matching asset): {cov['n_orphan']}
+        - **SQLite cross-check (optional):** {facts['sqlite']}
+
+        ## The join M1.5 relies on
+
+        `{facts['table']}.{facts['asset_key']}` → `{facts['asset_table']}.{facts['assets_pk']}` →
+        SQLite `assets.immich_id` (`shared/photoflow/core`). The Immich asset UUID is the
+        same key our store uses, so embeddings index straight onto tracked photos.
+
+        > ⚠️ Immich's asset table is **`{facts['asset_table']}`** in the observed version
+        > (it was `assets` in older versions). The probe discovers this from the FK; M1.5's
+        > reader must not hardcode the name.
+
+        ## Environment observed
+
+        - **Postgres:** {meta['postgres_version']}
+        - **pgvector extension:** {meta['pgvector_extension']}
+        - **Schema age proxy (latest migration):** {meta['latest_migration']}
+        - **Immich server version:** _record from the Immich UI / `GET /api/server/version`_
+          — not reliably available in the DB.
+        - **CLIP model:** _record the model from Immich's Machine-Learning settings_ — the
+          user is re-running CLIP with a stronger model, so dimension + coverage above are a
+          snapshot of whichever model was live at probe time.
+
+        ## Read-only access (recommended hardening for M1.5)
+
+        The probe enforces read-only at the session layer
+        (`SET default_transaction_read_only = on` + psycopg `read_only`), which neutralises
+        write capability even with Immich's write/DDL-capable `postgres` user. For M1.5,
+        provision a dedicated least-privilege role instead:
+
+        ```sql
+        CREATE ROLE photoflow_ro LOGIN PASSWORD '...';
+        GRANT CONNECT ON DATABASE immich TO photoflow_ro;
+        GRANT USAGE ON SCHEMA {facts['schema']} TO photoflow_ro;
+        GRANT SELECT ON {facts['schema']}.{facts['table']}, {facts['schema']}.{facts['asset_table']} TO photoflow_ro;
+        ```
+        """
+    )
+
+
+def print_report(facts: dict) -> None:
+    cov = facts["coverage"]
+    print("\n=== pgvector spike — report ===")
+    print(f"  table/column     : {facts['schema']}.{facts['table']}.{facts['column']}")
+    print(f"  dimension        : {facts['dimension']}  ({facts['dimension_note']})")
+    print(f"  distance operator: {facts['operator']}")
+    print(f"  operator sanity  : {facts['self_similarity']}")
+    print(f"  asset key        : {facts['asset_key_note']}")
+    print(f"  join             : {facts['join_summary']}  (orphans: {cov['n_orphan']})")
+    print(f"  coverage (raw)   : {_pct(cov['raw_ratio'])}  ({cov['n_joined_to_assets']}/{cov['n_assets_total']})")
+    print(f"  coverage (image) : {_pct(cov['image_ratio'])}  ({cov['n_embedded_image']}/{cov['n_assets_image']})")
+    print(f"  sqlite x-check   : {facts['sqlite']}")
+    print(f"  postgres         : {facts['meta']['postgres_version']}")
+    print(f"  pgvector ext     : {facts['meta']['pgvector_extension']}")
+    print("===============================\n")
+
+
+# ---------------------------------------------------------------------------- main
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Read-only pgvector embedding feasibility probe.")
+    ap.add_argument("--sqlite", default="data/trip-cluster.db",
+                    help="SQLite store path for the optional cross-check (default: data/trip-cluster.db)")
+    ap.add_argument("--findings", default=DEFAULT_FINDINGS, help="findings doc path to write")
+    ap.add_argument("--no-write", action="store_true", help="print the report but do not write the findings doc")
+    args = ap.parse_args()
+
+    dsn = find_dsn()
+    if not dsn:
+        print("ERROR: IMMICH_DB_URL not set (env or a .env up-tree). See .env.example.", file=sys.stderr)
+        return 2
+
+    try:
+        conn = connect_readonly(dsn)
+    except Exception as exc:  # noqa: BLE001
+        print(f"ERROR: could not connect read-only: {exc}", file=sys.stderr)
+        return 2
+
+    with conn:
+        vec_cols = discover_vector_columns(conn)
+        if not vec_cols:
+            print("FAIL (Req 1): no pgvector `vector` column found in this database.", file=sys.stderr)
+            print("  Immich may not have run CLIP yet, or the schema changed beyond the probe's discovery.",
+                  file=sys.stderr)
+            return 1
+        schema, table, column = vec_cols[0]
+        if len(vec_cols) > 1:
+            others = ", ".join(f"{s}.{t}.{c}" for s, t, c in vec_cols[1:])
+            print(f"NOTE: multiple vector columns found; probing {schema}.{table}.{column}. Others: {others}")
+
+        dim, dim_note = vector_dimension(conn, schema, table, column)
+        op, op_desc = distance_operator(conn, schema, table, column)
+        asset_key, asset_table, assets_pk, asset_key_note = detect_asset_key(conn, schema, table)
+        if not asset_key:
+            print(f"FAIL (Req 2): {asset_key_note}", file=sys.stderr)
+            return 1
+
+        self_sim = self_similarity_ok(conn, schema, table, column, asset_key, op)
+        cov = join_and_coverage(conn, schema, table, asset_key, asset_table, assets_pk)
+        sqlite_note = sqlite_crosscheck(conn, schema, table, asset_key, Path(args.sqlite))
+        meta = environment_meta(conn)
+
+        join_summary = (
+            f"{cov['n_joined_to_assets']}/{cov['n_embeddings_distinct']} embeddings join to "
+            f"{asset_table} via {asset_key}"
+        )
+        facts = {
+            "generated": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
+            "schema": schema, "table": table, "column": column,
+            "readable": f"yes — read {cov['n_embeddings_distinct']} distinct embedding(s)",
+            "dimension": dim if dim is not None else "unknown",
+            "dimension_note": dim_note,
+            "operator": op_desc,
+            "self_similarity": self_sim,
+            "asset_key": asset_key, "asset_table": asset_table, "assets_pk": assets_pk,
+            "asset_key_note": asset_key_note,
+            "join_summary": join_summary,
+            "coverage": cov, "sqlite": sqlite_note, "meta": meta,
+        }
+
+    print_report(facts)
+
+    join_clean = cov["n_orphan"] == 0 and cov["n_joined_to_assets"] > 0
+    if not args.no_write:
+        out = Path(args.findings)
+        out.parent.mkdir(parents=True, exist_ok=True)
+        out.write_text(render_findings(facts))
+        print(f"Wrote findings -> {out}")
+
+    if cov["n_embeddings_distinct"] == 0:
+        print("WARN: zero embeddings — readable+well-shaped but coverage 0 (CLIP re-run unfinished?).")
+    if not join_clean:
+        print(f"WARN (Req 2): {cov['n_orphan']} orphan embedding(s) did not join to assets.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/requirements-spike.txt

@@ -0,0 +1,5 @@
+# Dependencies for the throwaway pgvector feasibility spike (scripts/pgvector_spike.py).
+# Kept out of the app's runtime deps on purpose — only the spike / M1.5 needs Postgres access.
+# Install into a venv:  pip install -r scripts/requirements-spike.txt
+psycopg[binary]==3.3.4
+pgvector==0.4.2

shared/photoflow/core/store.py

@@ -67,6 +67,8 @@ CREATE TABLE IF NOT EXISTS writeback_log (
 CREATE INDEX IF NOT EXISTS idx_assets_taken_at ON assets(taken_at);
 CREATE INDEX IF NOT EXISTS idx_members_cluster ON cluster_members(cluster_id);
 CREATE INDEX IF NOT EXISTS idx_members_asset ON cluster_members(immich_id);
+CREATE INDEX IF NOT EXISTS idx_writeback_lookup
+  ON writeback_log(immich_id, action, tag, result);
 """
 
 
@@ -109,6 +111,12 @@ class Store:
             self._conn.close()
             self._conn = None
 
+    def __enter__(self) -> "Store":
+        return self
+
+    def __exit__(self, *exc) -> None:
+        self.close()
+
     def upsert_asset(self, a: Asset) -> None:
         has_gps = 1 if (a.gps_lat is not None and a.gps_lon is not None) else 0
         self.conn.execute(
@@ -284,6 +292,10 @@ class Store:
         if boundary_immich_id not in ids:
             raise ValueError(f"boundary {boundary_immich_id!r} not in cluster")
         idx = ids.index(boundary_immich_id)
+        if idx == 0:
+            raise ValueError(
+                f"boundary {boundary_immich_id!r} is the first member; "
+                "split would leave an empty cluster")
         base = self.get_cluster(cluster_id)
         left, right = pairs[:idx], pairs[idx:]
 

shared/photoflow/ui/templates/macros.html

@@ -10,7 +10,7 @@
   {% elif confidence < 0.75 %}
     <span class="badge badge-sm badge-info">med</span>
   {% else %}
-    <span class="badge badge-sm badge-success">high</span>
+    <span class="badge badge-sm badge-success" title="high-confidence">high</span>
   {% endif %}
 {% endmacro %}
 

shared/tests/test_store_clusters.py

@@ -97,3 +97,14 @@ def test_writeback_log_idempotency(tmp_path):
     s.log_writeback("b", "trip", "Venice", "error:boom")
     assert s.already_applied("b", "trip", "Venice") is False  # only ok counts
     s.close()
+
+
+def test_split_cluster_rejects_first_member_boundary(tmp_path):
+    import pytest
+    s = _seed(tmp_path)
+    cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03",
+                                   suggested_name="Trip"), _members(["a", "b", "c"]))
+    with pytest.raises(ValueError):
+        s.split_cluster(cid, "a")          # boundary == first member -> empty left
+    assert s.get_cluster(cid).status != "split"   # original untouched on rejection
+    s.close()