fix(review): harden write-back, confirmation gate, and edge cases

- write-back: call upsert_tag inside _apply_tag's try so a tag-create failure
  is caught per-asset and apply_all no longer aborts mid-batch (was P1)
- write-back: surface _pipeline/processed write failures in the result instead
  of discarding them (was reported as success)
- ui: add title to the high confidence badge so approve-high-confidence's
  pre-action count is non-zero; confirm() before single-cluster apply
- core: Store context manager; close DB connection even if a route raises;
  guard split_cluster against a first-member boundary (empty cluster); add
  writeback_log lookup index
- ingest: split thumbnail download/write error handling and clean up the
  .tmp file on a write failure
- tests: upsert/processed write-failure regression tests + split-guard test

apps/trip-cluster/tests/test_writeback.py

@@ -8,9 +8,12 @@ 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]
 
@@ -111,3 +114,35 @@ def test_apply_skipped(tmp_path):
     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()