fix(ingest): don't advance incremental cursor past a failed/sampled asset

A later successful asset could push last_ingest_at past an earlier asset whose
thumbnail download failed, permanently excluding it from later incremental runs
(only --full recovered it). Now the cursor never advances to/past the earliest
failed asset, never below the prior cursor, and not at all on a --subset run.
Regression tests added. (review finding #9)

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/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_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/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.

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()